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

Merge branch 'work/00001KZPQW4GJ-worker-remove-v3' into orchestration-merge-request-domain

Browse Source
This commit is contained in:
Hare
2026-08-12 17:01:24 +09:00
parent ebe0f93744 8ae930c5fc
commit 795f26fb51
12 changed files with 1565 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/http_server.rs
+81
View File
@@ -21,6 +21,9 @@ use crate::interaction::{WorkerInput, WorkerInteractionAck};
use crate::management::{RuntimeSummary, WorkerDeleteResult};
#[cfg(feature = "ws-server")]
use crate::observation::WorkerObservationCursor;
use crate::retention::{
WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, WorkerRetentionInventory,
};
#[cfg(feature = "ws-server")]
use crate::runtime::RuntimeSubscriptionRecvError;
use crate::{Runtime, RuntimeWorkspaceScope};
@@ -217,6 +220,14 @@ fn runtime_http_router_with_optional_auth(
"/v1/workers/{worker_id}",
get(get_worker).delete(delete_worker),
)
.route(
"/v1/workers/{worker_id}/retention/inventory",
get(worker_retention_inventory),
)
.route(
"/v1/workers/{worker_id}/retention/execute",
post(execute_worker_retention),
)
.route("/v1/workers/{worker_id}/input", post(send_worker_input))
.route("/v1/workers/{worker_id}/restore", post(restore_worker))
.route(
@@ -1220,6 +1231,61 @@ fn protocol_error_event(message: impl Into<String>) -> protocol::Event {
}
}
async fn worker_retention_inventory(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
) -> RestResult<WorkerRetentionInventory> {
let scope = auth_workspace_scope(&state, auth.as_ref())?.ok_or_else(|| {
RuntimeHttpRestError::new(
StatusCode::FORBIDDEN,
"workspace_scope_required",
"Worker retention inventory requires workspace-scoped authorization",
)
})?;
let worker_ref = worker_ref_for(&state.runtime, worker_id)?;
state
.runtime
.worker_retention_inventory(&scope.workspace_id, &worker_ref)
.map(Json)
.map_err(RuntimeHttpRestError::runtime)
}
async fn execute_worker_retention(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
body: Result<Json<WorkerRetentionExecutionRequest>, JsonRejection>,
) -> RestResult<WorkerRetentionExecutionResult> {
let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
if request.worker_id.to_string() != worker_id {
return Err(RuntimeHttpRestError::new(
StatusCode::BAD_REQUEST,
"worker_id_mismatch",
"Retention request worker_id does not match the route",
));
}
let scope = auth_workspace_scope(&state, auth.as_ref())?.ok_or_else(|| {
RuntimeHttpRestError::new(
StatusCode::FORBIDDEN,
"workspace_scope_required",
"Worker retention execution requires workspace-scoped authorization",
)
})?;
if request.workspace_id != scope.workspace_id {
return Err(RuntimeHttpRestError::new(
StatusCode::NOT_FOUND,
"worker_not_found",
"Worker was not found in the authenticated Workspace",
));
}
state
.runtime
.execute_worker_retention(&request)
.map(Json)
.map_err(RuntimeHttpRestError::runtime)
}
async fn send_worker_input(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1473,6 +1539,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
if path.ends_with("/completions") {
return Some("workers:read");
}
if path.contains("/retention/") {
return Some("workers:delete");
}
if path.starts_with("/v1/workers/") && *method == Method::DELETE {
return Some("workers:delete");
}
@@ -2066,6 +2135,18 @@ mod tests {
}
}
#[test]
fn retention_routes_require_worker_delete_permission() {
assert_eq!(
required_runtime_permission(&Method::GET, "/v1/workers/worker-1/retention/inventory",),
Some("workers:delete")
);
assert_eq!(
required_runtime_permission(&Method::POST, "/v1/workers/worker-1/retention/execute",),
Some("workers:delete")
);
}
#[test]
fn workdir_routes_require_dedicated_operation_permission() {
assert_eq!(
crates/worker-runtime/src/retention.rs
+4
View File
@@ -117,6 +117,7 @@ pub struct WorkerRetentionExecutionRequest {
pub workspace_id: String,
pub source_runtime_id: String,
pub worker_id: WorkerId,
pub expected_worker_revision: String,
pub expected_run_generation: u64,
pub source_created_at: String,
pub removed_at: String,
@@ -155,6 +156,7 @@ pub struct WorkerSessionArchiveManifest {
pub struct WorkerRetentionExecutionResult {
pub operation_id: String,
pub input_fingerprint: String,
pub expected_worker_revision: String,
pub worker_id: WorkerId,
pub session_disposition: SessionDisposition,
pub diagnostics_disposition: DiagnosticsDisposition,
@@ -521,6 +523,7 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider {
let mut result = WorkerRetentionExecutionResult {
operation_id: request.operation_id.clone(),
input_fingerprint: request.input_fingerprint.clone(),
expected_worker_revision: request.expected_worker_revision.clone(),
worker_id: request.worker_id,
session_disposition: request.session_disposition,
diagnostics_disposition: request.diagnostics_disposition,
@@ -1291,6 +1294,7 @@ mod tests {
WorkerRetentionExecutionRequest {
operation_id: "operation-a".to_string(),
input_fingerprint: "fingerprint-a".to_string(),
expected_worker_revision: "revision-a".to_string(),
archive_id: (disposition == SessionDisposition::Archive)
.then(|| "archive-a".to_string()),
workspace_id: "workspace-a".to_string(),
crates/worker-runtime/src/worker_source.rs
+45 -9
View File
@@ -124,6 +124,8 @@ pub trait EmbeddedWorkerMutationDispatcher: Send + Sync {
proof: InProcessWorkerMutationProof,
target_runtime_id: &str,
target_worker_id: &str,
expected_worker_revision: &str,
reason: &str,
) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError>;
}
@@ -186,6 +188,8 @@ impl RuntimeWorkerMutationForwarder {
&self,
target_runtime_id: &str,
target_worker_id: &str,
expected_worker_revision: &str,
reason: &str,
) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
let proof = self.authority.issue_worker_remove(
&self.scope,
@@ -205,6 +209,8 @@ impl RuntimeWorkerMutationForwarder {
let body = serde_json::json!({
"target_runtime_id": target_runtime_id,
"target_worker_id": target_worker_id,
"expected_worker_revision": expected_worker_revision,
"reason": reason,
});
let response = client
.post(url)
@@ -223,7 +229,13 @@ impl RuntimeWorkerMutationForwarder {
(
RuntimeWorkerMutationTransport::Embedded { dispatcher },
RuntimeOwnedWorkerMutationProof::InProcess(claims),
) => dispatcher.execute_worker_remove(claims, target_runtime_id, target_worker_id),
) => dispatcher.execute_worker_remove(
claims,
target_runtime_id,
target_worker_id,
expected_worker_revision,
reason,
),
_ => Err(RuntimeWorkerMutationForwardError::AuthorityTransportMismatch),
}
}
@@ -316,6 +328,8 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
&self,
target_runtime_id: &str,
target_worker_id: &str,
expected_worker_revision: &str,
reason: &str,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.worker_remove
.as_ref()
@@ -324,7 +338,12 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
"Runtime-owned WorkerRemove forwarding is unavailable".to_string(),
)
})?
.execute_worker_remove(target_runtime_id, target_worker_id)
.execute_worker_remove(
target_runtime_id,
target_worker_id,
expected_worker_revision,
reason,
)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))
}
}
@@ -500,16 +519,22 @@ mod tests {
format!("http://{address}"),
);
let response = forwarder
.execute_worker_remove("runtime-target", "worker-target")
.execute_worker_remove(
"runtime-target",
"worker-target",
"revision-7",
"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(
r#"{"target_runtime_id":"runtime-target","target_worker_id":"worker-target"}"#
));
assert!(request.contains("\"target_runtime_id\":\"runtime-target\""));
assert!(request.contains("\"target_worker_id\":\"worker-target\""));
assert!(request.contains("\"expected_worker_revision\":\"revision-7\""));
assert!(request.contains("\"reason\":\"retire obsolete Worker\""));
let token = request
.lines()
.find_map(|line| {
@@ -541,7 +566,7 @@ mod tests {
#[derive(Default)]
struct RecordingDispatcher {
seen: Mutex<Option<(WorkerMutationSourceClaims, String, String)>>,
seen: Mutex<Option<(WorkerMutationSourceClaims, String, String, String, String)>>,
}
impl EmbeddedWorkerMutationDispatcher for RecordingDispatcher {
fn execute_worker_remove(
@@ -549,11 +574,15 @@ mod tests {
proof: InProcessWorkerMutationProof,
target_runtime_id: &str,
target_worker_id: &str,
expected_worker_revision: &str,
reason: &str,
) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
*self.seen.lock().unwrap() = Some((
proof.into_claims(),
target_runtime_id.to_string(),
target_worker_id.to_string(),
expected_worker_revision.to_string(),
reason.to_string(),
));
Ok(WorkspaceResponse {
status: 202,
@@ -571,10 +600,15 @@ mod tests {
dispatcher.clone(),
);
let response = forwarder
.execute_worker_remove("runtime-target", "worker-target")
.execute_worker_remove(
"runtime-target",
"worker-target",
"revision-7",
"retire obsolete Worker",
)
.unwrap();
assert_eq!(response.status, 202);
let (claims, target_runtime_id, target_worker_id) =
let (claims, target_runtime_id, target_worker_id, expected_revision, reason) =
dispatcher.seen.lock().unwrap().take().unwrap();
assert_eq!(claims.iss, "runtime-embedded");
assert_eq!(claims.worker_id, "worker-source");
@@ -582,6 +616,8 @@ mod tests {
assert_eq!(claims.target_worker_id, "worker-target");
assert_eq!(target_runtime_id, "runtime-target");
assert_eq!(target_worker_id, "worker-target");
assert_eq!(expected_revision, "revision-7");
assert_eq!(reason, "retire obsolete Worker");
}
#[test]
crates/worker/src/feature/builtin/manage_worker.rs
+168 -6
View File
@@ -77,6 +77,11 @@ impl FeatureModule for ManageWorkerFeature {
self.client.clone(),
workspace_id.clone(),
),
WorkerOperation::Remove => definition::<WorkerRemoveInput>(
operation,
self.client.clone(),
workspace_id.clone(),
),
};
context
.tools()
@@ -149,6 +154,15 @@ struct WorkerStopInput {
reason: Option<String>,
}
#[derive(Debug, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct WorkerRemoveInput {
runtime_id: String,
worker_id: String,
expected_worker_revision: String,
reason: String,
}
struct WorkspaceWorkerTool {
operation: WorkerOperation,
client: Arc<dyn WorkspaceClient>,
@@ -161,10 +175,17 @@ enum WorkerOperation {
Spawn,
Stop,
Restore,
Remove,
}
impl WorkerOperation {
const ALL: [Self; 4] = [Self::List, Self::Spawn, Self::Stop, Self::Restore];
const ALL: [Self; 5] = [
Self::List,
Self::Spawn,
Self::Stop,
Self::Restore,
Self::Remove,
];
fn tool_name(self) -> &'static str {
match self {
@@ -172,6 +193,7 @@ impl WorkerOperation {
Self::Spawn => "WorkerSpawn",
Self::Stop => "WorkerStop",
Self::Restore => "WorkerRestore",
Self::Remove => "WorkerRemove",
}
}
@@ -187,6 +209,9 @@ impl WorkerOperation {
Self::Restore => {
"Restore a stopped Backend/Runtime Worker session in the current Workspace."
}
Self::Remove => {
"Remove an eligible stopped, unassigned, non-internal Worker. Supply the current Worker revision and a bounded reason; Backend validation and retention are authoritative."
}
}
}
}
@@ -198,7 +223,30 @@ impl Tool for WorkspaceWorkerTool {
input_json: &str,
ctx: ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let request = match self.operation {
let response = match self.operation {
WorkerOperation::Remove => {
let input = parse::<WorkerRemoveInput>(input_json, "WorkerRemove")?;
let runtime_id = authority_id(&input.runtime_id, "runtime_id")?;
let worker_id = authority_id(&input.worker_id, "worker_id")?;
let expected_worker_revision =
non_empty(input.expected_worker_revision, "expected_worker_revision")?;
let reason = non_empty(input.reason, "reason")?;
if reason.len() > 512 {
return Err(ToolError::ExecutionFailed(
"reason must contain at most 512 bytes".to_string(),
));
}
self.client
.execute_worker_remove(
&runtime_id,
&worker_id,
&expected_worker_revision,
&reason,
)
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?
}
operation => {
let request = match operation {
WorkerOperation::List => {
parse::<WorkerListInput>(input_json, "WorkerList")?;
WorkspaceRequest::get(format!("/api/w/{}/workers", self.workspace_id))
@@ -269,11 +317,13 @@ impl Tool for WorkspaceWorkerTool {
"{}",
)
}
WorkerOperation::Remove => unreachable!("handled above"),
};
let response = self
.client
self.client
.execute(request)
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?
}
};
if !response.is_success() {
return Err(ToolError::ExecutionFailed(format!(
"Workspace Worker operation returned HTTP {}: {}",
@@ -356,6 +406,7 @@ mod tests {
#[derive(Debug, Default)]
struct RecordingWorkspaceClient {
requests: Mutex<Vec<WorkspaceRequest>>,
removals: Mutex<Vec<(String, String, String, String)>>,
}
impl WorkspaceClient for RecordingWorkspaceClient {
@@ -381,6 +432,25 @@ mod tests {
body: "{}".to_string(),
})
}
fn execute_worker_remove(
&self,
target_runtime_id: &str,
target_worker_id: &str,
expected_worker_revision: &str,
reason: &str,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.removals.lock().unwrap().push((
target_runtime_id.to_string(),
target_worker_id.to_string(),
expected_worker_revision.to_string(),
reason.to_string(),
));
Ok(WorkspaceResponse {
status: 200,
body: r#"{"removed":true}"#.to_string(),
})
}
}
#[tokio::test]
@@ -433,7 +503,13 @@ mod tests {
fn worker_tool_family_is_distinct_from_sub_worker_tools() {
assert_eq!(
WorkerOperation::ALL.map(WorkerOperation::tool_name),
["WorkerList", "WorkerSpawn", "WorkerStop", "WorkerRestore"]
[
"WorkerList",
"WorkerSpawn",
"WorkerStop",
"WorkerRestore",
"WorkerRemove",
]
);
}
@@ -481,6 +557,92 @@ mod tests {
assert!(value.get("initial_text").is_none());
}
#[tokio::test]
async fn worker_remove_forwards_only_target_revision_and_bounded_reason() {
let client = Arc::new(RecordingWorkspaceClient::default());
let tool = WorkspaceWorkerTool {
operation: WorkerOperation::Remove,
client: client.clone(),
workspace_id: "workspace%2Ftest".to_string(),
};
tool.execute(
&serde_json::json!({
"runtime_id": "runtime-1",
"worker_id": "worker-7",
"expected_worker_revision": "2026-08-11T20:00:00Z",
"reason": " retire completed Worker "
})
.to_string(),
ToolExecutionContext::new("call-remove", "batch-remove", 0),
)
.await
.unwrap();
assert_eq!(
client.removals.lock().unwrap().as_slice(),
[(
"runtime-1".to_string(),
"worker-7".to_string(),
"2026-08-11T20:00:00Z".to_string(),
"retire completed Worker".to_string(),
)]
);
let schema = serde_json::to_value(schemars::schema_for!(WorkerRemoveInput))
.unwrap()
.to_string();
for field in [
"runtime_id",
"worker_id",
"expected_worker_revision",
"reason",
] {
assert!(schema.contains(field));
}
for forbidden in ["proof", "actor", "workspace_id", "policy", "plan", "stage"] {
assert!(!schema.contains(forbidden), "schema leaked {forbidden}");
}
}
#[tokio::test]
async fn worker_remove_rejects_empty_oversized_and_unknown_authority_input() {
let client = Arc::new(RecordingWorkspaceClient::default());
let tool = WorkspaceWorkerTool {
operation: WorkerOperation::Remove,
client: client.clone(),
workspace_id: "workspace%2Ftest".to_string(),
};
for reason in [" ".to_string(), "x".repeat(513)] {
let _error = tool
.execute(
&serde_json::json!({
"runtime_id": "runtime-1",
"worker_id": "worker-7",
"expected_worker_revision": "revision-1",
"reason": reason,
})
.to_string(),
ToolExecutionContext::new("call-invalid", "batch-remove", 0),
)
.await
.unwrap_err();
}
let _error = tool
.execute(
&serde_json::json!({
"runtime_id": "runtime-1",
"worker_id": "worker-7",
"expected_worker_revision": "revision-1",
"reason": "retire",
"source_proof": "caller-controlled"
})
.to_string(),
ToolExecutionContext::new("call-spoof", "batch-remove", 0),
)
.await
.unwrap_err();
assert!(client.removals.lock().unwrap().is_empty());
}
#[test]
fn worker_inputs_reject_paths_and_parent_traversal() {
assert!(authority_id("https://runtime.example", "runtime_id").is_err());
crates/worker/src/prompt/catalog.rs
+15
View File
@@ -746,6 +746,21 @@ compact_system = "PREFIX\n{% include \"$yoi/internal/compact_system\" %}"
assert!(rendered.contains("bypass user/Ticket authorization"));
}
#[test]
fn orchestrator_role_prompt_fences_worker_remove_authority() {
let source = include_str!("../../../../resources/prompts/role/orchestrator.md");
assert!(source.contains("Use `WorkerRemove` only for a terminal or authoritatively reassigned non-internal Coder"));
assert!(source.contains("exact current `updated_at` value"));
assert!(source.contains("must have no current Ticket assignment"));
assert!(source.contains("pending notification, Reviewer handoff, legal hold, or pin"));
assert!(source.contains("After removal, reread the Worker catalog and attachment state"));
assert!(source.contains("attachment-close, and attachment-release conflicts"));
assert!(source.contains("preserves the Workdir materialization"));
assert!(!source.contains("source proof"));
assert!(!source.contains("provider handle"));
assert!(!source.contains("retention plan"));
}
#[test]
fn sub_worker_spawn_tool_description_renders_profile_block() {
let cat = PromptCatalog::builtins_only().unwrap();
crates/worker/src/worker.rs
+2
View File
@@ -230,6 +230,8 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
&self,
_target_runtime_id: &str,
_target_worker_id: &str,
_expected_worker_revision: &str,
_reason: &str,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
Err(WorkspaceClientError::Unavailable(
"Runtime-owned WorkerRemove forwarding is unavailable".to_string(),
crates/workspace-server/src/hosts.rs
+109 -1
View File
@@ -54,6 +54,9 @@ use worker_runtime::interaction::{
};
use worker_runtime::management::{RuntimeOptions as EmbeddedRuntimeOptions, RuntimeStatus};
use worker_runtime::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveInput};
use worker_runtime::retention::{
WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, WorkerRetentionInventory,
};
pub(crate) const EMBEDDED_RUNTIME_ID: &str = "embedded-worker-runtime";
const EMBEDDED_HOST_KIND: &str = "embedded-worker-runtime-host";
@@ -856,6 +859,25 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
}
}
fn worker_retention_inventory(
&self,
worker_id: &str,
) -> Result<WorkerRetentionInventory, String> {
Err(format!(
"runtime does not implement retention inventory for '{worker_id}'"
))
}
fn execute_worker_retention(
&self,
request: WorkerRetentionExecutionRequest,
) -> Result<WorkerRetentionExecutionResult, String> {
Err(format!(
"runtime does not implement retention execution for '{}'",
request.worker_id
))
}
fn observation_source(
&self,
_worker_id: &str,
@@ -1399,6 +1421,44 @@ impl RuntimeRegistry {
Ok(runtime.delete_worker(worker_id))
}
pub fn worker_retention_inventory(
&self,
worker: &RuntimeWorkerRef,
) -> Result<WorkerRetentionInventory, RuntimeRegistryError> {
validate_backend_identifier("runtime_id", &worker.runtime_id)?;
validate_backend_identifier("worker_id", &worker.worker_id)?;
self.runtime(&worker.runtime_id)?
.worker_retention_inventory(&worker.worker_id)
.map_err(|message| RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: worker.runtime_id.clone(),
code: "worker_retention_inventory_failed".to_string(),
message,
})
}
pub fn execute_worker_retention(
&self,
worker: &RuntimeWorkerRef,
request: WorkerRetentionExecutionRequest,
) -> Result<WorkerRetentionExecutionResult, RuntimeRegistryError> {
validate_backend_identifier("runtime_id", &worker.runtime_id)?;
validate_backend_identifier("worker_id", &worker.worker_id)?;
if request.worker_id.to_string() != worker.worker_id {
return Err(RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: worker.runtime_id.clone(),
code: "worker_id_mismatch".to_string(),
message: "retention request worker_id does not match target".to_string(),
});
}
self.runtime(&worker.runtime_id)?
.execute_worker_retention(request)
.map_err(|message| RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: worker.runtime_id.clone(),
code: "worker_retention_execution_failed".to_string(),
message,
})
}
pub fn observation_source(
&self,
worker: &RuntimeWorkerRef,
@@ -1438,6 +1498,7 @@ impl RuntimeRegistry {
#[derive(Clone)]
pub struct EmbeddedWorkerRuntime {
workspace_id: String,
runtime_id: String,
host_id: String,
runtime: worker_runtime::Runtime,
@@ -1501,8 +1562,9 @@ impl EmbeddedWorkerRuntime {
.bind_runtime_identity(EMBEDDED_RUNTIME_ID)
.expect("fresh embedded Runtime must accept its Backend-owned identity");
Self {
runtime_id: EMBEDDED_RUNTIME_ID.to_string(),
host_id: host_id_for_embedded_workspace(&workspace_id),
workspace_id,
runtime_id: EMBEDDED_RUNTIME_ID.to_string(),
runtime,
execution_enabled: false,
resource_broker: BackendResourceBroker::default(),
@@ -2106,6 +2168,30 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
}
}
fn worker_retention_inventory(
&self,
worker_id: &str,
) -> Result<WorkerRetentionInventory, String> {
let worker_ref = self
.worker_ref(worker_id)
.ok_or_else(|| format!("invalid embedded Worker id '{worker_id}'"))?;
self.runtime
.worker_retention_inventory(&self.workspace_id, &worker_ref)
.map_err(|error| error.to_string())
}
fn execute_worker_retention(
&self,
request: WorkerRetentionExecutionRequest,
) -> Result<WorkerRetentionExecutionResult, String> {
if request.workspace_id != self.workspace_id {
return Err("retention request Workspace does not match embedded Runtime".to_string());
}
self.runtime
.execute_worker_retention(&request)
.map_err(|error| error.to_string())
}
fn observation_source(
&self,
worker_id: &str,
@@ -3128,6 +3214,28 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
}
}
fn worker_retention_inventory(
&self,
worker_id: &str,
) -> Result<WorkerRetentionInventory, String> {
self.get_json::<WorkerRetentionInventory>(&format!(
"/v1/workers/{worker_id}/retention/inventory"
))
.map_err(|diagnostic| diagnostic.message)
}
fn execute_worker_retention(
&self,
request: WorkerRetentionExecutionRequest,
) -> Result<WorkerRetentionExecutionResult, String> {
let worker_id = request.worker_id.to_string();
self.post_json::<_, WorkerRetentionExecutionResult>(
&format!("/v1/workers/{worker_id}/retention/execute"),
&request,
)
.map_err(|diagnostic| diagnostic.message)
}
fn observation_source(
&self,
worker_id: &str,
crates/workspace-server/src/retention.rs
+185 -1
View File
@@ -377,6 +377,7 @@ impl SqliteWorkspaceStore {
workspace_id: plan.workspace_id.clone(),
source_runtime_id: plan.worker.runtime_id.clone(),
worker_id: worker_runtime::identity::WorkerId::new(worker_number),
expected_worker_revision: plan.worker_revision.clone(),
expected_run_generation: plan.run_generation,
source_created_at: worker.created_at,
removed_at,
@@ -391,6 +392,85 @@ impl SqliteWorkspaceStore {
})
}
pub fn recover_worker_removal_execution(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
expected_worker_revision: &str,
reason: &str,
) -> Result<Option<PreparedWorkerRemoval>, WorkerRetentionError> {
bounded("workspace", workspace_id, 160)?;
bounded("revision", expected_worker_revision, 256)?;
bounded("reason", reason, 512)?;
let plan = self.with_conn(|conn| {
conn.query_row(
"SELECT plan_id FROM worker_removal_operations
WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3
AND worker_revision=?4 AND reason=?5
AND state IN ('executing','failed','succeeded')
ORDER BY CASE state WHEN 'succeeded' THEN 0 ELSE 1 END,
created_at DESC LIMIT 1",
params![
workspace_id,
worker.runtime_id,
worker.worker_id,
expected_worker_revision,
reason,
],
|row| row.get::<_, String>(0),
)
.optional()
.map_err(StoreError::from)
.and_then(|plan_id| match plan_id {
Some(plan_id) => load_plan(conn, &plan_id),
None => Ok(None),
})
})?;
let Some(plan) = plan else {
return Ok(None);
};
let worker_number = plan.worker.worker_id.parse::<u64>().map_err(|_| {
WorkerRetentionError::Invalid(
"Runtime Worker id is not a canonical unsigned integer".to_string(),
)
})?;
let worker = if plan.state == WorkerRemovalPlanState::Succeeded {
None
} else {
Some(
self.with_conn(|conn| load_worker(conn, workspace_id, &plan.worker))?
.ok_or(WorkerRetentionError::WorkerNotFound)?,
)
};
Ok(Some(PreparedWorkerRemoval {
runtime_request: WorkerRetentionExecutionRequest {
operation_id: plan.operation_id.clone(),
input_fingerprint: plan.input_fingerprint.clone(),
archive_id: plan.archive_id.clone(),
workspace_id: plan.workspace_id.clone(),
source_runtime_id: plan.worker.runtime_id.clone(),
worker_id: worker_runtime::identity::WorkerId::new(worker_number),
expected_worker_revision: plan.worker_revision.clone(),
expected_run_generation: plan.run_generation,
source_created_at: worker
.as_ref()
.map(|worker| worker.created_at.clone())
.unwrap_or_else(|| plan.created_at.clone()),
removed_at: plan.created_at.clone(),
effective_profile: worker
.as_ref()
.map(|worker| worker.profile.clone())
.unwrap_or_else(|| Some("removed".to_string())),
retention_class: None,
policy_id: plan.policy_id.clone(),
policy_revision: plan.policy_revision,
session_disposition: plan.session_disposition,
diagnostics_disposition: plan.diagnostics_disposition,
},
plan,
}))
}
pub fn fail_worker_removal(
&self,
workspace_id: &str,
@@ -424,7 +504,8 @@ impl SqliteWorkspaceStore {
if plan.state != WorkerRemovalPlanState::Executing {
return Err(StoreError::InvalidInput(format!("stale:{}:plan state {} is not committable", plan.plan_id, state_s(plan.state))));
}
if result.worker_id.to_string() != plan.worker.worker_id
if result.expected_worker_revision != plan.worker_revision
|| result.worker_id.to_string() != plan.worker.worker_id
|| result.session_disposition != plan.session_disposition
|| result.diagnostics_disposition != plan.diagnostics_disposition
{
@@ -1107,6 +1188,7 @@ mod tests {
let result = WorkerRetentionExecutionResult {
operation_id: p.operation_id.clone(),
input_fingerprint: p.input_fingerprint.clone(),
expected_worker_revision: p.worker_revision.clone(),
worker_id: WorkerId::new(1),
session_disposition: p.session_disposition,
diagnostics_disposition: p.diagnostics_disposition,
@@ -1247,6 +1329,7 @@ mod tests {
let r = WorkerRetentionExecutionResult {
operation_id: p.operation_id.clone(),
input_fingerprint: p.input_fingerprint.clone(),
expected_worker_revision: p.worker_revision.clone(),
worker_id: WorkerId::new(1),
session_disposition: SessionDisposition::Purge,
diagnostics_disposition: DiagnosticsDisposition::Purge,
@@ -1265,6 +1348,7 @@ mod tests {
let mut result = WorkerRetentionExecutionResult {
operation_id: plan.operation_id.clone(),
input_fingerprint: plan.input_fingerprint.clone(),
expected_worker_revision: plan.worker_revision.clone(),
worker_id: WorkerId::new(1),
session_disposition: plan.session_disposition,
diagnostics_disposition: plan.diagnostics_disposition,
@@ -1348,6 +1432,106 @@ mod tests {
);
}
#[test]
fn runtime_result_must_match_prepared_worker_revision() {
let s = setup();
let plan = s.plan_worker_removal(&req(), &inv()).unwrap();
let prepared = s
.prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint)
.unwrap();
let mut runtime_result = WorkerRetentionExecutionResult {
operation_id: prepared.plan.operation_id.clone(),
input_fingerprint: prepared.plan.input_fingerprint.clone(),
expected_worker_revision: prepared.plan.worker_revision.clone(),
worker_id: WorkerId::new(1),
session_disposition: prepared.plan.session_disposition,
diagnostics_disposition: prepared.plan.diagnostics_disposition,
archive: None,
source_removed: true,
diagnostics_retained: false,
};
runtime_result.expected_worker_revision = "stale-revision".to_string();
let error = s
.commit_worker_removal(
"w",
&plan.operation_id,
&plan.input_fingerprint,
&runtime_result,
)
.unwrap_err();
assert!(
!error.to_string().is_empty(),
"mismatched Runtime revision must be rejected"
);
}
#[test]
fn succeeded_worker_removal_recovers_after_registry_purge() {
let s = setup();
let request = req();
let plan = s.plan_worker_removal(&request, &inv()).unwrap();
let prepared = s
.prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint)
.unwrap();
let runtime_result = WorkerRetentionExecutionResult {
operation_id: prepared.plan.operation_id.clone(),
input_fingerprint: prepared.plan.input_fingerprint.clone(),
expected_worker_revision: prepared.plan.worker_revision.clone(),
worker_id: WorkerId::new(1),
session_disposition: prepared.plan.session_disposition,
diagnostics_disposition: prepared.plan.diagnostics_disposition,
archive: Some(worker_runtime::retention::WorkerSessionArchiveManifest {
schema_version: 1,
archive_id: prepared.plan.archive_id.clone().unwrap(),
workspace_id: "w".into(),
source_runtime_id: "r".into(),
source_worker_id: WorkerId::new(1),
source_session_id: "s".into(),
segment_ids: vec!["a".into()],
source_created_at: "created".into(),
removed_at: "removed".into(),
archived_at_unix_seconds: 1,
effective_profile: None,
retention_class: None,
content_checksum_sha256: "sum".into(),
content_bytes: 1,
content_file_count: 1,
policy_id: prepared.plan.policy_id.clone(),
policy_revision: prepared.plan.policy_revision,
operation_id: prepared.plan.operation_id.clone(),
input_fingerprint: prepared.plan.input_fingerprint.clone(),
}),
source_removed: true,
diagnostics_retained: false,
};
s.commit_worker_removal(
"w",
&plan.operation_id,
&plan.input_fingerprint,
&runtime_result,
)
.unwrap();
assert!(
s.get_worker_registry("w", &request.worker)
.unwrap()
.is_none()
);
let recovered = s
.recover_worker_removal_execution(
"w",
&request.worker,
&request.expected_worker_revision,
&request.reason,
)
.unwrap()
.unwrap();
assert_eq!(recovered.plan.state, WorkerRemovalPlanState::Succeeded);
assert_eq!(
recovered.runtime_request.expected_worker_revision,
request.expected_worker_revision
);
}
#[test]
fn old_schema_upgrade_seeds_existing_workspace() {
let temp = tempfile::tempdir().unwrap();
crates/workspace-server/src/server.rs
+690 -33
View File
@@ -1,7 +1,7 @@
use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::sync::{Arc, Mutex, Weak};
use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade};
use axum::extract::{Path as AxumPath, Query, Request, State};
@@ -262,21 +262,328 @@ pub struct WorkspaceApi {
resource_broker: BackendResourceBroker,
workdir_sessions: Arc<Mutex<HashMap<RuntimeWorkerRef, WorkdirSessionHandle>>>,
workdir_session_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
worker_remove_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
}
#[derive(Clone)]
struct WorkspaceWorkerRemoveExecutor {
workspace_id: String,
store: Arc<dyn ControlPlaneStore>,
runtime: Weak<RuntimeRegistry>,
workdir_sessions: Arc<Mutex<HashMap<RuntimeWorkerRef, WorkdirSessionHandle>>>,
workdir_session_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
worker_remove_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
}
impl WorkspaceWorkerRemoveExecutor {
fn new(api: &WorkspaceApi) -> Self {
Self {
workspace_id: api.config.workspace_id.clone(),
store: api.store.clone(),
runtime: Arc::downgrade(&api.runtime),
workdir_sessions: api.workdir_sessions.clone(),
workdir_session_locks: api.workdir_session_locks.clone(),
worker_remove_locks: api.worker_remove_locks.clone(),
}
}
async fn resume_worker_retention(
&self,
runtime: &RuntimeRegistry,
target: &RuntimeWorkerRef,
prepared: crate::retention::PreparedWorkerRemoval,
) -> std::result::Result<worker::WorkspaceResponse, String> {
let result =
match runtime.execute_worker_retention(target, prepared.runtime_request.clone()) {
Ok(result) => result,
Err(_) => {
return Ok(worker_remove_error_response(
StatusCode::SERVICE_UNAVAILABLE,
"runtime_retention_failed",
"Runtime retention recovery failed; removal can be retried",
));
}
};
match self.store.commit_worker_removal(
&self.workspace_id,
&prepared.plan.operation_id,
&prepared.plan.input_fingerprint,
&result,
) {
Ok(_) => Ok(worker_remove_success_response(target)),
Err(error) => Ok(worker_retention_error_response(error)),
}
}
async fn execute_async(
&self,
source: crate::worker_source::VerifiedWorkerMutationSource,
target_runtime_id: &str,
target_worker_id: &str,
expected_worker_revision: &str,
reason: &str,
) -> std::result::Result<worker::WorkspaceResponse, String> {
let reason = reason.trim();
if reason.is_empty() || reason.len() > 512 {
return Ok(worker_remove_error_response(
StatusCode::BAD_REQUEST,
"invalid_reason",
"WorkerRemove reason must be between 1 and 512 bytes",
));
}
let runtime = self.runtime.upgrade().ok_or_else(|| {
"Workspace Runtime registry is unavailable during WorkerRemove".to_string()
})?;
let source_is_current_orchestrator =
runtime.list_workers(1_000).items.into_iter().any(|worker| {
worker.worker.runtime_id == source.runtime_id
&& worker.worker.worker_id == source.worker_id
&& worker.singleton_key.as_deref()
== Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY)
});
if !source_is_current_orchestrator {
return Ok(worker_remove_error_response(
StatusCode::FORBIDDEN,
"orchestrator_required",
"WorkerRemove is restricted to the current Workspace Orchestrator",
));
}
if source.runtime_id == target_runtime_id && source.worker_id == target_worker_id {
return Ok(worker_remove_error_response(
StatusCode::CONFLICT,
"self_removal_forbidden",
"The current Orchestrator cannot remove itself",
));
}
let target = RuntimeWorkerRef {
runtime_id: target_runtime_id.to_string(),
worker_id: target_worker_id.to_string(),
};
let remove_lock = {
let mut locks = self
.worker_remove_locks
.lock()
.map_err(|_| "WorkerRemove lock registry was poisoned".to_string())?;
locks
.entry(target.clone())
.or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
.clone()
};
let _remove_guard = remove_lock.lock().await;
let workdir_session_lock = {
let mut locks = self
.workdir_session_locks
.lock()
.map_err(|_| "Workdir session lock registry was poisoned".to_string())?;
locks
.entry(target.clone())
.or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
.clone()
};
let _workdir_session_guard = workdir_session_lock.lock().await;
let prepared = self
.store
.recover_worker_removal_execution(
&self.workspace_id,
&target,
expected_worker_revision,
reason,
)
.map_err(|_| "Worker removal recovery authority is unavailable".to_string())?;
if let Some(prepared) = prepared {
if prepared.plan.state == crate::retention::WorkerRemovalPlanState::Succeeded {
return Ok(worker_remove_success_response(&target));
}
return self
.resume_worker_retention(&runtime, &target, prepared)
.await;
}
let worker = match runtime.worker(&target) {
Ok(worker) => worker,
Err(_) => {
return Ok(worker_remove_error_response(
StatusCode::NOT_FOUND,
"worker_not_found",
"Worker was not found in this Workspace",
));
}
};
if worker.singleton_key.is_some() {
return Ok(worker_remove_error_response(
StatusCode::CONFLICT,
"internal_worker_forbidden",
"Internal service Workers cannot be removed with WorkerRemove",
));
}
if !worker.state.eq_ignore_ascii_case("stopped") {
return Ok(worker_remove_error_response(
StatusCode::CONFLICT,
"worker_not_stopped",
"Worker must be stopped before removal",
));
}
let inventory = match runtime.worker_retention_inventory(&target) {
Ok(inventory) => inventory,
Err(_) => {
return Ok(worker_remove_error_response(
StatusCode::SERVICE_UNAVAILABLE,
"retention_inventory_unavailable",
"Retention inventory could not be loaded; removal can be retried",
));
}
};
let request = crate::retention::WorkerRemovalPlanRequest {
workspace_id: self.workspace_id.clone(),
worker: target.clone(),
expected_worker_revision: expected_worker_revision.to_string(),
reason: reason.to_string(),
};
let plan = match self.store.plan_worker_removal(&request, &inventory) {
Ok(plan) => plan,
Err(error) => return Ok(worker_retention_error_response(error)),
};
let prepared = match self.store.prepare_worker_removal_execution(
&self.workspace_id,
&plan.plan_id,
&plan.input_fingerprint,
) {
Ok(prepared) => prepared,
Err(error) => return Ok(worker_retention_error_response(error)),
};
let session = self
.workdir_sessions
.lock()
.map_err(|_| "Workdir session registry was poisoned".to_string())?
.get(&target)
.cloned();
if let Some(session) = session {
if session.close().await.is_err() {
let _ = self.store.fail_worker_removal(
&self.workspace_id,
&plan.operation_id,
&plan.input_fingerprint,
"workdir_session_close_failed",
);
return Ok(worker_remove_error_response(
StatusCode::SERVICE_UNAVAILABLE,
"attachment_close_failed",
"Worker Workdir session could not be closed; removal can be retried",
));
}
self.workdir_sessions
.lock()
.map_err(|_| "Workdir session registry was poisoned".to_string())?
.remove(&target);
}
if let Err(_) = self.store.detach_worker_workdir(
&self.workspace_id,
&target,
None,
&Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
) {
let _ = self.store.fail_worker_removal(
&self.workspace_id,
&plan.operation_id,
&plan.input_fingerprint,
"workdir_attachment_release_failed",
);
return Ok(worker_remove_error_response(
StatusCode::SERVICE_UNAVAILABLE,
"attachment_release_failed",
"Worker Workdir attachment could not be released; removal can be retried",
));
}
let retention_result =
match runtime.execute_worker_retention(&target, prepared.runtime_request.clone()) {
Ok(result) => result,
Err(_) => {
let _ = self.store.fail_worker_removal(
&self.workspace_id,
&plan.operation_id,
&plan.input_fingerprint,
"runtime_retention_failed",
);
return Ok(worker_remove_error_response(
StatusCode::SERVICE_UNAVAILABLE,
"runtime_retention_failed",
"Runtime retention execution failed; removal can be retried",
));
}
};
match self.store.commit_worker_removal(
&self.workspace_id,
&plan.operation_id,
&plan.input_fingerprint,
&retention_result,
) {
Ok(_) => {}
Err(error) => {
let _ = self.store.fail_worker_removal(
&self.workspace_id,
&plan.operation_id,
&plan.input_fingerprint,
"metadata_commit_failed",
);
return Ok(worker_retention_error_response(error));
}
};
Ok(worker_remove_success_response(&target))
}
}
impl crate::worker_source::VerifiedWorkerRemoveExecutor for WorkspaceWorkerRemoveExecutor {
fn execute(
&self,
source: crate::worker_source::VerifiedWorkerMutationSource,
target_runtime_id: &str,
target_worker_id: &str,
expected_worker_revision: &str,
reason: &str,
) -> std::result::Result<worker::WorkspaceResponse, String> {
let executor = self.clone();
let target_runtime_id = target_runtime_id.to_string();
let target_worker_id = target_worker_id.to_string();
let expected_worker_revision = expected_worker_revision.to_string();
let reason = reason.to_string();
std::thread::spawn(move || {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| error.to_string())?
.block_on(executor.execute_async(
source,
&target_runtime_id,
&target_worker_id,
&expected_worker_revision,
&reason,
))
})
.join()
.map_err(|_| "embedded WorkerRemove executor thread panicked".to_string())?
}
}
impl WorkspaceApi {
pub async fn new(config: ServerConfig, store: Arc<dyn ControlPlaneStore>) -> Result<Self> {
let resource_broker = BackendResourceBroker::default();
let execution_backend = WorkerRuntimeExecutionBackend::new(
ProfileRuntimeWorkerFactory::new(config.workspace_root.clone())
.with_embedded_worker_mutation_dispatcher(
EMBEDDED_RUNTIME_ID,
Arc::new(
let worker_remove_dispatcher = Arc::new(
crate::worker_source::EmbeddedServerWorkerMutationDispatcher::new(
config.clone(),
store.clone(),
),
),
);
let execution_backend = WorkerRuntimeExecutionBackend::new(
ProfileRuntimeWorkerFactory::new(config.workspace_root.clone())
.with_embedded_worker_mutation_dispatcher(
EMBEDDED_RUNTIME_ID,
worker_remove_dispatcher.clone(),
)
.with_runtime_store_dir(config.embedded_runtime_store_root.clone())
.with_resource_client(Arc::new(resource_broker.clone())),
@@ -291,6 +598,7 @@ impl WorkspaceApi {
store,
Arc::new(execution_backend),
resource_broker,
Some(worker_remove_dispatcher),
)
.await
}
@@ -306,6 +614,7 @@ impl WorkspaceApi {
store,
execution_backend,
BackendResourceBroker::default(),
None,
)
.await
}
@@ -315,6 +624,9 @@ impl WorkspaceApi {
store: Arc<dyn ControlPlaneStore>,
execution_backend: Arc<dyn worker_runtime::execution::WorkerExecutionBackend>,
resource_broker: BackendResourceBroker,
worker_remove_dispatcher: Option<
Arc<crate::worker_source::EmbeddedServerWorkerMutationDispatcher>,
>,
) -> Result<Self> {
store
.upsert_workspace(&WorkspaceRecord {
@@ -376,7 +688,7 @@ impl WorkspaceApi {
let runtime = Arc::new(runtime);
let companion = Arc::new(CompanionConsole::disabled());
let observation_proxy = BackendObservationProxy::new(config.runtime_event_sources.clone());
Ok(Self {
let api = Self {
authority: SqliteWorkspaceAuthority::new(
config.database_path.clone(),
config.workspace_id.clone(),
@@ -392,7 +704,14 @@ impl WorkspaceApi {
resource_broker,
workdir_sessions: Arc::new(Mutex::new(HashMap::new())),
workdir_session_locks: Arc::new(Mutex::new(HashMap::new())),
})
worker_remove_locks: Arc::new(Mutex::new(HashMap::new())),
};
if let Some(dispatcher) = worker_remove_dispatcher {
dispatcher
.install_executor(Arc::new(WorkspaceWorkerRemoveExecutor::new(&api)))
.map_err(|message| Error::Config(message.to_string()))?;
}
Ok(api)
}
pub fn workspace_id(&self) -> &str {
@@ -4823,10 +5142,93 @@ async fn scoped_workspace_protocol_ws(
.into_response())
}
#[derive(Debug, Serialize)]
struct WorkerRemoveSuccessResponse<'a> {
removed: bool,
runtime_id: &'a str,
worker_id: &'a str,
}
#[derive(Debug, Serialize)]
struct WorkerRemoveErrorResponse<'a> {
code: &'a str,
message: &'a str,
}
fn worker_remove_success_response(worker: &RuntimeWorkerRef) -> worker::WorkspaceResponse {
let body = serde_json::to_string(&WorkerRemoveSuccessResponse {
removed: true,
runtime_id: &worker.runtime_id,
worker_id: &worker.worker_id,
})
.unwrap_or_else(|_| r#"{"removed":true}"#.to_string());
worker::WorkspaceResponse {
status: StatusCode::OK.as_u16(),
body,
}
}
fn worker_remove_error_response(
status: StatusCode,
code: &str,
message: &str,
) -> worker::WorkspaceResponse {
let body =
serde_json::to_string(&WorkerRemoveErrorResponse { code, message }).unwrap_or_else(|_| {
r#"{"code":"worker_remove_failed","message":"Worker removal failed"}"#.to_string()
});
worker::WorkspaceResponse {
status: status.as_u16(),
body,
}
}
fn worker_retention_error_response(
error: crate::retention::WorkerRetentionError,
) -> worker::WorkspaceResponse {
match error {
crate::retention::WorkerRetentionError::WorkerNotFound
| crate::retention::WorkerRetentionError::CrossWorkspace => worker_remove_error_response(
StatusCode::NOT_FOUND,
"worker_not_found",
"Worker was not found in this Workspace",
),
crate::retention::WorkerRetentionError::WorkerRevisionConflict { .. }
| crate::retention::WorkerRetentionError::PolicyRevisionConflict { .. }
| crate::retention::WorkerRetentionError::StalePlan { .. }
| crate::retention::WorkerRetentionError::OperationFingerprintConflict { .. } => {
worker_remove_error_response(
StatusCode::CONFLICT,
"worker_revision_conflict",
"Worker removal state changed; reread the Worker and retry",
)
}
crate::retention::WorkerRetentionError::Blocked(_) => worker_remove_error_response(
StatusCode::CONFLICT,
"worker_removal_blocked",
"Worker removal is blocked by current assignment, hold, or retention policy",
),
crate::retention::WorkerRetentionError::Invalid(_) => worker_remove_error_response(
StatusCode::BAD_REQUEST,
"invalid_worker_remove",
"Worker removal request is invalid",
),
crate::retention::WorkerRetentionError::PolicyMissing { .. }
| crate::retention::WorkerRetentionError::Store(_) => worker_remove_error_response(
StatusCode::SERVICE_UNAVAILABLE,
"worker_removal_authority_unavailable",
"Worker removal authority is unavailable; removal can be retried",
),
}
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct WorkerRemoveBoundaryRequest {
target_runtime_id: String,
target_worker_id: String,
expected_worker_revision: String,
reason: String,
}
async fn scoped_worker_remove_source_boundary(
@@ -4856,19 +5258,41 @@ async fn scoped_worker_remove_source_boundary(
)
.await
{
Ok(source) => (
StatusCode::NOT_IMPLEMENTED,
Json(serde_json::json!({
"error": "WorkerRemove lifecycle is deferred to its consumer Ticket",
"source": {
"runtime_id": source.runtime_id,
"worker_id": source.worker_id,
"actor_kind": source.actor_kind,
"permission": source.permission,
}
})),
Ok(source) => {
let executor = WorkspaceWorkerRemoveExecutor::new(&api);
match executor
.execute_async(
source,
&request.target_runtime_id,
&request.target_worker_id,
&request.expected_worker_revision,
&request.reason,
)
.await
{
Ok(response) => (
StatusCode::from_u16(response.status)
.unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
[(CONTENT_TYPE, "application/json")],
response.body,
)
.into_response(),
Err(_) => {
let response = worker_remove_error_response(
StatusCode::INTERNAL_SERVER_ERROR,
"worker_remove_failed",
"Worker removal failed before lifecycle execution",
);
(
StatusCode::from_u16(response.status)
.unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
[(CONTENT_TYPE, "application/json")],
response.body,
)
.into_response()
}
}
}
Err(error) => {
let status = match error {
crate::worker_source::WorkerMutationSourceProofError::Replay => {
@@ -6009,7 +6433,7 @@ async fn scoped_check_runtime_config_bundle(
async fn scoped_get_runtime_worker(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
) -> ApiResult<Json<WorkerSummary>> {
) -> ApiResult<Json<WorkerShowProjection>> {
validate_workspace_scope(&api, &path.workspace_id)?;
get_runtime_worker(
State(api),
@@ -7715,10 +8139,17 @@ async fn post_companion_cancel(
Ok(Json(api.companion.cancel(request)))
}
#[derive(Debug, Serialize)]
struct WorkerShowProjection {
#[serde(flatten)]
worker: WorkerSummary,
updated_at: String,
}
async fn get_runtime_worker(
State(api): State<WorkspaceApi>,
AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>,
) -> ApiResult<Json<WorkerSummary>> {
) -> ApiResult<Json<WorkerShowProjection>> {
let worker_ref = RuntimeWorkerRef::new(runtime_id, worker_id);
let worker = api
.runtime
@@ -7731,12 +8162,11 @@ async fn get_runtime_worker(
let workdirs = api
.store
.list_workdir_registry(&api.config.workspace_id, 500)?;
Ok(Json(merge_worker_registry_projection(
Some(&worker),
&record,
links,
&workdirs,
)))
let updated_at = record.updated_at.clone();
Ok(Json(WorkerShowProjection {
worker: merge_worker_registry_projection(Some(&worker), &record, links, &workdirs),
updated_at,
}))
}
async fn restore_runtime_worker(
@@ -13242,7 +13672,7 @@ mod tests {
let temp = tempfile::tempdir().unwrap();
let app = build_router(test_api(temp.path()).await);
let body = r#"{"target_runtime_id":"runtime-target","target_worker_id":"target-worker"}"#;
let body = r#"{"target_runtime_id":"runtime-target","target_worker_id":"target-worker","expected_worker_revision":"revision-1","reason":"retire target Worker"}"#;
let browser = app
.clone()
.oneshot(
@@ -13257,6 +13687,7 @@ mod tests {
.unwrap();
assert_eq!(browser.status(), StatusCode::UNAUTHORIZED);
let legacy = app
.clone()
.oneshot(
Request::builder()
.method("POST")
@@ -13270,6 +13701,20 @@ mod tests {
.await
.unwrap();
assert_eq!(legacy.status(), StatusCode::UNAUTHORIZED);
let body_spoof = app
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/api/w/{TEST_WORKSPACE_ID}/workers/remove"))
.header(CONTENT_TYPE, "application/json")
.body(Body::from(
r#"{"target_runtime_id":"runtime-target","target_worker_id":"target-worker","expected_worker_revision":"revision-1","reason":"retire target Worker","source_proof":"browser-controlled","actor":"orchestrator","policy":"purge"}"#,
))
.unwrap(),
)
.await
.unwrap();
assert_eq!(body_spoof.status(), StatusCode::UNPROCESSABLE_ENTITY);
}
#[tokio::test]
@@ -13338,15 +13783,220 @@ mod tests {
api.config.clone(),
api.store.clone(),
);
let response =
let error =
worker_runtime::worker_source::EmbeddedWorkerMutationDispatcher::execute_worker_remove(
&dispatcher,
fresh_proof,
"runtime-target",
"target-worker",
"revision-1",
"retire target Worker",
)
.unwrap_err();
assert!(error.to_string().contains("executor is unavailable"));
}
#[tokio::test]
async fn worker_remove_rejects_self_running_and_stale_revision_at_caller_boundary() {
let temp = tempfile::tempdir().unwrap();
let api = test_api(temp.path()).await;
let Json(orchestrator) = scoped_start_workspace_orchestrator(
State(api.clone()),
AxumPath(ScopedWorkspacePath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
}),
)
.await
.unwrap();
let source = orchestrator.worker.unwrap().worker;
let verified_source = || crate::worker_source::VerifiedWorkerMutationSource {
runtime_id: source.runtime_id.clone(),
worker_id: source.worker_id.clone(),
actor_kind: worker_runtime::auth::WorkerMutationActorKind::Worker,
permission: worker_runtime::auth::WORKER_REMOVE_PERMISSION.to_string(),
jti: "caller-guard-proof".to_string(),
};
let executor = WorkspaceWorkerRemoveExecutor::new(&api);
let self_response = executor
.execute_async(
verified_source(),
&source.runtime_id,
&source.worker_id,
"irrelevant",
"must reject self",
)
.await
.unwrap();
assert_eq!(self_response.status, StatusCode::CONFLICT.as_u16());
assert!(self_response.body.contains("self_removal_forbidden"));
let spawned = api
.spawn_workspace_worker(
EMBEDDED_WORKER_RUNTIME_ID,
WorkerSpawnRequest {
intent: WorkerSpawnIntent::WorkspaceCompanion,
requested_worker_name: Some("guard-target".to_string()),
acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
expected_segments: 0,
},
profile: worker_runtime::catalog::ProfileSelector::Builtin(
"builtin:companion".to_string(),
),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
resolved_config_bundle: Some(runtime_test_bundle()),
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
resolved_workspace_api: None,
},
)
.unwrap();
assert_eq!(response.status, 501);
let target = spawned.worker.unwrap().worker;
let running_response = executor
.execute_async(
verified_source(),
&target.runtime_id,
&target.worker_id,
"irrelevant",
"must reject a live Worker",
)
.await
.unwrap();
assert_eq!(running_response.status, StatusCode::CONFLICT.as_u16());
assert!(running_response.body.contains("worker_not_stopped"));
api.runtime
.stop_worker(
&target,
WorkerLifecycleRequest {
reason: Some("prepare stale revision guard".to_string()),
ticket_assignment: None,
},
)
.unwrap();
let summary = api.runtime.worker(&target).unwrap();
let record = sync_worker_observation(&api, &summary).unwrap();
let stale_response = executor
.execute_async(
verified_source(),
&target.runtime_id,
&target.worker_id,
&format!("{}-stale", record.updated_at),
"must reject stale revision",
)
.await
.unwrap();
assert_eq!(stale_response.status, StatusCode::CONFLICT.as_u16());
assert!(stale_response.body.contains("worker_revision_conflict"));
}
#[tokio::test]
async fn embedded_worker_remove_executes_retention_and_returns_bounded_result() {
let temp = tempfile::tempdir().unwrap();
let api = test_api(temp.path()).await;
let Json(orchestrator) = scoped_start_workspace_orchestrator(
State(api.clone()),
AxumPath(ScopedWorkspacePath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
}),
)
.await
.unwrap();
let source = orchestrator.worker.unwrap().worker;
let spawned = api
.spawn_workspace_worker(
EMBEDDED_WORKER_RUNTIME_ID,
WorkerSpawnRequest {
intent: WorkerSpawnIntent::WorkspaceCompanion,
requested_worker_name: Some("remove-target".to_string()),
acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
expected_segments: 0,
},
profile: worker_runtime::catalog::ProfileSelector::Builtin(
"builtin:companion".to_string(),
),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
resolved_config_bundle: Some(runtime_test_bundle()),
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
resolved_workspace_api: None,
},
)
.unwrap();
let target = spawned.worker.unwrap().worker;
let stopped = api
.runtime
.stop_worker(
&target,
WorkerLifecycleRequest {
reason: Some("prepare WorkerRemove regression".to_string()),
ticket_assignment: None,
},
)
.unwrap();
assert_eq!(stopped.state, WorkerOperationState::Accepted);
let worker_root = temp
.path()
.join(".test-embedded-runtime-store/workers")
.join(&target.worker_id);
fs::create_dir_all(worker_root.join("session/segments")).unwrap();
fs::write(
worker_root.join("session/session.json"),
serde_json::to_vec_pretty(&json!({
"schema_version": 1,
"session_id": "worker-remove-session"
}))
.unwrap(),
)
.unwrap();
fs::write(
worker_root.join("session/segments/segment-a.jsonl"),
b"retained evidence\n",
)
.unwrap();
let summary = api.runtime.worker(&target).unwrap();
let record = sync_worker_observation(&api, &summary).unwrap();
let response = WorkspaceWorkerRemoveExecutor::new(&api)
.execute_async(
crate::worker_source::VerifiedWorkerMutationSource {
runtime_id: source.runtime_id,
worker_id: source.worker_id,
actor_kind: worker_runtime::auth::WorkerMutationActorKind::Worker,
permission: worker_runtime::auth::WORKER_REMOVE_PERMISSION.to_string(),
jti: "embedded-valid-proof".to_string(),
},
&target.runtime_id,
&target.worker_id,
&record.updated_at,
"retire completed Worker",
)
.await
.unwrap();
assert_eq!(
response.status,
StatusCode::OK.as_u16(),
"{}",
response.body
);
assert!(response.body.contains("\"removed\":true"));
assert!(!response.body.contains("disposition"));
assert!(!response.body.contains("stage"));
assert!(!response.body.contains("path"));
assert!(
api.store
.get_worker_registry(TEST_WORKSPACE_ID, &target)
.unwrap()
.is_none()
);
}
#[tokio::test]
@@ -13524,13 +14174,20 @@ mod tests {
route_token,
)
.body(Body::from(
r#"{"target_runtime_id":"runtime-target","target_worker_id":"target-worker"}"#,
r#"{"target_runtime_id":"runtime-target","target_worker_id":"target-worker","expected_worker_revision":"revision-1","reason":"retire target Worker"}"#,
))
.unwrap(),
)
.await
.unwrap();
assert_eq!(route_response.status(), StatusCode::NOT_IMPLEMENTED);
assert_eq!(route_response.status(), StatusCode::FORBIDDEN);
let route_body = axum::body::to_bytes(route_response.into_body(), usize::MAX)
.await
.unwrap();
let route_body = String::from_utf8(route_body.to_vec()).unwrap();
assert!(route_body.contains("orchestrator_required"));
assert!(!route_body.contains("source"));
assert!(!route_body.contains("proof"));
let mut revoked = trust;
revoked.revoked_at = Some("2026-08-11T00:01:00Z".to_string());
crates/workspace-server/src/store.rs
+149
View File
@@ -474,6 +474,73 @@ pub trait ControlPlaneStore: Send + Sync {
now_seconds: u64,
consumed_at: &str,
) -> Result<bool>;
fn plan_worker_removal(
&self,
request: &crate::retention::WorkerRemovalPlanRequest,
inventory: &worker_runtime::retention::WorkerRetentionInventory,
) -> std::result::Result<
crate::retention::WorkerRemovalPlan,
crate::retention::WorkerRetentionError,
> {
let _ = (request, inventory);
Err(crate::retention::WorkerRetentionError::Invalid(
"Worker retention authority is unavailable".to_string(),
))
}
fn prepare_worker_removal_execution(
&self,
workspace_id: &str,
plan_id: &str,
input_fingerprint: &str,
) -> std::result::Result<
crate::retention::PreparedWorkerRemoval,
crate::retention::WorkerRetentionError,
> {
let _ = (workspace_id, plan_id, input_fingerprint);
Err(crate::retention::WorkerRetentionError::Invalid(
"Worker retention authority is unavailable".to_string(),
))
}
fn recover_worker_removal_execution(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
expected_worker_revision: &str,
reason: &str,
) -> std::result::Result<
Option<crate::retention::PreparedWorkerRemoval>,
crate::retention::WorkerRetentionError,
> {
let _ = (workspace_id, worker, expected_worker_revision, reason);
Ok(None)
}
fn fail_worker_removal(
&self,
workspace_id: &str,
operation_id: &str,
input_fingerprint: &str,
category: &str,
) -> std::result::Result<(), crate::retention::WorkerRetentionError> {
let _ = (workspace_id, operation_id, input_fingerprint, category);
Err(crate::retention::WorkerRetentionError::Invalid(
"Worker retention authority is unavailable".to_string(),
))
}
fn commit_worker_removal(
&self,
workspace_id: &str,
operation_id: &str,
input_fingerprint: &str,
result: &worker_runtime::retention::WorkerRetentionExecutionResult,
) -> std::result::Result<
crate::retention::WorkerRemovalPlan,
crate::retention::WorkerRetentionError,
> {
let _ = (workspace_id, operation_id, input_fingerprint, result);
Err(crate::retention::WorkerRetentionError::Invalid(
"Worker retention authority is unavailable".to_string(),
))
}
fn list_workspaces(&self) -> Result<Vec<WorkspaceRecord>>;
fn upsert_repository(&self, record: &RepositoryRecord) -> Result<()>;
fn get_repository(
@@ -948,6 +1015,88 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn plan_worker_removal(
&self,
request: &crate::retention::WorkerRemovalPlanRequest,
inventory: &worker_runtime::retention::WorkerRetentionInventory,
) -> std::result::Result<
crate::retention::WorkerRemovalPlan,
crate::retention::WorkerRetentionError,
> {
SqliteWorkspaceStore::plan_worker_removal(self, request, inventory)
}
fn prepare_worker_removal_execution(
&self,
workspace_id: &str,
plan_id: &str,
input_fingerprint: &str,
) -> std::result::Result<
crate::retention::PreparedWorkerRemoval,
crate::retention::WorkerRetentionError,
> {
SqliteWorkspaceStore::prepare_worker_removal_execution(
self,
workspace_id,
plan_id,
input_fingerprint,
)
}
fn recover_worker_removal_execution(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
expected_worker_revision: &str,
reason: &str,
) -> std::result::Result<
Option<crate::retention::PreparedWorkerRemoval>,
crate::retention::WorkerRetentionError,
> {
SqliteWorkspaceStore::recover_worker_removal_execution(
self,
workspace_id,
worker,
expected_worker_revision,
reason,
)
}
fn fail_worker_removal(
&self,
workspace_id: &str,
operation_id: &str,
input_fingerprint: &str,
category: &str,
) -> std::result::Result<(), crate::retention::WorkerRetentionError> {
SqliteWorkspaceStore::fail_worker_removal(
self,
workspace_id,
operation_id,
input_fingerprint,
category,
)
}
fn commit_worker_removal(
&self,
workspace_id: &str,
operation_id: &str,
input_fingerprint: &str,
result: &worker_runtime::retention::WorkerRetentionExecutionResult,
) -> std::result::Result<
crate::retention::WorkerRemovalPlan,
crate::retention::WorkerRetentionError,
> {
SqliteWorkspaceStore::commit_worker_removal(
self,
workspace_id,
operation_id,
input_fingerprint,
result,
)
}
fn list_workspaces(&self) -> Result<Vec<WorkspaceRecord>> {
self.with_conn(|conn| {
let mut stmt = conn.prepare(
crates/workspace-server/src/worker_source.rs
+46 -9
View File
@@ -1,3 +1,4 @@
use std::sync::{Arc, OnceLock};
use std::time::{SystemTime, UNIX_EPOCH};
use axum::http::HeaderMap;
@@ -168,18 +169,43 @@ async fn verify_worker_remove_source_with(
})
}
pub(crate) trait VerifiedWorkerRemoveExecutor: Send + Sync {
fn execute(
&self,
source: VerifiedWorkerMutationSource,
target_runtime_id: &str,
target_worker_id: &str,
expected_worker_revision: &str,
reason: &str,
) -> Result<worker::WorkspaceResponse, String>;
}
#[derive(Clone)]
pub(crate) struct EmbeddedServerWorkerMutationDispatcher {
config: crate::server::ServerConfig,
store: std::sync::Arc<dyn crate::store::ControlPlaneStore>,
store: Arc<dyn crate::store::ControlPlaneStore>,
executor: Arc<OnceLock<Arc<dyn VerifiedWorkerRemoveExecutor>>>,
}
impl EmbeddedServerWorkerMutationDispatcher {
pub(crate) fn new(
config: crate::server::ServerConfig,
store: std::sync::Arc<dyn crate::store::ControlPlaneStore>,
store: Arc<dyn crate::store::ControlPlaneStore>,
) -> Self {
Self { config, store }
Self {
config,
store,
executor: Arc::new(OnceLock::new()),
}
}
pub(crate) fn install_executor(
&self,
executor: Arc<dyn VerifiedWorkerRemoveExecutor>,
) -> Result<(), &'static str> {
self.executor
.set(executor)
.map_err(|_| "WorkerRemove executor is already installed")
}
}
@@ -191,11 +217,13 @@ impl worker_runtime::worker_source::EmbeddedWorkerMutationDispatcher
proof: InProcessWorkerMutationProof,
target_runtime_id: &str,
target_worker_id: &str,
expected_worker_revision: &str,
reason: &str,
) -> Result<
worker::WorkspaceResponse,
worker_runtime::worker_source::RuntimeWorkerMutationForwardError,
> {
futures::executor::block_on(verify_worker_remove_source_with(
let source = futures::executor::block_on(verify_worker_remove_source_with(
&self.config,
&self.store,
PresentedWorkerMutationSourceProof::InProcess(proof),
@@ -207,11 +235,20 @@ impl worker_runtime::worker_source::EmbeddedWorkerMutationDispatcher
error.to_string(),
)
})?;
Ok(worker::WorkspaceResponse {
status: 501,
body: "WorkerRemove lifecycle is not implemented by this operation boundary"
.to_string(),
})
let executor = self.executor.get().ok_or_else(|| {
worker_runtime::worker_source::RuntimeWorkerMutationForwardError::Embedded(
"WorkerRemove executor is unavailable".to_string(),
)
})?;
executor
.execute(
source,
target_runtime_id,
target_worker_id,
expected_worker_revision,
reason,
)
.map_err(worker_runtime::worker_source::RuntimeWorkerMutationForwardError::Embedded)
}
}
resources/prompts/role/orchestrator.md
+2
View File
@@ -5,3 +5,5 @@ Keep durable orchestration behavior here and treat the first committed user mess
Do not create or delegate an implementation worktree/branch until the Ticket records enough agreed intent, requirements, and acceptance criteria to bound the work.
Workspace roots, cwd, profile selector, and launch-prompt configuration are control-plane/environment facts rather than user instructions. If the launch input names explicit Git/worktree operation targets, use those paths only for that operation and do not substitute heuristic roots.
Use `WorkerRemove` only for a terminal or authoritatively reassigned non-internal Coder after implementation, review, fix, merge/commit, and report handoffs are complete. Do not remove a Coder merely because one turn completed or it is temporarily idle; retain it while review or request-changes work can still return. The Worker must already be stopped, must not be restoring, must have no current Ticket assignment, pending notification, Reviewer handoff, legal hold, or pin, and must not be this Orchestrator. Immediately before removal, reread authoritative Ticket state, assignment, thread/review evidence, and the target Worker with `WorkerShow`; pass the exact current `updated_at` value as `expected_worker_revision` with a concise reason. After removal, reread the Worker catalog and attachment state. Treat revision, assignment, running/restoring, retention-policy, attachment-close, and attachment-release conflicts as authoritative failures: do not guess policy or retry with stale input. `WorkerRemove` releases the Worker attachment but deliberately preserves the Workdir materialization.