refactor(worker-runtime): route by WorkerRef and reconcile owned execution cleanup

crates/worker-runtime/src/execution.rs

@@ -207,34 +207,6 @@ impl WorkerExecutionResult {
     }
 }
 
-/// Opaque per-Worker execution handle returned by a backend.
-///
-/// The handle is a typed token for routing calls back into the same backend. It
-/// intentionally contains no socket path, process id, credential, manifest path,
-/// or session path.
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct WorkerExecutionHandle {
-    worker_ref: WorkerRef,
-    backend_id: String,
-}
-
-impl WorkerExecutionHandle {
-    pub fn new(worker_ref: WorkerRef, backend_id: impl Into<String>) -> Self {
-        Self {
-            worker_ref,
-            backend_id: backend_id.into(),
-        }
-    }
-
-    pub fn worker_ref(&self) -> &WorkerRef {
-        &self.worker_ref
-    }
-
-    pub fn backend_id(&self) -> &str {
-        &self.backend_id
-    }
-}
-
 /// Runtime hooks available to an execution backend for one Worker.
 #[cfg(feature = "ws-server")]
 type WorkerObservationPublisher = Arc<
@@ -348,7 +320,17 @@ impl WorkerExecutionContext {
             let Some(event) = event else {
                 return Ok(());
             };
-            (self.observation_publisher)(self.worker_ref.clone(), event)?;
+            if let Err(error) = (self.observation_publisher)(self.worker_ref.clone(), event.clone())
+            {
+                let mut state = self
+                    .publication_state
+                    .lock()
+                    .map_err(|_| RuntimeError::StatePoisoned)?;
+                if let WorkerExecutionPublicationState::Activating(events) = &mut *state {
+                    events.push_front(event);
+                }
+                return Err(error);
+            }
         }
     }
 
@@ -443,7 +425,6 @@ pub struct WorkerExecutionRestoreRequest {
 pub struct WorkerExecutionStopRequest {
     pub operation_id: WorkerLifecycleOperationId,
     pub worker_ref: WorkerRef,
-    pub handle: Option<WorkerExecutionHandle>,
 }
 
 /// Runtime-side request to refresh the latest Workspace Config before Worker creation.
@@ -466,7 +447,6 @@ pub enum WorkspaceConfigFetchResult {
 #[derive(Clone, Debug)]
 pub enum WorkerExecutionSpawnResult {
     Connected {
-        handle: WorkerExecutionHandle,
         worker_state: protocol::WorkerStateSnapshot,
         workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
     },
@@ -475,11 +455,10 @@ pub enum WorkerExecutionSpawnResult {
     /// Live work started, failed, and all operation-owned resources were joined/released.
     RolledBack(WorkerExecutionResult),
     /// The operation crossed the live side-effect boundary and could not prove
-    /// either commit or complete cleanup. Optional handle/state fields preserve
+    /// either commit or complete cleanup. Optional state fields preserve
     /// concrete execution evidence for retry reconciliation.
     ReconciliationRequired {
         result: WorkerExecutionResult,
-        handle: Option<WorkerExecutionHandle>,
         worker_state: Option<protocol::WorkerStateSnapshot>,
         workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
     },
@@ -490,12 +469,10 @@ pub enum WorkerExecutionSpawnResult {
 
 impl WorkerExecutionSpawnResult {
     pub fn connected(
-        handle: WorkerExecutionHandle,
         worker_state: protocol::WorkerStateSnapshot,
         workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
     ) -> Self {
         Self::Connected {
-            handle,
             worker_state,
             workdir_attachments,
         }
@@ -521,6 +498,57 @@ pub struct WorkerSessionAttachmentRequest {
     pub attachment_id: String,
 }
 
+/// One real protocol connection, bound to the execution that was attached.
+/// This is not a routing token and is never persisted as Worker authority.
+#[cfg(feature = "ws-server")]
+pub struct WorkerProtocolTransport {
+    pub(crate) worker_ref: WorkerRef,
+    pub snapshot: protocol::Event,
+    pub events: tokio::sync::mpsc::Receiver<protocol::Event>,
+    dispatch:
+        Arc<dyn Fn(Method) -> Result<Vec<protocol::Event>, WorkerExecutionResult> + Send + Sync>,
+    validate: Arc<dyn Fn() -> Result<(), WorkerExecutionResult> + Send + Sync>,
+}
+
+#[cfg(feature = "ws-server")]
+impl WorkerProtocolTransport {
+    pub fn new(
+        worker_ref: WorkerRef,
+        snapshot: protocol::Event,
+        events: tokio::sync::mpsc::Receiver<protocol::Event>,
+        dispatch: Arc<
+            dyn Fn(Method) -> Result<Vec<protocol::Event>, WorkerExecutionResult> + Send + Sync,
+        >,
+        validate: Arc<dyn Fn() -> Result<(), WorkerExecutionResult> + Send + Sync>,
+    ) -> Self {
+        Self {
+            worker_ref,
+            snapshot,
+            events,
+            dispatch,
+            validate,
+        }
+    }
+
+    pub fn dispatch(&self, method: Method) -> Result<Vec<protocol::Event>, WorkerExecutionResult> {
+        (self.dispatch)(method)
+    }
+
+    pub fn validate(&self) -> Result<(), WorkerExecutionResult> {
+        (self.validate)()
+    }
+}
+
+#[cfg(feature = "ws-server")]
+impl fmt::Debug for WorkerProtocolTransport {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("WorkerProtocolTransport")
+            .finish_non_exhaustive()
+    }
+}
+
+/// The connection selects this backend; WorkerRef identifies an execution,
+/// never a capability or authorization token.
 pub trait WorkerExecutionBackend: Send + Sync + 'static {
     fn backend_id(&self) -> &str;
 
@@ -580,8 +608,9 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
     fn activate_restored_worker(
         &self,
         _operation_id: WorkerLifecycleOperationId,
-        _handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
     ) -> Result<(), String> {
+        let _ = worker_ref;
         Ok(())
     }
 
@@ -673,20 +702,17 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
         Ok(())
     }
 
-    fn dispatch_input(
+    fn dispatch_input(&self, worker_ref: &WorkerRef, input: WorkerInput) -> WorkerExecutionResult;
-        &self,
-        handle: &WorkerExecutionHandle,
-        input: WorkerInput,
-    ) -> WorkerExecutionResult;
 
     fn upload_file(
         &self,
-        _handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
         _file_name: &str,
         _media_type: &str,
         _content: &[u8],
         _context: Option<&session_store::UploadedFileUploadContext>,
     ) -> Result<UploadedFileRef, WorkerExecutionResult> {
+        let _ = worker_ref;
         Err(WorkerExecutionResult::unsupported(
             WorkerExecutionOperation::UploadFile,
             "execution backend does not support file upload",
@@ -695,20 +721,18 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
 
     fn delete_uploaded_file(
         &self,
-        _handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
         _artifact_id: &str,
     ) -> WorkerExecutionResult {
+        let _ = worker_ref;
         WorkerExecutionResult::unsupported(
             WorkerExecutionOperation::DeleteUploadedFile,
             "execution backend does not support uploaded-file deletion",
         )
     }
 
-    fn dispatch_method(
+    fn dispatch_method(&self, worker_ref: &WorkerRef, _method: Method) -> WorkerExecutionResult {
-        &self,
+        let _ = worker_ref;
-        _handle: &WorkerExecutionHandle,
-        _method: Method,
-    ) -> WorkerExecutionResult {
         WorkerExecutionResult::unsupported(
             WorkerExecutionOperation::ProtocolMethod,
             "execution backend does not support direct Worker protocol methods",
@@ -717,32 +741,29 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
 
     fn worker_completions(
         &self,
-        _handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
         _kind: protocol::CompletionKind,
         _prefix: &str,
         _context: Option<&protocol::CompletionContext>,
     ) -> Vec<protocol::CompletionEntry> {
+        let _ = worker_ref;
         Vec::new()
     }
 
     fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult {
-        let Some(handle) = request.handle.as_ref() else {
+        self.stop_worker(&request.worker_ref)
-            return WorkerExecutionResult::unsupported(
-                WorkerExecutionOperation::Stop,
-                "execution backend cannot reconcile a stopped Worker without a live handle",
-            );
-        };
-        self.stop_worker(handle)
     }
 
-    fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
+    fn stop_worker(&self, worker_ref: &WorkerRef) -> WorkerExecutionResult {
+        let _ = worker_ref;
         WorkerExecutionResult::unsupported(
             WorkerExecutionOperation::Stop,
             "execution backend does not support stopping workers",
         )
     }
 
-    fn cancel_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
+    fn cancel_worker(&self, worker_ref: &WorkerRef) -> WorkerExecutionResult {
+        let _ = worker_ref;
         WorkerExecutionResult::unsupported(
             WorkerExecutionOperation::Cancel,
             "execution backend does not support cancelling workers",
@@ -750,7 +771,20 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
     }
 
     #[cfg(feature = "ws-server")]
-    fn worker_snapshot(&self, _handle: &WorkerExecutionHandle) -> Option<protocol::Event> {
+    fn attach_worker_protocol(
+        self: Arc<Self>,
+        worker_ref: &WorkerRef,
+    ) -> Result<WorkerProtocolTransport, WorkerExecutionResult> {
+        let _ = worker_ref;
+        Err(WorkerExecutionResult::unsupported(
+            WorkerExecutionOperation::ProtocolMethod,
+            "execution backend does not support bound Worker protocol connections",
+        ))
+    }
+
+    #[cfg(feature = "ws-server")]
+    fn worker_snapshot(&self, worker_ref: &WorkerRef) -> Option<protocol::Event> {
+        let _ = worker_ref;
         None
     }
 }
@@ -828,9 +862,10 @@ impl WorkerExecutionBackendRef {
     pub(crate) fn activate_restored_worker(
         &self,
         operation_id: WorkerLifecycleOperationId,
-        handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
     ) -> Result<(), String> {
-        self.backend.activate_restored_worker(operation_id, handle)
+        self.backend
+            .activate_restored_worker(operation_id, worker_ref)
     }
 
     pub(crate) fn restore_worker(
@@ -896,57 +931,62 @@ impl WorkerExecutionBackendRef {
 
     pub(crate) fn dispatch_input(
         &self,
-        handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
         input: WorkerInput,
     ) -> WorkerExecutionResult {
-        self.backend.dispatch_input(handle, input)
+        self.backend.dispatch_input(worker_ref, input)
     }
 
     pub(crate) fn upload_file(
         &self,
-        handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
         file_name: &str,
         media_type: &str,
         content: &[u8],
         context: Option<&session_store::UploadedFileUploadContext>,
     ) -> Result<UploadedFileRef, WorkerExecutionResult> {
         self.backend
-            .upload_file(handle, file_name, media_type, content, context)
+            .upload_file(worker_ref, file_name, media_type, content, context)
     }
 
     pub(crate) fn delete_uploaded_file(
         &self,
-        handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
         artifact_id: &str,
     ) -> WorkerExecutionResult {
-        self.backend.delete_uploaded_file(handle, artifact_id)
+        self.backend.delete_uploaded_file(worker_ref, artifact_id)
     }
 
     pub(crate) fn dispatch_method(
         &self,
-        handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
         method: Method,
     ) -> WorkerExecutionResult {
-        self.backend.dispatch_method(handle, method)
+        self.backend.dispatch_method(worker_ref, method)
     }
 
     #[cfg(feature = "ws-server")]
-    pub(crate) fn worker_snapshot(
+    pub(crate) fn attach_worker_protocol(
         &self,
-        handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
-    ) -> Option<protocol::Event> {
+    ) -> Result<WorkerProtocolTransport, WorkerExecutionResult> {
-        self.backend.worker_snapshot(handle)
+        Arc::clone(&self.backend).attach_worker_protocol(worker_ref)
+    }
+
+    #[cfg(feature = "ws-server")]
+    pub(crate) fn worker_snapshot(&self, worker_ref: &WorkerRef) -> Option<protocol::Event> {
+        self.backend.worker_snapshot(worker_ref)
     }
 
     pub(crate) fn worker_completions(
         &self,
-        handle: &WorkerExecutionHandle,
+        worker_ref: &WorkerRef,
         kind: protocol::CompletionKind,
         prefix: &str,
         context: Option<&protocol::CompletionContext>,
     ) -> Vec<protocol::CompletionEntry> {
         self.backend
-            .worker_completions(handle, kind, prefix, context)
+            .worker_completions(worker_ref, kind, prefix, context)
     }
 
     pub(crate) fn stop_worker_operation(
@@ -956,12 +996,12 @@ impl WorkerExecutionBackendRef {
         self.backend.stop_worker_operation(request)
     }
 
-    pub(crate) fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
+    pub(crate) fn stop_worker(&self, worker_ref: &WorkerRef) -> WorkerExecutionResult {
-        self.backend.stop_worker(handle)
+        self.backend.stop_worker(worker_ref)
     }
 
-    pub(crate) fn cancel_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
+    pub(crate) fn cancel_worker(&self, worker_ref: &WorkerRef) -> WorkerExecutionResult {
-        self.backend.cancel_worker(handle)
+        self.backend.cancel_worker(worker_ref)
     }
 }
 

crates/worker-runtime/src/fs_store.rs

@@ -493,7 +493,12 @@ pub(crate) struct PersistedWorkerRestoreOperation {
 #[serde(deny_unknown_fields)]
 pub(crate) struct PersistedWorkerStopOperation {
     pub(crate) operation_id: WorkerLifecycleOperationId,
-    pub(crate) request: CreateWorkerRequest,
+    /// Restore resources remain operation-owned until stop confirms cleanup.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) pending_restore: Option<PersistedWorkerRestoreOperation>,
+    /// Stop can reconcile an identity whose execution metadata is unavailable.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) request: Option<CreateWorkerRequest>,
     pub(crate) binding: Option<PersistedWorkerExecutionBinding>,
     pub(crate) restore_intent: WorkerRestoreIntent,
     pub(crate) last_settled_status: WorkerStatus,
@@ -1942,7 +1947,7 @@ impl WorkerAggregateRecord {
                             });
                         }
                         (
-                            &restore.request,
+                            Some(&restore.request),
                             &restore.binding,
                             if restore.last_settled_status.is_active() {
                                 WorkerRestoreIntent::Automatic
@@ -1953,7 +1958,7 @@ impl WorkerAggregateRecord {
                         )
                     }
                     PersistedWorkerLifecycleOperation::Stop(stop) => {
-                        if !stop.last_settled_status.is_active() {
+                        if !stop.last_settled_status.is_active() && stop.pending_restore.is_none() {
                             return Err(RuntimeError::StoreCorrupt {
                                 operation: "read Worker stop reconciliation",
                                 path: path.to_path_buf(),
@@ -1961,8 +1966,18 @@ impl WorkerAggregateRecord {
                                     .to_string(),
                             });
                         }
+                        if stop.request.is_none()
+                            && (stop.binding.is_some() || stop.pending_restore.is_some())
+                        {
+                            return Err(RuntimeError::StoreCorrupt {
+                                operation: "read Worker stop reconciliation",
+                                path: path.to_path_buf(),
+                                message: "metadata-free stop cannot carry restore authority"
+                                    .to_string(),
+                            });
+                        }
                         (
-                            &stop.request,
+                            stop.request.as_ref(),
                             &stop.binding,
                             stop.restore_intent,
                             stop.last_settled_status,
@@ -1978,13 +1993,15 @@ impl WorkerAggregateRecord {
                                 .to_string(),
                     });
                 }
-                WorkerExecutionRecord {
+                if let Some(request) = request {
-                    schema_version: SCHEMA_VERSION,
+                    WorkerExecutionRecord {
-                    request: request.clone(),
+                        schema_version: SCHEMA_VERSION,
-                    binding: binding.clone(),
+                        request: request.clone(),
-                    restore_intent,
+                        binding: binding.clone(),
+                        restore_intent,
+                    }
+                    .validate_for_schema(&identity, path, SCHEMA_VERSION)?;
                 }
-                .validate_for_schema(&identity, path, SCHEMA_VERSION)?;
                 PersistedWorkerExecutionState::ReconciliationRequired(operation)
             }
             WorkerExecutionStateRecord::Unavailable => PersistedWorkerExecutionState::Unavailable,

crates/workspace-server/src/hosts.rs

@@ -6869,7 +6869,7 @@ mod tests {
 
         fn dispatch_input(
             &self,
-            _handle: &worker_runtime::execution::WorkerExecutionHandle,
+            _worker_ref: &EmbeddedWorkerRef,
             _input: EmbeddedWorkerInput,
         ) -> worker_runtime::execution::WorkerExecutionResult {
             worker_runtime::execution::WorkerExecutionResult::rejected(
@@ -6898,7 +6898,7 @@ mod tests {
 
         fn dispatch_input(
             &self,
-            _handle: &worker_runtime::execution::WorkerExecutionHandle,
+            _worker_ref: &EmbeddedWorkerRef,
             _input: EmbeddedWorkerInput,
         ) -> worker_runtime::execution::WorkerExecutionResult {
             unreachable!("Repository observation test does not dispatch Worker input")
@@ -6978,10 +6978,6 @@ mod tests {
                 .unwrap()
                 .insert(request.worker_ref.clone(), request.context);
             worker_runtime::execution::WorkerExecutionSpawnResult::Connected {
-                handle: worker_runtime::execution::WorkerExecutionHandle::new(
-                    request.worker_ref,
-                    self.backend_id(),
-                ),
                 worker_state: protocol::WorkerStateSnapshot {
                     ..protocol::WorkerStatus::Idle.into()
                 },
@@ -7000,15 +6996,10 @@ mod tests {
 
         fn dispatch_input(
             &self,
-            handle: &worker_runtime::execution::WorkerExecutionHandle,
+            worker_ref: &EmbeddedWorkerRef,
             input: EmbeddedWorkerInput,
         ) -> worker_runtime::execution::WorkerExecutionResult {
-            let context = self
+            let context = self.contexts.lock().unwrap().get(worker_ref).cloned();
-                .contexts
-                .lock()
-                .unwrap()
-                .get(handle.worker_ref())
-                .cloned();
             let Some(context) = context else {
                 return worker_runtime::execution::WorkerExecutionResult::rejected(
                     worker_runtime::execution::WorkerExecutionOperation::Input,

crates/workspace-server/src/runtime_subscription_tests.rs

@@ -3,10 +3,10 @@ use protocol::subscription::{SubscriptionWorkerIds, SubscriptionWorkerState};
 use worker_runtime::Runtime;
 use worker_runtime::catalog::{CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource};
 use worker_runtime::execution::{
-    WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult,
+    WorkerExecutionBackend, WorkerExecutionOperation, WorkerExecutionResult,
     WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult,
 };
-use worker_runtime::identity::WorkerId;
+use worker_runtime::identity::{WorkerId, WorkerRef};
 
 #[derive(Debug)]
 struct TestExecutionBackend;
@@ -16,9 +16,8 @@ impl WorkerExecutionBackend for TestExecutionBackend {
         "runtime-subscription-test"
     }
 
-    fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
+    fn spawn_worker(&self, _request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
         WorkerExecutionSpawnResult::connected(
-            WorkerExecutionHandle::new(request.worker_ref, self.backend_id()),
             protocol::WorkerStateSnapshot {
                 ..protocol::WorkerStatus::Idle.into()
             },
@@ -28,7 +27,7 @@ impl WorkerExecutionBackend for TestExecutionBackend {
 
     fn dispatch_input(
         &self,
-        _handle: &WorkerExecutionHandle,
+        _worker_ref: &WorkerRef,
         input: worker_runtime::interaction::WorkerInput,
     ) -> WorkerExecutionResult {
         if let Some(submission_request_id) = input.submission_request_id {
@@ -43,7 +42,7 @@ impl WorkerExecutionBackend for TestExecutionBackend {
         }
     }
 
-    fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
+    fn stop_worker(&self, _worker_ref: &WorkerRef) -> WorkerExecutionResult {
         WorkerExecutionResult::accepted(WorkerExecutionOperation::Stop)
     }
 }

docs/report/T-700-subworker-followup-observation.md

@@ -0,0 +1,15 @@
+# SubWorker follow-up後のcommitted sessionを確認できなかった
+
+T-700の実装中、既存のIdleなdirect child `backend`へ`WorkerSendInput`で追加修正を依頼した。toolは成功を返し、その後Idle通知と`WorkerList`のIdle表示を確認した。
+
+一方、`ViewSessionOverview`と`SearchSessionEntries`が返した最後のentryは、follow-up前の保存Manifest fixture修正報告（entry range 744）だった。今回送った追加依頼の文字列の検索は0件で、offset 744からもその旧報告1件しか見えなかった。したがって、通知だけから追加作業の完了は認定できなかった。これは最新turnの実行失敗、captureの遅延、観測面のfreshnessのいずれかを切り分ける証拠ではない。
+
+親は現在のdiffを直接確認し、子をStopしてdelegated scopeを解放した後、残りのShutdown admission修正を新しいdirect childへ限定して委譲した。新しい子では最初の依頼と修正報告をsession observationから確認できた。
+
+改善案:
+
+- Idle通知に、対象turnとcommitted captureの末尾entryの対応が分かる情報を含める。
+- follow-upの開始またはcaptureに失敗した場合、成功した旧turnだけを返す代わりにbounded diagnosticを表示する。
+- observationで最新のcommitted captureが未取得の場合、古いsnapshotであることを明示する。
+
+credential、raw prompt、Backend保存領域の直接読取で観測制限を迂回することは行っていない。