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

auth: enforce workspace worker credentials over ticket REST

Browse Source
This commit is contained in:
Hare
2026-08-01 17:23:59 +09:00
parent 3412f1c0ed
commit 81e631e640
11 changed files with 1853 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ticket/src/lib.rs
-11
View File
@@ -1791,17 +1791,6 @@ where
})
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "snake_case")]
pub enum TicketBackendHttpResponse {
Ok {
result: TicketBackendOperationResult,
},
Error {
message: String,
},
}
#[derive(Debug, Clone)]
pub struct LocalTicketBackend {
root: PathBuf,
crates/worker-runtime/src/execution.rs
+21
View File
@@ -31,6 +31,7 @@ pub enum WorkerExecutionOperation {
Restore,
Input,
ProtocolMethod,
ReplaceWorkspaceAccessToken,
Stop,
Cancel,
}
@@ -331,6 +332,17 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
Vec::new()
}
fn replace_workspace_access_token(
&self,
_handle: &WorkerExecutionHandle,
_access_token: String,
) -> WorkerExecutionResult {
WorkerExecutionResult::unsupported(
WorkerExecutionOperation::ReplaceWorkspaceAccessToken,
"execution backend does not support replacing Workspace access tokens",
)
}
fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
WorkerExecutionResult::unsupported(
WorkerExecutionOperation::Stop,
@@ -443,6 +455,15 @@ impl WorkerExecutionBackendRef {
self.backend.worker_completions(handle, kind, prefix)
}
pub(crate) fn replace_workspace_access_token(
&self,
handle: &WorkerExecutionHandle,
access_token: String,
) -> WorkerExecutionResult {
self.backend
.replace_workspace_access_token(handle, access_token)
}
pub(crate) fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
self.backend.stop_worker(handle)
}
crates/worker-runtime/src/http_server.rs
+64 -1
View File
@@ -12,7 +12,7 @@ use crate::auth::{
};
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary,
WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkingDirectoryRequest, WorkingDirectoryStatus, WorkspaceApiRef,
};
use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
use crate::error::RuntimeError;
@@ -193,6 +193,10 @@ fn runtime_http_router_with_optional_auth(
)
.route("/v1/workers/{worker_id}/input", post(send_worker_input))
.route("/v1/workers/{worker_id}/restore", post(restore_worker))
.route(
"/v1/workers/{worker_id}/workspace-api",
post(replace_worker_workspace_api),
)
.route(
"/v1/workers/{worker_id}/completions",
post(worker_completions),
@@ -282,6 +286,12 @@ pub struct RuntimeHttpWorkerResponse {
pub worker: WorkerDetail,
}
/// Replace the Workspace API binding for an existing Worker.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkerWorkspaceApiRequest {
pub workspace_api: WorkspaceApiRef,
}
/// Worker delete response.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkerDeleteResponse {
@@ -509,6 +519,28 @@ async fn create_worker(
Ok(Json(RuntimeHttpWorkerResponse { worker }))
}
async fn replace_worker_workspace_api(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
body: Result<Json<RuntimeHttpWorkerWorkspaceApiRequest>, JsonRejection>,
) -> RestResult<RuntimeHttpWorkerResponse> {
let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
let worker_ref = worker_ref_for(&state.runtime, worker_id)?;
let worker = match auth_workspace_scope(&state, auth.as_ref())? {
Some(scope) => state.runtime.replace_worker_workspace_api_scoped(
&scope,
&worker_ref,
request.workspace_api,
),
None => state
.runtime
.replace_worker_workspace_api(&worker_ref, request.workspace_api),
}
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpWorkerResponse { worker }))
}
async fn restore_worker(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -960,6 +992,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
if path.starts_with("/v1/config-bundles") || path.starts_with("/v1/working-directories") {
return Some("workers:create");
}
if path.ends_with("/workspace-api") {
return Some("workers:create");
}
if path.ends_with("/input") || path.ends_with("/restore") {
return Some("workers:input");
}
@@ -1484,6 +1519,17 @@ mod tests {
)
}
fn replace_workspace_access_token(
&self,
_handle: &WorkerExecutionHandle,
_access_token: String,
) -> WorkerExecutionResult {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::ReplaceWorkspaceAccessToken,
WorkerExecutionRunState::Idle,
)
}
fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::Stop,
@@ -1575,6 +1621,23 @@ mod tests {
created.worker.worker_id
);
let response = authed_json_request(
app.clone(),
Method::POST,
&format!("/v1/workers/{}/workspace-api", created.worker.worker_id),
token,
&RuntimeHttpWorkerWorkspaceApiRequest {
workspace_api: WorkspaceApiRef {
workspace_id: "local".to_string(),
base_url: "http://127.0.0.1:8787".to_string(),
runtime_id: None,
access_token: Some("workspace-access-token".to_string()),
},
},
)
.await;
assert_eq!(response.status(), StatusCode::OK);
let input = WorkerInput::user("hello from backend");
let response = authed_json_request(
app.clone(),
crates/worker-runtime/src/runtime.rs
+147 -2
View File
@@ -1,7 +1,7 @@
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerStatus,
WorkerSummary, WorkingDirectoryRequest,
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus,
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkspaceApiRef,
};
use crate::config_bundle::{
ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle,
@@ -589,6 +589,94 @@ impl Runtime {
Ok(())
}
/// Replace the Workspace API binding persisted for a Worker and update the
/// live execution when one is connected.
pub fn replace_worker_workspace_api_scoped(
&self,
scope: &RuntimeWorkspaceScope,
worker_ref: &WorkerRef,
workspace_api: WorkspaceApiRef,
) -> Result<WorkerDetail, RuntimeError> {
self.ensure_worker_in_workspace(scope, worker_ref)?;
if workspace_api.workspace_id != scope.workspace_id {
return Err(RuntimeError::InvalidRequest(format!(
"Workspace API scope `{}` does not match authorized workspace `{}`",
workspace_api.workspace_id, scope.workspace_id
)));
}
self.replace_worker_workspace_api(worker_ref, workspace_api)
}
pub fn replace_worker_workspace_api(
&self,
worker_ref: &WorkerRef,
workspace_api: WorkspaceApiRef,
) -> Result<WorkerDetail, RuntimeError> {
let access_token = workspace_api
.access_token
.as_ref()
.filter(|token| !token.trim().is_empty())
.cloned()
.ok_or_else(|| {
RuntimeError::InvalidRequest(
"Workspace API replacement requires an access token".to_string(),
)
})?;
let (previous_workspace_api, live_execution) = {
let state = self.lock()?;
let worker = state.worker(worker_ref)?;
if let Some(existing) = worker.request.workspace_api.as_ref()
&& (existing.workspace_id != workspace_api.workspace_id
|| existing.base_url.trim_end_matches('/')
!= workspace_api.base_url.trim_end_matches('/')
|| existing.runtime_id.as_ref().is_some_and(|runtime_id| {
workspace_api.runtime_id.as_ref() != Some(runtime_id)
}))
{
return Err(RuntimeError::InvalidRequest(
"Workspace API replacement cannot change Worker Workspace identity, Runtime identity, or base URL"
.to_string(),
));
}
let live_execution = match (
state.execution_backend.clone(),
worker.execution_handle.clone(),
) {
(Some(backend), Some(handle)) => Some((backend, handle)),
_ => None,
};
(worker.request.workspace_api.clone(), live_execution)
};
{
let mut state = self.lock()?;
state.worker_mut(worker_ref)?.request.workspace_api = Some(workspace_api);
if let Err(error) = state.persist_runtime_snapshot() {
state.worker_mut(worker_ref)?.request.workspace_api = previous_workspace_api;
return Err(error);
}
}
if let Some((backend, handle)) = live_execution {
let result = backend.replace_workspace_access_token(&handle, access_token);
if !result.is_accepted() {
let mut state = self.lock()?;
state.worker_mut(worker_ref)?.request.workspace_api = previous_workspace_api;
state.persist_runtime_snapshot()?;
return Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id.clone(),
operation: result.operation,
outcome: result.outcome,
message: result.message_or_default(),
result,
});
}
}
let state = self.lock()?;
Ok(state.worker(worker_ref)?.detail())
}
/// Attach a live execution through a workspace-scoped Runtime authorization context.
pub fn restore_worker_scoped(
&self,
@@ -953,7 +1041,8 @@ impl Runtime {
WorkerExecutionOperation::Spawn
| WorkerExecutionOperation::Restore
| WorkerExecutionOperation::Input
| WorkerExecutionOperation::ProtocolMethod => return Ok(()),
| WorkerExecutionOperation::ProtocolMethod
| WorkerExecutionOperation::ReplaceWorkspaceAccessToken => return Ok(()),
};
if result.is_accepted() {
return Ok(());
@@ -2228,6 +2317,7 @@ mod tests {
restore_result: Mutex<Option<WorkerExecutionSpawnResult>>,
restore_count: Mutex<u64>,
contexts: Mutex<BTreeMap<WorkerId, WorkerExecutionContext>>,
workspace_access_tokens: Mutex<BTreeMap<WorkerId, String>>,
#[cfg(feature = "ws-server")]
snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
}
@@ -2316,6 +2406,21 @@ mod tests {
})
}
fn replace_workspace_access_token(
&self,
handle: &WorkerExecutionHandle,
access_token: String,
) -> WorkerExecutionResult {
self.workspace_access_tokens
.lock()
.unwrap()
.insert(handle.worker_ref().worker_id.clone(), access_token);
WorkerExecutionResult::accepted(
WorkerExecutionOperation::ReplaceWorkspaceAccessToken,
WorkerExecutionRunState::Idle,
)
}
fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::Stop,
@@ -2442,6 +2547,46 @@ mod tests {
);
}
#[test]
fn workspace_api_replacement_updates_live_execution_and_persisted_request() {
let (runtime, backend) = runtime_and_backend();
let scope = scope("workspace-a", "server-a");
let worker = runtime
.create_worker_scoped(
&scope,
scoped_task_request("repair credential", "workspace-a"),
)
.unwrap();
let replacement = WorkspaceApiRef {
workspace_id: "workspace-a".to_string(),
base_url: "https://workspace.example/workspace-a/".to_string(),
runtime_id: Some("runtime-a".to_string()),
access_token: Some("replacement-token".to_string()),
};
runtime
.replace_worker_workspace_api_scoped(&scope, &worker.worker_ref, replacement.clone())
.unwrap();
assert_eq!(
backend
.workspace_access_tokens
.lock()
.unwrap()
.get(&worker.worker_ref.worker_id),
Some(&"replacement-token".to_string())
);
let state = runtime.lock().unwrap();
assert_eq!(
state
.worker(&worker.worker_ref)
.unwrap()
.request
.workspace_api,
Some(replacement)
);
}
#[test]
fn workspace_owner_binding_rejects_other_backend_and_forgets_after_last_worker_delete() {
let runtime = runtime_with_backend();
crates/worker-runtime/src/worker_backend.rs
+28
View File
@@ -1124,6 +1124,34 @@ where
result
}
fn replace_workspace_access_token(
&self,
handle: &WorkerExecutionHandle,
access_token: String,
) -> WorkerExecutionResult {
let (worker, _busy) = match self.get_execution(handle) {
Ok(execution) => execution,
Err(mut result) => {
result.operation = WorkerExecutionOperation::ReplaceWorkspaceAccessToken;
return result;
}
};
worker
.replace_workspace_access_token(access_token)
.map(|_| {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::ReplaceWorkspaceAccessToken,
WorkerExecutionRunState::Idle,
)
})
.unwrap_or_else(|error| {
WorkerExecutionResult::errored(
WorkerExecutionOperation::ReplaceWorkspaceAccessToken,
error.to_string(),
)
})
}
fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
if handle.backend_id() != self.backend_id() {
return WorkerExecutionResult::rejected(
crates/worker/src/controller.rs
+15 -1
View File
@@ -26,7 +26,10 @@ use crate::shutdown_after_idle::{
use crate::spawn::comm_tools::{read_worker_output_tool, send_to_worker_tool, stop_worker_tool};
use crate::spawn::registry::SpawnedWorkerRegistry;
use crate::spawn::tool::spawn_worker_tool;
use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult};
use crate::worker::{
SystemItemCommitter, Worker, WorkerError, WorkerRunResult, WorkspaceClient,
WorkspaceClientError,
};
use protocol::{
AlertLevel, AlertSource, ErrorCode, Event, Method, RewindTargetId, RunResult, Segment,
TurnResult, WorkerStatus,
@@ -40,6 +43,7 @@ use protocol::{
pub struct WorkerHandle {
method_tx: mpsc::Sender<Method>,
event_tx: broadcast::Sender<Event>,
workspace_client: Arc<dyn WorkspaceClient>,
pub shared_state: Arc<WorkerSharedState>,
pub runtime_dir: Arc<RuntimeDir>,
pub alerter: Alerter,
@@ -113,6 +117,14 @@ impl WorkerHandle {
pub fn alert(&self, level: AlertLevel, source: AlertSource, message: String) {
self.alerter.alert(level, source, message);
}
/// Replace the Runtime-issued Workspace access token used by this live Worker.
pub fn replace_workspace_access_token(
&self,
access_token: String,
) -> Result<(), WorkspaceClientError> {
self.workspace_client.replace_access_token(access_token)
}
}
async fn set_controller_status(
@@ -234,6 +246,7 @@ impl WorkerController {
let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
let (method_tx, method_rx) = mpsc::channel::<Method>(32);
let (event_tx, _) = broadcast::channel::<Event>(256);
let workspace_client = worker.workspace_client_handle();
let alerter = Alerter::new(event_tx.clone());
let in_flight = InFlightEvents::new(event_tx.clone());
worker.attach_in_flight_events(in_flight.clone());
@@ -352,6 +365,7 @@ impl WorkerController {
let handle = WorkerHandle {
method_tx,
event_tx: event_tx.clone(),
workspace_client,
shared_state: shared_state.clone(),
runtime_dir: runtime_dir.clone(),
alerter: alerter.clone(),
crates/worker/src/feature/builtin/ticket.rs
+328 -42
View File
@@ -12,10 +12,10 @@ use std::{
use ticket::{
LocalTicketBackend, MarkdownText, NewOrchestrationPlanRecord, NewTicket, NewTicketEvent,
NewTicketRelation, OrchestrationPlanKind, OrchestrationPlanRecord, Result as TicketResult,
Ticket, TicketBackend, TicketBackendHttpResponse, TicketBackendOperation,
TicketBackendOperationResult, TicketDoctorReport, TicketError, TicketIdOrSlug,
TicketIntakeSummary, TicketListQuery, TicketRef, TicketRelation, TicketRelationKind,
TicketRelationView, TicketReview, TicketStateChange, TicketSummary,
Ticket, TicketBackend, TicketBackendOperation, TicketBackendOperationResult,
TicketDoctorReport, TicketError, TicketIdOrSlug, TicketIntakeSummary, TicketListQuery,
TicketRef, TicketRelation, TicketRelationKind, TicketRelationView, TicketReview,
TicketStateChange, TicketSummary,
config::{DEFAULT_TICKET_BACKEND_RELATIVE_PATH, TicketConfig},
tool::{TICKET_TOOL_NAMES, TicketToolBackend, ticket_tool_description, ticket_tools},
};
@@ -387,57 +387,310 @@ impl WorkspaceHttpTicketBackend {
Self { client }
}
fn endpoint(&self) -> String {
format!(
"/api/w/{}/tickets/backend",
self.client.workspace_id().unwrap_or_default()
)
}
fn invoke(
&self,
operation: TicketBackendOperation,
) -> TicketResult<TicketBackendOperationResult> {
let client = self.client.clone();
let endpoint = self.endpoint();
let workspace_id = self.client.workspace_id().unwrap_or_default().to_string();
if tokio::runtime::Handle::try_current().is_ok() {
return std::thread::spawn(move || Self::invoke_client(client, endpoint, operation))
return std::thread::spawn(move || {
Self::invoke_client(client, workspace_id, operation)
})
.join()
.map_err(|_| {
TicketError::Conflict("ticket backend request thread panicked".to_string())
TicketError::Conflict("ticket REST request thread panicked".to_string())
})?;
}
Self::invoke_client(client, endpoint, operation)
Self::invoke_client(client, workspace_id, operation)
}
fn ticket_path(id: &TicketIdOrSlug) -> String {
let value = match id {
TicketIdOrSlug::Id(value)
| TicketIdOrSlug::Slug(value)
| TicketIdOrSlug::Query(value) => value,
};
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(byte as char);
} else {
use std::fmt::Write as _;
let _ = write!(encoded, "%{byte:02X}");
}
}
encoded
}
fn request<T: serde::de::DeserializeOwned>(
client: Arc<dyn WorkspaceClient>,
method: WorkspaceRequestMethod,
endpoint: String,
body: Option<serde_json::Value>,
) -> TicketResult<T> {
let request = match body {
Some(body) => WorkspaceRequest::json(method, endpoint, body.to_string()),
None if method == WorkspaceRequestMethod::Get => WorkspaceRequest::get(endpoint),
None => WorkspaceRequest {
method,
path: endpoint,
body: None,
},
};
let response = client.execute(request).map_err(|error| {
TicketError::Conflict(format!("ticket REST request failed: {error}"))
})?;
if !response.is_success() {
return Err(TicketError::Conflict(format!(
"ticket REST API returned HTTP {}: {}",
response.status, response.body
)));
}
serde_json::from_str(&response.body)
.map_err(|error| TicketError::Conflict(format!("decode ticket REST response: {error}")))
}
fn request_unit(
client: Arc<dyn WorkspaceClient>,
method: WorkspaceRequestMethod,
endpoint: String,
body: Option<serde_json::Value>,
) -> TicketResult<TicketBackendOperationResult> {
let request = match body {
Some(body) => WorkspaceRequest::json(method, endpoint, body.to_string()),
None => WorkspaceRequest {
method,
path: endpoint,
body: None,
},
};
let response = client.execute(request).map_err(|error| {
TicketError::Conflict(format!("ticket REST request failed: {error}"))
})?;
if !response.is_success() {
return Err(TicketError::Conflict(format!(
"ticket REST API returned HTTP {}: {}",
response.status, response.body
)));
}
Ok(TicketBackendOperationResult::Unit)
}
fn invoke_client(
client: Arc<dyn WorkspaceClient>,
endpoint: String,
workspace_id: String,
operation: TicketBackendOperation,
) -> TicketResult<TicketBackendOperationResult> {
let body = serde_json::to_string(&operation).map_err(|error| {
TicketError::Conflict(format!("serialize ticket operation: {error}"))
})?;
let response = client
.execute(WorkspaceRequest::json(
let base = format!("/api/w/{workspace_id}/tickets");
match operation {
TicketBackendOperation::DefaultIntakeReadyStateChangeBody { from } => {
let value = Self::request::<String>(
client,
WorkspaceRequestMethod::Post,
endpoint,
body,
format!("{base}/default-intake-ready-body"),
Some(serde_json::json!({ "from": from })),
)?;
Ok(TicketBackendOperationResult::Text(value))
}
TicketBackendOperation::List { filter } => {
let state = match filter.state {
ticket::TicketStateSelector::Active => "active".to_string(),
ticket::TicketStateSelector::All => "all".to_string(),
ticket::TicketStateSelector::States(states) => states
.into_iter()
.map(|state| state.as_str().to_string())
.collect::<Vec<_>>()
.join(","),
};
let tickets = Self::request(
client,
WorkspaceRequestMethod::Get,
format!("{base}/search?state={state}"),
None,
)?;
Ok(TicketBackendOperationResult::Tickets(tickets))
}
TicketBackendOperation::Show { id } => {
let ticket = Self::request(
client,
WorkspaceRequestMethod::Get,
format!("{base}/{}/record", Self::ticket_path(&id)),
None,
)?;
Ok(TicketBackendOperationResult::Ticket(ticket))
}
TicketBackendOperation::Create { input } => {
let ticket = Self::request(
client,
WorkspaceRequestMethod::Post,
base,
Some(serde_json::to_value(input).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket create: {error}"))
})?),
)?;
Ok(TicketBackendOperationResult::TicketRef(ticket))
}
TicketBackendOperation::EditItem { id, edit } => {
let ticket = Self::request(
client,
WorkspaceRequestMethod::Patch,
format!("{base}/{}/item", Self::ticket_path(&id)),
Some(serde_json::to_value(edit).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket edit: {error}"))
})?),
)?;
Ok(TicketBackendOperationResult::Ticket(ticket))
}
TicketBackendOperation::DependencyCheck { id } => {
let check = Self::request(
client,
WorkspaceRequestMethod::Get,
format!("{base}/{}/dependency-check", Self::ticket_path(&id)),
None,
)?;
Ok(TicketBackendOperationResult::DependencyCheck(check))
}
TicketBackendOperation::AddEvent { id, event } => Self::request_unit(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/thread-events", Self::ticket_path(&id)),
Some(serde_json::to_value(event).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket event: {error}"))
})?),
),
TicketBackendOperation::AddStateChanged { id, change } => Self::request_unit(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/state-changes", Self::ticket_path(&id)),
Some(serde_json::to_value(change).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket state change: {error}"))
})?),
),
TicketBackendOperation::AddIntakeSummary { id, summary } => Self::request_unit(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/intake-summaries", Self::ticket_path(&id)),
Some(serde_json::to_value(summary).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket intake summary: {error}"))
})?),
),
TicketBackendOperation::SetStateField { id, field, change } => Self::request_unit(
client,
WorkspaceRequestMethod::Post,
format!(
"{base}/{}/state-fields/{}",
Self::ticket_path(&id),
Self::ticket_path(&TicketIdOrSlug::Query(field))
),
Some(serde_json::to_value(change).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket state field change: {error}"))
})?),
),
TicketBackendOperation::SetWorkflowState { id, change } => Self::request_unit(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/workflow-state", Self::ticket_path(&id)),
Some(serde_json::to_value(change).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket workflow change: {error}"))
})?),
),
TicketBackendOperation::MarkIntakeReady {
id,
summary,
change,
} => Self::request_unit(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/intake-ready", Self::ticket_path(&id)),
Some(serde_json::json!({ "summary": summary, "change": change })),
),
TicketBackendOperation::QueueReady { id, .. } => Self::request_unit(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/workflow/queue", Self::ticket_path(&id)),
None,
),
TicketBackendOperation::Review { id, review } => Self::request_unit(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/workflow/review", Self::ticket_path(&id)),
Some(serde_json::to_value(review).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket review: {error}"))
})?),
),
TicketBackendOperation::Close { id, resolution } => Self::request_unit(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/workflow/close", Self::ticket_path(&id)),
Some(serde_json::to_value(resolution).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket close: {error}"))
})?),
),
TicketBackendOperation::AddTicketRelation { id, relation } => {
let relation = Self::request(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/relations", Self::ticket_path(&id)),
Some(serde_json::to_value(relation).map_err(|error| {
TicketError::Conflict(format!("serialize Ticket relation: {error}"))
})?),
)?;
Ok(TicketBackendOperationResult::Relation(relation))
}
TicketBackendOperation::QueryTicketRelations { ticket, kind } => {
let relations = Self::request(
client,
WorkspaceRequestMethod::Post,
format!("{base}/relations/search"),
Some(serde_json::json!({ "ticket": ticket, "kind": kind })),
)?;
Ok(TicketBackendOperationResult::Relations(relations))
}
TicketBackendOperation::RelationView { id } => {
let view = Self::request(
client,
WorkspaceRequestMethod::Get,
format!("{base}/{}/relation-view", Self::ticket_path(&id)),
None,
)?;
Ok(TicketBackendOperationResult::RelationView(view))
}
TicketBackendOperation::AddOrchestrationPlanRecord { id, record } => {
let record = Self::request(
client,
WorkspaceRequestMethod::Post,
format!("{base}/{}/orchestration-plans", Self::ticket_path(&id)),
Some(serde_json::to_value(record).map_err(|error| {
TicketError::Conflict(format!(
"serialize Ticket orchestration plan: {error}"
))
.map_err(|error| {
TicketError::Conflict(format!("ticket backend request failed: {error}"))
})?;
if !response.is_success() {
return Err(TicketError::Conflict(format!(
"ticket backend returned HTTP {}: {}",
response.status, response.body
)));
})?),
)?;
Ok(TicketBackendOperationResult::OrchestrationPlanRecord(
record,
))
}
TicketBackendOperation::QueryOrchestrationPlanRecords { ticket, kind } => {
let records = Self::request(
client,
WorkspaceRequestMethod::Post,
format!("{base}/orchestration-plans/search"),
Some(serde_json::json!({ "ticket": ticket, "kind": kind })),
)?;
Ok(TicketBackendOperationResult::OrchestrationPlanRecords(
records,
))
}
TicketBackendOperation::Doctor => {
let report = Self::request(
client,
WorkspaceRequestMethod::Get,
format!("{base}/doctor"),
None,
)?;
Ok(TicketBackendOperationResult::DoctorReport(report))
}
match serde_json::from_str::<TicketBackendHttpResponse>(&response.body).map_err(
|error| TicketError::Conflict(format!("decode ticket backend response: {error}")),
)? {
TicketBackendHttpResponse::Ok { result } => Ok(result),
TicketBackendHttpResponse::Error { message } => Err(TicketError::Conflict(message)),
}
}
}
@@ -1129,7 +1382,42 @@ provider = "github"
})
.unwrap_err();
assert!(error.to_string().contains("ticket backend request failed"));
assert!(error.to_string().contains("ticket REST request failed"));
}
#[test]
fn workspace_http_backend_posts_ticket_event_subresource() {
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 buffer = [0_u8; 8192];
let size = stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..size]);
assert!(
request
.starts_with("POST /api/w/workspace-a/tickets/01TEST/thread-events HTTP/1.1")
);
assert!(!request.contains("\"operation\""));
assert!(request.contains("\"kind\":\"comment\""));
stream
.write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n")
.unwrap();
});
let client = Arc::new(crate::worker::RuntimeWorkspaceHttpClient::new(
"workspace-a",
format!("http://{address}"),
"worker-a",
));
let backend = WorkspaceHttpTicketBackend::new(client);
backend
.add_event(
TicketIdOrSlug::Id("01TEST".to_string()),
NewTicketEvent::new(ticket::TicketEventKind::Comment, "REST comment"),
)
.unwrap();
server.join().unwrap();
}
#[test]
@@ -1141,15 +1429,13 @@ provider = "github"
let mut buffer = [0_u8; 8192];
let len = stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..len]);
assert!(request.starts_with("POST /api/w/workspace-a/tickets/backend HTTP/1.1"));
assert!(request.contains("\"operation\":\"create\""));
assert!(request.starts_with("POST /api/w/workspace-a/tickets HTTP/1.1"));
assert!(!request.contains("\"operation\""));
assert!(request.contains("\"title\":\"HTTP ticket\""));
let response_body = serde_json::to_string(&TicketBackendHttpResponse::Ok {
result: TicketBackendOperationResult::TicketRef(TicketRef {
let response_body = serde_json::to_string(&TicketRef {
id: "01TEST".to_string(),
slug: "http-ticket".to_string(),
status: ticket::TicketStatus::Open,
}),
})
.unwrap();
write!(
crates/worker/src/worker.rs
+30 -1
View File
@@ -215,6 +215,13 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
fn is_available(&self) -> bool;
fn execute(&self, request: WorkspaceRequest)
-> Result<WorkspaceResponse, WorkspaceClientError>;
/// Replace the Runtime-issued Workspace access token for this live client.
fn replace_access_token(&self, _access_token: String) -> Result<(), WorkspaceClientError> {
Err(WorkspaceClientError::Unavailable(
"Workspace client does not support access token replacement".to_string(),
))
}
}
/// HTTP forwarding client created by Runtime for one concrete Worker execution.
@@ -323,6 +330,13 @@ impl WorkspaceClient for RuntimeWorkspaceHttpClient {
}
Ok(result.0)
}
fn replace_access_token(&self, access_token: String) -> Result<(), WorkspaceClientError> {
*self.access_token.lock().map_err(|_| {
WorkspaceClientError::Request("workspace credential lock poisoned".to_string())
})? = Some(access_token);
Ok(())
}
}
fn execute_runtime_workspace_http_with_refresh(
@@ -6318,13 +6332,28 @@ mod build_summary_prompt_tests {
.with_access_token(Some("expired-token".to_string()));
let response = client
.execute(WorkspaceRequest::get(
"/api/w/workspace-refresh/tickets/backend",
"/api/w/workspace-refresh/tickets/search",
))
.unwrap();
assert_eq!(response.status, 200);
server.join().unwrap();
}
#[test]
fn runtime_workspace_client_can_install_missing_access_token() {
let client =
RuntimeWorkspaceHttpClient::new("workspace-a", "https://workspace.example", "worker-a");
client
.replace_access_token("replacement-token".to_string())
.unwrap();
assert_eq!(
client.access_token.lock().unwrap().as_deref(),
Some("replacement-token")
);
}
fn minimal_manifest() -> WorkerManifest {
let toml_str = r#"
[worker]
crates/workspace-server/src/hosts.rs
+178 -32
View File
@@ -34,7 +34,8 @@ use worker_runtime::http_server::{
RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerCompletionsRequest,
RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse,
RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest,
RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse,
RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse,
RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
};
use worker_runtime::identity::{WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef};
@@ -260,6 +261,14 @@ pub struct WorkerRestoreResult {
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerWorkspaceApiResult {
pub state: WorkerOperationState,
#[serde(skip_serializing_if = "Option::is_none")]
pub worker: Option<WorkerSummary>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RuntimeList<T> {
pub items: Vec<T>,
@@ -415,6 +424,30 @@ pub struct ConfigBundleListResult {
pub diagnostics: Vec<RuntimeDiagnostic>,
}
fn required_worker_workspace_api(
request: &WorkerSpawnRequest,
) -> Result<WorkspaceApiRef, RuntimeDiagnostic> {
let workspace_api = request.resolved_workspace_api.clone().ok_or_else(|| {
diagnostic(
"worker_workspace_credential_missing",
DiagnosticSeverity::Error,
"Workspace-bound Worker spawn requires a resolved Workspace API credential",
)
})?;
if workspace_api
.access_token
.as_deref()
.is_none_or(|token| token.trim().is_empty())
{
return Err(diagnostic(
"worker_workspace_credential_missing",
DiagnosticSeverity::Error,
"Workspace-bound Worker spawn requires a non-empty Workspace API access token",
));
}
Ok(workspace_api)
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum WorkerOperationState {
@@ -622,6 +655,24 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
}
}
fn replace_worker_workspace_api(
&self,
worker_id: &str,
_workspace_api: WorkspaceApiRef,
) -> WorkerWorkspaceApiResult {
WorkerWorkspaceApiResult {
state: WorkerOperationState::Unsupported,
worker: None,
diagnostics: vec![diagnostic(
"worker_workspace_api_replace_unsupported",
DiagnosticSeverity::Info,
format!(
"runtime does not support replacing the Workspace API for worker `{worker_id}`"
),
)],
}
}
fn create_working_directory(
&self,
_request: WorkingDirectoryRequest,
@@ -1072,6 +1123,18 @@ impl RuntimeRegistry {
Ok(runtime.restore_worker(worker_id))
}
pub fn replace_worker_workspace_api(
&self,
runtime_id: &str,
worker_id: &str,
workspace_api: WorkspaceApiRef,
) -> Result<WorkerWorkspaceApiResult, RuntimeRegistryError> {
validate_backend_identifier("runtime_id", runtime_id)?;
validate_backend_identifier("worker_id", worker_id)?;
let runtime = self.runtime(runtime_id)?;
Ok(runtime.replace_worker_workspace_api(worker_id, workspace_api))
}
pub fn spawn_worker(
&self,
runtime_id: &str,
@@ -1312,8 +1375,6 @@ impl RuntimeRegistry {
pub struct EmbeddedWorkerRuntime {
runtime_id: String,
host_id: String,
workspace_id: String,
backend_base_url: Option<String>,
runtime: worker_runtime::Runtime,
execution_enabled: bool,
resource_broker: BackendResourceBroker,
@@ -1366,18 +1427,11 @@ impl EmbeddedWorkerRuntime {
self
}
pub fn with_backend_base_url(mut self, backend_base_url: impl Into<String>) -> Self {
self.backend_base_url = Some(backend_base_url.into().trim_end_matches('/').to_string());
self
}
pub fn from_runtime(workspace_id: impl AsRef<str>, runtime: worker_runtime::Runtime) -> Self {
let workspace_id = workspace_id.as_ref().to_string();
Self {
runtime_id: EMBEDDED_RUNTIME_ID.to_string(),
host_id: host_id_for_embedded_workspace(&workspace_id),
workspace_id,
backend_base_url: None,
runtime,
execution_enabled: false,
resource_broker: BackendResourceBroker::default(),
@@ -1626,6 +1680,39 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
}
}
fn replace_worker_workspace_api(
&self,
worker_id: &str,
workspace_api: WorkspaceApiRef,
) -> WorkerWorkspaceApiResult {
let Some(worker_ref) = self.worker_ref(worker_id) else {
return WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![diagnostic(
"embedded_worker_id_invalid",
DiagnosticSeverity::Warning,
"Worker id was empty and cannot receive Workspace access".to_string(),
)],
};
};
match self
.runtime
.replace_worker_workspace_api(&worker_ref, workspace_api)
{
Ok(detail) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(detail)),
diagnostics: Vec::new(),
},
Err(err) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![embedded_runtime_diagnostic(&err)],
},
}
}
fn create_working_directory(
&self,
_request: WorkingDirectoryRequest,
@@ -1727,6 +1814,18 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
.map_or((None, None), |(key, fingerprint)| {
(Some(key), Some(fingerprint))
});
let workspace_api = match required_worker_workspace_api(&request) {
Ok(workspace_api) => workspace_api,
Err(diagnostic) => {
diagnostics.push(diagnostic);
return WorkerSpawnResult {
state: WorkerOperationState::Rejected,
worker: None,
acceptance_evidence: Vec::new(),
diagnostics,
};
}
};
let create_request = CreateWorkerRequest {
idempotency_key,
idempotency_fingerprint,
@@ -1737,16 +1836,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
initial_input: request.initial_input.clone(),
working_directory_request: request.resolved_working_directory_request.clone(),
working_directory: request.resolved_working_directory.clone(),
workspace_api: request.resolved_workspace_api.clone().or_else(|| {
self.backend_base_url
.as_ref()
.map(|base_url| WorkspaceApiRef {
workspace_id: self.workspace_id.clone(),
base_url: base_url.clone(),
runtime_id: Some(self.runtime_id.clone()),
access_token: None,
})
}),
workspace_api: Some(workspace_api),
};
match self.runtime.create_worker(create_request) {
Ok(detail) => WorkerSpawnResult {
@@ -2602,6 +2692,28 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
}
}
fn replace_worker_workspace_api(
&self,
worker_id: &str,
workspace_api: WorkspaceApiRef,
) -> WorkerWorkspaceApiResult {
match self.post_json::<_, RuntimeHttpWorkerResponse>(
&format!("/v1/workers/{worker_id}/workspace-api"),
&RuntimeHttpWorkerWorkspaceApiRequest { workspace_api },
) {
Ok(response) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(response.worker)),
diagnostics: Vec::new(),
},
Err(diagnostic) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![diagnostic],
},
}
}
fn create_working_directory(
&self,
request: WorkingDirectoryRequest,
@@ -2711,6 +2823,17 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
.map_or((None, None), |(key, fingerprint)| {
(Some(key), Some(fingerprint))
});
let workspace_api = match required_worker_workspace_api(&request) {
Ok(workspace_api) => workspace_api,
Err(diagnostic) => {
return WorkerSpawnResult {
state: WorkerOperationState::Rejected,
worker: None,
acceptance_evidence: Vec::new(),
diagnostics: vec![diagnostic],
};
}
};
let create = CreateWorkerRequest {
idempotency_key,
idempotency_fingerprint,
@@ -2721,14 +2844,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
initial_input: request.initial_input.clone(),
working_directory_request: request.resolved_working_directory_request.clone(),
working_directory: request.resolved_working_directory.clone(),
workspace_api: request.resolved_workspace_api.clone().or_else(|| {
Some(WorkspaceApiRef {
workspace_id: self.workspace_id.clone(),
base_url: self.backend_base_url.clone(),
runtime_id: Some(self.runtime_id.clone()),
access_token: None,
})
}),
workspace_api: Some(workspace_api),
};
match self.post_json::<_, RuntimeHttpWorkerResponse>("/v1/workers", &create) {
Ok(response) => WorkerSpawnResult {
@@ -3700,6 +3816,15 @@ mod tests {
use std::sync::{Arc, Mutex};
use std::thread;
fn test_workspace_api() -> WorkspaceApiRef {
WorkspaceApiRef {
workspace_id: "workspace-test".to_string(),
base_url: "http://127.0.0.1:8787".to_string(),
runtime_id: Some("runtime-test".to_string()),
access_token: Some("workspace-access-token".to_string()),
}
}
#[test]
fn embedded_builtin_decodal_profiles_resolve_through_archive() {
let root = tempfile::tempdir().unwrap();
@@ -4222,10 +4347,31 @@ mod tests {
resolved_working_directory_request: None,
resolved_working_directory: None,
resolved_config_bundle: None,
resolved_workspace_api: None,
resolved_workspace_api: Some(test_workspace_api()),
}
}
#[test]
fn embedded_runtime_rejects_tokenless_workspace_spawn() {
let runtime = EmbeddedWorkerRuntime::new_memory_with_execution_backend(
"local:test",
Arc::new(AcceptingExecutionBackend::default()),
)
.expect("test backend should connect");
let mut request = embedded_spawn_request();
request.resolved_workspace_api = None;
let spawned = runtime.spawn_worker(request);
assert_eq!(spawned.state, WorkerOperationState::Rejected);
assert!(
spawned
.diagnostics
.iter()
.any(|diagnostic| { diagnostic.code == "worker_workspace_credential_missing" })
);
}
#[test]
fn embedded_runtime_spawn_execution_failure_is_rejected_and_not_input_capable() {
let runtime = EmbeddedWorkerRuntime::new_memory_with_execution_backend(
@@ -4350,7 +4496,7 @@ mod tests {
resolved_working_directory_request: None,
resolved_working_directory: None,
resolved_config_bundle: None,
resolved_workspace_api: None,
resolved_workspace_api: Some(test_workspace_api()),
},
)
.unwrap();
@@ -4448,7 +4594,7 @@ mod tests {
resolved_working_directory_request: None,
resolved_working_directory: None,
resolved_config_bundle: None,
resolved_workspace_api: None,
resolved_workspace_api: Some(test_workspace_api()),
},
)
.unwrap();
@@ -4482,7 +4628,7 @@ mod tests {
resolved_working_directory_request: None,
resolved_working_directory: None,
resolved_config_bundle: None,
resolved_workspace_api: None,
resolved_workspace_api: Some(test_workspace_api()),
},
)
.unwrap();
crates/workspace-server/src/server.rs
+942 -126
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/store.rs
+33
View File
@@ -650,6 +650,14 @@ pub trait ControlPlaneStore: Send + Sync {
new_token: &str,
new_expires_at: &str,
) -> Result<Option<WorkerWorkspaceCredentialRecord>>;
fn revoke_worker_workspace_credentials_except(
&self,
workspace_id: &str,
runtime_id: &str,
worker_id: &str,
active_credential_id: &str,
revoked_at: &str,
) -> Result<()>;
fn revoke_worker_workspace_credentials(
&self,
workspace_id: &str,
@@ -2338,6 +2346,31 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn revoke_worker_workspace_credentials_except(
&self,
workspace_id: &str,
runtime_id: &str,
worker_id: &str,
active_credential_id: &str,
revoked_at: &str,
) -> Result<()> {
self.with_conn(|conn| {
conn.execute(
r#"UPDATE worker_workspace_credentials SET revoked_at = ?5
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND credential_id <> ?4 AND revoked_at IS NULL"#,
params![
workspace_id,
runtime_id,
worker_id,
active_credential_id,
revoked_at
],
)?;
Ok(())
})
}
fn revoke_worker_workspace_credentials(
&self,
workspace_id: &str,