refactor: integrate registry-bound worker operations

crates/server-api/src/lib.rs

@@ -2077,6 +2077,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_restore_alias(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] runtime_id: String,
         #[path] worker_id: String,
     ) -> Result<WorkerRestoreResponse, RepositoryApiError>;
@@ -2091,6 +2092,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_restore(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] workspace_id: String,
         #[path] runtime_id: String,
         #[path] worker_id: String,
@@ -2107,6 +2109,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_pin(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] workspace_id: String,
         #[path] runtime_id: String,
         #[path] worker_id: String,
@@ -2122,6 +2125,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_unpin(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] workspace_id: String,
         #[path] runtime_id: String,
         #[path] worker_id: String,
@@ -2151,6 +2155,7 @@ pub trait ServerApi {
     )]
     async fn runtime_cleanup_execute(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] workspace_id: String,
         #[path] runtime_id: String,
         #[body] request: ExecuteRuntimeCleanupRequest,
@@ -2166,6 +2171,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_input_alias(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] runtime_id: String,
         #[path] worker_id: String,
         #[body] request: RuntimeWorkerInputRequest,
@@ -2181,6 +2187,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_input(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] workspace_id: String,
         #[path] runtime_id: String,
         #[path] worker_id: String,
@@ -2297,6 +2304,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_stop_alias(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] runtime_id: String,
         #[path] worker_id: String,
         #[body] request: RuntimeWorkerLifecycleRequest,
@@ -2312,6 +2320,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_stop(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] workspace_id: String,
         #[path] runtime_id: String,
         #[path] worker_id: String,
@@ -2328,6 +2337,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_cancel_alias(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] runtime_id: String,
         #[path] worker_id: String,
         #[body] request: RuntimeWorkerLifecycleRequest,
@@ -2343,6 +2353,7 @@ pub trait ServerApi {
     )]
     async fn runtime_worker_cancel(
         &self,
+        #[extension] context: ServerRequestContext,
         #[path] workspace_id: String,
         #[path] runtime_id: String,
         #[path] worker_id: String,

crates/workspace-server/src/server/tests/worker_operations_tests.rs

@@ -0,0 +1,864 @@
+use super::*;
+
+fn operation_input(kind: WorkerInputKind) -> WorkerInputRequest {
+    WorkerInputRequest {
+        kind,
+        content: "operation-boundary-test".into(),
+        submission_request_id: None,
+        segments: None,
+    }
+}
+
+fn operation_lifecycle() -> WorkerLifecycleRequest {
+    WorkerLifecycleRequest {
+        reason: Some("operation-boundary-test".into()),
+        ticket_assignment: None,
+    }
+}
+
+fn operation_cleanup_candidate(identity: &RuntimeWorkerRef) -> CleanupWorkerCandidate {
+    CleanupWorkerCandidate {
+        target_id: format!("worker:{}:{}", identity.runtime_id, identity.worker_id),
+        action: server_api::CleanupTargetKind::WorkerDelete,
+        worker_id: identity.worker_id.clone(),
+        runtime_worker_id: identity.worker_id.clone(),
+        runtime_id: identity.runtime_id.clone(),
+        reason: "operation-boundary-test".into(),
+        blocking_reason: None,
+        pinned: false,
+        retention_state: "normal".into(),
+        linked_workdir_ids: Vec::new(),
+        running_linked: false,
+        estimated_reclaim_bytes: None,
+    }
+}
+
+async fn browser_operation_context(api: &WorkspaceApi) -> (WorkerOperationContext, String) {
+    seed_test_api_token(api.store.as_ref(), "worker-operation-browser");
+    let token = "worker-operation-browser-session";
+    api.store
+        .create_browser_session(&BrowserSessionRecord {
+            token_hash: crate::auth::token_hash(token),
+            session_id: "worker-operation-session".into(),
+            user_id: "user-worker-operation-browser".into(),
+            created_at: "2026-01-01T00:00:00Z".into(),
+            expires_at: "2099-01-01T00:00:00Z".into(),
+            revoked_at: None,
+        })
+        .unwrap();
+    let AuthConfig::Passkey {
+        cookie_name,
+        origin,
+        ..
+    } = &api.config.auth;
+    let mut headers = HeaderMap::new();
+    headers.insert(ORIGIN, origin.parse().unwrap());
+    headers.insert(
+        axum::http::header::COOKIE,
+        format!("{cookie_name}={token}").parse().unwrap(),
+    );
+    let actor = resolve_actor(&ServerAuthApi::from(api), &headers)
+        .await
+        .unwrap()
+        .expect("valid browser session");
+    (
+        WorkerOperationContext::Browser {
+            headers,
+            source: authenticated_browser_input_source(&actor),
+        },
+        crate::auth::token_hash(token),
+    )
+}
+
+fn browser_operation_methods() -> Vec<protocol::Method> {
+    vec![
+        protocol::Method::Pause {
+            command: protocol::WorkerCommandEnvelope::new(701),
+        },
+        protocol::Method::Resume {
+            command: protocol::WorkerCommandEnvelope::new(702),
+        },
+        protocol::Method::Cancel {
+            command: protocol::WorkerCommandEnvelope::new(703),
+        },
+        protocol::Method::Compact {
+            command: protocol::WorkerCommandEnvelope::new(704),
+        },
+        protocol::Method::Submit {
+            submission_request_id: "worker-operation-submit".into(),
+            input: vec![protocol::Segment::Text {
+                content: "must not reach execution".into(),
+            }],
+        },
+        protocol::Method::Notify {
+            notification_request_id: "worker-operation-notify".into(),
+            message: "must not reach execution".into(),
+        },
+        protocol_test_completion("worker-operation-completion"),
+    ]
+}
+
+#[tokio::test]
+async fn handle_resource_keys_preserve_identity_for_equal_local_ids_across_runtimes() {
+    let workspace = tempfile::tempdir().unwrap();
+    let api = test_api(workspace.path()).await;
+    let a = RuntimeWorkerRef::new("offline-runtime-a", "shared-local-id");
+    let b = RuntimeWorkerRef::new("offline-runtime-b", "shared-local-id");
+    for identity in [&a, &b] {
+        seed_worker_source_member(&api, &identity.runtime_id, &identity.worker_id);
+    }
+    let key = |identity: &RuntimeWorkerRef| {
+        api.store
+            .worker_resource_key(TEST_WORKSPACE_ID, identity)
+            .unwrap()
+            .unwrap()
+    };
+    assert_ne!(key(&a), key(&b));
+
+    for identity in [&a, &b] {
+        let direct =
+            WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap();
+        let canonical =
+            WorkspaceWorker::resolve(&api, &identity.runtime_id, &key(identity)).unwrap();
+        let pinned = canonical
+            .set_pinned(&WorkerOperationContext::Backend, true)
+            .await
+            .unwrap();
+        assert_eq!(pinned.runtime_id, identity.runtime_id);
+        assert_eq!(pinned.worker_id, identity.worker_id);
+        let unpinned = direct
+            .set_pinned(&WorkerOperationContext::Backend, false)
+            .await
+            .unwrap();
+        assert_eq!(unpinned.runtime_id, pinned.runtime_id);
+        assert_eq!(unpinned.worker_id, pinned.worker_id);
+    }
+    assert!(matches!(
+        WorkspaceWorker::resolve(&api, &b.runtime_id, &key(&a)),
+        Err(Error::UnknownWorker { .. })
+    ));
+}
+
+#[tokio::test]
+async fn handle_resolution_rejects_unknown_foreign_and_internal_only_workers() {
+    let workspace = tempfile::tempdir().unwrap();
+    let api = test_api(workspace.path()).await;
+    let known = spawn_ticket_check_source(&api, "operation-resolution-parent");
+    let mut foreign = api
+        .store
+        .get_worker_registry(TEST_WORKSPACE_ID, &known)
+        .unwrap()
+        .unwrap();
+    let mut other_workspace = api
+        .store
+        .get_workspace(TEST_WORKSPACE_ID)
+        .await
+        .unwrap()
+        .unwrap();
+    other_workspace.workspace_id = "other-workspace".into();
+    api.store.upsert_workspace(&other_workspace).await.unwrap();
+    foreign.workspace_id = other_workspace.workspace_id;
+    foreign.worker.worker_id = "foreign-only-worker".into();
+    api.store.upsert_worker_registry(&foreign).unwrap();
+    assert!(
+        api.store
+            .get_worker_registry(TEST_WORKSPACE_ID, &foreign.worker)
+            .unwrap()
+            .is_none()
+    );
+
+    let RuntimeObservationSource::Embedded(source) =
+        api.runtime.observation_source(&known).unwrap()
+    else {
+        panic!("expected embedded fixture");
+    };
+    let internal = protocol::InternalWorkerRef {
+        session_id: "internal-only-session".into(),
+        parent_session_id: None,
+        name: "internal-only-child".into(),
+        kind: protocol::InternalWorkerKind::SubWorker,
+    };
+    source
+        .runtime
+        .observe_worker_event(
+            &source.worker_ref,
+            protocol::Event::InternalWorker {
+                worker: internal.clone(),
+                revision: 1,
+                event: Box::new(protocol::Event::WorkerState {
+                    snapshot: protocol::WorkerStateSnapshot::initial(),
+                }),
+            },
+        )
+        .unwrap();
+
+    for reference in [
+        "unknown-worker",
+        &foreign.worker.worker_id,
+        &internal.session_id,
+        &internal.name,
+    ] {
+        assert!(
+            matches!(
+                WorkspaceWorker::resolve(&api, &known.runtime_id, reference),
+                Err(Error::UnknownWorker { .. })
+            ),
+            "unexpected registry handle for {reference}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn offline_handle_can_pin_but_deleted_handle_cannot_mutate_or_connect() {
+    let workspace = tempfile::tempdir().unwrap();
+    let api = test_api(workspace.path()).await;
+    let identity = RuntimeWorkerRef::new("unregistered-offline-runtime", "42");
+    seed_worker_source_member(&api, &identity.runtime_id, &identity.worker_id);
+    assert!(
+        api.runtime.worker(&identity).is_err(),
+        "fixture must have no live Runtime"
+    );
+    let handle = WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap();
+    let retained = handle
+        .set_pinned(&WorkerOperationContext::Backend, true)
+        .await
+        .unwrap();
+    assert!(retained.pinned);
+    assert_eq!(
+        api.store
+            .get_worker_registry(TEST_WORKSPACE_ID, &identity)
+            .unwrap()
+            .unwrap()
+            .retention_state,
+        "pinned"
+    );
+
+    let mut wrong_candidate = operation_cleanup_candidate(&identity);
+    wrong_candidate.runtime_worker_id = "different-worker".into();
+    assert!(matches!(
+        handle.delete_from_plan(&WorkerOperationContext::Backend, &wrong_candidate)
+            .await.unwrap_err().error,
+        Error::InvalidInput(message) if message.contains("resolved Worker")
+    ));
+
+    api.store
+        .delete_worker_registry(TEST_WORKSPACE_ID, &identity)
+        .unwrap();
+    assert!(
+        matches!(handle.set_pinned(&WorkerOperationContext::Backend, false)
+        .await.unwrap_err().error, Error::UnknownWorker { worker } if worker == identity)
+    );
+    assert!(matches!(
+        handle
+            .restore(&WorkerOperationContext::Backend)
+            .await
+            .unwrap_err()
+            .error,
+        Error::UnknownWorker { .. }
+    ));
+    assert!(matches!(
+        handle
+            .stop(&WorkerOperationContext::Backend, operation_lifecycle())
+            .await
+            .unwrap_err()
+            .error,
+        Error::UnknownWorker { .. }
+    ));
+    assert!(matches!(
+        handle
+            .cancel(&WorkerOperationContext::Backend, operation_lifecycle())
+            .await
+            .unwrap_err()
+            .error,
+        Error::UnknownWorker { .. }
+    ));
+    for kind in [WorkerInputKind::User, WorkerInputKind::Notify] {
+        assert!(matches!(
+            handle
+                .input(&WorkerOperationContext::Backend, operation_input(kind))
+                .await
+                .unwrap_err()
+                .error,
+            Error::UnknownWorker { .. }
+        ));
+    }
+    assert!(matches!(
+        handle
+            .delete_from_plan(
+                &WorkerOperationContext::Backend,
+                &operation_cleanup_candidate(&identity)
+            )
+            .await
+            .unwrap_err()
+            .error,
+        Error::UnknownWorker { .. }
+    ));
+    assert!(matches!(
+        handle
+            .connect_protocol(&WorkerOperationContext::Backend)
+            .await,
+        Err(Error::UnknownWorker { .. })
+    ));
+    assert!(
+        api.store
+            .get_worker_registry(TEST_WORKSPACE_ID, &identity)
+            .unwrap()
+            .is_none(),
+        "a stale handle must not recreate its registry row"
+    );
+}
+
+#[tokio::test]
+async fn worker_control_handle_rechecks_revoked_grants_for_each_operation() {
+    let workspace = tempfile::tempdir().unwrap();
+    let api = test_api(workspace.path()).await;
+    let controller = RuntimeWorkerRef::new("offline-runtime", "controller");
+    let subject = RuntimeWorkerRef::new("offline-runtime", "subject");
+    for identity in [&controller, &subject] {
+        seed_worker_source_member(&api, &identity.runtime_id, &identity.worker_id);
+    }
+    api.store
+        .create_worker_control_grant(&WorkerControlGrantRecord {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+            grant_id: "operation-grant".into(),
+            controller: controller.clone(),
+            subject: subject.clone(),
+            relation: "spawned".into(),
+            origin: "test".into(),
+            permissions: [
+                "restore",
+                "stop",
+                "cancel",
+                "send_input",
+                "notify",
+                "pin",
+                "remove",
+            ]
+            .into_iter()
+            .map(str::to_owned)
+            .collect(),
+            operation_id: "operation-grant-test".into(),
+            created_at: now_registry_timestamp(),
+            revoked_at: None,
+        })
+        .unwrap();
+    let handle = WorkspaceWorker::resolve(&api, &subject.runtime_id, &subject.worker_id).unwrap();
+    let context = WorkerOperationContext::WorkerControl { controller };
+    handle.set_pinned(&context, true).await.unwrap();
+    api.store
+        .revoke_worker_control_grant(
+            TEST_WORKSPACE_ID,
+            "operation-grant",
+            &now_registry_timestamp(),
+        )
+        .unwrap();
+
+    for permission in [
+        "restore",
+        "stop",
+        "cancel",
+        "send_input",
+        "notify",
+        "pin",
+        "remove",
+    ] {
+        assert!(
+            matches!(
+                handle.authorize_operation(&context, permission).await,
+                Err(ApiError {
+                    error: Error::UnknownWorker { .. },
+                    ..
+                })
+            ),
+            "revoked grant authorized {permission}"
+        );
+    }
+    assert!(matches!(
+        handle.restore(&context).await.unwrap_err().error,
+        Error::UnknownWorker { .. }
+    ));
+    assert!(matches!(
+        handle
+            .stop(&context, operation_lifecycle())
+            .await
+            .unwrap_err()
+            .error,
+        Error::UnknownWorker { .. }
+    ));
+    assert!(matches!(
+        handle
+            .cancel(&context, operation_lifecycle())
+            .await
+            .unwrap_err()
+            .error,
+        Error::UnknownWorker { .. }
+    ));
+    for kind in [WorkerInputKind::User, WorkerInputKind::Notify] {
+        assert!(matches!(
+            handle
+                .input(&context, operation_input(kind))
+                .await
+                .unwrap_err()
+                .error,
+            Error::UnknownWorker { .. }
+        ));
+    }
+    assert!(matches!(
+        handle.set_pinned(&context, false).await.unwrap_err().error,
+        Error::UnknownWorker { .. }
+    ));
+    assert!(matches!(
+        handle
+            .delete_from_plan(&context, &operation_cleanup_candidate(&subject))
+            .await
+            .unwrap_err()
+            .error,
+        Error::UnknownWorker { .. }
+    ));
+    assert_eq!(
+        api.store
+            .get_worker_registry(TEST_WORKSPACE_ID, &subject)
+            .unwrap()
+            .unwrap()
+            .retention_state,
+        "pinned"
+    );
+}
+
+#[tokio::test]
+async fn generic_and_scoped_input_aliases_match_handle_idle_only_and_notify_semantics() {
+    let fixture = guarded_spawn_fixture().await;
+    let response = post_guarded_spawn(
+        &fixture,
+        guarded_spawn_payload(&fixture, "operation-alias-parity"),
+    )
+    .await;
+    assert_eq!(response.status(), StatusCode::OK);
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    let created: BrowserCreateWorkerResponse = serde_json::from_slice(&bytes).unwrap();
+    let api = fixture.api.clone();
+    let runtime = &fixture.runtime;
+    let subject = RuntimeWorkerRef::new(created.runtime_id, created.worker_id);
+    let controller = fixture.controller.clone();
+    let key = api
+        .store
+        .worker_resource_key(TEST_WORKSPACE_ID, &subject)
+        .unwrap()
+        .unwrap();
+    let handle = WorkspaceWorker::resolve(&api, &subject.runtime_id, &key).unwrap();
+    let context = WorkerOperationContext::WorkerControl { controller };
+    let idle_only = handle
+        .input(&context, operation_input(WorkerInputKind::User))
+        .await
+        .unwrap();
+    assert_eq!(idle_only.state, InternalWorkerOperationState::Rejected);
+    let notify = handle
+        .input(&context, operation_input(WorkerInputKind::Notify))
+        .await
+        .unwrap();
+    assert_eq!(notify.state, InternalWorkerOperationState::Accepted);
+
+    let request = || server_api::RuntimeWorkerInputRequest {
+        kind: Some("user".into()),
+        content: "human input".into(),
+        segments: None,
+    };
+    let Json(generic) = send_runtime_worker_input(
+        State(api.clone()),
+        AxumPath((subject.runtime_id.clone(), key.clone())),
+        Json(request()),
+    )
+    .await
+    .unwrap();
+    let Json(scoped) = scoped_send_runtime_worker_input(
+        State(api.clone()),
+        AxumPath(ScopedRuntimeWorkerPath {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+            worker: RuntimeWorkerRef::new(&subject.runtime_id, &key),
+        }),
+        Json(request()),
+    )
+    .await
+    .unwrap();
+    assert_eq!(generic.state, scoped.state);
+    assert_eq!(generic.state, server_api::WorkerOperationState::Accepted);
+    assert_eq!(
+        runtime
+            .input_requests
+            .lock()
+            .unwrap()
+            .iter()
+            .map(|request| request.kind.clone())
+            .collect::<Vec<_>>(),
+        vec![
+            WorkerInputKind::UserIfIdle,
+            WorkerInputKind::Notify,
+            WorkerInputKind::User,
+            WorkerInputKind::User
+        ]
+    );
+}
+
+#[tokio::test]
+async fn browser_protocol_sender_rejects_job_input_and_all_live_methods() {
+    let workspace = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(workspace.path()).await;
+    let request = BackendJobRequest {
+        job_id: "worker-operation-read-only-job".into(),
+        purpose: "operation_boundary_test".into(),
+        input_revision: "revision-1".into(),
+        input_ref: "test://worker-operation/input".into(),
+        input: serde_json::json!({"immutable": true}),
+        instruction: "Return a structured result.".into(),
+        profile: "builtin:backend-job".into(),
+        source_worker: None,
+        notification_target: None,
+        limits: crate::backend_job::BackendJobLimits::default(),
+    };
+    let dispatched = api.dispatch_backend_job(&request).unwrap();
+    let identity = dispatched.attempt.worker.unwrap();
+    assert_eq!(execution.take_inputs().len(), 1);
+    let handle = WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap();
+    let (context, _) = browser_operation_context(&api).await;
+    let mut connection = handle.connect_protocol(&context).await.unwrap();
+    assert!(matches!(
+        tokio::time::timeout(std::time::Duration::from_secs(2), connection.events.recv())
+            .await
+            .unwrap()
+            .unwrap(),
+        protocol::Event::Snapshot { .. }
+    ));
+    for method in browser_operation_methods() {
+        assert!(matches!(connection.sender.send(&context, method).await,
+            Err(Error::InvalidInput(message)) if message.contains("read-only")));
+    }
+    for kind in [WorkerInputKind::User, WorkerInputKind::Notify] {
+        assert!(
+            matches!(handle.input(&context, operation_input(kind)).await.unwrap_err().error,
+            Error::InvalidInput(message) if message.contains("read-only"))
+        );
+    }
+    assert!(execution.protocol_methods().is_empty());
+    assert!(execution.take_inputs().is_empty());
+}
+
+#[tokio::test]
+async fn connected_browser_sender_rechecks_revoked_credentials_before_each_method() {
+    let workspace = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(workspace.path()).await;
+    let identity = spawn_ticket_check_source(&api, "operation-browser-revocation");
+    let handle = WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap();
+    let (context, token_hash) = browser_operation_context(&api).await;
+    let mut connection = handle.connect_protocol(&context).await.unwrap();
+    assert!(matches!(
+        tokio::time::timeout(std::time::Duration::from_secs(2), connection.events.recv())
+            .await
+            .unwrap()
+            .unwrap(),
+        protocol::Event::Snapshot { .. }
+    ));
+    // Prove these credentials were usable before revocation, then keep the same sender.
+    handle.set_pinned(&context, true).await.unwrap();
+    assert!(
+        api.store
+            .revoke_browser_session(&token_hash, &now_registry_timestamp())
+            .unwrap()
+    );
+    assert!(matches!(
+        handle.connect_protocol(&context).await,
+        Err(Error::WorkspacePermissionDenied(_))
+    ));
+    for method in browser_operation_methods() {
+        assert!(matches!(
+            connection.sender.send(&context, method).await,
+            Err(Error::WorkspacePermissionDenied(_))
+        ));
+    }
+    assert!(matches!(
+        handle.set_pinned(&context, false).await.unwrap_err().error,
+        Error::WorkspacePermissionDenied(_)
+    ));
+    assert!(
+        execution.protocol_methods().is_empty(),
+        "revoked credentials must not reach the execution channel"
+    );
+    assert_eq!(
+        api.store
+            .get_worker_registry(TEST_WORKSPACE_ID, &identity)
+            .unwrap()
+            .unwrap()
+            .retention_state,
+        "pinned"
+    );
+}
+
+fn operation_request_context(
+    source: Option<server_api::ServerRuntimeSource>,
+) -> server_api::ServerRequestContext {
+    server_api::ServerRequestContext {
+        actor: None,
+        worker_source: None,
+        runtime_source: source,
+        origin: None,
+        transport_headers: Vec::new(),
+    }
+}
+
+#[tokio::test]
+async fn request_context_uses_verified_subject_not_headers_or_account_fallback() {
+    let workspace = tempfile::tempdir().unwrap();
+    let api = test_api(workspace.path()).await;
+    let mut headers_only = operation_request_context(None);
+    headers_only.transport_headers = vec![
+        ("x-yoi-runtime-id".into(), b"runtime-a".to_vec()),
+        ("x-yoi-worker-id".into(), b"controller-a".to_vec()),
+    ];
+    assert!(WorkerOperationContext::from_request(&api, &headers_only).is_err());
+    let mut runtime_only = operation_request_context(Some(server_api::ServerRuntimeSource {
+        runtime_id: "runtime-a".into(),
+        worker_id: None,
+    }));
+    runtime_only.actor = Some(test_browser_request_actor());
+    assert!(WorkerOperationContext::from_request(&api, &runtime_only).is_err());
+    let mut mismatched = operation_request_context(Some(server_api::ServerRuntimeSource {
+        runtime_id: "runtime-a".into(),
+        worker_id: Some("controller-a".into()),
+    }));
+    mismatched.actor = Some(test_browser_request_actor());
+    mismatched.transport_headers = vec![("x-yoi-worker-id".into(), b"controller-b".to_vec())];
+    assert!(WorkerOperationContext::from_request(&api, &mismatched).is_err());
+    seed_worker_source_member(&api, "runtime-a", "controller-a");
+    let context = operation_request_context(Some(server_api::ServerRuntimeSource {
+        runtime_id: "runtime-a".into(),
+        worker_id: Some("controller-a".into()),
+    }));
+    assert!(
+        matches!(WorkerOperationContext::from_request(&api, &context).unwrap(),
+        WorkerOperationContext::WorkerControl { controller } if controller == RuntimeWorkerRef::new("runtime-a", "controller-a"))
+    );
+}
+
+#[tokio::test]
+async fn generic_management_aliases_cannot_bypass_worker_grants_or_target_bound_remove() {
+    use server_api::ServerApi;
+    let workspace = tempfile::tempdir().unwrap();
+    let api = test_api(workspace.path()).await;
+    seed_worker_source_member(&api, "offline-runtime", "controller");
+    seed_worker_source_member(&api, "offline-runtime", "subject");
+    let service = ServerApiContractService::Workspace(api.clone());
+    let context = || {
+        operation_request_context(Some(server_api::ServerRuntimeSource {
+            runtime_id: "offline-runtime".into(),
+            worker_id: Some("controller".into()),
+        }))
+    };
+    let input = || server_api::RuntimeWorkerInputRequest {
+        kind: Some("user".into()),
+        content: "must not execute".into(),
+        segments: None,
+    };
+    assert!(
+        service
+            .runtime_worker_input_alias(
+                context(),
+                "offline-runtime".into(),
+                "subject".into(),
+                input()
+            )
+            .await
+            .is_err()
+    );
+    assert!(
+        service
+            .runtime_worker_input(
+                context(),
+                TEST_WORKSPACE_ID.into(),
+                "offline-runtime".into(),
+                "subject".into(),
+                input()
+            )
+            .await
+            .is_err()
+    );
+    assert!(
+        service
+            .runtime_worker_restore_alias(context(), "offline-runtime".into(), "subject".into())
+            .await
+            .is_err()
+    );
+    let lifecycle = || server_api::RuntimeWorkerLifecycleRequest {
+        reason: None,
+        ticket_assignment: None,
+    };
+    assert!(
+        service
+            .runtime_worker_stop_alias(
+                context(),
+                "offline-runtime".into(),
+                "subject".into(),
+                lifecycle()
+            )
+            .await
+            .is_err()
+    );
+    assert!(
+        service
+            .runtime_worker_cancel_alias(
+                context(),
+                "offline-runtime".into(),
+                "subject".into(),
+                lifecycle()
+            )
+            .await
+            .is_err()
+    );
+    assert!(
+        service
+            .runtime_worker_pin(
+                context(),
+                TEST_WORKSPACE_ID.into(),
+                "offline-runtime".into(),
+                "subject".into()
+            )
+            .await
+            .is_err()
+    );
+    let cleanup = service
+        .runtime_cleanup_execute(
+            context(),
+            TEST_WORKSPACE_ID.into(),
+            "offline-runtime".into(),
+            ExecuteRuntimeCleanupRequest {
+                expected_plan_revision: "irrelevant".into(),
+                expected_plan_digest: "irrelevant".into(),
+                worker_target_ids: Vec::new(),
+                workdir_target_ids: Vec::new(),
+                confirm_dirty_discard_target_ids: Vec::new(),
+            },
+        )
+        .await
+        .unwrap_err();
+    assert!(cleanup.message.contains("target-bound WorkerRemove"));
+    assert_eq!(
+        api.store
+            .get_worker_registry(
+                TEST_WORKSPACE_ID,
+                &RuntimeWorkerRef::new("offline-runtime", "subject")
+            )
+            .unwrap()
+            .unwrap()
+            .retention_state,
+        "normal"
+    );
+}
+
+#[test]
+fn uncertain_unary_input_keeps_a_distinct_diagnostic_without_claiming_acceptance() {
+    let result = worker_input_result_to_api(WorkerInputResult {
+        state: InternalWorkerOperationState::Rejected,
+        disposition: WorkerInputDisposition::Unknown,
+        worker: RuntimeWorkerRef::new("runtime-a", "worker-a"),
+        runtime_run_id: None,
+        diagnostics: Vec::new(),
+    });
+    assert_eq!(result.state, server_api::WorkerOperationState::Rejected);
+    assert!(
+        result
+            .diagnostics
+            .iter()
+            .any(|diagnostic| diagnostic.code == "worker_input_outcome_unknown")
+    );
+}
+
+#[tokio::test]
+async fn grant_replacement_while_admission_waits_cannot_use_another_grants_lock() {
+    let workspace = tempfile::tempdir().unwrap();
+    let api = test_api(workspace.path()).await;
+    let controller = RuntimeWorkerRef::new("offline-runtime", "controller");
+    let subject = RuntimeWorkerRef::new("offline-runtime", "subject");
+    for identity in [&controller, &subject] {
+        seed_worker_source_member(&api, &identity.runtime_id, &identity.worker_id);
+    }
+    let grant = |id: &str| WorkerControlGrantRecord {
+        workspace_id: TEST_WORKSPACE_ID.into(),
+        grant_id: id.into(),
+        controller: controller.clone(),
+        subject: subject.clone(),
+        relation: "spawned".into(),
+        origin: "test".into(),
+        permissions: vec!["pin".into()],
+        operation_id: id.into(),
+        created_at: now_registry_timestamp(),
+        revoked_at: None,
+    };
+    api.store
+        .create_worker_control_grant(&grant("original-grant"))
+        .unwrap();
+    let guard = worker_control_lock(&api, "original-grant")
+        .lock_owned()
+        .await;
+    let handle = WorkspaceWorker::resolve(&api, &subject.runtime_id, &subject.worker_id).unwrap();
+    let context = WorkerOperationContext::WorkerControl {
+        controller: controller.clone(),
+    };
+    let admission = handle.set_pinned(&context, true);
+    tokio::pin!(admission);
+    // Poll deterministically to the held lock before replacing grant authority.
+    assert!(futures::poll!(admission.as_mut()).is_pending());
+    api.store
+        .revoke_worker_control_grant(
+            TEST_WORKSPACE_ID,
+            "original-grant",
+            &now_registry_timestamp(),
+        )
+        .unwrap();
+    api.store
+        .create_worker_control_grant(&grant("replacement-grant"))
+        .unwrap();
+    drop(guard);
+    assert!(matches!(
+        admission.await.unwrap_err().error,
+        Error::RepositoryConflict(_)
+    ));
+    assert_eq!(
+        api.store
+            .get_worker_registry(TEST_WORKSPACE_ID, &subject)
+            .unwrap()
+            .unwrap()
+            .retention_state,
+        "normal"
+    );
+}
+
+#[tokio::test]
+async fn browser_protocol_connection_and_cleanup_reject_cookie_origin_mismatch() {
+    let workspace = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(workspace.path()).await;
+    let identity = spawn_ticket_check_source(&api, "operation-origin");
+    let handle = WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap();
+    let (mut context, _) = browser_operation_context(&api).await;
+    let WorkerOperationContext::Browser { headers, .. } = &mut context else {
+        unreachable!()
+    };
+    headers.insert(ORIGIN, "https://untrusted.example".parse().unwrap());
+    assert!(matches!(
+        handle.connect_protocol(&context).await,
+        Err(Error::WorkspacePermissionDenied(_))
+    ));
+    let result = execute_runtime_cleanup_with_context(
+        &api,
+        EMBEDDED_WORKER_RUNTIME_ID,
+        ExecuteRuntimeCleanupRequest {
+            expected_plan_revision: "irrelevant".into(),
+            expected_plan_digest: "irrelevant".into(),
+            worker_target_ids: Vec::new(),
+            workdir_target_ids: Vec::new(),
+            confirm_dirty_discard_target_ids: Vec::new(),
+        },
+        &context,
+    )
+    .await
+    .unwrap_err();
+    assert!(matches!(result.error, Error::WorkspacePermissionDenied(_)));
+    assert!(execution.protocol_methods().is_empty());
+}

crates/workspace-server/src/server/worker_operations.rs

@@ -0,0 +1,591 @@
+//! Workspace identity-bound operation reception. Runtime/controller remain execution authority.
+use super::*;
+
+/// Caller authority supplied by authenticated Server adapters, never public JSON/identity hints.
+/// The unchecked Backend variant exists only for lower-level regression fixtures.
+#[derive(Clone)]
+pub(crate) enum WorkerOperationContext {
+    #[cfg(test)]
+    Backend,
+    WorkerControl {
+        controller: RuntimeWorkerRef,
+    },
+    Browser {
+        headers: HeaderMap,
+        source: protocol::AuthenticatedInputSource,
+    },
+}
+
+impl WorkerOperationContext {
+    pub(super) async fn authorize_workspace(&self, api: &WorkspaceApi) -> ApiResult<()> {
+        if let Self::Browser { headers, source } = self {
+            let actor = resolve_actor(&ServerAuthApi::from(api), headers)
+                .await?
+                .ok_or_else(|| {
+                    Error::WorkspacePermissionDenied(
+                        "Worker connection credentials were revoked".into(),
+                    )
+                })?;
+            if matches!(actor.auth_method, ActorAuthMethod::BrowserSession) {
+                let AuthConfig::Passkey { origin, .. } = &api.config.auth;
+                if headers.get(ORIGIN).and_then(|value| value.to_str().ok())
+                    != Some(origin.as_str())
+                {
+                    return Err(Error::WorkspacePermissionDenied(
+                        "cross-origin Worker mutation denied".into(),
+                    )
+                    .into());
+                }
+            }
+            if authenticated_browser_input_source(&actor) != *source {
+                return Err(Error::WorkspacePermissionDenied(
+                    "Worker connection Account changed".into(),
+                )
+                .into());
+            }
+            if !api
+                .store
+                .get_workspace(api.workspace_id())
+                .await?
+                .is_some_and(|workspace| workspace.state == "active")
+            {
+                return Err(Error::WorkspacePermissionDenied(
+                    "Workspace is not accepting Worker mutations".into(),
+                )
+                .into());
+            }
+        }
+        Ok(())
+    }
+    pub(super) fn from_request(
+        api: &WorkspaceApi,
+        context: &server_api::ServerRequestContext,
+    ) -> std::result::Result<Self, server_api::RepositoryApiError> {
+        if let Some(source) = context.runtime_source.as_ref() {
+            // Check hints against middleware's verified subject without replaying its proof
+            // or requiring the target Runtime to be online just to represent identity.
+            current_worker_contract_headers(context)?;
+            let controller = RuntimeWorkerRef::new(
+                &source.runtime_id,
+                source
+                    .worker_id
+                    .as_deref()
+                    .expect("source binding required a Worker identity"),
+            );
+            if api
+                .store
+                .get_worker_registry(api.workspace_id(), &controller)
+                .map_err(|error| ApiError::from(error).into_repository_api_error())?
+                .is_none()
+            {
+                return Err(ApiError::from(Error::UnknownWorker { worker: controller })
+                    .into_repository_api_error());
+            }
+            return Ok(Self::WorkerControl { controller });
+        }
+        let actor = context.actor.as_ref().ok_or_else(|| {
+            ApiError::from(Error::WorkspacePermissionDenied(
+                "authenticated Worker operation source required".into(),
+            ))
+            .into_repository_api_error()
+        })?;
+        let headers = contract_request_headers(context)?;
+        Ok(Self::Browser {
+            headers,
+            source: authenticated_browser_input_source(actor),
+        })
+    }
+}
+
+/// A registry identity, not an authorization lease or a snapshot of live execution.
+#[derive(Clone)]
+pub(crate) struct WorkspaceWorker<Services = WorkspaceApi> {
+    api: Services,
+    removal: WorkerRemovalService,
+    identity: RuntimeWorkerRef,
+}
+
+// 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<Services> WorkspaceWorker<Services> {
+    fn resolve_with_services(
+        api: Services,
+        removal: WorkerRemovalService,
+        runtime_id: &str,
+        reference: &str,
+    ) -> Result<Self> {
+        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,
+            removal,
+            identity,
+        })
+    }
+    pub(super) fn identity(&self) -> &RuntimeWorkerRef {
+        &self.identity
+    }
+
+    fn current_record(&self) -> ApiResult<WorkerRegistryRecord> {
+        self.removal
+            .store
+            .get_worker_registry(&self.removal.workspace_id, &self.identity)?
+            .ok_or_else(|| {
+                Error::UnknownWorker {
+                    worker: self.identity.clone(),
+                }
+                .into()
+            })
+    }
+
+    pub(super) async fn remove(
+        &self,
+        source: crate::worker_source::VerifiedWorkerMutationSource,
+        reason: &str,
+    ) -> std::result::Result<worker::WorkspaceResponse, String> {
+        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<worker::WorkspaceResponse, String> {
+        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> {
+        Self::resolve_with_services(
+            api.clone(),
+            WorkerRemovalService::new(api),
+            runtime_id,
+            reference,
+        )
+    }
+
+    pub(super) async fn authorize_operation(
+        &self,
+        context: &WorkerOperationContext,
+        permission: &str,
+    ) -> ApiResult<Option<tokio::sync::OwnedMutexGuard<()>>> {
+        self.current_record()?;
+        match context {
+            #[cfg(test)]
+            WorkerOperationContext::Backend => Ok(None),
+            WorkerOperationContext::Browser { .. } => {
+                context.authorize_workspace(&self.api).await?;
+                Ok(None)
+            }
+            WorkerOperationContext::WorkerControl { controller } => {
+                if self
+                    .api
+                    .store
+                    .get_worker_registry(self.api.workspace_id(), controller)?
+                    .is_none()
+                {
+                    return Err(Error::UnknownWorker {
+                        worker: controller.clone(),
+                    }
+                    .into());
+                }
+                let grant = authorize_known_worker_permission(
+                    &self.api,
+                    self.api.workspace_id(),
+                    controller,
+                    &self.identity,
+                    permission,
+                )?;
+                let guard = worker_control_lock(&self.api, &grant.grant_id)
+                    .lock_owned()
+                    .await;
+                self.current_record()?;
+                let current_grant = authorize_known_worker_permission(
+                    &self.api,
+                    self.api.workspace_id(),
+                    controller,
+                    &self.identity,
+                    permission,
+                )?;
+                if current_grant.grant_id != grant.grant_id {
+                    return Err(Error::RepositoryConflict(
+                        "Worker control grant changed while waiting for admission".into(),
+                    )
+                    .into());
+                }
+                Ok(Some(guard))
+            }
+        }
+    }
+
+    pub(super) async fn restore(
+        &self,
+        context: &WorkerOperationContext,
+    ) -> ApiResult<InternalWorkerRestoreResult> {
+        let _guard = self.authorize_operation(context, "restore").await?;
+        // Preserve singleton, Workdir, credentials, bindings and pending operation reconciliation.
+        self.api.restore_workspace_worker(&self.identity)
+    }
+
+    pub(super) async fn stop(
+        &self,
+        context: &WorkerOperationContext,
+        request: WorkerLifecycleRequest,
+    ) -> ApiResult<WorkerLifecycleResult> {
+        let _guard = self.authorize_operation(context, "stop").await?;
+        let result = self
+            .api
+            .runtime
+            .stop_worker(&self.identity, request)
+            .map_err(|e| e.into_error())?;
+        if result.state != InternalWorkerOperationState::Accepted {
+            return Ok(result);
+        }
+        parse_runtime_worker_id_for_registry(&self.identity.worker_id)?;
+        let session_lock = current_worker_session_lock(&self.api, &self.identity);
+        let _session_guard = session_lock.lock().await;
+        close_current_worker_session_locked(&self.api, &self.identity).await?;
+        if let Some(record) = self
+            .api
+            .store
+            .get_worker_registry(self.api.workspace_id(), &self.identity)?
+        {
+            sync_linked_workdir_after_worker_stop(&self.api, &self.identity.runtime_id, &record)?;
+        }
+        Ok(result)
+    }
+
+    pub(super) async fn cancel(
+        &self,
+        context: &WorkerOperationContext,
+        request: WorkerLifecycleRequest,
+    ) -> ApiResult<WorkerLifecycleResult> {
+        let _guard = self.authorize_operation(context, "cancel").await?;
+        Ok(self
+            .api
+            .runtime
+            .cancel_worker(&self.identity, request)
+            .map_err(|e| e.into_error())?)
+    }
+
+    pub(super) async fn input(
+        &self,
+        context: &WorkerOperationContext,
+        mut request: WorkerInputRequest,
+    ) -> ApiResult<WorkerInputResult> {
+        let permission = if request.kind == WorkerInputKind::Notify {
+            "notify"
+        } else {
+            "send_input"
+        };
+        let _guard = self.authorize_operation(context, permission).await?;
+        self.ensure_interactive()?;
+        if matches!(context, WorkerOperationContext::WorkerControl { .. })
+            && request.kind == WorkerInputKind::User
+        {
+            request.kind = WorkerInputKind::UserIfIdle;
+        }
+        Ok(self
+            .api
+            .runtime
+            .send_input(&self.identity, request)
+            .map_err(|e| e.into_error())?)
+    }
+
+    fn ensure_interactive(&self) -> ApiResult<()> {
+        if self
+            .api
+            .store
+            .worker_registry_projection(self.api.workspace_id(), &self.identity)?
+            .and_then(|projection| projection.job)
+            .is_some()
+        {
+            return Err(Error::InvalidInput("Backend Job Workers accept only their Backend-owned immutable input; Console is read-only".into()).into());
+        }
+        Ok(())
+    }
+
+    pub(super) async fn set_pinned(
+        &self,
+        context: &WorkerOperationContext,
+        pinned: bool,
+    ) -> ApiResult<WorkerRetentionResponse> {
+        let _guard = self.authorize_operation(context, "pin").await?;
+        let retention_state = if pinned { "pinned" } else { "normal" };
+        let changed = self.api.worker_projection.publish_ordered(|| {
+            self.api
+                .store
+                .update_worker_retention(
+                    self.api.workspace_id(),
+                    &self.identity,
+                    retention_state,
+                    now_registry_timestamp().as_str(),
+                )
+                .map(|commit| (!commit.changes.is_empty(), commit))
+        })?;
+        if !changed {
+            return Err(cleanup_api_error(
+                &self.identity.runtime_id,
+                "workspace_worker_retention_unknown_worker",
+                "Worker is not known to the Backend registry",
+            ));
+        }
+        Ok(WorkerRetentionResponse {
+            workspace_id: self.api.workspace_id().to_owned(),
+            runtime_id: self.identity.runtime_id.clone(),
+            worker_id: self.identity.worker_id.clone(),
+            pinned,
+            retention_state: retention_state.into(),
+        })
+    }
+
+    /// The caller owns the full cleanup-plan revision/digest check. The removal service
+    /// rechecks live assignment/retention/use at its existing commit boundary.
+    pub(super) async fn delete_from_plan(
+        &self,
+        context: &WorkerOperationContext,
+        candidate: &CleanupWorkerCandidate,
+    ) -> ApiResult<()> {
+        let _guard = self.authorize_operation(context, "remove").await?;
+        if candidate.runtime_id != self.identity.runtime_id
+            || candidate.runtime_worker_id != self.identity.worker_id
+        {
+            return Err(Error::InvalidInput(
+                "cleanup candidate does not address the resolved Worker".into(),
+            )
+            .into());
+        }
+        WorkerRemovalService::new(&self.api)
+            .execute_cleanup_removal(candidate)
+            .await
+    }
+
+    pub(crate) async fn connect_protocol(
+        &self,
+        context: &WorkerOperationContext,
+    ) -> Result<WorkspaceWorkerConnection> {
+        let _guard = self
+            .authorize_operation(context, "observe")
+            .await
+            .map_err(|error| error.error)?;
+        let source = match context {
+            WorkerOperationContext::Browser { source, .. } => Some(source),
+            _ => None,
+        };
+        let connection =
+            connect_workspace_worker_protocol(&self.api, &self.identity, source).await?;
+        Ok(WorkspaceWorkerConnection {
+            sender: WorkspaceWorkerMethodSender {
+                worker: self.clone(),
+                methods: connection.methods,
+            },
+            events: connection.events,
+        })
+    }
+}
+
+/// An execution connection is bound to exactly the identity which created it. No caller
+/// can pair a different target handle with a subscription's raw sending capability.
+#[derive(Clone)]
+pub(crate) struct WorkspaceWorkerMethodSender {
+    worker: WorkspaceWorker,
+    methods: tokio::sync::mpsc::Sender<protocol::Method>,
+}
+pub(crate) struct WorkspaceWorkerConnection {
+    pub(crate) sender: WorkspaceWorkerMethodSender,
+    pub(crate) events: tokio::sync::mpsc::Receiver<protocol::Event>,
+}
+impl WorkspaceWorkerMethodSender {
+    pub(crate) async fn send(
+        &self,
+        context: &WorkerOperationContext,
+        method: protocol::Method,
+    ) -> Result<()> {
+        match method {
+            protocol::Method::Pause { command } => self.pause(context, command).await,
+            protocol::Method::Resume { command } => self.resume(context, command).await,
+            protocol::Method::Cancel { command } => self.cancel(context, command).await,
+            protocol::Method::Compact { command } => self.compact(context, command).await,
+            protocol::Method::Submit {
+                submission_request_id,
+                input,
+            } => self.submit(context, submission_request_id, input).await,
+            protocol::Method::Notify {
+                notification_request_id,
+                message,
+            } => self.notify(context, notification_request_id, message).await,
+            other => self.dispatch(context, other).await,
+        }
+    }
+    async fn dispatch(
+        &self,
+        context: &WorkerOperationContext,
+        method: protocol::Method,
+    ) -> Result<()> {
+        self.send_checked(context, method)
+            .await
+            .map_err(|error| error.error)
+    }
+    pub(crate) async fn pause(
+        &self,
+        context: &WorkerOperationContext,
+        command: protocol::WorkerCommandEnvelope,
+    ) -> Result<()> {
+        self.dispatch(context, protocol::Method::Pause { command })
+            .await
+    }
+    pub(crate) async fn resume(
+        &self,
+        context: &WorkerOperationContext,
+        command: protocol::WorkerCommandEnvelope,
+    ) -> Result<()> {
+        self.dispatch(context, protocol::Method::Resume { command })
+            .await
+    }
+    pub(crate) async fn cancel(
+        &self,
+        context: &WorkerOperationContext,
+        command: protocol::WorkerCommandEnvelope,
+    ) -> Result<()> {
+        self.dispatch(context, protocol::Method::Cancel { command })
+            .await
+    }
+    pub(crate) async fn compact(
+        &self,
+        context: &WorkerOperationContext,
+        command: protocol::WorkerCommandEnvelope,
+    ) -> Result<()> {
+        self.dispatch(context, protocol::Method::Compact { command })
+            .await
+    }
+    pub(crate) async fn submit(
+        &self,
+        context: &WorkerOperationContext,
+        submission_request_id: String,
+        input: Vec<protocol::Segment>,
+    ) -> Result<()> {
+        self.dispatch(
+            context,
+            protocol::Method::Submit {
+                submission_request_id,
+                input,
+            },
+        )
+        .await
+    }
+    pub(crate) async fn notify(
+        &self,
+        context: &WorkerOperationContext,
+        notification_request_id: String,
+        message: String,
+    ) -> Result<()> {
+        self.dispatch(
+            context,
+            protocol::Method::Notify {
+                notification_request_id,
+                message,
+            },
+        )
+        .await
+    }
+
+    async fn send_checked(
+        &self,
+        context: &WorkerOperationContext,
+        method: protocol::Method,
+    ) -> ApiResult<()> {
+        let _guard = self
+            .worker
+            .authorize_operation(context, "send_input")
+            .await?;
+        self.worker.ensure_interactive()?;
+        let WorkerOperationContext::Browser { source, .. } = context else {
+            return Err(Error::WorkspacePermissionDenied(
+                "Browser Worker protocol context required".into(),
+            )
+            .into());
+        };
+        let method = authorize_browser_worker_method(method, source)
+            .map_err(|message| Error::InvalidInput(message.into()))?;
+        self.methods.send(method).await.map_err(|_| {
+            Error::RuntimeOperationFailed {
+                runtime_id: self.worker.identity.runtime_id.clone(),
+                code: "worker_protocol_closed".into(),
+                message:
+                    "Worker execution connection closed; no implicit Restore or retry was performed"
+                        .into(),
+            }
+            .into()
+        })
+    }
+}

crates/workspace-server/src/workspace_subscription.rs

@@ -12,7 +12,7 @@ use worker_runtime::identity::RuntimeWorkerRef;
 
 use crate::runtime_subscription::RuntimeSubscriptionBroker;
 use crate::server::{
-    WorkspaceApi, authorize_browser_worker_method, connect_workspace_worker_protocol,
+    WorkerOperationContext, WorkspaceApi, WorkspaceWorker, WorkspaceWorkerMethodSender,
 };
 
 const OUTBOUND_CAPACITY: usize = 256;
@@ -20,13 +20,13 @@ const WORKER_PROTOCOL_SNAPSHOT_TIMEOUT: std::time::Duration = std::time::Duratio
 
 struct ActiveSubscription {
     task: tokio::task::JoinHandle<()>,
-    methods: Option<mpsc::Sender<protocol::Method>>,
+    methods: Option<WorkspaceWorkerMethodSender>,
 }
 
 pub(crate) async fn serve_workspace_subscription(
     api: WorkspaceApi,
     socket: WebSocket,
-    input_source: protocol::AuthenticatedInputSource,
+    context: WorkerOperationContext,
 ) {
     let broker = api.runtime_subscription_broker().clone();
     let (mut socket_sender, mut socket_receiver) = socket.split();
@@ -90,16 +90,20 @@ pub(crate) async fn serve_workspace_subscription(
                                 worker_id,
                                 runtime_id: Some(runtime_id),
                             } => {
-                                let worker = RuntimeWorkerRef::new(&runtime_id, worker_id.as_str());
-                                match connect_workspace_worker_protocol(
+                                let connection = match WorkspaceWorker::resolve(
                                     &api,
-                                    &worker,
-                                    Some(&input_source),
-                                )
-                                .await
-                                {
+                                    &runtime_id,
+                                    worker_id.as_str(),
+                                ) {
+                                    Ok(worker) => worker
+                                        .connect_protocol(&context)
+                                        .await
+                                        .map_err(|error| error.to_string()),
+                                    Err(error) => Err(error.to_string()),
+                                };
+                                match connection {
                                     Ok(connection) => {
-                                        let methods = connection.methods.clone();
+                                        let methods = connection.sender.clone();
                                         let task = tokio::spawn(run_worker_protocol(
                                             request_id,
                                             subscription_id.clone(),
@@ -165,12 +169,7 @@ pub(crate) async fn serve_workspace_subscription(
                         else {
                             break;
                         };
-                        let Ok(method) =
-                            authorize_browser_worker_method(message.method, &input_source)
-                        else {
-                            break;
-                        };
-                        if methods.send(method).await.is_err() {
+                        if methods.send(&context, message.method).await.is_err() {
                             break;
                         }
                     }

docs/architecture/workspace-worker-operations.md

@@ -0,0 +1,140 @@
+# Workspace Worker operations
+
+T-702 documents the implemented Workspace operation-reception boundary. It does not move execution authority into the Workspace registry or make all Worker-related routes use one API. The common implementation is [`worker_operations.rs`](../../crates/workspace-server/src/server/worker_operations.rs); production adapters and service orchestration are in [`server.rs`](../../crates/workspace-server/src/server.rs).
+
+## Identity reception is not execution authority
+
+`WorkspaceWorker::resolve` resolves a Workspace registry reference to a `RuntimeWorkerRef` (Runtime ID and Worker ID). Resolution first tries that exact pair, then a Workspace resource reference constrained to the requested Runtime. It returns an identity-bound handle, not a live execution snapshot, authorization lease, or promise that the Worker is running. Each operation rereads the target registry record and authorizes its own context.
+
+The owners have different responsibilities:
+
+- **Workspace registry and domain services:** durable Workspace identity, retention/pinning, Workdir links and sessions, Ticket assignments, credentials/bindings, control grants, restore coordination, and removal plans/catalog publication. Registry projections and observation streams describe Workers; they do not admit controller commands.
+- **`RuntimeRegistry` and Runtime adapters:** select the Runtime owning the resolved pair and delegate execution. Remote and embedded adapters preserve their existing transport/error behavior. Runtime owns lifecycle operation settlement and checks Workspace scope at the scoped embedded execution boundary.
+- **Execution backend and Worker controller/session:** own the actual execution endpoint, command admission, authoritative Idle/Busy state, in-flight run transitions, Submit FIFO and durable receipts, Notify receipts, and compaction. A catalog status is not sufficient evidence for command admission.
+
+The term *controller* in a `WorkerControl` context means the calling Runtime Worker holding a grant over a target. It is not the target's in-process execution controller.
+
+## Production ingress → common method → execution owner
+
+For the HTTP rows below, `R` abbreviates `/api/runtimes/{runtime_id}/workers/{worker_id}` and `W` abbreviates `/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}`. Aliases operate in the server's configured Workspace; scoped routes additionally validate the Workspace path. The declarations live in [`server-api`](../../crates/server-api/src/lib.rs), and the generated contract service implementations in `server.rs` are the production unary adapters.
+
+| Ingress | Common reception / coordination | Execution or persistence owner |
+| --- | --- | --- |
+| `POST R/restore` and `POST W/restore` | `WorkerOperationContext::from_request` → `restore_runtime_worker_with_context` → `WorkspaceWorker::restore`; scoped route first uses `scoped_restore_runtime_worker_with_context` | `WorkspaceApi::restore_workspace_worker` retains the existing singleton, Workdir, credential/binding and pending-operation reconciliation flow; Runtime/backend owns restored execution |
+| `POST R/input` and `POST W/input` | Resolve → convert request → `WorkspaceWorker::input` | `RuntimeRegistry::send_input` → Runtime adapter → execution backend/controller/session |
+| `POST R/stop` and `POST W/stop` | `WorkspaceWorker::stop` | Runtime owns Stop; on an **Accepted** result Workspace closes the current Workdir session and synchronizes linked Workdir state. Rejected/Unsupported results do not perform that finalization |
+| `POST R/cancel` and `POST W/cancel` | `WorkspaceWorker::cancel` | `RuntimeRegistry::cancel_worker` → target execution controller; not registry removal |
+| `PUT W/pin` / `DELETE W/pin` | `WorkspaceWorker::set_pinned(true/false)` | Workspace store retention update, ordered through `worker_projection`; no execution command |
+| `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 |
+
+Both WS adapters call `WorkspaceWorkerMethodSender::send`, which selects the following **named methods** rather than implementing separate command semantics in each transport:
+
+| Public protocol method | Named common sender method | Distinct owner/receipt |
+| --- | --- | --- |
+| `Pause { command }` | `pause` | Controller command admission/acknowledgement; pause is not Stop |
+| `Resume { command }` | `resume` | Controller resumes the paused run; not Restore |
+| `Cancel { command }` | `cancel` | Controller command admission for cancelling the current run; distinct from HTTP lifecycle request/result |
+| `Compact { command }` | `compact` | Controller maintenance/compaction, with command acknowledgement and subsequent lifecycle events |
+| `Submit { submission_request_id, input }` | `submit` | Session durable Submit receipt/FIFO and later run execution |
+| `Notify { notification_request_id, message }` | `notify` | Session durable Notify receipt and advisory delivery, not the Submit FIFO |
+
+All six methods funnel through `dispatch` / `send_checked`. Other browser-allowed protocol methods also use that checked dispatch; the six named methods are not an exhaustive allowlist. For example pending-submission operations, rewind, Shutdown, completions, ListWorkers, RestoreWorker and RegisterPeer are allowed by `authorize_browser_worker_method`. This forwarding does not replace their lower-layer semantics with Workspace registry operations.
+
+`build_inner_router` installs the three production WS routes using typed `websocket_route` registrations and merges the generated Workspace contract router for unary routes. [`workspace_subscription.rs`](../../crates/workspace-server/src/workspace_subscription.rs) handles multiplexing. Its `WorkspaceWorkers` subscription is a separate catalog stream with no method sender; only a runtime-scoped `WorkerProtocol` subscription carries an operational sender.
+
+## Context and authorization restrictions
+
+`WorkerOperationContext` is crate-private. Clients cannot submit this enum to select their authority.
+
+- **Request-derived Worker control:** `from_request` uses the middleware-verified Runtime Worker subject and current Workspace registry membership. The shared source-binding helper rejects Runtime-only proofs and conflicting identity hints; no hint-only request or Account fallback can impersonate a Worker. Known-worker adapters bind their source through the same verified request extension before constructing the operation context.
+- **Account/browser context:** despite the `Browser` name, this also represents account-authenticated unary requests. It retains request headers and the server-derived Account input source. On each common authorization call, the server re-resolves credentials, rejects revoked credentials or an Account change, requires the configured Origin for `BrowserSession` actors, and requires an active Workspace. It does not interpret the operation's permission string as a Worker-control grant check for an Account actor; request middleware remains part of the route's authorization boundary.
+- **Worker-control context:** the target and calling Worker must be registered. The server requires an active grant for this controller/target pair and the operation's permission, locks by grant ID, then rereads the target and rechecks permission and the exact grant ID after acquiring the lock. A replacement grant requires new admission under its own lock. A missing active grant is reported as UnknownWorker; a present grant lacking the permission is InvalidInput. Merely knowing an ID does not grant control.
+- **Test Backend context:** `WorkerOperationContext::Backend` exists only under `#[cfg(test)]`. It bypasses context authorization after the target-record check. There is no production Backend enum variant for a client, Worker Tool, or generic internal caller to select. Production Backend services continue to own their explicit domain flows.
+
+Worker-control grants are Workspace-owned records, not inferred parentage or universal access to every Runtime Worker. The production spawn flow persists a controller/subject grant with `send_input`, `notify`, `cancel`, `stop`, `restore`, `remove`, and `observe`; that default does not include `pin`. Known-worker discovery follows active grants, while Internal children use their separate parent-owned registry. Other grant origins/permission sets must be read from their own authority rather than assumed from this spawn default.
+
+Permissions are `restore`, `stop`, `cancel`, `pin`, and `remove` for those identity operations. HTTP input uses `notify` for Notify and `send_input` for all other accepted kinds. `connect_protocol` checks `observe`; every connected method dispatch checks `send_input`, requires an interactive target, then explicitly requires a **Browser** context. Thus an observe grant is not a mutation grant, and a Worker-control grant does not enable browser WS method forwarding. WS Notify also uses this browser-only `send_input` check, not HTTP input's `notify` grant mapping.
+
+`input` and all connected protocol method sends reject Workers with a Backend Job projection: their immutable input is Backend-owned and their Console is read-only. Connecting for observation is not itself rejected by that interactive check.
+
+For Worker-control user input, `input` converts `User` to `UserIfIdle`; admission belongs to the target controller and never queues behind an active run. Browser Submit may durably queue while Running/Paused. Notify is advisory input, not a queued Submit; it can be incorporated at a between-turn boundary or drive the notification flow while Idle. The known-worker input adapter turns definite non-Accepted input results into errors, but keeps uncertain outcomes as the existing typed result with a `worker_input_outcome_unknown` diagnostic, and supplies the Idle-only guidance for user input.
+
+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.
+
+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<WorkspaceApi>` 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
+
+There is no universal Worker command token or universal success result.
+
+- `RuntimeWorkerRef` identifies the registry target; a Workspace resource reference is an ingress lookup form. Neither identifies a particular run or proves authority.
+- `WorkerCommandEnvelope.command_id` orders controller lifecycle commands. `WorkerCommandAcknowledgement` contains command kind, disposition (`Accepted`, `StaleCommandId`, `Conflict`, `InvalidState`) and authoritative state. These IDs belong to controller command admission, not durable Submit identity or Workspace removal planning.
+- `submission_request_id` is the caller's Submit retry key in an authenticated-source namespace. A durable `SubmissionAccepted` receipt also contains the Worker-allocated `submission_id` and `Started`/`Queued` disposition. `SubmissionRejected` is correlated by request ID. Acceptance is not a successful completed run; `last_finished_submission_request_id` is a teardown fence, not a result-success claim.
+- `notification_request_id` correlates `NotificationAccepted` / `NotificationRejected`. Notify has its own durable receipt and exact-payload retry behavior; it does not allocate a Submit FIFO entry. Browser Submit/Notify are rewritten to `SubmitTracked`/`NotifyTracked` with the server's Account source. Browser-supplied tracked forms, `SubmitIfIdle`, and `WorkerEvent` are rejected.
+- Runtime `WorkerLifecycleOperationId` is a durable lifecycle/reconciliation identity, separate from a controller command ID. Ticket `assignment_operation_id`, cleanup plan revision/digest, and retention plan/operation IDs and fingerprints likewise belong to their own domain checks. A multiplexed `subscription_id` only routes frames to an already-bound sender.
+
+The common sender returns `Result<()>` after authorization and enqueueing on its transport channel. This is **not** a controller acknowledgement, durable receipt, or proof of upstream delivery. Actual outcomes arrive on the event stream. The dedicated WS reports dispatch errors as protocol error events; the multiplexed adapter currently terminates the connection when sending a method fails. It also requires an initial session Snapshot before accepting the WorkerProtocol subscription.
+
+Unary input/lifecycle responses preserve their existing API projections: `WorkerOperationState` is `Accepted`, `Unsupported`, or `Rejected`, with Runtime/Worker IDs and diagnostics. Restore instead has `Accepted`, `Rejected`, `RolledBack`, and `ReconciliationRequired`. Pin returns retention state; cleanup returns per-target actions/statuses; WorkerRemove has its own response/error body. These are not interchangeable acknowledgements.
+
+A transport or store failure can leave an **unknown outcome** after effects have started. Lower-layer input results have `WorkerInputDisposition::{Accepted,Rejected,Unknown}` and an optional `runtime_run_id`, but `worker_input_result_to_api` projects only state, target IDs and diagnostics: the public unary input result does not expose those two fields. Unknown delivery is explicitly preserved as a `worker_input_outcome_unknown` diagnostic; neither HTTP adapter changes it into definite rejection or retries it. Runtime execution results also distinguish Busy/Rejected/Errored/Unsupported; error handling and durable restore/removal recovery remain lower-layer responsibilities. Do not infer that a rejected-looking projected response or a closed WS proves that nothing was accepted. The new reception layer does not add automatic replay or an exactly-once guarantee across every ingress. In particular, `worker_input_request_from_api` sets `submission_request_id` to `None`; the unary Workspace input contract does not provide the browser WS Submit retry-key contract.
+
+## Connected execution lifetime
+
+`WorkspaceWorkerMethodSender` stores both the resolved Worker handle and the sending channel privately. A WS caller cannot pair a different Worker handle with a subscription's raw sending capability. Registry identity can outlive an execution, but an operational connection does not follow it to a replacement execution.
+
+`connect_workspace_worker_protocol` obtains an observation source from the Workspace proxy (falling back to the Runtime only for WorkerNotFound). It then connects to one remote execution WS or attaches one embedded `WorkerProtocolTransport`. The embedded path captures the real endpoint and its initial snapshot, not the WorkerRef-wide catalog observation bus.
+
+Embedded dispatch uses `send_connected_protocol_method_scoped`. Runtime verifies that the transport target equals the authorized WorkerRef, checks scope and pending-operation settlement, and validates the captured endpoint before any durable mutation. Even a delayed terminal-close observation cannot make the old endpoint Stop or send to a new execution under the same identity.
+
+Shutdown and a closed execution relay terminate the operational stream. Embedded lifecycle/store/transport uncertainty is terminal; a settled Busy/Rejected/Unsupported rejection may leave it usable only if the endpoint still validates. Remote bridge failure closes its channels. `send_checked` reports `worker_protocol_closed` when its channel is closed and explicitly states that no implicit Restore or retry occurred. No bridge in this path retargets, restores, or reconnects to the next execution. The caller must explicitly Restore when authorized and establish a new connection; this is different from catalog observation continuing to describe the same registry identity.
+
+## Internal Workers are parent-owned
+
+Internal SubWorkers and service Internal Workers are not Workspace registry targets or independently addressable Runtime protocol subjects. They remain separate, parent-owned sessions. Their visible identity is `InternalWorkerRef.session_id` (`name` is display-only), and snapshots/events are presented through the parent's protocol stream; service-private Internal Workers are excluded from the public child inventory.
+
+Direct-child control/observation stays in the parent-owned Worker machinery. A Workspace Runtime Worker grant is not a grant to an arbitrary Internal Worker, and this registry handle does not create an Internal Worker Runtime record, independent transport, or implicit Restore path. The Workspace observation capture adapter explicitly rejects a non-RuntimeWorker subject; it does not resolve a SubWorker as a registry Worker.
+
+## Regression helpers are not production routes
+
+Lower-level test functions in `server.rs` must not be mistaken for installed unauthenticated endpoints:
+
+- `set_worker_retention`, `send_runtime_worker_input` / `dispatch_runtime_worker_input`, and `execute_runtime_cleanup` / `scoped_execute_runtime_cleanup` are `#[cfg(test)]` wrappers using the test-only Backend context.
+- `worker_protocol_ws_session`, `remote_worker_protocol_ws_session`, and `embedded_worker_protocol_ws_session` are `#[cfg(test)]` transport regression helpers. Production dedicated WS uses `workspace_worker_protocol_session` with the common bound sender; production multiplexed WS uses `serve_workspace_subscription` with that same sender type.
+- Production `connect_workspace_worker_protocol`, `connect_remote_worker_protocol`, and `connect_embedded_worker_protocol` are real shared connection helpers, not those test-only sessions. The grant-checked observation capture service also calls the connection helper to read a committed Snapshot; that read-only capture is outside the six operational WS methods.
+
+Source presence alone is not route wiring. Follow `build_inner_router`, generated contract implementations, and the typed route declarations when auditing production ingress.
+
+## Minimum scope, limitations, and future guard
+
+The boundary above covers the T-702 minimum operation set. It is not a claim that every session/file/completion API has been migrated:
+
+- Session snapshot/history and retained session attachment reads remain their existing observation/retention paths. They are not controller command admission.
+- Attachment upload grants and upload cancellation keep their Account/target-bound grant checks; file upload and uploaded-file deletion still resolve the target and delegate through `api.runtime`, rather than calling a new `WorkspaceWorker` file method.
+- Unary completions still resolve the registry reference and call `api.runtime.worker_completions`. Protocol `ListCompletions` is also browser-allowed through checked WS dispatch. Those are different ingress paths.
+- Worker creation, Backend Job execution, rollback compensation, Workspace deletion, and other domain-owned orchestration are not universally rewritten to this handle. Legitimate lower-level Runtime calls still exist.
+
+Known limitations worth follow-up, rather than undocumented guarantees:
+
+1. Browser credentials/Account are rechecked for each common operation or WS method send, but not continuously for every outgoing observation event. This is not a promise of immediate revocation-driven stream teardown.
+2. Cleanup is sequential rather than an all-target atomic transaction. The existing cleanup plan revision/digest and each removal service's live revalidation remain distinct protections.
+3. Unary input has its existing narrower wire contract, rather than the WS authenticated receipt envelope. Identity unification does not create a new retry contract or manufacture caller request IDs across transports.
+
+**T-701's stale-observation Restore guard is not implemented here.** T-701 will connect the caller's observed revision/generation to Runtime's existing operation lock, distinguish a new Restore intent from pending reconciliation, and handle UI conflicts. This boundary leaves those Restore inputs/results and the pending-operation state machine intact rather than creating a duplicate guard.
+
+### Implementation references
+
+- [`server/worker_operations.rs`](../../crates/workspace-server/src/server/worker_operations.rs): identity handle, contexts, authorization, named methods and private bound sender.
+- [`server.rs`](../../crates/workspace-server/src/server.rs): contract adapters, `build_inner_router`, scoped Restore/cleanup, WorkerRemovalService, dedicated WS and shared transport bridges.
+- [`workspace_subscription.rs`](../../crates/workspace-server/src/workspace_subscription.rs): multiplexed WorkerProtocol binding and catalog subscriptions.
+- [`hosts.rs`](../../crates/workspace-server/src/hosts.rs): Runtime registry/adapters and distinct input/lifecycle result projections.
+- [`protocol`](../../crates/protocol/src/lib.rs): command envelopes/acknowledgements, Submit/Notify receipts and Internal Worker presentation.
+- [`worker-runtime/runtime.rs`](../../crates/worker-runtime/src/runtime.rs) and [`execution.rs`](../../crates/worker-runtime/src/execution.rs): scoped connected endpoint validation, lifecycle identities and execution results.

docs/report/2026-10-06-t702-command-output-readback.md

@@ -0,0 +1,7 @@
+# Command-output readback during T-702
+
+During a cold `cargo test -p yoi-workspace-server` run, Bash bounded its inline output and reported a spill file under the Worker's declared readable `bash-output` boundary. A subsequent `Grep` of that exact returned absolute path was rejected as outside allowed scope. I did not attempt another path to bypass the rejection; instead I reran the relevant tests with bounded output filtering.
+
+This made a timed-out crate test's failure details unavailable through the advertised readback path. Suggested improvement: issue a readback capability/reference for Bash spills usable by Read/Grep, or ensure the Workdir attachment routes the declared readable spill directory consistently. Include the permitted tool/attachment in the spill message if it differs from the repository attachment.
+
+The first whole-crate test also reached the provider's 600-second timeout while an outdated WebSocket fixture awaited a subscription that was rejected by the new operation-time authentication. After providing genuine fixture credentials, the targeted subscription test completed; subsequent whole-crate runs use bounded test concurrency. A timeout was treated as incomplete validation, not a passing suite.

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.