From 849a6b4c1c4c7ec94c8440c61bccdbefd6d41db6 Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 6 Oct 2026 14:50:31 +0900 Subject: [PATCH] fix(server): route embedded WorkerRemove through registry-bound reception --- crates/workspace-server/src/server.rs | 133 +++++++++-------- .../src/server/worker_operations.rs | 141 ++++++++++++++---- .../workspace-worker-operations.md | 5 +- ...026-10-06-t702-stale-checker-validation.md | 28 ++++ 4 files changed, 213 insertions(+), 94 deletions(-) create mode 100644 docs/report/2026-10-06-t702-stale-checker-validation.md diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 2e9a24470..fac133171 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2135,35 +2135,6 @@ impl WorkerRemovalService { } } -impl crate::worker_source::VerifiedWorkerRemoveExecutor for WorkerRemovalService { - fn execute( - &self, - source: crate::worker_source::VerifiedWorkerMutationSource, - target_runtime_id: &str, - target_worker_id: &str, - reason: &str, - ) -> std::result::Result { - let executor = self.clone(); - let target_runtime_id = target_runtime_id.to_string(); - let target_worker_id = target_worker_id.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, - &reason, - )) - }) - .join() - .map_err(|_| "embedded WorkerRemove executor thread panicked".to_string())? - } -} - #[derive(Clone)] pub struct WorkspaceServerApi { template: Arc, @@ -3458,7 +3429,9 @@ impl WorkspaceApi { }; if let Some(dispatcher) = worker_remove_dispatcher { dispatcher - .install_executor(Arc::new(WorkerRemovalService::new(&api))) + .install_executor(Arc::new( + worker_operations::EmbeddedWorkspaceWorkerRemoveExecutor::new(&api), + )) .map_err(|message| Error::Config(message.to_string()))?; } let expired_external_workdirs = api.store.reconcile_external_workdir_grants_after_restart( @@ -28915,27 +28888,9 @@ fn resolve_workspace_worker_reference( runtime_id: &str, reference: &str, ) -> ApiResult { - let direct = RuntimeWorkerRef::new(runtime_id, reference); - if let Some(record) = api - .store - .get_worker_registry(&api.config.workspace_id, &direct)? - { - return Ok(record.worker); - } - let worker = api - .store - .resolve_worker_resource_reference(&api.config.workspace_id, reference)? - .filter(|worker| worker.runtime_id == runtime_id) - .ok_or_else(|| Error::UnknownWorker { - worker: RuntimeWorkerRef::new(runtime_id, reference), - })?; - let record = api - .store - .get_worker_registry(&api.config.workspace_id, &worker)? - .ok_or_else(|| Error::UnknownWorker { - worker: RuntimeWorkerRef::new(runtime_id, reference), - })?; - Ok(record.worker) + Ok(WorkspaceWorker::resolve(api, runtime_id, reference)? + .identity() + .clone()) } async fn get_runtime_worker( @@ -51491,7 +51446,27 @@ mod tests { #[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 config = test_server_config(temp.path()); + let store = Arc::new(test_control_store(&config)); + seed_test_registered_workspace(store.as_ref(), &config) + .await + .unwrap(); + let dispatcher = Arc::new( + crate::worker_source::EmbeddedServerWorkerMutationDispatcher::new( + config.clone(), + store.clone(), + ), + ); + // Use the same constructor-installed executor as the production embedded factory. + let api = WorkspaceApi::new_with_execution_backend_and_broker( + config, + store, + Arc::new(DeterministicExecutionBackend::default()), + BackendResourceBroker::default(), + Some(dispatcher.clone()), + ) + .await + .unwrap(); let Json(orchestrator) = scoped_start_workspace_orchestrator( State(api.clone()), AxumPath(ScopedWorkspacePath { @@ -51532,6 +51507,27 @@ mod tests { ) .unwrap(); let target = spawned.worker.unwrap().worker; + seed_worker_control_grant(&api, &source, &target, "embedded-valid-proof"); + let target_key = api + .store + .worker_resource_key(TEST_WORKSPACE_ID, &target) + .unwrap() + .unwrap(); + let forwarder = worker_runtime::worker_source::RuntimeWorkerMutationForwarder::embedded( + &source.runtime_id, + worker_runtime::RuntimeWorkspaceScope::new(TEST_WORKSPACE_ID, "embedded-backend"), + &source.worker_id, + dispatcher, + ); + let running = forwarder + .execute_worker_remove( + &target.runtime_id, + &target_key, + "must retain running Worker", + ) + .unwrap(); + assert_eq!(running.status, StatusCode::CONFLICT.as_u16()); + assert!(running.body.contains("worker_not_stopped")); let stopped = api .runtime .stop_worker( @@ -51564,25 +51560,30 @@ mod tests { .unwrap(); let summary = api.runtime.worker(&target).unwrap(); sync_worker_observation(&api, &summary).unwrap(); - seed_worker_control_grant(&api, &source, &target, "embedded-valid-proof"); + let mut subscriber = api.worker_projection.subscribe_ordered(10).unwrap(); let _ = subscriber.take_snapshot(); - let response = WorkerRemovalService::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, - "retire completed Worker", - ) + let handle = WorkspaceWorker::resolve(&api, &target.runtime_id, &target_key).unwrap(); + handle + .set_pinned(&WorkerOperationContext::Backend, true) .await .unwrap(); + let pinned = forwarder + .execute_worker_remove(&target.runtime_id, &target_key, "must retain pinned Worker") + .unwrap(); + assert_eq!(pinned.status, StatusCode::CONFLICT.as_u16()); + assert!(pinned.body.contains("worker_removal_blocked")); + handle + .set_pinned(&WorkerOperationContext::Backend, false) + .await + .unwrap(); + + // A resource key must be normalized by common reception after its exact + // target-bound proof is verified; the old raw service executor cannot do this. + let response = forwarder + .execute_worker_remove(&target.runtime_id, &target_key, "retire completed Worker") + .unwrap(); assert_eq!( response.status, StatusCode::OK.as_u16(), diff --git a/crates/workspace-server/src/server/worker_operations.rs b/crates/workspace-server/src/server/worker_operations.rs index 5d83b7b99..d8ada5fec 100644 --- a/crates/workspace-server/src/server/worker_operations.rs +++ b/crates/workspace-server/src/server/worker_operations.rs @@ -99,29 +99,56 @@ impl WorkerOperationContext { /// A registry identity, not an authorization lease or a snapshot of live execution. #[derive(Clone)] -pub(crate) struct WorkspaceWorker { - api: WorkspaceApi, +pub(crate) struct WorkspaceWorker { + api: Services, + removal: WorkerRemovalService, identity: RuntimeWorkerRef, } -impl WorkspaceWorker { - pub(crate) fn resolve(api: &WorkspaceApi, runtime_id: &str, reference: &str) -> Result { - let identity = resolve_workspace_worker_reference(api, runtime_id, reference) - .map_err(|error| error.error)?; +// The removal-only specialization keeps the installed embedded dispatcher from +// retaining WorkspaceApi -> Runtime -> dispatcher -> WorkspaceApi as a strong cycle. +// Both specializations resolve the same registry identity and call the same remove method. +impl WorkspaceWorker { + fn resolve_with_services( + api: Services, + removal: WorkerRemovalService, + runtime_id: &str, + reference: &str, + ) -> Result { + let direct = RuntimeWorkerRef::new(runtime_id, reference); + let identity = if let Some(record) = removal + .store + .get_worker_registry(&removal.workspace_id, &direct)? + { + record.worker + } else { + let worker = removal + .store + .resolve_worker_resource_reference(&removal.workspace_id, reference)? + .filter(|worker| worker.runtime_id == runtime_id) + .ok_or_else(|| Error::UnknownWorker { + worker: direct.clone(), + })?; + removal + .store + .get_worker_registry(&removal.workspace_id, &worker)? + .ok_or_else(|| Error::UnknownWorker { worker: direct })? + .worker + }; Ok(Self { - api: api.clone(), + api, + removal, identity, }) } - pub(super) fn identity(&self) -> &RuntimeWorkerRef { &self.identity } fn current_record(&self) -> ApiResult { - self.api + self.removal .store - .get_worker_registry(self.api.workspace_id(), &self.identity)? + .get_worker_registry(&self.removal.workspace_id, &self.identity)? .ok_or_else(|| { Error::UnknownWorker { worker: self.identity.clone(), @@ -130,6 +157,83 @@ impl WorkspaceWorker { }) } + pub(super) async fn remove( + &self, + source: crate::worker_source::VerifiedWorkerMutationSource, + reason: &str, + ) -> std::result::Result { + self.current_record() + .map_err(|error| error.error.to_string())?; + self.removal + .execute_async( + source, + &self.identity.runtime_id, + &self.identity.worker_id, + reason, + ) + .await + } +} + +/// Server-owned adapter for Runtime's independent embedded Tool dispatcher. +/// The proof was target-bound and verified before this synchronous transport seam. +pub(super) struct EmbeddedWorkspaceWorkerRemoveExecutor { + removal: WorkerRemovalService, +} +impl EmbeddedWorkspaceWorkerRemoveExecutor { + pub(super) fn new(api: &WorkspaceApi) -> Self { + Self { + removal: WorkerRemovalService::new(api), + } + } +} +impl crate::worker_source::VerifiedWorkerRemoveExecutor for EmbeddedWorkspaceWorkerRemoveExecutor { + fn execute( + &self, + source: crate::worker_source::VerifiedWorkerMutationSource, + target_runtime_id: &str, + target_worker_id: &str, + reason: &str, + ) -> std::result::Result { + let worker = match WorkspaceWorker::resolve_with_services( + (), + self.removal.clone(), + target_runtime_id, + target_worker_id, + ) { + Ok(worker) => worker, + Err(Error::UnknownWorker { .. }) => { + return Ok(worker_remove_error_response( + StatusCode::NOT_FOUND, + "unknown_worker", + "The target Worker is not known in this Workspace", + )); + } + Err(error) => return Err(error.to_string()), + }; + let reason = reason.to_owned(); + std::thread::spawn(move || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|error| error.to_string())? + .block_on(worker.remove(source, &reason)) + }) + .join() + .map_err(|_| "embedded WorkerRemove executor thread panicked".to_owned())? + } +} + +impl WorkspaceWorker { + pub(crate) fn resolve(api: &WorkspaceApi, runtime_id: &str, reference: &str) -> Result { + Self::resolve_with_services( + api.clone(), + WorkerRemovalService::new(api), + runtime_id, + reference, + ) + } + pub(super) async fn authorize_operation( &self, context: &WorkerOperationContext, @@ -326,23 +430,6 @@ impl WorkspaceWorker { .await } - pub(super) async fn remove( - &self, - source: crate::worker_source::VerifiedWorkerMutationSource, - reason: &str, - ) -> std::result::Result { - self.current_record() - .map_err(|error| error.error.to_string())?; - WorkerRemovalService::new(&self.api) - .execute_async( - source, - &self.identity.runtime_id, - &self.identity.worker_id, - reason, - ) - .await - } - pub(crate) async fn connect_protocol( &self, context: &WorkerOperationContext, diff --git a/docs/architecture/workspace-worker-operations.md b/docs/architecture/workspace-worker-operations.md index 36fc29cb0..ea247b845 100644 --- a/docs/architecture/workspace-worker-operations.md +++ b/docs/architecture/workspace-worker-operations.md @@ -28,6 +28,7 @@ For the HTTP rows below, `R` abbreviates `/api/runtimes/{runtime_id}/workers/{wo | `GET /api/w/{workspace_id}/runtimes/{runtime_id}/cleanup-plan` | `build_runtime_cleanup_plan` (preview, not a Worker mutation) | Workspace cleanup planner/projections | | `POST /api/w/{workspace_id}/runtimes/{runtime_id}/cleanup-executions` | `execute_runtime_cleanup_with_context` validates plan revision/digest and selected candidates → each selected Worker's `delete_from_plan` | `WorkerRemovalService::execute_cleanup_removal` → existing retention planning, Runtime retention execution, and Workspace catalog commit. Workdir-only actions remain in the cleanup service | | `POST /api/w/{workspace_id}/workers/remove` (`WorkerRemove`, not a browser removal alias) | `worker_remove_contract` verifies its source proof against the requested target → `WorkspaceWorker::remove` | `WorkerRemovalService::execute_async` checks the `remove` grant and coordinates retention/removal. It rejects self-removal and requires a stopped target for a new removal; existing durable removal operations have their own recovery path | +| Embedded `WorkerRemove` Tool | Runtime-owned in-process target-bound proof → `EmbeddedServerWorkerMutationDispatcher` verifies/consumes proof → installed `EmbeddedWorkspaceWorkerRemoveExecutor` → `WorkspaceWorker::resolve_with_services` → the same `WorkspaceWorker::remove` | Same `WorkerRemovalService::execute_async`, grant-lock/recheck, self/live/retention guards, durable plan/recovery and catalog commit as HTTP; no Runtime-to-Server dependency | | Dedicated WS `R/protocol/ws` and `W/protocol/ws` | `worker_protocol_ws` (scoped wrapper for `W`) → `WorkspaceWorker::connect_protocol` → `workspace_worker_protocol_session` → bound sender | One remote WS or embedded execution transport; controller/session owns protocol operation results | | Multiplexed WS `/api/w/{workspace_id}/protocol/ws`, runtime-scoped `WorkerProtocol` selector | `scoped_workspace_protocol_ws` → `serve_workspace_subscription` → resolve + `connect_protocol`; subscription stores the bound sender | Same connected execution transport as dedicated WS; the subscription layer wraps its snapshot/events, not a different execution implementation | | Worker-control `POST /api/w/{workspace_id}/worker-control/workers/{runtime_id}/{worker_id}/{input,cancel,stop,restore}` | `send_known_worker_input`, `cancel_known_worker`, `stop_known_worker`, `restore_known_worker` authenticate the Worker source, construct `WorkerControl`, and call the same identity methods/restore adapter | Same Runtime execution owners, under the applicable active target grant; known-worker input adds its existing non-Accepted-result error translation | @@ -66,7 +67,9 @@ For Worker-control user input, `input` converts `User` to `UserIfIdle`; admissio Scoped Restore may coordinate Ticket assignment only when both `ticket_id` and `assignment_operation_id` are supplied. A Worker-control context cannot use either assignment parameter. Aggregate cleanup rejects all Worker-control requests: Worker callers must retain the stronger target-bound WorkerRemove proof/retention path. A control grant is not authority over Ticket assignments or Workdir cleanup. The cleanup coordinator revalidates Account credentials, Origin, and active Workspace even for a Workdir-only request. -`WorkerRemove` is intentionally a different authorization entry point. It accepts a verified mutation source rather than `WorkerOperationContext`, verifies its target-bound proof before calling the identity method, and keeps source-specific and grant revalidation in the removal service. It is not permission-free just because it does not call `authorize_operation`. +The HTTP and embedded `WorkerRemove` entry points share the registry-bound removal method. `WorkerRemove` intentionally has a distinct authorization contract from the other operations. It accepts a verified mutation source rather than `WorkerOperationContext`, verifies its target-bound proof before calling the identity method, and keeps source-specific and grant revalidation in the removal service. It is not permission-free just because it does not call `authorize_operation`. + +The embedded executor retains only the removal services with a **weak** RuntimeRegistry reference. Its removal-only `WorkspaceWorker<()>` specialization shares resolution, current-record checks and the removal method with the full `WorkspaceWorker` handle. This service projection avoids a strong `WorkspaceApi → Runtime → dispatcher → WorkspaceApi` ownership cycle; it is not another Worker ledger or an authorization fallback. It cannot call the full handle's restore/input/protocol methods. The existing independent Runtime-owned proof forwarder and verification/replay seam stay unchanged. ## Tokens and outcomes are operation-specific diff --git a/docs/report/2026-10-06-t702-stale-checker-validation.md b/docs/report/2026-10-06-t702-stale-checker-validation.md new file mode 100644 index 000000000..c70e53c12 --- /dev/null +++ b/docs/report/2026-10-06-t702-stale-checker-validation.md @@ -0,0 +1,28 @@ +# T-702 validation: intermittent stale Ticket checker assertion + +The full `yoi-workspace-server` library suite intermittently fails +`server::tests::ticket_checker_suppresses_stale_and_clean_results_and_diagnoses_invalid_output` +at its post-stale-result `assert!(execution.take_inputs().is_empty())`. The same unchanged +assertion passed in some isolated/full runs and failed in others during T-702 validation. + +To separate this from Worker operation reception, I archived original base +`b96bd110c623fafdd7e8df1b153e8c640c793abe` under the ignored `target/t702-baseline` +directory without moving/replacing refs or discarding the Ticket Workdir. A first +shared-target build failed with stale dependency-interface artifacts and was **not** +baseline evidence. I then used an isolated `CARGO_TARGET_DIR=target/t702-baseline-build` +with `CARGO_BUILD_JOBS=3`. The baseline compiled successfully; its isolated checker test +passed. Its full library run with `--test-threads=4` reproduced the same assertion: +**599 passed, 1 failed, 600 total** (79.19s), at baseline `server.rs:44331`. +The post-review T-702 fix's full library run reported **611 passed, 1 failed, 612 total** +(79.01s), at the corresponding `server.rs:44266` assertion. This demonstrates an inherited +intermittent validation failure, not a green suite or proof of its root cause. +A subsequent post-review-fix whole-crate run, with the same implementation and no skips, +passed **612 library + 11 binary tests and doc tests** (library 81.24s). That successful +run does not erase the failed runs or resolve the underlying checker race. + +The Worker reception fix's 12 operation-boundary tests, 5 WorkerRemove tests (including +constructor-installed embedded dispatcher, target-bound proof, resource-key normalization, +running/pinned refusals, retention and catalog removal), root `cargo check`, formatting and +diff checks passed. The checker implementation/test was not changed or skipped to conceal +the failure. Follow-up should instrument competing checker-job dispatch and deterministic +execution-input consumption so this assertion has an explicit synchronization boundary.