fix: align Worker operation grants and known-target observation

Compose the published T-803 candidate 51f0c7e019 with T-766 without target adoption. Preserve operation-use versus directional control, notify-only replies, live revocation, and T-766 receipt/provenance boundaries.

crates/runtime-api/src/lib.rs

@@ -450,6 +450,9 @@ pub struct CreateWorkerRequest {
     pub workdir_attachment_requests: Vec<WorkingDirectoryAttachmentRequest>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub workdir_attachments: Vec<WorkingDirectoryAttachmentClaim>,
+    /// Enables the optional Runtime-local peer observation provider, not operation
+    /// authority. Scoped Backend observation follows the Worker Feature and live
+    /// operation/control checks, including when this saved flag is false.
     #[serde(default, skip_serializing_if = "std::ops::Not::not")]
     pub worker_observation_enabled: bool,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]

crates/server-api/src/authorization.rs

@@ -42,8 +42,10 @@
 //! terminal confirmation and does not undo completed effects.
 //!
 //! A child receives no Subject body identity, singleton, whole Memory or public
-//! conversation-post authority. `worker.result` is explicit notify authority to a
-//! named origin Worker, not arbitrary control or publication. Chat delivery and
+//! conversation-post authority. `worker.messaging` permits the messaging operation;
+//! the Worker Feature's Backend control grant separately limits its target and direction.
+//! Stored `worker.result` and `worker.notify` names are read compatibility only, not
+//! result publication, arbitrary control or current request names. Chat delivery and
 //! conversation correlation belong to the caller integration, not this contract.
 //! Unset existing Subject policies deny new Subject execution/delegation; ordinary
 //! non-Subject Workers explicitly retain existing Backend/provider contracts.
@@ -115,6 +117,11 @@ pub enum PermissionResource {
         path: String,
         recursive: bool,
     },
+    /// Authorizes operation use, not a target Worker or a Workspace collection.
+    /// The Worker Feature must still enforce its directional target control grant.
+    WorkerOperation {
+        workspace_id: String,
+    },
     Worker {
         runtime_id: String,
         worker_id: String,
@@ -186,6 +193,9 @@ impl<'de> Deserialize<'de> for PermissionResource {
                 path: String,
                 recursive: bool,
             },
+            WorkerOperation {
+                workspace_id: String,
+            },
             Worker {
                 runtime_id: String,
                 worker_id: String,
@@ -285,6 +295,7 @@ impl<'de> Deserialize<'de> for PermissionResource {
                 path,
                 recursive,
             },
+            Wire::WorkerOperation { workspace_id } => Self::WorkerOperation { workspace_id },
             Wire::Worker {
                 runtime_id,
                 worker_id,
@@ -310,6 +321,7 @@ impl PermissionResource {
             | Self::Repository { workspace_id, .. }
             | Self::WorkdirAttachment { workspace_id, .. }
             | Self::WorkspaceConfig { workspace_id, .. }
+            | Self::WorkerOperation { workspace_id }
             | Self::Workspace { workspace_id } => Some(workspace_id),
             _ => None,
         }
@@ -414,7 +426,9 @@ impl PermissionResource {
                 validate_authorization_identifier(name)
             }
             Self::Feature => Ok(()),
-            Self::Workspace { workspace_id } => validate_authorization_identifier(workspace_id),
+            Self::Workspace { workspace_id } | Self::WorkerOperation { workspace_id } => {
+                validate_authorization_identifier(workspace_id)
+            }
             Self::Workdir {
                 working_directory_id,
                 path,
@@ -567,6 +581,10 @@ impl PermissionResource {
             }
             (Self::Feature, Self::Feature) => true,
             (Self::Workspace { workspace_id: a }, Self::Workspace { workspace_id: b }) => a == b,
+            (
+                Self::WorkerOperation { workspace_id: a },
+                Self::WorkerOperation { workspace_id: b },
+            ) => a == b,
             (
                 Self::Worker {
                     runtime_id: ar,
@@ -605,6 +623,14 @@ impl PermissionResource {
     }
 }
 
+/// Stored scopes retain their original resources and exact operation names.
+///
+/// A legacy Worker read adapter may use [`canonical_worker_operation`] to project
+/// operation use, but must retain the original target separately for the Worker
+/// Feature's control-grant check, including direction, revocation and delegation
+/// limits. Replacing an old Worker resource with WorkerOperation and discarding
+/// that target would broaden authority. Neither validation nor `covers` performs
+/// that conversion.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
 #[serde(deny_unknown_fields)]
@@ -620,6 +646,14 @@ impl ScopedPermission {
         if matches!(self.resource, PermissionResource::Feature) && self.operation.contains('.') {
             return Err("Feature resources authorize activation names only, not operations".into());
         }
+        if matches!(self.resource, PermissionResource::WorkerOperation { .. })
+            && canonical_worker_operation(&self.operation) != Some(self.operation.as_str())
+        {
+            return Err(
+                "WorkerOperation resources authorize only exact current Worker control operations"
+                    .into(),
+            );
+        }
         Ok(())
     }
 
@@ -691,6 +725,18 @@ pub struct OperationAuthorizationRequest {
 
 impl OperationAuthorizationRequest {
     pub fn validate(&self) -> Result<(), String> {
+        if let Some(operation) = canonical_worker_operation(&self.operation) {
+            if operation != self.operation {
+                return Err(
+                    "legacy Worker message names are storage-only; request worker.messaging".into(),
+                );
+            }
+            if !matches!(self.resource, PermissionResource::WorkerOperation { .. }) {
+                return Err(
+                    "current Worker control requests require a WorkerOperation resource".into(),
+                );
+            }
+        }
         ScopedPermission {
             operation: self.operation.clone(),
             resource: self.resource.clone(),
@@ -736,6 +782,26 @@ pub struct BoundOperationAuthorizationRequest {
     pub target: BoundAuthorizationTarget,
 }
 
+/// Maps exact runtime Worker control names to the current operation-use contract.
+///
+/// `worker.notify` and `worker.result` are accepted here only to read saved
+/// policy/grants; current requests must use `worker.messaging`. This helper does
+/// not grant authority or convert resources: adapters must preserve legacy Worker
+/// target restrictions separately. Spawn, discovery/list and Internal SubWorker
+/// operations are not part of this contract. Unknown names return `None`.
+pub fn canonical_worker_operation(operation: &str) -> Option<&'static str> {
+    match operation {
+        "worker.messaging" | "worker.notify" | "worker.result" => Some("worker.messaging"),
+        "worker.send_input" => Some("worker.send_input"),
+        "worker.cancel" => Some("worker.cancel"),
+        "worker.stop" => Some("worker.stop"),
+        "worker.restore" => Some("worker.restore"),
+        "worker.remove" => Some("worker.remove"),
+        "worker.observe" => Some("worker.observe"),
+        _ => None,
+    }
+}
+
 pub fn validate_operation_name(value: &str) -> Result<(), String> {
     if value.is_empty()
         || value.len() > 256
@@ -898,6 +964,220 @@ mod tests {
         );
     }
 
+    #[test]
+    fn worker_operation_wire_and_schema_require_only_workspace_identity() {
+        let wire = serde_json::json!({"kind":"worker_operation","workspace_id":"ws"});
+        let resource: PermissionResource = serde_json::from_value(wire.clone()).unwrap();
+        assert_eq!(serde_json::to_value(&resource).unwrap(), wire);
+        assert_eq!(resource.workspace_id(), Some("ws"));
+        assert!(resource.validate().is_ok());
+        for invalid in [
+            serde_json::json!({"kind":"worker_operation"}),
+            serde_json::json!({"kind":"worker_operation","workspace_id":"ws","worker_id":"w"}),
+            serde_json::json!({"kind":"worker_operation","workspace_id":"ws","runtime_id":"r"}),
+        ] {
+            assert!(serde_json::from_value::<PermissionResource>(invalid).is_err());
+        }
+        for workspace_id in ["", " ws", "ws\n", "*", &"x".repeat(257)] {
+            let resource: PermissionResource = serde_json::from_value(
+                serde_json::json!({"kind":"worker_operation","workspace_id":workspace_id}),
+            )
+            .unwrap();
+            assert!(resource.validate().is_err(), "{workspace_id:?}");
+            assert!(!resource.covers(&resource));
+        }
+        let schema = serde_json::to_value(schemars::schema_for!(PermissionResource)).unwrap();
+        let variant = schema["oneOf"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .find(|variant| variant["properties"]["kind"]["const"] == "worker_operation")
+            .unwrap();
+        assert_eq!(variant["properties"]["workspace_id"]["type"], "string");
+        assert_eq!(
+            variant["required"],
+            serde_json::json!(["kind", "workspace_id"])
+        );
+        assert_eq!(variant["additionalProperties"], false);
+    }
+
+    #[test]
+    fn worker_operation_containment_never_crosses_workspace_or_resource_kind() {
+        let resource = PermissionResource::WorkerOperation {
+            workspace_id: "ws".into(),
+        };
+        assert!(resource.covers(&resource));
+        for other in [
+            PermissionResource::WorkerOperation {
+                workspace_id: "other".into(),
+            },
+            PermissionResource::Workspace {
+                workspace_id: "ws".into(),
+            },
+            PermissionResource::Worker {
+                runtime_id: "r".into(),
+                worker_id: "w".into(),
+            },
+            PermissionResource::SubWorker {
+                runtime_id: "r".into(),
+                worker_id: "w".into(),
+                name: "child".into(),
+            },
+        ] {
+            assert!(!resource.covers(&other), "{other:?}");
+            assert!(!other.covers(&resource), "{other:?}");
+        }
+    }
+
+    #[test]
+    fn worker_operation_scopes_accept_only_seven_exact_current_operations() {
+        let resource = PermissionResource::WorkerOperation {
+            workspace_id: "ws".into(),
+        };
+        for operation in [
+            "worker.messaging",
+            "worker.send_input",
+            "worker.cancel",
+            "worker.stop",
+            "worker.restore",
+            "worker.remove",
+            "worker.observe",
+        ] {
+            assert_eq!(canonical_worker_operation(operation), Some(operation));
+            let permission = ScopedPermission {
+                operation: operation.into(),
+                resource: resource.clone(),
+            };
+            assert!(permission.validate().is_ok(), "{operation}");
+            assert!(permission.covers(&permission));
+            assert!(
+                OperationAuthorizationRequest {
+                    operation: operation.into(),
+                    resource: resource.clone(),
+                }
+                .validate()
+                .is_ok()
+            );
+            for old_resource in [
+                PermissionResource::Worker {
+                    runtime_id: "r".into(),
+                    worker_id: "w".into(),
+                },
+                PermissionResource::Workspace {
+                    workspace_id: "ws".into(),
+                },
+                PermissionResource::SubWorker {
+                    runtime_id: "r".into(),
+                    worker_id: "w".into(),
+                    name: "child".into(),
+                },
+            ] {
+                assert!(
+                    OperationAuthorizationRequest {
+                        operation: operation.into(),
+                        resource: old_resource,
+                    }
+                    .validate()
+                    .is_err()
+                );
+            }
+            let descendant = format!("{operation}.extra");
+            assert_eq!(canonical_worker_operation(&descendant), None);
+            assert!(
+                ScopedPermission {
+                    operation: descendant,
+                    resource: resource.clone(),
+                }
+                .validate()
+                .is_err()
+            );
+        }
+        for operation in [
+            "worker",
+            "worker.spawn",
+            "worker.lifecycle",
+            "worker.list",
+            "worker.input",
+            "subworker.stop",
+            "sub_worker.stop",
+            "ticket.read",
+            "worker.notify",
+            "worker.result",
+        ] {
+            assert!(
+                ScopedPermission {
+                    operation: operation.into(),
+                    resource: resource.clone(),
+                }
+                .validate()
+                .is_err(),
+                "{operation}"
+            );
+            if !matches!(operation, "worker.notify" | "worker.result") {
+                assert_eq!(canonical_worker_operation(operation), None);
+            }
+        }
+    }
+
+    #[test]
+    fn legacy_worker_message_storage_stays_exact_and_cannot_authorize_current_requests() {
+        let current = ScopedPermission {
+            operation: "worker.messaging".into(),
+            resource: PermissionResource::WorkerOperation {
+                workspace_id: "ws".into(),
+            },
+        };
+        for operation in ["worker.notify", "worker.result"] {
+            assert_eq!(
+                canonical_worker_operation(operation),
+                Some("worker.messaging")
+            );
+            let wire = serde_json::json!({
+                "operation":operation,
+                "resource":{"kind":"worker","runtime_id":"r","worker_id":"w"}
+            });
+            let stored: ScopedPermission = serde_json::from_value(wire.clone()).unwrap();
+            assert!(stored.validate().is_ok());
+            assert_eq!(serde_json::to_value(&stored).unwrap(), wire);
+            assert!(stored.covers(&stored));
+            assert!(!stored.covers(&current));
+            assert!(!current.covers(&stored));
+            let renamed = ScopedPermission {
+                operation: "worker.messaging".into(),
+                resource: stored.resource.clone(),
+            };
+            assert!(!stored.covers(&renamed));
+            assert!(!renamed.covers(&stored));
+            for resource in [
+                stored.resource,
+                current.resource.clone(),
+                PermissionResource::Workspace {
+                    workspace_id: "ws".into(),
+                },
+            ] {
+                assert!(
+                    OperationAuthorizationRequest {
+                        operation: operation.into(),
+                        resource,
+                    }
+                    .validate()
+                    .is_err()
+                );
+            }
+        }
+        // Creation and discovery remain outside runtime control operation-use scopes.
+        assert!(
+            OperationAuthorizationRequest {
+                operation: "worker.lifecycle".into(),
+                resource: PermissionResource::Workspace {
+                    workspace_id: "ws".into()
+                },
+            }
+            .validate()
+            .is_ok()
+        );
+    }
+
     #[test]
     fn authorization_response_carries_matching_scope_and_defaults_legacy_absence() {
         let legacy: OperationAuthorizationResponse =

crates/server-api/src/lib.rs

@@ -1833,6 +1833,21 @@ pub trait ServerApi {
         #[extension] context: ServerRequestContext,
         #[path] workspace_id: String,
     ) -> Result<WorkerControlListResponse, RepositoryApiError>;
+    #[get(
+        "/api/w/{workspace_id}/worker-control/workers/{runtime_id}/{worker_id}/permissions/{permission}",
+        status = 204,
+        error_status = 400,
+        additional_error_statuses = [401, 403, 404, 500],
+        openapi = false,
+    )]
+    async fn worker_control_authorize(
+        &self,
+        #[extension] context: ServerRequestContext,
+        #[path] workspace_id: String,
+        #[path] runtime_id: String,
+        #[path] worker_id: String,
+        #[path] permission: String,
+    ) -> Result<(), RepositoryApiError>;
     #[post(
         "/api/w/{workspace_id}/worker-control/workers",
         status = 200,
@@ -14966,6 +14981,7 @@ mod openapi_artifact_tests {
             "merge_request_review_submit",
             "merge_request_reviewer_child_register",
             "runtime_resource_fetch",
+            "worker_control_authorize",
             "worker_control_cancel",
             "worker_control_input",
             "worker_control_list",

crates/worker-runtime/src/catalog.rs

@@ -284,8 +284,9 @@ pub struct CreateWorkerRequest {
     pub workdir_attachment_requests: Vec<WorkingDirectoryAttachmentRequest>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub workdir_attachments: Vec<WorkingDirectoryAttachmentClaim>,
-    /// Backend-only feature enablement. Grants still define local Runtime peers;
-    /// the Workspace provider reauthorizes its dynamic set per operation.
+    /// Backend-only enablement of the optional Runtime-local peer provider.
+    /// Scoped Backend observation is composed by the Worker Feature, independently
+    /// of this saved flag, and rechecks operation use plus current target control.
     #[serde(default, skip_serializing_if = "is_false")]
     pub worker_observation_enabled: bool,
     /// Backend-authored, bounded peer session grants. Runtime revalidates each

crates/worker-runtime/src/worker_backend.rs

@@ -143,9 +143,6 @@ use workdir::{
 #[cfg(test)]
 use worker::WorkerController;
 use worker::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession;
-use worker::feature::builtin::{
-    CompositeWorkerObservationProvider, WorkspaceClientWorkerObservationProvider,
-};
 use worker::feature::contracts::{
     WorkerObservationError, WorkerObservationProvider, WorkerObservationSubject,
     WorkerObservationSubjectRef, WorkerSessionCapture,
@@ -2161,20 +2158,15 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
         if let (Some(runtime_id), Some(workspace_id)) =
             (observation_runtime_id, observation_workspace_id.clone())
             && observation_enabled
+            && !observation_grants.is_empty()
         {
-            let mut providers: Vec<Arc<dyn WorkerObservationProvider>> = vec![Arc::new(
-                WorkspaceClientWorkerObservationProvider::new(worker.workspace_client_handle()),
-            )];
-            if !observation_grants.is_empty() {
-                providers.push(Arc::new(RuntimeGrantedWorkerObservationProvider {
+            worker.bind_worker_observation_provider(Some(Arc::new(
+                RuntimeGrantedWorkerObservationProvider {
                     runtime_id,
                     workspace_id,
                     grants: observation_grants.into_iter().take(100).collect(),
                     hub: self.observation_hub.clone(),
-                }));
-            }
-            worker.bind_worker_observation_provider(Some(Arc::new(
-                CompositeWorkerObservationProvider::new(providers),
+                },
             )));
         }
         if flow_transition_enabled {
@@ -2877,16 +2869,13 @@ impl ProfileRuntimeWorkerFactory {
                 .expect("retained restore owner")
                 .sessions
                 .push(workdir_sessions.clone());
-            let mut providers: Vec<Arc<dyn WorkerObservationProvider>> = vec![Arc::new(
-                WorkspaceClientWorkerObservationProvider::new(workspace_client.clone()),
-            )];
             let observation_provider = if let (Some(runtime_id), Some(api)) = (
                 self.runtime_id.clone(),
                 request.request.workspace_api.as_ref(),
             ) && request.request.worker_observation_enabled
+                && !request.request.worker_observation_grants.is_empty()
             {
-                if !request.request.worker_observation_grants.is_empty() {
-                    providers.push(Arc::new(RuntimeGrantedWorkerObservationProvider {
+                Some(Arc::new(RuntimeGrantedWorkerObservationProvider {
                     runtime_id,
                     workspace_id: api.workspace_id.clone(),
                     grants: request
@@ -2897,10 +2886,7 @@ impl ProfileRuntimeWorkerFactory {
                         .cloned()
                         .collect(),
                     hub: self.observation_hub.clone(),
-                    }));
-                }
-                Some(Arc::new(CompositeWorkerObservationProvider::new(providers))
-                    as Arc<dyn WorkerObservationProvider>)
+                }) as Arc<dyn WorkerObservationProvider>)
             } else {
                 None
             };

crates/worker-runtime/src/worker_source.rs

@@ -705,6 +705,20 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
                                 client.worker_control_list(workspace_id).await,
                             )
                         }
+                        WorkspaceServerOperation::WorkerControlAuthorize {
+                            runtime_id,
+                            worker_id,
+                            permission,
+                        } => generated_workspace_response(
+                            client
+                                .worker_control_authorize(
+                                    workspace_id,
+                                    runtime_id,
+                                    worker_id,
+                                    permission,
+                                )
+                                .await,
+                        ),
                         WorkspaceServerOperation::WorkerControlSpawn(request) => {
                             generated_workspace_response(
                                 client.worker_control_spawn(workspace_id, request).await,

crates/worker/src/controller.rs

@@ -2061,7 +2061,7 @@ where
         crate::feature::builtin::objective::composition::compose(
             &mut feature_registry,
             feature_config.objective.enabled,
-            workspace_client,
+            workspace_client.clone(),
             objective_namespace.map(|route| (route, wip_permissions.clone())),
         )?;
 
@@ -2082,6 +2082,15 @@ where
         ) {
             observation_providers.push(provider);
         }
+        // A scoped Backend provider belongs to the Worker Feature's composition,
+        // including ordinary spawn and restore. Runtime's optional local peer
+        // provider is not its enablement or authority: every actual capture still
+        // requires current operation use and the target's Backend control grant.
+        if worker_enabled {
+            observation_providers.push(Arc::new(
+                crate::feature::builtin::worker_observation::WorkspaceClientWorkerObservationProvider::new(workspace_client),
+            ));
+        }
         if let Some(provider) = host_worker_observation_provider {
             observation_providers.push(provider);
         }
@@ -4163,6 +4172,82 @@ mod tests {
             }
         }
 
+        #[tokio::test]
+        async fn scoped_observation_is_composed_only_with_runtime_worker_service_without_startup_access()
+         {
+            for enabled in [false, true] {
+                let root = TempDir::new().unwrap();
+                let mut manifest = manifest::WorkerManifest::from_toml(
+                    "[worker]\nname = 'observation-wiring'\n[model]\nscheme = 'anthropic'\nmodel_id = 'test'\n[engine]\n[scope]\n"
+                ).unwrap();
+                manifest.feature.worker.enabled = enabled;
+                manifest.feature.sub_worker.enabled = false;
+                manifest.feature.workdir_catalog.enabled = false;
+                let model = ScriptedClient::new(vec![]);
+                let engine =
+                    Engine::<_, agen::state::Mutable, crate::SessionHistoryMetadata>::new_annotated(
+                        model,
+                    );
+                let store = session_store::CombinedStore::new(
+                    session_store::FsStore::new(root.path().join("sessions")).unwrap(),
+                    session_store::FsWorkerStore::new(root.path().join("workers")).unwrap(),
+                );
+                let operations = Arc::new(std::sync::Mutex::new(Vec::new()));
+                let host = Arc::new(DriveWorkspaceHost {
+                    available: true,
+                    workspace: Some("workspace-1"),
+                    granted: false,
+                    operations: operations.clone(),
+                });
+                let mut worker = Worker::new(
+                    manifest,
+                    engine,
+                    store,
+                    crate::worker::WorkerWorkspaceContext::with_client(
+                        Some(crate::worker::WorkspaceId::new("workspace-1").unwrap()),
+                        host,
+                    ),
+                    crate::worker::WorkerFilesystemAuthority::None,
+                    manifest::Scope::empty(),
+                )
+                .await
+                .unwrap();
+                worker.set_system_prompt_template(
+                    crate::SystemPromptTemplate::parse(
+                        "default",
+                        crate::PromptCatalogSource::builtins_only(),
+                    )
+                    .unwrap(),
+                );
+                let prepared = WorkerController::initialize(worker, root.path(), None)
+                    .await
+                    .unwrap();
+                let Event::Snapshot { greeting, .. } =
+                    crate::ipc::protocol_session::subscribe_worker_protocol_session(
+                        &prepared.handle,
+                    )
+                    .snapshot_event
+                else {
+                    panic!("initial committed greeting");
+                };
+                for name in [
+                    "ViewSessionOverview",
+                    "SearchSessionEntries",
+                    "ReadSessionEntry",
+                ] {
+                    assert_eq!(
+                        greeting.tools.iter().any(|tool| tool == name),
+                        enabled,
+                        "Worker service configuration must determine observation publication, not client availability alone"
+                    );
+                }
+                assert!(
+                    operations.lock().unwrap().is_empty(),
+                    "provider composition may not authorize or read protected resources"
+                );
+            }
+        }
+
         #[tokio::test]
         async fn drive_controller_registers_without_authorization_and_checks_each_access_in_both_modes()
          {

crates/worker/src/feature/builtin/manage_worker.rs

@@ -41,22 +41,18 @@ struct WorkspaceWorkerControlService {
 }
 
 impl WorkspaceWorkerControlService {
-    fn authorize_runtime_worker(
-        &self,
-        operation: &str,
-        runtime_id: &str,
-        worker_id: &str,
-    ) -> Result<(), WorkspaceClientError> {
+    fn authorize_runtime_operation(&self, operation: &str) -> Result<(), WorkspaceClientError> {
         if !self.runtime_worker_control {
             return Err(WorkspaceClientError::Unavailable(
                 "Runtime Worker control is not enabled for this Worker".into(),
             ));
         }
+        // Common authorization permits using the operation. The Backend separately
+        // checks the authenticated caller's control grant for the actual target.
         self.client.authorize_operation(
             operation,
-            &PermissionResource::Worker {
-                runtime_id: runtime_id.into(),
-                worker_id: worker_id.into(),
+            &PermissionResource::WorkerOperation {
+                workspace_id: self.workspace_id.clone(),
             },
         )
     }
@@ -201,14 +197,10 @@ impl WorkerControlService for WorkspaceWorkerControlService {
                 runtime_id,
                 worker_id,
             } => {
-                self.authorize_runtime_worker(
-                    match kind {
+                self.authorize_runtime_operation(match kind {
                     WorkerControlInputKind::User => "worker.send_input",
-                        WorkerControlInputKind::Notify => "worker.notify",
-                    },
-                    &runtime_id,
-                    &worker_id,
-                )?;
+                    WorkerControlInputKind::Notify => "worker.messaging",
+                })?;
                 let kind = match kind {
                     WorkerControlInputKind::User => "user",
                     WorkerControlInputKind::Notify => "notify",
@@ -273,14 +265,10 @@ impl WorkerControlService for WorkspaceWorkerControlService {
                 runtime_id,
                 worker_id,
             } => {
-                self.authorize_runtime_worker(
-                    match kind {
+                self.authorize_runtime_operation(match kind {
                     WorkerControlStopKind::Cancel => "worker.cancel",
                     WorkerControlStopKind::Stop => "worker.stop",
-                    },
-                    &runtime_id,
-                    &worker_id,
-                )?;
+                })?;
                 let operation = match kind {
                     WorkerControlStopKind::Cancel => {
                         crate::worker::WorkspaceServerOperation::WorkerControlCancel {
@@ -361,7 +349,7 @@ impl WorkerControlService for WorkspaceWorkerControlService {
                 "Runtime Worker control is not enabled for this Worker".to_string(),
             ));
         }
-        self.authorize_runtime_worker("worker.restore", &runtime_id, &worker_id)?;
+        self.authorize_runtime_operation("worker.restore")?;
         self.client.execute_server_operation(
             crate::worker::WorkspaceServerOperation::WorkerControlRestore {
                 runtime_id,
@@ -382,7 +370,7 @@ impl WorkerControlService for WorkspaceWorkerControlService {
                 "Runtime Worker control is not enabled for this Worker".to_string(),
             ));
         }
-        self.authorize_runtime_worker("worker.remove", runtime_id, worker_id)?;
+        self.authorize_runtime_operation("worker.remove")?;
         self.client
             .execute_worker_remove(runtime_id, worker_id, reason)
     }
@@ -416,50 +404,26 @@ impl WorkerControlService for WorkspaceWorkerControlService {
                         "Runtime Worker control is not enabled for this Worker".to_string(),
                     ));
                 }
-                self.authorize_runtime_worker(
-                    &format!("worker.{permission}"),
-                    runtime_id,
-                    worker_id,
-                )?;
+                let operation = if permission == "notify" {
+                    "worker.messaging".to_owned()
+                } else {
+                    format!("worker.{permission}")
+                };
+                self.authorize_runtime_operation(&operation)?;
                 let response = self.client.execute_server_operation(
-                    crate::worker::WorkspaceServerOperation::WorkerControlList,
+                    crate::worker::WorkspaceServerOperation::WorkerControlAuthorize {
+                        runtime_id: runtime_id.clone(),
+                        worker_id: worker_id.clone(),
+                        permission: permission.into(),
+                    },
                 )?;
-                if !response.is_success() {
-                    return Err(WorkspaceClientError::Request(format!(
-                        "Workspace control request returned {}: {}",
-                        response.status, response.body
-                    )));
-                }
-                let body: serde_json::Value =
-                    serde_json::from_str(&response.body).map_err(|error| {
-                        WorkspaceClientError::Request(format!(
-                            "invalid Workspace control response: {error}"
-                        ))
-                    })?;
-                let granted = body
-                    .get("items")
-                    .and_then(serde_json::Value::as_array)
-                    .into_iter()
-                    .flatten()
-                    .any(|item| {
-                        item.get("subject")
-                            .and_then(|value| value.get("runtime_id"))
-                            == Some(&serde_json::Value::String(runtime_id.clone()))
-                            && item.get("subject").and_then(|value| value.get("worker_id"))
-                                == Some(&serde_json::Value::String(worker_id.clone()))
-                            && item
-                                .get("permissions")
-                                .and_then(serde_json::Value::as_array)
-                                .is_some_and(|permissions| {
-                                    permissions.iter().any(|candidate| candidate == permission)
-                                })
-                    });
-                if granted {
+                if response.is_success() {
                     Ok(())
                 } else {
-                    Err(WorkspaceClientError::Request(
-                        "unknown Worker or permission not granted".to_string(),
-                    ))
+                    Err(WorkspaceClientError::Request(format!(
+                        "Worker target permission check returned {}: {}",
+                        response.status, response.body
+                    )))
                 }
             }
         }
@@ -1106,6 +1070,32 @@ fn tool_output(
             response.status, response.body
         )));
     }
+    if operation == WorkerOperation::Notify {
+        let notified: server_api::RuntimeWorkerInputResult = serde_json::from_str(&response.body)
+            .map_err(|error| {
+            ToolError::ExecutionFailed(format!("invalid WorkerNotify response: {error}"))
+        })?;
+        // HTTP transport success is not Runtime acceptance. Preserve uncertain
+        // diagnostics in the error; callers must inspect history, not resend or
+        // implicitly Restore a stopped destination.
+        if notified.state != server_api::WorkerOperationState::Accepted
+            || notified
+                .diagnostics
+                .iter()
+                .any(|diagnostic| diagnostic.code == "worker_input_outcome_unknown")
+        {
+            return Err(ToolError::ExecutionFailed(format!(
+                "WorkerNotify was not confirmed accepted (no automatic retry or Restore): {}",
+                response.body
+            )));
+        }
+        return Ok(ToolOutput {
+            summary: "WorkerNotify accepted; model processing and work completion are unconfirmed"
+                .into(),
+            content: Some(response.body),
+            attachments: Vec::new(),
+        });
+    }
     if operation == WorkerOperation::Restore {
         let restored: server_api::WorkerRestoreResponse = serde_json::from_str(&response.body)
             .map_err(|error| {
@@ -1288,6 +1278,8 @@ mod tests {
     struct RecordingWorkspaceClient {
         requests: Mutex<Vec<WorkspaceRequest>>,
         authorizations: Mutex<Vec<server_api::authorization::OperationAuthorizationRequest>>,
+        bound_authorizations:
+            Mutex<Vec<server_api::authorization::BoundOperationAuthorizationRequest>>,
         denied: std::sync::atomic::AtomicBool,
         removals: Mutex<Vec<(String, String, String)>>,
     }
@@ -1321,7 +1313,16 @@ mod tests {
                     body: r#"{"scope":null}"#.into(),
                 });
             }
+            if request.path.ends_with("/authorization/bound-check") {
+                let auth = serde_json::from_str(request.body.as_deref().unwrap()).unwrap();
+                self.bound_authorizations.lock().unwrap().push(auth);
+                return Ok(WorkspaceResponse {
+                    status: 200,
+                    body: r#"{"scope":null}"#.into(),
+                });
+            }
             let is_worker_list = request.path.ends_with("/worker-control/workers");
+            let is_worker_input = request.path.ends_with("/input");
             self.requests.lock().unwrap().push(request);
             Ok(WorkspaceResponse {
                 status: 200,
@@ -1333,6 +1334,7 @@ mod tests {
                                     "runtime_id": "runtime-a",
                                     "worker_id": "worker-a"
                                 },
+                                "permissions": ["observe"],
                                 "summary": { "status": "idle" }
                             },
                             {
@@ -1340,11 +1342,14 @@ mod tests {
                                     "runtime_id": "runtime-a",
                                     "worker_id": "worker-a"
                                 },
+                                "permissions": ["observe"],
                                 "summary": { "status": "idle" }
                             }
                         ]
                     })
                     .to_string()
+                } else if is_worker_input {
+                    serde_json::json!({"state":"accepted","runtime_id":"runtime-a","worker_id":"worker-a","diagnostics":[]}).to_string()
                 } else {
                     "{}".to_string()
                 },
@@ -1476,12 +1481,48 @@ mod tests {
     fn test_control(client: Arc<RecordingWorkspaceClient>) -> Arc<dyn WorkerControlService> {
         Arc::new(WorkspaceWorkerControlService {
             client,
-            workspace_id: "workspace%2Ftest".to_string(),
+            workspace_id: "workspace/test".to_string(),
             registry: None,
             runtime_worker_control: true,
         })
     }
 
+    #[test]
+    fn notify_transport_success_requires_runtime_acceptance_and_preserves_unknown_diagnostics() {
+        for body in [
+            serde_json::json!({"state":"rejected","runtime_id":"r","worker_id":"w","diagnostics":[]}),
+            serde_json::json!({"state":"unsupported","runtime_id":"r","worker_id":"w","diagnostics":[]}),
+            serde_json::json!({"state":"rejected","runtime_id":"r","worker_id":"w","diagnostics":[{"code":"worker_input_outcome_unknown","severity":"warning","message":"inspect current session before retrying"}]}),
+        ] {
+            let error = tool_output(
+                WorkerOperation::Notify,
+                WorkspaceResponse {
+                    status: 200,
+                    body: body.to_string(),
+                },
+            )
+            .unwrap_err()
+            .to_string();
+            assert!(error.contains("not confirmed accepted"), "{error}");
+            assert!(error.contains("no automatic retry or Restore"), "{error}");
+            if !body["diagnostics"].as_array().unwrap().is_empty() {
+                assert!(error.contains("worker_input_outcome_unknown"));
+            }
+        }
+        assert!(
+            tool_output(
+                WorkerOperation::Notify,
+                WorkspaceResponse {
+                    status: 200,
+                    body: "{}".into()
+                }
+            )
+            .is_err()
+        );
+        let accepted = tool_output(WorkerOperation::Notify,WorkspaceResponse{status:200,body:serde_json::json!({"state":"accepted","runtime_id":"r","worker_id":"w","diagnostics":[]}).to_string()}).unwrap();
+        assert!(accepted.summary.contains("unconfirmed"));
+    }
+
     #[tokio::test]
     async fn worker_restore_forwards_exact_observation_and_request_identity() {
         let client = Arc::new(RecordingWorkspaceClient::default());
@@ -1679,7 +1720,7 @@ mod tests {
         let client = Arc::new(RecordingWorkspaceClient::default());
         let control = WorkspaceWorkerControlService {
             client: client.clone(),
-            workspace_id: "workspace%2Ftest".to_string(),
+            workspace_id: "workspace/test".to_string(),
             registry: None,
             runtime_worker_control: true,
         };
@@ -1715,7 +1756,7 @@ mod tests {
         let client = Arc::new(RecordingWorkspaceClient::default());
         let control = WorkspaceWorkerControlService {
             client: client.clone(),
-            workspace_id: "workspace%2Ftest".to_string(),
+            workspace_id: "workspace/test".to_string(),
             registry: None,
             runtime_worker_control: true,
         };
@@ -1764,7 +1805,7 @@ mod tests {
         let registry = SpawnedWorkerRegistry::new(runtime_dir);
         let control: Arc<dyn WorkerControlService> = Arc::new(WorkspaceWorkerControlService {
             client: client.clone(),
-            workspace_id: "workspace%2Ftest".to_string(),
+            workspace_id: "workspace/test".to_string(),
             registry: Some(registry),
             runtime_worker_control: false,
         });
@@ -1832,7 +1873,7 @@ mod tests {
         let client = Arc::new(RecordingWorkspaceClient::default());
         let control: Arc<dyn WorkerControlService> = Arc::new(WorkspaceWorkerControlService {
             client: client.clone(),
-            workspace_id: "workspace%2Ftest".to_string(),
+            workspace_id: "workspace/test".to_string(),
             registry: None,
             runtime_worker_control: false,
         });
@@ -2138,6 +2179,28 @@ mod tests {
             )
             .await
             .unwrap();
+            let check = client
+                .authorizations
+                .lock()
+                .unwrap()
+                .last()
+                .cloned()
+                .unwrap();
+            assert_eq!(
+                check.operation,
+                match operation {
+                    WorkerOperation::SendInput => "worker.send_input",
+                    WorkerOperation::Notify => "worker.messaging",
+                    WorkerOperation::Cancel => "worker.cancel",
+                    _ => unreachable!(),
+                }
+            );
+            assert_eq!(
+                check.resource,
+                PermissionResource::WorkerOperation {
+                    workspace_id: "workspace/test".into(),
+                }
+            );
             let request = client.requests.lock().unwrap().last().cloned().unwrap();
             assert!(request.path.ends_with(expected_suffix));
             assert!(request.path.contains("/worker-control/workers/"));
@@ -2326,12 +2389,12 @@ mod tests {
             };
             match op {
                 "worker.list" => control.list_workers().await,
-                "worker.send_input" | "worker.notify" => {
+                "worker.send_input" | "worker.messaging" => {
                     control
                         .send_input(
                             subject,
                             "input".into(),
-                            if op == "worker.notify" {
+                            if op == "worker.messaging" {
                                 WorkerControlInputKind::Notify
                             } else {
                                 WorkerControlInputKind::User
@@ -2369,6 +2432,21 @@ mod tests {
                     "target-worker",
                     "requested removal",
                 ),
+                "worker.observe" => {
+                    control
+                        .ensure_permission(
+                            &WorkerObservationSubjectRef::RuntimeWorker {
+                                runtime_id: "runtime-a".into(),
+                                worker_id: "worker-a".into(),
+                            },
+                            "observe",
+                        )
+                        .await?;
+                    Ok(WorkspaceResponse {
+                        status: 200,
+                        body: "{}".into(),
+                    })
+                }
                 "worker.create" => {
                     control
                         .spawn_worker(WorkerLifecycleSpawnRequest {
@@ -2391,11 +2469,12 @@ mod tests {
         for op in [
             "worker.list",
             "worker.send_input",
-            "worker.notify",
+            "worker.messaging",
             "worker.cancel",
             "worker.stop",
             "worker.restore",
             "worker.remove",
+            "worker.observe",
             "worker.create",
         ] {
             let client = Arc::new(RecordingWorkspaceClient::default());
@@ -2433,23 +2512,134 @@ mod tests {
                     check.resource,
                     if matches!(op, "worker.list" | "worker.create") {
                         PermissionResource::Workspace {
-                            workspace_id: if op == "worker.list" {
-                                "workspace%2Ftest"
-                            } else {
-                                "workspace/test"
-                            }
-                            .into(),
+                            workspace_id: "workspace/test".into(),
                         }
                     } else {
-                        PermissionResource::Worker {
-                            runtime_id: "target-runtime".into(),
-                            worker_id: "target-worker".into(),
+                        PermissionResource::WorkerOperation {
+                            workspace_id: "workspace/test".into(),
                         }
                     }
                 );
             }
         }
     }
+    // Connection: published Tool registration and dispatch, production typed control
+    // service and authorization transport; the Host and Backend are fixture boundaries.
+    #[tokio::test]
+    async fn published_runtime_tools_reject_missing_operation_grant_without_backend_effects() {
+        let client = Arc::new(RecordingWorkspaceClient::default());
+        let mut pending_tools = Vec::new();
+        let mut hooks = crate::HookRegistryBuilder::default();
+        let report = crate::feature::FeatureRegistryBuilder::new()
+            .with_module(manage_worker_feature(client.clone(), None, true))
+            .install_into_pending(&mut pending_tools, &mut hooks);
+        assert!(report.reports[0].installed);
+        let tools: Vec<_> = pending_tools
+            .into_iter()
+            .map(|definition| definition())
+            .collect();
+        client
+            .denied
+            .store(true, std::sync::atomic::Ordering::SeqCst);
+        for (tool_name, operation, fields) in [
+            (
+                "WorkerSendInput",
+                "worker.send_input",
+                serde_json::json!({"content":"continue"}),
+            ),
+            (
+                "WorkerNotify",
+                "worker.messaging",
+                serde_json::json!({"content":"result ready"}),
+            ),
+            (
+                "WorkerCancel",
+                "worker.cancel",
+                serde_json::json!({"reason":"cancel"}),
+            ),
+            (
+                "WorkerStop",
+                "worker.stop",
+                serde_json::json!({"reason":"stop"}),
+            ),
+            (
+                "WorkerRestore",
+                "worker.restore",
+                serde_json::json!({"expected_observation_token":"observed"}),
+            ),
+            (
+                "WorkerRemove",
+                "worker.remove",
+                serde_json::json!({"reason":"retire"}),
+            ),
+        ] {
+            let (_, tool) = tools
+                .iter()
+                .find(|(meta, _)| meta.name == tool_name)
+                .unwrap();
+            for worker_id in ["worker-a", "worker-b"] {
+                let mut input = fields.clone();
+                input["subject"] = serde_json::json!({
+                    "kind":"runtime_worker", "runtime_id":"runtime-a", "worker_id":worker_id,
+                });
+                let error = tool
+                    .execute(&input.to_string(), ToolExecutionContext::direct())
+                    .await
+                    .unwrap_err();
+                assert!(
+                    error.to_string().contains("authorization denied"),
+                    "{tool_name}: {error}"
+                );
+                let checks = client.authorizations.lock().unwrap();
+                let check = checks.last().unwrap();
+                assert_eq!(check.operation, operation);
+                assert_eq!(
+                    check.resource,
+                    PermissionResource::WorkerOperation {
+                        workspace_id: "workspace/test".into(),
+                    }
+                );
+                assert!(client.requests.lock().unwrap().is_empty(), "{tool_name}");
+                assert!(client.removals.lock().unwrap().is_empty(), "{tool_name}");
+            }
+        }
+        assert_eq!(client.authorizations.lock().unwrap().len(), 12);
+    }
+
+    #[test]
+    fn runtime_operation_denial_does_not_replace_host_bound_subworker_authorization() {
+        let client = Arc::new(RecordingWorkspaceClient::default());
+        client
+            .denied
+            .store(true, std::sync::atomic::Ordering::SeqCst);
+        let control = WorkspaceWorkerControlService {
+            client: client.clone(),
+            workspace_id: "workspace/test".into(),
+            registry: None,
+            runtime_worker_control: true,
+        };
+        for operation in [
+            "subworker.send_input",
+            "subworker.stop",
+            "subworker.observe",
+        ] {
+            control.authorize_subworker(operation, "child").unwrap();
+            let checks = client.bound_authorizations.lock().unwrap();
+            let check = checks.last().unwrap();
+            assert_eq!(check.operation, operation);
+            assert_eq!(
+                check.target,
+                server_api::authorization::BoundAuthorizationTarget::SubWorker {
+                    name: "child".into(),
+                }
+            );
+        }
+        assert_eq!(client.bound_authorizations.lock().unwrap().len(), 3);
+        assert!(client.authorizations.lock().unwrap().is_empty());
+        assert!(client.requests.lock().unwrap().is_empty());
+        assert!(client.removals.lock().unwrap().is_empty());
+    }
+
     #[test]
     fn only_explicit_standalone_retains_local_subworker_authority_without_host() {
         for workspace_id in [None, Some("bound-workspace".to_string())] {

crates/worker/src/feature/builtin/worker_observation.rs

@@ -125,14 +125,10 @@ impl WorkerObservationProvider for WorkspaceClientWorkerObservationProvider {
             )
         })?;
         match subject {
-            WorkerObservationSubjectRef::RuntimeWorker {
-                runtime_id,
-                worker_id,
-            } => self.client.authorize_operation(
+            WorkerObservationSubjectRef::RuntimeWorker { .. } => self.client.authorize_operation(
                 "worker.observe",
-                &server_api::authorization::PermissionResource::Worker {
-                    runtime_id: runtime_id.clone(),
-                    worker_id: worker_id.clone(),
+                &server_api::authorization::PermissionResource::WorkerOperation {
+                    workspace_id: _workspace_id.into(),
                 },
             ),
             WorkerObservationSubjectRef::SubWorker { name } => self

crates/worker/src/worker.rs

@@ -762,6 +762,11 @@ pub enum WorkspaceServerOperation {
         result_id: String,
     },
     WorkerControlList,
+    WorkerControlAuthorize {
+        runtime_id: String,
+        worker_id: String,
+        permission: String,
+    },
     WorkerControlSpawn(server_api::CreateWorkspaceWorkerRequest),
     WorkerControlInput {
         runtime_id: String,
@@ -960,6 +965,16 @@ fn workspace_server_operation_request(
         WorkspaceServerOperation::WorkerControlList => Ok(WorkspaceRequest::get(format!(
             "{base}/worker-control/workers"
         ))),
+        WorkspaceServerOperation::WorkerControlAuthorize {
+            runtime_id,
+            worker_id,
+            permission,
+        } => Ok(WorkspaceRequest::get(format!(
+            "{base}/worker-control/workers/{}/{}/permissions/{}",
+            encode_workspace_path_segment(&runtime_id),
+            encode_workspace_path_segment(&worker_id),
+            encode_workspace_path_segment(&permission)
+        ))),
         WorkspaceServerOperation::WorkerControlSpawn(request) => {
             workspace_server_json_request(format!("{base}/worker-control/workers"), &request)
         }

crates/workspace-server/src/server.rs

@@ -2067,6 +2067,7 @@ struct WorkerRemovalService {
     journal_storage: crate::WorkspaceFeatureStorage,
     journal_registration: crate::RegisteredFeature,
     journal_dispatch_api: Weak<WorkspaceApi>,
+    worker_authority: subject_authorization::WorkerAuthorityView,
     store: Arc<dyn ControlPlaneStore>,
     host_executors: HostExecutors,
     runtime: Weak<RuntimeRegistry>,
@@ -2081,6 +2082,7 @@ impl WorkerRemovalService {
     fn new(api: &WorkspaceApi) -> Self {
         Self {
             workspace_id: api.config.workspace_id.clone(),
+            worker_authority: subject_authorization::WorkerAuthorityView::from(api),
             journal_storage: api.feature_storage.clone(),
             journal_registration: api.subjektiv_registration.clone(),
             journal_dispatch_api: api
@@ -2403,22 +2405,13 @@ impl WorkerRemovalService {
             ));
         }
         let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id);
-        let grant = self
-            .store
-            .get_active_worker_control_grant(&self.workspace_id, &controller, &target)
-            .map_err(|_| "Worker control grant authority is unavailable".to_string())?
-            .filter(|grant| {
-                grant
-                    .permissions
-                    .iter()
-                    .any(|permission| permission == "remove")
-            });
-        let Some(grant) = grant else {
-            return Ok(worker_remove_error_response(
-                StatusCode::NOT_FOUND,
-                "unknown_worker",
-                "The target Worker is not known to the current Worker",
-            ));
+        let grant =
+            match self
+                .worker_authority
+                .authorize_control_permission(&controller, &target, "remove")
+            {
+                Ok(grant) => grant,
+                Err(error) => return Ok(worker_control_denied_response(error)),
             };
         let control_lock = {
             let mut locks = self
@@ -2431,17 +2424,15 @@ impl WorkerRemovalService {
                 .clone()
         };
         let _control_guard = control_lock.lock().await;
-        let still_granted = self
-            .store
-            .get_active_worker_control_grant(&self.workspace_id, &controller, &target)
-            .map_err(|_| "Worker control grant authority is unavailable".to_string())?
-            .is_some_and(|current| {
-                current.grant_id == grant.grant_id
-                    && current
-                        .permissions
-                        .iter()
-                        .any(|permission| permission == "remove")
-            });
+        let current =
+            match self
+                .worker_authority
+                .authorize_control_permission(&controller, &target, "remove")
+            {
+                Ok(grant) => grant,
+                Err(error) => return Ok(worker_control_denied_response(error)),
+            };
+        let still_granted = current.grant_id == grant.grant_id;
         if !still_granted {
             return Ok(worker_remove_error_response(
                 StatusCode::NOT_FOUND,
@@ -7827,6 +7818,9 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
         .merge(server_api::server_api_axum::worker_control_list(
             service.clone(),
         ))
+        .merge(server_api::server_api_axum::worker_control_authorize(
+            service.clone(),
+        ))
         .merge(server_api::server_api_axum::worker_control_spawn(
             service.clone(),
         ))
@@ -10394,6 +10388,46 @@ impl server_api::ServerApi for ServerApiContractService {
         Ok(response)
     }
 
+    async fn worker_control_authorize(
+        &self,
+        context: server_api::ServerRequestContext,
+        workspace_id: String,
+        runtime_id: String,
+        worker_id: String,
+        permission: String,
+    ) -> std::result::Result<(), server_api::RepositoryApiError> {
+        let api = self.workspace_api()?.clone();
+        let headers = current_worker_contract_headers(&context)?;
+        api.host_executors
+            .host()
+            .map_err(|error| ApiError::from(error).into_repository_api_error())?
+            .run(move || -> ApiResult<()> {
+                validate_workspace_scope(&api, &workspace_id)?;
+                let source = authenticate_worker_mutation_source(&api, &workspace_id, &headers)?;
+                if !matches!(
+                    permission.as_str(),
+                    "observe" | "notify" | "send_input" | "cancel" | "stop" | "restore" | "remove"
+                ) {
+                    return Err(
+                        Error::InvalidInput("unknown Worker control permission".into()).into(),
+                    );
+                }
+                // This is a current, target-only preflight for providers that read
+                // locally. It discloses no registry/session data and issues no
+                // reusable receipt. Actual Backend effects still recheck authority.
+                authorize_known_worker_permission(
+                    &api,
+                    &workspace_id,
+                    &RuntimeWorkerRef::new(source.runtime_id, source.worker_id),
+                    &RuntimeWorkerRef::new(runtime_id, worker_id),
+                    &permission,
+                )?;
+                Ok(())
+            })
+            .await
+            .map_err(ApiError::into_repository_api_error)
+    }
+
     async fn worker_control_spawn(
         &self,
         context: server_api::ServerRequestContext,
@@ -26515,6 +26549,16 @@ async fn list_known_workers(
             validate_workspace_scope(&api, &path.workspace_id)?;
             let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?;
             let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id);
+            subject_authorization::authorize(
+                &api,
+                &controller,
+                &server_api::authorization::OperationAuthorizationRequest {
+                    operation: "worker.list".into(),
+                    resource: server_api::authorization::PermissionResource::Workspace {
+                        workspace_id: path.workspace_id.clone(),
+                    },
+                },
+            )?;
             let limit = api.config.max_records.clamp(1, 500);
             let grants = api.store.list_active_worker_control_grants(
                 &path.workspace_id,
@@ -26637,6 +26681,15 @@ async fn spawn_known_worker_authenticated_on_host(
         context,
     )
     .await?;
+    // Creation has now committed the real registry row. Keep reply-grant failure
+    // outside spawn compensation so an exact operation replay repairs only the
+    // missing grant, without a second Worker or repeated initial Submit.
+    subject_authorization::persist_reply_grant(
+        &api,
+        &controller,
+        &RuntimeWorkerRef::new(&response.0.runtime_id, &response.0.worker_id),
+        &scoped_worker_control_operation_id(&controller, &operation_id),
+    )?;
     if let Err(error) = api
         .store
         .create_worker_control_grant(&WorkerControlGrantRecord {
@@ -26681,6 +26734,26 @@ fn worker_control_lock(api: &WorkspaceApi, grant_id: &str) -> Arc<tokio::sync::M
     )
 }
 
+fn worker_control_denied_response(error: Error) -> worker::WorkspaceResponse {
+    // Keep removal's existing opaque target-grant refusal; operation-policy
+    // refusals remain distinct and do not impersonate an unknown target.
+    if matches!(&error, Error::UnknownWorker { .. })
+        || matches!(&error, Error::InvalidInput(message)
+            if message.starts_with("worker control permission "))
+    {
+        return worker_remove_error_response(
+            StatusCode::NOT_FOUND,
+            "unknown_worker",
+            "The target Worker is not known to the current Worker",
+        );
+    }
+    worker_remove_error_response(
+        api_error_status(&error),
+        "worker_control_denied",
+        &error.to_string(),
+    )
+}
+
 fn authorize_known_worker_permission(
     api: &WorkspaceApi,
     workspace_id: &str,
@@ -26688,54 +26761,13 @@ fn authorize_known_worker_permission(
     subject: &RuntimeWorkerRef,
     permission: &str,
 ) -> Result<WorkerControlGrantRecord> {
-    // Child result return is deliberately narrower than control or public chat
-    // posting. The execution grant names only this origin Worker and only notify.
-    if permission == "notify" {
-        if let Some(permissions) = subject_authorization::worker_permissions(api, controller)
-            .map_err(|error| error.error)?
-        {
-            let requested = server_api::authorization::ScopedPermission {
-                operation: "worker.result".into(),
-                resource: server_api::authorization::PermissionResource::Worker {
-                    runtime_id: subject.runtime_id.clone(),
-                    worker_id: subject.worker_id.clone(),
-                },
-            };
-            if permissions.iter().any(|granted| granted.covers(&requested)) {
-                return Ok(WorkerControlGrantRecord {
-                    workspace_id: workspace_id.into(),
-                    grant_id: format!(
-                        "execution-result:{}/{}",
-                        controller.runtime_id, controller.worker_id
-                    ),
-                    controller: controller.clone(),
-                    subject: subject.clone(),
-                    relation: "result_return".into(),
-                    origin: "execution_grant".into(),
-                    permissions: vec!["notify".into()],
-                    operation_id: "execution-result".into(),
-                    created_at: String::new(),
-                    revoked_at: None,
-                });
+    if workspace_id != api.workspace_id() {
+        return Err(Error::WorkspacePermissionDenied(
+            "resource belongs to another Workspace".into(),
+        ));
     }
-        }
-    }
-    let grant = api
-        .store
-        .get_active_worker_control_grant(workspace_id, controller, subject)?
-        .ok_or_else(|| Error::UnknownWorker {
-            worker: subject.clone(),
-        })?;
-    if !grant
-        .permissions
-        .iter()
-        .any(|candidate| candidate == permission)
-    {
-        return Err(Error::InvalidInput(format!(
-            "worker control permission `{permission}` was not granted"
-        )));
-    }
-    Ok(grant)
+    subject_authorization::WorkerAuthorityView::from(api)
+        .authorize_control_permission(controller, subject, permission)
 }
 
 async fn send_known_worker_input(
@@ -26854,6 +26886,16 @@ async fn scoped_list_worker_observation_sessions(
             validate_workspace_scope(&api, &path.workspace_id)?;
             let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?;
             let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id);
+            subject_authorization::authorize(
+                &api,
+                &controller,
+                &server_api::authorization::OperationAuthorizationRequest {
+                    operation: "worker.observe.list".into(),
+                    resource: server_api::authorization::PermissionResource::Workspace {
+                        workspace_id: path.workspace_id.clone(),
+                    },
+                },
+            )?;
             let sessions = api
                 .store
                 .list_active_worker_control_grants(&path.workspace_id, &controller, 100)?
@@ -26865,6 +26907,14 @@ async fn scoped_list_worker_observation_sessions(
                         .any(|permission| permission == "observe")
                 })
                 .filter_map(|grant| {
+                    authorize_known_worker_permission(
+                        &api,
+                        &path.workspace_id,
+                        &controller,
+                        &grant.subject,
+                        "observe",
+                    )
+                    .ok()?;
                     let worker = api.runtime.worker(&grant.subject).ok()?;
                     if matches!(
                         worker.state.as_str(),

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

@@ -16,7 +16,7 @@
 use super::*;
 use server_api::authorization::{
     OperationAuthorizationRequest, PermissionResource, ScopedPermission, SubjectPermissions,
-    WorkerDelegationRequest, WorkerExecutionGrant,
+    WorkerDelegationRequest, WorkerExecutionGrant, canonical_worker_operation,
 };
 use std::collections::BTreeSet;
 
@@ -24,8 +24,57 @@ fn denied(message: impl Into<String>) -> ApiError {
     Error::WorkspacePermissionDenied(message.into()).into()
 }
 
-fn policy(api: &WorkspaceApi, subject_id: &str) -> ApiResult<Option<SubjectPermissions>> {
-    let store = open_subjektiv_store(api)?;
+/// Minimal shared authority for Worker control and removal. In particular this
+/// must not retain WorkspaceApi or Runtime: the embedded removal dispatcher is
+/// itself owned by Runtime. Policy, lineage and body ownership keep one checker.
+#[derive(Clone)]
+pub(super) struct WorkerAuthorityView {
+    workspace_id: String,
+    store: Arc<dyn ControlPlaneStore>,
+    config_store: Arc<crate::SqliteWorkspaceStore>,
+    feature_storage: crate::WorkspaceFeatureStorage,
+    subjektiv_registration: crate::RegisteredFeature,
+}
+
+impl From<&WorkspaceApi> for WorkerAuthorityView {
+    fn from(api: &WorkspaceApi) -> Self {
+        Self {
+            workspace_id: api.workspace_id().into(),
+            store: api.store.clone(),
+            config_store: api.config_store.clone(),
+            feature_storage: api.feature_storage.clone(),
+            subjektiv_registration: api.subjektiv_registration.clone(),
+        }
+    }
+}
+
+impl WorkerAuthorityView {
+    fn subject_store(&self) -> ApiResult<crate::subjektiv::SubjektivStore> {
+        crate::subjektiv::SubjektivStore::open(&self.feature_storage, &self.subjektiv_registration)
+            .map_err(|error| Error::Store(error.to_string()).into())
+    }
+
+    fn require_not_deleted(&self, worker: &RuntimeWorkerRef) -> ApiResult<()> {
+        let deleting = self.config_store.with_conn(|connection| {
+            Ok(connection.query_row(
+                "SELECT EXISTS(SELECT 1 FROM subject_deletion_workers WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3)",
+                rusqlite::params![self.workspace_id, worker.runtime_id, worker.worker_id],
+                |row| row.get::<_, bool>(0),
+            )?)
+        })?;
+        if deleting
+            || self
+                .subject_store()?
+                .worker_subject_deleted(&worker.runtime_id, &worker.worker_id)
+                .map_err(subjektiv_store_error)?
+        {
+            return Err(denied("Worker belonged to an explicitly deleted Subject"));
+        }
+        Ok(())
+    }
+
+    fn policy(&self, subject_id: &str) -> ApiResult<Option<SubjectPermissions>> {
+        let store = self.subject_store()?;
         let subject = store
             .subject(subject_id)
             .map_err(subjektiv_store_error)?
@@ -38,13 +87,10 @@ fn policy(api: &WorkspaceApi, subject_id: &str) -> ApiResult<Option<SubjectPermi
             .map_err(subjektiv_store_error)?)
     }
 
-fn current_body_subject(
-    api: &WorkspaceApi,
-    worker: &RuntimeWorkerRef,
-) -> ApiResult<Option<String>> {
-    let retained = api
+    fn current_body_subject(&self, worker: &RuntimeWorkerRef) -> ApiResult<Option<String>> {
+        let retained = self
             .store
-        .retained_worker_singleton_lease(&api.config.workspace_id, worker)?;
+            .retained_worker_singleton_lease(&self.workspace_id, worker)?;
         let Some(retained) = retained else {
             return Ok(None);
         };
@@ -53,9 +99,9 @@ fn current_body_subject(
         };
         // This also rejects a retained lease from a released/replaced body. Never
         // fall through to non-Subject compatibility after ownership has ended.
-    let current = api
+        let current = self
             .store
-        .require_current_worker_singleton_owner(&api.config.workspace_id, worker)?
+            .require_current_worker_singleton_owner(&self.workspace_id, worker)?
             .ok_or_else(|| denied("Subject body no longer owns its singleton"))?;
         if current.key != retained.key {
             return Err(denied("Subject body ownership changed"));
@@ -64,18 +110,19 @@ fn current_body_subject(
     }
 
     fn live_child_grant(
-    api: &WorkspaceApi,
+        &self,
         worker: &RuntimeWorkerRef,
     ) -> ApiResult<Option<WorkerExecutionGrant>> {
-    let store = open_subjektiv_store(api)?;
+        let store = self.subject_store()?;
         let Some(grant) = store
             .worker_execution_grant(&worker.runtime_id, &worker.worker_id)
             .map_err(subjektiv_store_error)?
         else {
             return Ok(None);
         };
-    let source =
-        policy(api, &grant.subject_id)?.ok_or_else(|| denied("Subject permissions are unset"))?;
+        let source = self
+            .policy(&grant.subject_id)?
+            .ok_or_else(|| denied("Subject permissions are unset"))?;
         if source.change_id != grant.source_change_id {
             return Err(denied(
                 "delegated execution grant was revoked by permission replacement",
@@ -104,7 +151,8 @@ fn live_child_grant(
                     "delegated execution exceeds the current Subject ceiling",
                 ));
             }
-        let origin = RuntimeWorkerRef::new(&current.origin_runtime_id, &current.origin_worker_id);
+            let origin =
+                RuntimeWorkerRef::new(&current.origin_runtime_id, &current.origin_worker_id);
             match store
                 .worker_execution_grant(&origin.runtime_id, &origin.worker_id)
                 .map_err(subjektiv_store_error)?
@@ -128,7 +176,8 @@ fn live_child_grant(
                     current = parent;
                 }
                 None => {
-                if current_body_subject(api, &origin)?.as_deref() != Some(grant.subject_id.as_str())
+                    if self.current_body_subject(&origin)?.as_deref()
+                        != Some(grant.subject_id.as_str())
                     {
                         return Err(denied(
                             "delegation origin is no longer the current Subject body",
@@ -143,6 +192,116 @@ fn live_child_grant(
         ))
     }
 
+    fn subject_permissions(
+        &self,
+        worker: &RuntimeWorkerRef,
+    ) -> ApiResult<Option<Vec<ScopedPermission>>> {
+        if let Some(grant) = self.live_child_grant(worker)? {
+            return Ok(Some(grant.permissions));
+        }
+        let Some(subject) = self.current_body_subject(worker)? else {
+            return Ok(None);
+        };
+        Ok(Some(
+            self.policy(&subject)?
+                .map(|policy| policy.execution)
+                .unwrap_or_default(),
+        ))
+    }
+
+    /// Saved individual-Worker permissions remain unchanged. Their canonical
+    /// projection authorizes operation use only; without an explicit current
+    /// WorkerOperation capability, the original exact target still restricts
+    /// Backend control. Neither covers nor delegation ceilings are widened.
+    fn authorize_control_use(
+        &self,
+        controller: &RuntimeWorkerRef,
+        target: &RuntimeWorkerRef,
+        permission: &str,
+    ) -> ApiResult<()> {
+        self.require_not_deleted(controller)?;
+        if self
+            .store
+            .worker_registry_projection(&self.workspace_id, controller)?
+            .is_some_and(|record| record.job.is_some())
+        {
+            return Err(denied("Job domain grants do not authorize Worker control"));
+        }
+        let Some(permissions) = self.subject_permissions(controller)? else {
+            return Ok(()); // Preserve ordinary Workers' existing Backend contract.
+        };
+        let operation = if permission == "notify" {
+            "worker.messaging".to_owned()
+        } else {
+            format!("worker.{permission}")
+        };
+        let requested = ScopedPermission {
+            operation: operation.clone(),
+            resource: PermissionResource::WorkerOperation {
+                workspace_id: self.workspace_id.clone(),
+            },
+        };
+        if permissions.iter().any(|granted| granted.covers(&requested)) {
+            return Ok(());
+        }
+        if permissions.iter().any(|granted| {
+            granted.validate().is_ok()
+                && canonical_worker_operation(&granted.operation) == Some(operation.as_str())
+                && matches!(&granted.resource, PermissionResource::Worker { runtime_id, worker_id }
+                    if runtime_id == &target.runtime_id && worker_id == &target.worker_id)
+        }) {
+            return Ok(());
+        }
+        Err(denied(format!(
+            "Worker operation {operation} is not granted for this target"
+        )))
+    }
+
+    pub(super) fn authorize_control_permission(
+        &self,
+        controller: &RuntimeWorkerRef,
+        target: &RuntimeWorkerRef,
+        permission: &str,
+    ) -> Result<WorkerControlGrantRecord> {
+        self.authorize_control_use(controller, target, permission)
+            .map_err(|error| error.error)?;
+        let grant = self
+            .store
+            .get_active_worker_control_grant(&self.workspace_id, controller, target)?
+            .ok_or_else(|| Error::UnknownWorker {
+                worker: target.clone(),
+            })?;
+        if !grant
+            .permissions
+            .iter()
+            .any(|candidate| candidate == permission)
+        {
+            return Err(Error::InvalidInput(format!(
+                "worker control permission {permission} was not granted"
+            )));
+        }
+        Ok(grant)
+    }
+}
+
+fn policy(api: &WorkspaceApi, subject_id: &str) -> ApiResult<Option<SubjectPermissions>> {
+    WorkerAuthorityView::from(api).policy(subject_id)
+}
+
+fn current_body_subject(
+    api: &WorkspaceApi,
+    worker: &RuntimeWorkerRef,
+) -> ApiResult<Option<String>> {
+    WorkerAuthorityView::from(api).current_body_subject(worker)
+}
+
+fn live_child_grant(
+    api: &WorkspaceApi,
+    worker: &RuntimeWorkerRef,
+) -> ApiResult<Option<WorkerExecutionGrant>> {
+    WorkerAuthorityView::from(api).live_child_grant(worker)
+}
+
 /// Job domain capabilities attenuate this policy; they are never non-Subject
 /// compatibility. A missing frozen origin cannot mint new operation permissions.
 fn job_permissions(
@@ -291,27 +450,12 @@ pub(super) fn worker_permissions(
     api: &WorkspaceApi,
     worker: &RuntimeWorkerRef,
 ) -> ApiResult<Option<Vec<ScopedPermission>>> {
-    if subjektiv_deletion::deleting_worker(api, worker)?
-        || open_subjektiv_store(api)?
-            .worker_subject_deleted(&worker.runtime_id, &worker.worker_id)
-            .map_err(subjektiv_store_error)?
-    {
-        return Err(denied("Worker belonged to an explicitly deleted Subject"));
-    }
+    let authority = WorkerAuthorityView::from(api);
+    authority.require_not_deleted(worker)?;
     if let Some(permissions) = job_permissions(api, worker)? {
         return Ok(Some(permissions));
     }
-    if let Some(grant) = live_child_grant(api, worker)? {
-        return Ok(Some(grant.permissions));
-    }
-    let Some(subject) = current_body_subject(api, worker)? else {
-        return Ok(None);
-    };
-    Ok(Some(
-        policy(api, &subject)?
-            .map(|policy| policy.execution)
-            .unwrap_or_default(),
-    ))
+    authority.subject_permissions(worker)
 }
 
 pub(super) fn authorize(
@@ -383,6 +527,20 @@ pub(super) fn authorize_scope(
     let Some(permissions) = worker_permissions(api, worker)? else {
         return Ok(None);
     };
+    // Read-only migration of saved target-scoped permissions. Do not mutate the
+    // assignment/grant or use this projection for delegation; Backend separately
+    // checks the original exact target at every control effect.
+    if matches!(&needed.resource, PermissionResource::WorkerOperation { workspace_id }
+        if workspace_id == api.workspace_id())
+        && permissions.iter().any(|permission| {
+            permission.validate().is_ok()
+                && canonical_worker_operation(&permission.operation)
+                    == Some(needed.operation.as_str())
+                && matches!(permission.resource, PermissionResource::Worker { .. })
+        })
+    {
+        return Ok(Some(needed.resource));
+    }
     for permission in &permissions {
         if permission.operation == needed.operation
             && resources_cover(api, &permission.resource, &needed.resource)?
@@ -518,9 +676,59 @@ pub(super) fn bind_reserved_grant(
 ) -> ApiResult<()> {
     let mut bound = grant.clone();
     bound.worker_id = worker_id.to_owned();
-    Ok(open_subjektiv_store(api)?
+    open_subjektiv_store(api)?
         .bind_worker_execution_grant(&bound)
-        .map_err(subjektiv_store_error)?)
+        .map_err(subjektiv_store_error)?;
+    Ok(())
+}
+
+/// Persist only the confirmed execution delegation's reply direction, never an
+/// inferred inverse of an ordinary A -> B control grant. A reservation is not a
+/// Worker registry row: this runs after normal creation has committed that row.
+/// Initial Runtime execution may already have started; replies are unavailable
+/// until this write, and initial input is not queued or redispatched to hide that
+/// window. Spawn replay repairs a failed write under the same allocation identity
+/// without replacing the Worker or reissuing a revoked saved grant.
+pub(super) fn persist_reply_grant(
+    api: &WorkspaceApi,
+    controller: &RuntimeWorkerRef,
+    child: &RuntimeWorkerRef,
+    allocation_key: &str,
+) -> ApiResult<()> {
+    let authority = WorkerAuthorityView::from(api);
+    authority.require_not_deleted(child)?;
+    let Some(bound) = authority.live_child_grant(child)? else {
+        return Ok(()); // Ordinary Workers do not acquire reverse control.
+    };
+    let origin = RuntimeWorkerRef::new(&bound.origin_runtime_id, &bound.origin_worker_id);
+    if &origin != controller {
+        return Err(denied(
+            "reply grant origin does not match authenticated spawn source",
+        ));
+    }
+    if api
+        .store
+        .get_worker_registry(api.workspace_id(), child)?
+        .is_none()
+    {
+        return Err(denied(
+            "reply grant requires the committed Worker registry identity",
+        ));
+    }
+    api.store
+        .create_worker_control_grant(&WorkerControlGrantRecord {
+            workspace_id: api.workspace_id().into(),
+            grant_id: new_id("wcg"),
+            controller: child.clone(),
+            subject: origin,
+            relation: "reply".into(),
+            origin: "execution_grant".into(),
+            permissions: vec!["notify".into()],
+            operation_id: format!("worker-reply:{allocation_key}"),
+            created_at: Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
+            revoked_at: None,
+        })?;
+    Ok(())
 }
 
 /// Host-owned binding for an already authenticated Backend operation. No model

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

@@ -9,6 +9,10 @@ use super::*;
 mod chat_tests;
 #[path = "subject_job_model_supply_tests.rs"]
 mod subject_job_model_supply;
+#[path = "worker_messaging_connected_tests.rs"]
+mod worker_messaging_connected;
+#[path = "worker_operation_authorization_tests.rs"]
+mod worker_operation_authorization;
 use agen::tool::{Tool, ToolExecutionContext, ToolOutput};
 use server_api::authorization::{
     PermissionResource, ScopedPermission, SubjectPermissions, SubjectPermissionsUpdate,
@@ -570,7 +574,7 @@ fn concrete_resource_permissions(f: &Fixture) -> Vec<ScopedPermission> {
             },
         ),
         (
-            "worker.observe",
+            "subworker.observe",
             PermissionResource::SubWorker {
                 runtime_id: f.source.runtime_id.clone(),
                 worker_id: f.source.worker_id.clone(),
@@ -2604,12 +2608,34 @@ impl Drop for ConnectedRuntimeServer {
     }
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
-async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_read_but_not_write()
-{
-    use worker_runtime::working_directory::WorkingDirectoryMaterializer;
+const CONNECTED_RUNTIME: &str = "connected-subject-authorization";
 
-    const RUNTIME: &str = "connected-subject-authorization";
+/// Shared signed Backend/Runtime connection. Only the supplied model is replaced;
+/// temporary roots, request authentication and Controller/tool dispatch stay real.
+struct ConnectedWorkerHarness {
+    api: WorkspaceApi,
+    runtime: worker_runtime::Runtime,
+    factory: worker_runtime::worker_backend::ProfileRuntimeWorkerFactory,
+    identity: RuntimeIdentityMaterial,
+    backend_url: String,
+    owner: String,
+    materializer: worker_runtime::working_directory::RuntimeGitMaterializer,
+    authorizations: Arc<Mutex<Vec<(Value, StatusCode)>>>,
+    runtime_errors: Arc<Mutex<Vec<String>>>,
+    _backend_server: AuthorizationCallbackServer,
+    _runtime_server: ConnectedRuntimeServer,
+    _dir: tempfile::TempDir,
+}
+impl ConnectedWorkerHarness {
+    async fn start<M: agen::llm_client::LlmClient + Clone + 'static>(model: M) -> Self {
+        Self::start_with_layer(model, |router| router).await
+    }
+
+    async fn start_with_layer<M: agen::llm_client::LlmClient + Clone + 'static>(
+        model: M,
+        app_layer: impl FnOnce(Router) -> Router,
+    ) -> Self {
+        const RUNTIME: &str = CONNECTED_RUNTIME;
         let dir = tempfile::tempdir().unwrap();
         init_clean_git_workspace(dir.path());
         let backend_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -2621,8 +2647,6 @@ async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_r
         let materializer = worker_runtime::working_directory::RuntimeGitMaterializer::new(
             dir.path().join("connected-workdirs"),
         );
-    let (completed, mut completion) = tokio::sync::mpsc::channel(4);
-    let model = ConnectedAuthorizationModel { completed };
         let factory = worker_runtime::worker_backend::ProfileRuntimeWorkerFactory::new(
             dir.path().join("connected-profiles"),
         )
@@ -2637,7 +2661,8 @@ async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_r
         )
         .with_runtime_store_dir(&runtime_root)
         .with_spawn_model_client_factory(move || Box::new(model.clone()));
-    let execution = worker_runtime::worker_backend::WorkerRuntimeExecutionBackend::new(factory)
+        let execution =
+            worker_runtime::worker_backend::WorkerRuntimeExecutionBackend::new(factory.clone())
                 .unwrap()
                 .with_working_directory_materializer(materializer.clone());
         let runtime = worker_runtime::Runtime::with_fs_store_and_execution_backend(
@@ -2705,7 +2730,10 @@ async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_r
                     key_id: workspace_identity.key_id.clone(),
                     algorithm: workspace_identity.algorithm.clone(),
                     public_key: workspace_identity.public_key.clone().unwrap(),
-                public_key_fingerprint: workspace_identity.public_key_fingerprint.clone().unwrap(),
+                    public_key_fingerprint: workspace_identity
+                        .public_key_fingerprint
+                        .clone()
+                        .unwrap(),
                     trust_id: "connected-authorization-workspace-trust".into(),
                     state: worker_runtime::workspace_issuer::WorkspaceIssuerTrustState::Active,
                     registered_at_unix: 1,
@@ -2719,7 +2747,7 @@ async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_r
         .unwrap();
         let runtime_errors = Arc::new(Mutex::new(Vec::<String>::new()));
         let captured_errors = runtime_errors.clone();
-    let _runtime_server = ConnectedRuntimeServer::start(
+        let runtime_server = ConnectedRuntimeServer::start(
             runtime_listener,
             worker_runtime::http_server::runtime_http_router_with_workspace_auth(
                 runtime.clone(),
@@ -2734,7 +2762,9 @@ async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_r
                     let errors = captured_errors.clone();
                     async move {
                         let response = next.run(request).await;
-                    if response.status().is_success() {
+                        if !response.status().is_client_error()
+                            && !response.status().is_server_error()
+                        {
                             return response;
                         }
                         let (parts, body) = response.into_parts();
@@ -2759,12 +2789,14 @@ async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_r
                     display_name: RUNTIME.into(),
                     base_url: runtime_url,
                     bearer_token: None,
-                workspace_authorization: Some(crate::hosts::WorkspaceRuntimeAuthorization::new(
+                    workspace_authorization: Some(
+                        crate::hosts::WorkspaceRuntimeAuthorization::new(
                             api.store.clone(),
                             api.signing_identities.clone(),
                             backend_url.clone(),
                             binding,
-                )),
+                        ),
+                    ),
                     strict_public_egress: false,
                     cached_worker_creation_available: true,
                     cached_os: "linux".into(),
@@ -2777,7 +2809,6 @@ async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_r
             )
             .unwrap(),
         );
-    install_subject_profiles(&api, "workspace_write");
         seed_test_delivery_binding(&api);
 
         let authorizations = Arc::new(Mutex::new(Vec::<(Value, StatusCode)>::new()));
@@ -2803,9 +2834,44 @@ async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_r
                 }
             },
         ));
-    let _backend_server = AuthorizationCallbackServer::start(backend_listener, app);
+        let backend_server = AuthorizationCallbackServer::start(backend_listener, app_layer(app));
         let owner =
-        seed_test_workspace_owner_token(api.store.as_ref(), "connected-authorization-owner").await;
+            seed_test_workspace_owner_token(api.store.as_ref(), "connected-authorization-owner")
+                .await;
+
+        Self {
+            api,
+            runtime,
+            factory,
+            identity,
+            backend_url,
+            owner,
+            materializer,
+            authorizations,
+            runtime_errors,
+            _backend_server: backend_server,
+            _runtime_server: runtime_server,
+            _dir: dir,
+        }
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_read_but_not_write()
+{
+    use worker_runtime::working_directory::WorkingDirectoryMaterializer;
+
+    const RUNTIME: &str = CONNECTED_RUNTIME;
+    let (completed, mut completion) = tokio::sync::mpsc::channel(4);
+    let harness = ConnectedWorkerHarness::start(ConnectedAuthorizationModel { completed }).await;
+    let api = &harness.api;
+    let identity = &harness.identity;
+    let backend_url = &harness.backend_url;
+    let owner = &harness.owner;
+    let materializer = &harness.materializer;
+    let authorizations = &harness.authorizations;
+    let runtime_errors = &harness.runtime_errors;
+    install_subject_profiles(api, "workspace_write");
     let (status, created_workdir) = response(&api, human_request(&owner, "POST",
         &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"),
         Some(json!({"runtime_id":RUNTIME, "repository_key":"test-repository", "selector":"HEAD", "operation_id":"connected-authorization-workdir"})))).await;

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

@@ -0,0 +1,637 @@
+//! Public Worker tools over the shared signed Backend/Runtime connection. The
+//! external model alone is scripted; permissions API, Controller, tool engine,
+//! control grants, notification admission and canonical Session storage are real.
+//! No developer model credentials or live Runtime state are used.
+use super::*;
+use agen::llm_client::{
+    event::{Event, StopReason},
+    types::Item,
+};
+use worker_runtime::worker_backend::RuntimeWorkerFactory;
+
+const ORIGIN: &str = "ROUNDTRIP_ORIGIN";
+const PING: &str = "ROUNDTRIP_QUESTION";
+const REPLY: &str = "ROUNDTRIP_PROGRESS_RESULT";
+const EARLY: &str = "ROUNDTRIP_INITIAL_CHILD";
+
+fn operation(operation: &str) -> ScopedPermission {
+    ScopedPermission {
+        operation: operation.into(),
+        resource: PermissionResource::WorkerOperation {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+        },
+    }
+}
+fn collection(operation: &str) -> ScopedPermission {
+    ScopedPermission {
+        operation: operation.into(),
+        resource: PermissionResource::Workspace {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+        },
+    }
+}
+fn result(request: &agen::llm_client::Request, id: &str) -> Option<Value> {
+    request.items.iter().find_map(|item| match item {
+        Item::ToolResult {
+            call_id,
+            is_error,
+            content,
+            summary,
+            ..
+        } if call_id == id => {
+            assert!(!is_error, "public tool {id}: {summary}; {content:?}");
+            Some(serde_json::from_str(content.as_deref().expect("public JSON result")).unwrap())
+        }
+        _ => None,
+    })
+}
+fn call(id: &str, name: &str, input: Value) -> Vec<Event> {
+    vec![
+        Event::tool_use_start(0, id, name),
+        Event::tool_input_delta(0, input.to_string()),
+        Event::tool_use_stop(0),
+    ]
+}
+fn done() -> Vec<Event> {
+    vec![
+        Event::text_block_start(0),
+        Event::text_delta(0, "roundtrip observed"),
+        Event::text_block_stop(0, Some(StopReason::EndTurn)),
+    ]
+}
+
+struct RepairSignals {
+    early_observation: tokio::sync::mpsc::Sender<bool>,
+    spawn_refusal: tokio::sync::mpsc::Sender<()>,
+    allow_replay: tokio::sync::Notify,
+    origin: Mutex<Option<RuntimeWorkerRef>>,
+    origin_ready: tokio::sync::Notify,
+}
+#[derive(Clone)]
+struct MessagingModel {
+    child_permissions: Arc<Mutex<Vec<ScopedPermission>>>,
+    workdir: Arc<Mutex<Option<String>>>,
+    completed: tokio::sync::mpsc::Sender<Value>,
+    child_replied: Arc<tokio::sync::Notify>,
+    repair: Option<Arc<RepairSignals>>,
+}
+#[async_trait::async_trait]
+impl agen::llm_client::LlmClient for MessagingModel {
+    fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
+        Box::new(self.clone())
+    }
+    async fn stream(
+        &self,
+        request: agen::llm_client::Request,
+    ) -> std::result::Result<agen::llm_client::client::ResponseStream, agen::llm_client::ClientError>
+    {
+        let messages = request
+            .items
+            .iter()
+            .filter_map(|item| match item {
+                Item::Message { content, .. } => Some(serde_json::to_string(content).unwrap()),
+                _ => None,
+            })
+            .collect::<Vec<_>>()
+            .join("\n");
+        let origin = messages.contains(ORIGIN);
+        let events = if origin {
+            if let Some(stopped) = result(&request, "roundtrip-stop") {
+                assert_eq!(stopped["state"], "accepted", "{stopped}");
+                let child = result(&request, "roundtrip-spawn").unwrap();
+                if let Some(is_error) = request.items.iter().find_map(|item| match item {
+                    Item::ToolResult {
+                        call_id, is_error, ..
+                    } if call_id == "stopped-notify" => Some(*is_error),
+                    _ => None,
+                }) {
+                    assert!(
+                        is_error,
+                        "Notify to stopped child must refuse, never restore or claim success"
+                    );
+                    let _ = self
+                        .completed
+                        .try_send(json!({"spawn":child,"messages":messages}));
+                    done()
+                } else {
+                    call(
+                        "stopped-notify",
+                        "WorkerNotify",
+                        json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]},"content":"MUST_NOT_REVIVE_STOPPED_CHILD"}),
+                    )
+                }
+            } else if messages.contains(REPLY) {
+                let child = result(&request, "roundtrip-spawn").unwrap();
+                call(
+                    "roundtrip-stop",
+                    "WorkerStop",
+                    json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]},"reason":"connected roundtrip completed"}),
+                )
+            } else if result(&request, "roundtrip-ping").is_some() {
+                // Hold this model boundary while the real child replies to a Busy
+                // origin. This is a deterministic external-model barrier, not a
+                // replacement for notification admission or Controller scheduling.
+                tokio::time::timeout(
+                    std::time::Duration::from_secs(30),
+                    self.child_replied.notified(),
+                )
+                .await
+                .expect("child public reply result");
+                done()
+            } else if let Some(child) = result(&request, "roundtrip-spawn") {
+                if result(&request, "roundtrip-observe").is_some() {
+                    call(
+                        "roundtrip-ping",
+                        "WorkerNotify",
+                        json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]},"content":PING}),
+                    )
+                } else {
+                    // The origin has observe, not worker.list. A known target must
+                    // remain readable without collection discovery authority.
+                    call(
+                        "roundtrip-observe",
+                        "ViewSessionOverview",
+                        json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]}}),
+                    )
+                }
+            } else {
+                let call_id = if let Some(repair) = &self.repair {
+                    if let Some((is_error, summary)) =
+                        request.items.iter().find_map(|item| match item {
+                            Item::ToolResult {
+                                call_id,
+                                is_error,
+                                summary,
+                                ..
+                            } if call_id == "roundtrip-spawn-first" => Some((*is_error, summary)),
+                            _ => None,
+                        })
+                    {
+                        assert!(
+                            is_error,
+                            "grant-save fault must not report successful Spawn: {summary}"
+                        );
+                        let _ = repair.spawn_refusal.try_send(());
+                        tokio::time::timeout(
+                            std::time::Duration::from_secs(30),
+                            repair.allow_replay.notified(),
+                        )
+                        .await
+                        .expect("explicit original-operation replay authorization");
+                        "roundtrip-spawn"
+                    } else {
+                        "roundtrip-spawn-first"
+                    }
+                } else {
+                    "roundtrip-spawn"
+                };
+                let mut input = json!({"runtime_id":CONNECTED_RUNTIME,"profile":"project:messaging-peer","operation_id":"public-roundtrip-spawn","display_name":"Notify-only delegated peer","initial_submit":if self.repair.is_some() {vec![protocol::Segment::Text {content:EARLY.into()}]} else {vec![]},"execution_grant":{"permissions":self.child_permissions.lock().unwrap().clone(),"redelegation":[]}});
+                if let Some(workdir) = self.workdir.lock().unwrap().as_ref() {
+                    input["working_directory_id"] = json!(workdir);
+                }
+                call(call_id, "WorkerSpawn", input)
+            }
+        } else if self.repair.is_some() && !messages.contains(PING) {
+            assert!(
+                messages.contains(EARLY),
+                "early child started through original initial Submit"
+            );
+            if let Some(available) = request.items.iter().find_map(|item| match item {
+                Item::ToolResult {
+                    call_id,
+                    is_error,
+                    content,
+                    ..
+                } if call_id == "early-origin-list" => {
+                    let available = !is_error
+                        && content.as_deref().is_some_and(|text| {
+                            serde_json::from_str::<Value>(text).unwrap()["items"]
+                                .as_array()
+                                .is_some_and(|items| !items.is_empty())
+                        });
+                    Some(available)
+                }
+                _ => None,
+            }) {
+                assert!(
+                    !available,
+                    "an unpersisted reply grant cannot be discovered as authority"
+                );
+                let repair = self.repair.as_ref().unwrap();
+                if let Some(is_error) = request.items.iter().find_map(|item| match item {
+                    Item::ToolResult {
+                        call_id, is_error, ..
+                    } if call_id == "early-denied-notify" => Some(*is_error),
+                    _ => None,
+                }) {
+                    assert!(
+                        is_error,
+                        "missing reply grant must deny public Notify, not report acceptance"
+                    );
+                    let _ = repair.early_observation.try_send(available);
+                    done()
+                } else {
+                    tokio::time::timeout(
+                        std::time::Duration::from_secs(30),
+                        repair.origin_ready.notified(),
+                    )
+                    .await
+                    .expect("test obtains original origin identity");
+                    // The harness supplies only a target identity, never authority.
+                    // Discovery was empty; even knowing the ID cannot authorize Notify.
+                    let origin = repair.origin.lock().unwrap().clone().unwrap();
+                    call(
+                        "early-denied-notify",
+                        "WorkerNotify",
+                        json!({"subject":{"kind":"runtime_worker","runtime_id":origin.runtime_id,"worker_id":origin.worker_id},"content":"MUST_NOT_ACCEPT_EARLY_REPLY"}),
+                    )
+                }
+            } else {
+                call("early-origin-list", "WorkerList", json!({}))
+            }
+        } else {
+            assert!(
+                messages.contains(PING),
+                "child starts from real Idle Notify, not hidden initial Submit: {messages}"
+            );
+            if let Some(reply) = result(&request, "roundtrip-reply") {
+                assert_eq!(reply["state"], "accepted", "{reply}");
+                self.child_replied.notify_one();
+                done()
+            } else if let Some(list) = result(&request, "roundtrip-origin-list") {
+                let origins = list["items"]
+                    .as_array()
+                    .unwrap()
+                    .iter()
+                    .filter(|item| item["relation"] == "reply")
+                    .collect::<Vec<_>>();
+                assert_eq!(
+                    origins.len(),
+                    1,
+                    "public discovery must expose only the confirmed reply origin: {list}"
+                );
+                assert_eq!(origins[0]["permissions"], json!(["notify"]));
+                call(
+                    "roundtrip-reply",
+                    "WorkerNotify",
+                    json!({"subject":origins[0]["subject"],"content":REPLY}),
+                )
+            } else {
+                call("roundtrip-origin-list", "WorkerList", json!({}))
+            }
+        };
+        Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok))))
+    }
+}
+
+fn install_messaging_profiles(api: &WorkspaceApi) {
+    let state = api
+        .config_store
+        .load_workspace_config(TEST_WORKSPACE_ID)
+        .unwrap()
+        .unwrap();
+    let main = state
+        .snapshot
+        .get(&config_source::VirtualPath::parse("main.dcdl").unwrap())
+        .unwrap();
+    let content = r#"{ profile = { default_profile = "project:messaging-origin"; entries = [
+        { selector = "project:messaging-origin"; label = "Messaging origin"; description = "Public delegation boundary";
+          profile = (import "$builtin/profiles/companion.dcdl") // { slug = "messaging-origin"; scope = "workspace_read"; delegation_scope = "workspace_read"; feature = { worker = { enabled = true; direct_spawn = true; }; }; };
+        },
+        { selector = "project:messaging-peer"; label = "Messaging peer"; description = "Notify-only delegated peer";
+          profile = (import "$builtin/profiles/companion.dcdl") // { slug = "messaging-peer"; scope = "workspace_read"; feature = { subjektiv = { enabled = false; }; worker = { enabled = true; direct_spawn = false; }; }; };
+        }
+    ]; }; }"#;
+    commit_workspace_config_tree(
+        api,
+        TEST_WORKSPACE_ID,
+        &ConfigCommitRequest {
+            base_digest: state.snapshot.digest.clone(),
+            entrypoints: state.contract.entrypoints.clone(),
+            changes: vec![config_source::ConfigTreeChange::Update {
+                path: main.path.clone(),
+                expected_digest: main.content_digest.clone(),
+                content: content.into(),
+            }],
+        },
+    )
+    .unwrap();
+}
+
+async fn connected_roundtrip(explicit_workdir: bool, repair: bool) {
+    use worker_runtime::working_directory::WorkingDirectoryMaterializer;
+    let (completed, mut completion) = tokio::sync::mpsc::channel(2);
+    let permissions = Arc::new(Mutex::new(vec![
+        operation("worker.messaging"),
+        collection("worker.list"),
+    ]));
+    let workdir = Arc::new(Mutex::new(None));
+    let (early_observation, mut early) = tokio::sync::mpsc::channel(1);
+    let (spawn_refusal, mut refused) = tokio::sync::mpsc::channel(1);
+    let repair_signals = repair.then(|| {
+        Arc::new(RepairSignals {
+            early_observation,
+            spawn_refusal,
+            allow_replay: tokio::sync::Notify::new(),
+            origin: Mutex::new(None),
+            origin_ready: tokio::sync::Notify::new(),
+        })
+    });
+    let harness = ConnectedWorkerHarness::start(MessagingModel {
+        child_permissions: permissions.clone(),
+        workdir: workdir.clone(),
+        completed,
+        child_replied: Arc::new(tokio::sync::Notify::new()),
+        repair: repair_signals.clone(),
+    })
+    .await;
+    let api = &harness.api;
+    install_messaging_profiles(api);
+    if explicit_workdir {
+        let (status, created) = response(api, human_request(&harness.owner, "POST", &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"), Some(json!({"runtime_id":CONNECTED_RUNTIME,"repository_key":"test-repository","selector":"HEAD","operation_id":"public-messaging-workdir"})))).await;
+        assert_eq!(status, StatusCode::CREATED, "{created}");
+        let id = created["item"]["working_directory_id"]
+            .as_str()
+            .unwrap()
+            .to_owned();
+        permissions.lock().unwrap().push(ScopedPermission {
+            operation: "filesystem.read".into(),
+            resource: PermissionResource::Workdir {
+                working_directory_id: id.clone(),
+                path: ".".into(),
+                recursive: true,
+            },
+        });
+        *workdir.lock().unwrap() = Some(id);
+    }
+    let store = open_subjektiv_store(api).unwrap();
+    let subject = store
+        .create_subject(crate::subjektiv::SubjectRole::new("public-messaging").unwrap())
+        .unwrap()
+        .id;
+    let mut execution = [
+        "session.record",
+        "memory.resident.read",
+        "memory.resident.supply",
+        "journal.startup.read",
+        "journal.startup.supply",
+    ]
+    .into_iter()
+    .map(|op| subject_permission(&subject, op))
+    .collect::<Vec<_>>();
+    execution.extend([
+        collection("worker.create"),
+        operation("worker.messaging"),
+        operation("worker.stop"),
+        operation("worker.observe"),
+    ]);
+    let (status, assigned) = response(api, human_request(&harness.owner,"POST",&format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/permissions"),Some(json!({"operation_id":"public-messaging-policy","expected_change_id":null,"execution":execution,"delegation":permissions.lock().unwrap().clone()})))).await;
+    assert_eq!(status, StatusCode::OK, "{assigned}");
+    let before = store.subject_permissions(&subject).unwrap().unwrap();
+    if repair {
+        api.config_store.with_conn(|connection| {
+            connection.execute_batch("CREATE TRIGGER t803_reply_save_fault BEFORE INSERT ON worker_control_grants WHEN NEW.relation='reply' BEGIN SELECT RAISE(ABORT,'injected reply grant save failure'); END;")?;
+            Ok(())
+        }).unwrap();
+    }
+    let Json(source) = create_workspace_worker(
+        State(api.clone()),
+        HeaderMap::new(),
+        Json(CreateWorkspaceWorkerRequest {
+            runtime_id: CONNECTED_RUNTIME.into(),
+            display_name: "Public WorkerSpawn Subject origin".into(),
+            singleton_key: None,
+            profile: Some("project:messaging-origin".into()),
+            ticket_assignment: None,
+            initial_submit: vec![protocol::Segment::Text {
+                content: ORIGIN.into(),
+            }],
+            workdir_attachments: vec![],
+            feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
+                subjektiv: Some(server_api::SubjektivWorkerConnectionRequest {
+                    subject_id: subject.clone(),
+                }),
+            },
+            control_operation_id: None,
+            execution_grant: None,
+        }),
+    )
+    .await
+    .unwrap_or_else(|error| {
+        panic!(
+            "origin: {error:?}; Runtime: {:?}",
+            harness.runtime_errors.lock().unwrap()
+        )
+    });
+    if let Some(signals) = &repair_signals {
+        *signals.origin.lock().unwrap() =
+            Some(RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id));
+        signals.origin_ready.notify_one();
+        assert!(
+            !tokio::time::timeout(std::time::Duration::from_secs(60), early.recv())
+                .await
+                .expect("earliest child public discovery")
+                .unwrap()
+        );
+        tokio::time::timeout(std::time::Duration::from_secs(60), refused.recv())
+            .await
+            .expect("public Spawn save-failure refusal")
+            .unwrap();
+        assert_eq!(
+            harness.runtime.list_workers().unwrap().len(),
+            2,
+            "confirmed original child is retained despite reply-save failure"
+        );
+        api.config_store
+            .with_conn(|connection| {
+                connection.execute_batch("DROP TRIGGER t803_reply_save_fault")?;
+                Ok(())
+            })
+            .unwrap();
+        signals.allow_replay.notify_one();
+    }
+    let observed = tokio::time::timeout(std::time::Duration::from_secs(60), completion.recv())
+        .await
+        .expect("public Spawn/Notify/reply/Stop complete")
+        .expect("model completion evidence");
+    let child = RuntimeWorkerRef::new(
+        observed["spawn"]["runtime_id"].as_str().unwrap(),
+        observed["spawn"]["worker_id"].as_str().unwrap(),
+    );
+    let origin = RuntimeWorkerRef::new(source.runtime_id, source.worker_id);
+    assert!(observed["messages"].as_str().unwrap().contains(REPLY));
+    assert_eq!(
+        store.subject_permissions(&subject).unwrap().unwrap(),
+        before,
+        "allocated ID was never appended to Subject policy"
+    );
+    let forward = api
+        .store
+        .get_active_worker_control_grant(TEST_WORKSPACE_ID, &origin, &child)
+        .unwrap()
+        .unwrap();
+    assert!(forward.permissions.contains(&"stop".into()));
+    let reverse = api
+        .store
+        .get_active_worker_control_grant(TEST_WORKSPACE_ID, &child, &origin)
+        .unwrap()
+        .unwrap();
+    assert_eq!(reverse.permissions, vec!["notify".to_owned()]);
+    let child_ref = worker_runtime::identity::WorkerRef::try_from(&child).unwrap();
+    let snapshot = harness
+        .factory
+        .retained_session_snapshot(&child_ref)
+        .unwrap();
+    let history = serde_json::to_string(&snapshot.snapshot).unwrap();
+    assert!(
+        !history.contains("MUST_NOT_REVIVE_STOPPED_CHILD"),
+        "stopped destination received no notification/mailbox input"
+    );
+    assert_eq!(
+        harness.runtime.worker_detail(&child_ref).unwrap().status,
+        worker_runtime::catalog::WorkerStatus::Stopped
+    );
+    if repair {
+        assert_eq!(
+            snapshot
+                .snapshot
+                .entries
+                .iter()
+                .filter(|item| serde_json::to_string(item).unwrap().contains(EARLY))
+                .count(),
+            1,
+            "original initial input was not redispatched on recovery"
+        );
+        assert_eq!(
+            harness.runtime.list_workers().unwrap().len(),
+            2,
+            "one origin and one child after original-operation replay"
+        );
+    }
+    assert!(
+        history.contains(PING),
+        "Idle notification persisted before model supply"
+    );
+    let source_ref = worker_runtime::identity::WorkerRef::try_from(&origin).unwrap();
+    let origin_history = serde_json::to_string(
+        &harness
+            .factory
+            .retained_session_snapshot(&source_ref)
+            .unwrap()
+            .snapshot,
+    )
+    .unwrap();
+    assert!(
+        !origin_history.contains("MUST_NOT_ACCEPT_EARLY_REPLY"),
+        "unauthorized early Notify caused no origin input"
+    );
+    assert!(
+        origin_history.contains(REPLY),
+        "Busy reply persisted before origin model supply"
+    );
+    assert!(
+        api.store
+            .list_worker_workdir_links(TEST_WORKSPACE_ID, &origin)
+            .unwrap()
+            .is_empty()
+    );
+    let links = api
+        .store
+        .list_worker_workdir_links(TEST_WORKSPACE_ID, &child)
+        .unwrap();
+    assert_eq!(links.len(), usize::from(explicit_workdir));
+    if let Some(id) = workdir.lock().unwrap().as_ref() {
+        assert_eq!(
+            fs::read_to_string(
+                harness
+                    .materializer
+                    .bind_working_directory(id, None)
+                    .unwrap()
+                    .root()
+                    .join("README.md")
+            )
+            .unwrap(),
+            "clean\n"
+        );
+    }
+    api.store
+        .revoke_worker_control_grant(
+            TEST_WORKSPACE_ID,
+            &forward.grant_id,
+            &now_registry_timestamp(),
+        )
+        .unwrap();
+    let (denied, body) = connected_authorization_post(
+        &harness.backend_url,
+        &harness.identity,
+        &origin,
+        &format!(
+            "/api/w/{TEST_WORKSPACE_ID}/worker-control/workers/{}/{}/input",
+            child.runtime_id, child.worker_id
+        ),
+        &json!({"kind":"notify","content":"MUST_NOT_ACCEPT_REVOKED_CONTROL"}),
+    )
+    .await;
+    assert_eq!(
+        denied,
+        StatusCode::NOT_FOUND,
+        "revoked forward control: {body}"
+    );
+    assert_eq!(
+        harness.runtime.worker_detail(&child_ref).unwrap().status,
+        worker_runtime::catalog::WorkerStatus::Stopped
+    );
+    assert!(
+        !serde_json::to_string(
+            &harness
+                .factory
+                .retained_session_snapshot(&child_ref)
+                .unwrap()
+                .snapshot
+        )
+        .unwrap()
+        .contains("MUST_NOT_ACCEPT_REVOKED_CONTROL")
+    );
+    let calls = harness.authorizations.lock().unwrap();
+    for op in [
+        "worker.create",
+        "worker.messaging",
+        "worker.stop",
+        "worker.observe",
+    ] {
+        assert!(
+            calls
+                .iter()
+                .any(|(request, status)| request["operation"] == op && *status == StatusCode::OK),
+            "real signed common authorization for {op}: {calls:?}"
+        );
+    }
+    assert!(
+        calls
+            .iter()
+            .filter(|(request, _)| request["operation"] == "worker.messaging"
+                || request["operation"] == "worker.stop")
+            .all(|(request, _)| request["resource"]["kind"] == "worker_operation")
+    );
+    assert!(
+        browser_ticket_backend(api)
+            .unwrap()
+            .list(Default::default())
+            .unwrap()
+            .is_empty()
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn public_spawn_notify_reply_and_stop_without_ticket_or_attachment() {
+    connected_roundtrip(false, false).await;
+}
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn public_spawn_notify_reply_and_stop_with_explicit_workdir() {
+    connected_roundtrip(true, false).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn public_initial_reply_unavailable_until_original_spawn_replay_repairs_grant() {
+    connected_roundtrip(false, true).await;
+}

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

@@ -0,0 +1,945 @@
+//! Signed-router component tests for operation use and independent directional control.
+//! Subject policy APIs, SQLite grants, Feature installation and public Tools are real.
+//! WorkdirlessFixtureRuntime substitutes execution; Accepted is not Controller processing.
+//! Run with `cargo test -p yoi-workspace-server --lib worker_operation_authorization`.
+use super::*;
+use worker::feature::builtin::manage_worker::manage_worker_feature;
+
+struct OperationFixture {
+    inner: Fixture,
+    stops: Arc<std::sync::atomic::AtomicUsize>,
+}
+
+impl std::ops::Deref for OperationFixture {
+    type Target = Fixture;
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl OperationFixture {
+    async fn new() -> Self {
+        let mut inner = Fixture::new().await;
+        let stops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let counter = stops.clone();
+        inner.runtime.stop_callback = Some(Arc::new(move || {
+            counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        }));
+        inner.api.runtime.register_or_replace(inner.runtime.clone());
+        Self { inner, stops }
+    }
+}
+
+const CONTROL_OPERATIONS: &[(&str, &str)] = &[
+    ("worker.messaging", "notify"),
+    ("worker.send_input", "send_input"),
+    ("worker.cancel", "cancel"),
+    ("worker.stop", "stop"),
+    ("worker.restore", "restore"),
+    ("worker.remove", "remove"),
+    ("worker.observe", "observe"),
+];
+
+fn workspace_permission(operation: &str) -> ScopedPermission {
+    ScopedPermission {
+        operation: operation.into(),
+        resource: PermissionResource::Workspace {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+        },
+    }
+}
+
+fn operation_permission(operation: &str) -> ScopedPermission {
+    ScopedPermission {
+        operation: operation.into(),
+        resource: PermissionResource::WorkerOperation {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+        },
+    }
+}
+
+fn control_permissions() -> Vec<ScopedPermission> {
+    CONTROL_OPERATIONS
+        .iter()
+        .map(|(operation, _)| operation_permission(operation))
+        .collect()
+}
+
+fn target_permission(operation: &str, target: &RuntimeWorkerRef) -> ScopedPermission {
+    ScopedPermission {
+        operation: operation.into(),
+        resource: PermissionResource::Worker {
+            runtime_id: target.runtime_id.clone(),
+            worker_id: target.worker_id.clone(),
+        },
+    }
+}
+
+fn subject_value(worker: &RuntimeWorkerRef) -> Value {
+    json!({"kind":"runtime_worker", "runtime_id":worker.runtime_id, "worker_id":worker.worker_id})
+}
+
+fn peer_grant(
+    f: &OperationFixture,
+    controller: &RuntimeWorkerRef,
+    target: &RuntimeWorkerRef,
+    permissions: &[&str],
+) -> String {
+    let id = Uuid::now_v7().to_string();
+    f.api
+        .store
+        .create_worker_control_grant(&WorkerControlGrantRecord {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+            grant_id: id.clone(),
+            controller: controller.clone(),
+            subject: target.clone(),
+            relation: "peer".into(),
+            origin: "explicit_test_delegation".into(),
+            permissions: permissions
+                .iter()
+                .map(|permission| (*permission).into())
+                .collect(),
+            operation_id: id.clone(),
+            created_at: now_registry_timestamp(),
+            revoked_at: None,
+        })
+        .unwrap();
+    id
+}
+
+async fn backend_operation(
+    f: &OperationFixture,
+    controller: &RuntimeWorkerRef,
+    target: &RuntimeWorkerRef,
+    permission: &str,
+) -> (StatusCode, Value) {
+    let base = format!(
+        "/api/w/{TEST_WORKSPACE_ID}/worker-control/workers/{}/{}",
+        target.runtime_id, target.worker_id
+    );
+    let (path, body) = match permission {
+        "notify" | "send_input" => (
+            format!("{base}/input"),
+            json!({"kind":if permission == "notify" {"notify"} else {"user"}, "content":"authority matrix input"}),
+        ),
+        "restore" => {
+            let observation = f
+                .api
+                .runtime
+                .worker(target)
+                .unwrap()
+                .restore_observation_token
+                .unwrap();
+            (
+                format!("{base}/restore"),
+                json!({"expected_observation_token":observation, "request_id":Uuid::now_v7().to_string()}),
+            )
+        }
+        "observe" => (
+            format!("/api/w/{TEST_WORKSPACE_ID}/worker-observation/session"),
+            subject_value(target),
+        ),
+        "remove" => (
+            format!("/api/w/{TEST_WORKSPACE_ID}/workers/remove"),
+            json!({"target_runtime_id":target.runtime_id,"target_worker_id":target.worker_id,"reason":"authorization matrix"}),
+        ),
+        "cancel" | "stop" => (
+            format!("{base}/{permission}"),
+            json!({"reason":"authorization matrix"}),
+        ),
+        _ => panic!("unknown matrix operation {permission}"),
+    };
+    let mut request = runtime_source_request(
+        &f.identity,
+        Some(&controller.worker_id),
+        "POST",
+        &path,
+        serde_json::to_vec(&body).unwrap(),
+    );
+    if permission == "remove" {
+        let proof =
+            worker_runtime::auth::RuntimeWorkerMutationSourceSigner::from_identity(&f.identity)
+                .issue_worker_remove(
+                    "server-test",
+                    TEST_WORKSPACE_ID,
+                    &controller.worker_id,
+                    &target.runtime_id,
+                    &target.worker_id,
+                    30,
+                )
+                .unwrap();
+        request.headers_mut().insert(
+            worker_runtime::auth::WORKER_MUTATION_SOURCE_PROOF_HEADER,
+            proof.parse().unwrap(),
+        );
+    }
+    tokio::time::timeout(
+        std::time::Duration::from_secs(20),
+        response(&f.api, request),
+    )
+    .await
+    .unwrap_or_else(|_| panic!("Backend {permission} did not complete"))
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct RuntimeEffects {
+    inputs: usize,
+    stops: usize,
+    spawns: usize,
+    restores: usize,
+    workers: usize,
+}
+
+fn effects(f: &OperationFixture) -> RuntimeEffects {
+    RuntimeEffects {
+        inputs: f.runtime.input_requests.lock().unwrap().len(),
+        stops: f.stops.load(std::sync::atomic::Ordering::SeqCst),
+        spawns: f.runtime.spawn_requests().len(),
+        restores: *f.runtime.restore_attempts.lock().unwrap(),
+        workers: f.runtime.workers.lock().unwrap().len(),
+    }
+}
+
+async fn assert_backend_denied(
+    f: &OperationFixture,
+    controller: &RuntimeWorkerRef,
+    target: &RuntimeWorkerRef,
+    expected: Option<StatusCode>,
+) {
+    let before = effects(f);
+    let record = f
+        .api
+        .store
+        .get_worker_registry(TEST_WORKSPACE_ID, target)
+        .unwrap();
+    for (_, permission) in CONTROL_OPERATIONS {
+        let (status, body) = backend_operation(f, controller, target, permission).await;
+        if let Some(expected) = expected {
+            assert_eq!(status, expected, "{permission}: {body}");
+        } else {
+            assert!(status.is_client_error(), "{permission}: {status} {body}");
+        }
+        assert_eq!(
+            effects(f),
+            before,
+            "denied {permission} reached Runtime effects"
+        );
+        assert_eq!(
+            f.api
+                .store
+                .get_worker_registry(TEST_WORKSPACE_ID, target)
+                .unwrap(),
+            record,
+            "denied {permission} mutated target registry"
+        );
+    }
+}
+
+async fn worker_tool_host(
+    root: &Path,
+    client: Arc<dyn WorkspaceClient>,
+) -> worker::Worker<UnusedModel, session_store::FsStore> {
+    worker_tool_host_with_observation(root, client, None).await
+}
+
+async fn worker_tool_host_with_observation(
+    root: &Path,
+    client: Arc<dyn WorkspaceClient>,
+    observation: Option<Arc<dyn worker::feature::contracts::WorkerObservationProvider>>,
+) -> worker::Worker<UnusedModel, session_store::FsStore> {
+    let manifest = worker::WorkerManifest::from_toml(
+        r#"
+        [worker]
+        name = "worker-operation-tools"
+        [model]
+        scheme = "anthropic"
+        model_id = "unused"
+        [engine]
+        [scope]
+        allow = []
+        [feature.workdir_catalog]
+        enabled = false
+        [feature.worker]
+        enabled = true
+    "#,
+    )
+    .unwrap();
+    let engine =
+        agen::Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(
+            UnusedModel,
+        );
+    let mut host = worker::Worker::new(
+        manifest,
+        engine,
+        session_store::FsStore::new(root).unwrap(),
+        worker::WorkerWorkspaceContext::with_client(
+            Some(worker::WorkspaceId::new(TEST_WORKSPACE_ID).unwrap()),
+            client.clone(),
+        ),
+        worker::WorkerFilesystemAuthority::None,
+        manifest::Scope::empty(),
+    )
+    .await
+    .unwrap();
+    let mut features = worker::feature::FeatureRegistryBuilder::new()
+        .with_module(manage_worker_feature(client, None, true));
+    if let Some(provider) = observation {
+        features = features.with_module(
+            worker::feature::builtin::worker_observation::WorkerObservationFeature::new(provider),
+        );
+    }
+    let report = host.install_features(features);
+    assert!(
+        !report.has_errors(),
+        "Worker Feature must install without granting operation use"
+    );
+    host.engine().tool_server_handle().flush_pending();
+    host
+}
+
+fn installed_tool(
+    host: &worker::Worker<UnusedModel, session_store::FsStore>,
+    name: &str,
+) -> Arc<dyn Tool> {
+    host.engine()
+        .tool_server_handle()
+        .get_tool(name)
+        .unwrap_or_else(|| panic!("missing installed public {name}"))
+        .1
+}
+
+fn execute_tool_result(
+    host: &worker::Worker<UnusedModel, session_store::FsStore>,
+    name: &str,
+    input: Value,
+) -> std::result::Result<ToolOutput, agen::tool::ToolError> {
+    let tool = installed_tool(host, name);
+    // The signed fixture transport is synchronous. Leave Tokio's async driver
+    // before executing the public Tool, just as the real blocking adapter does.
+    tokio::task::block_in_place(|| {
+        futures::executor::block_on(
+            tool.execute(&input.to_string(), ToolExecutionContext::direct()),
+        )
+    })
+}
+
+async fn execute_tool(
+    host: &worker::Worker<UnusedModel, session_store::FsStore>,
+    name: &str,
+    input: Value,
+) -> Value {
+    value(execute_tool_result(host, name, input).unwrap_or_else(|error| panic!("{name}: {error}")))
+}
+
+fn tool_arguments(name: &str, target: &RuntimeWorkerRef) -> Value {
+    let mut args = json!({"subject":subject_value(target)});
+    match name {
+        "WorkerSendInput" | "WorkerNotify" => args["content"] = json!("public tool input"),
+        "WorkerRemove" => args["reason"] = json!("must not remove"),
+        "WorkerRestore" => {
+            args["expected_observation_token"] = json!("not-used-before-authorization")
+        }
+        _ => {}
+    }
+    args
+}
+
+// Only the protected Session read is substituted here. Installed observation /
+// control Features and signed Host / Backend target checks remain production code.
+struct CountedObservation(std::sync::atomic::AtomicUsize);
+#[async_trait::async_trait]
+impl worker::feature::contracts::WorkerObservationProvider for CountedObservation {
+    async fn list_worker_sessions(
+        &self,
+    ) -> std::result::Result<
+        Vec<worker::feature::contracts::WorkerObservationSubject>,
+        worker::feature::contracts::WorkerObservationError,
+    > {
+        panic!("known-target capture must not discover the collection")
+    }
+    async fn capture_worker_session(
+        &self,
+        _: &worker::feature::contracts::WorkerObservationSubjectRef,
+    ) -> std::result::Result<
+        worker::feature::contracts::WorkerSessionCapture,
+        worker::feature::contracts::WorkerObservationError,
+    > {
+        self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        Ok(worker::feature::contracts::WorkerSessionCapture {
+            segment_id: "target-session".into(),
+            session: protocol::SessionSnapshot {
+                pending_submissions: Default::default(),
+                entries: vec![protocol::SessionSnapshotEntry {
+                    use_policy: None,
+                    entry_id: "observed-entry".into(),
+                    timestamp: 1,
+                    provenance: protocol::SessionEntryProvenance::LegacyUnknown,
+                    derived_from: vec![],
+                    data: protocol::SessionSnapshotEntryData::Message {
+                        role: protocol::SessionMessageRole::User,
+                        content: vec![protocol::SessionContentPart::Text {
+                            text: "observed input".into(),
+                        }],
+                    },
+                }],
+            },
+        })
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn installed_observation_needs_only_observe_and_target_grant_not_collection_discovery() {
+    let f = OperationFixture::new().await;
+    f.assign(vec![workspace_permission("worker.create")], vec![])
+        .await;
+    let target = f.child("observed-target", vec![]).await;
+    f.assign(vec![operation_permission("worker.observe")], vec![])
+        .await;
+    let provider = Arc::new(CountedObservation(std::sync::atomic::AtomicUsize::new(0)));
+    let client = f.client(&f.source);
+    let host = worker_tool_host_with_observation(
+        &f._dir.path().join("observation-tools"),
+        client.clone(),
+        Some(provider.clone()),
+    )
+    .await;
+    assert!(
+        execute_tool_result(&host, "WorkerList", json!({})).is_err(),
+        "observe never grants collection discovery"
+    );
+    let overview = execute_tool(
+        &host,
+        "ViewSessionOverview",
+        json!({"subject":subject_value(&target)}),
+    )
+    .await;
+    let entry_ref = overview["entries"][0]["entry_ref"].as_str().unwrap();
+    let tools = [
+        (
+            "ViewSessionOverview",
+            json!({"subject":subject_value(&target)}),
+        ),
+        (
+            "SearchSessionEntries",
+            json!({"subject":subject_value(&target), "query":"observed"}),
+        ),
+        (
+            "ReadSessionEntry",
+            json!({"subject":subject_value(&target), "entry_ref":entry_ref}),
+        ),
+    ];
+    client.requests.lock().unwrap().clear();
+    for (name, input) in &tools {
+        execute_tool(&host, name, input.clone()).await;
+    }
+    assert_eq!(provider.0.load(std::sync::atomic::Ordering::SeqCst), 4);
+    assert!(
+        client
+            .requests
+            .lock()
+            .unwrap()
+            .iter()
+            .all(|(_, path)| !path.ends_with("/worker-control/workers")),
+        "known-target public observation must not invoke WorkerList"
+    );
+    let grant = f
+        .api
+        .store
+        .get_active_worker_control_grant(TEST_WORKSPACE_ID, &f.source, &target)
+        .unwrap()
+        .unwrap();
+    for missing in ["operation", "target", "revoked"] {
+        match missing {
+            "operation" => {
+                f.assign(vec![], vec![]).await;
+            }
+            "target" => {
+                f.assign(vec![operation_permission("worker.observe")], vec![])
+                    .await;
+                f.api
+                    .store
+                    .revoke_worker_control_grant(
+                        TEST_WORKSPACE_ID,
+                        &grant.grant_id,
+                        &now_registry_timestamp(),
+                    )
+                    .unwrap();
+            }
+            "revoked" => {}
+            _ => unreachable!(),
+        }
+        for (name, input) in &tools {
+            let mut input = input.clone();
+            if missing == "target" {
+                input["subject"]["worker_id"] = json!("unknown-target");
+            }
+            let error = execute_tool_result(&host, name, input)
+                .unwrap_err()
+                .to_string();
+            assert!(
+                error.contains("not found or is not accessible"),
+                "{missing} {name}: {error}"
+            );
+        }
+        assert_eq!(
+            provider.0.load(std::sync::atomic::Ordering::SeqCst),
+            4,
+            "denied {missing} must not read or supply protected Session data"
+        );
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn backend_control_grant_does_not_replace_exact_operation_use_permission() {
+    let f = OperationFixture::new().await;
+    f.assign(vec![workspace_permission("worker.create")], vec![])
+        .await;
+    let target = f.child("permission-matrix-target", vec![]).await;
+    assert!(
+        f.api
+            .store
+            .get_active_worker_control_grant(TEST_WORKSPACE_ID, &f.source, &target)
+            .unwrap()
+            .is_some()
+    );
+    assert_backend_denied(&f, &f.source, &target, Some(StatusCode::FORBIDDEN)).await;
+    // Old collection/Feature permissions and another Workspace never authorize
+    // individual current Worker operations, even with a real target grant.
+    let mut wrong_workspace = control_permissions();
+    for permission in &mut wrong_workspace {
+        permission.resource = PermissionResource::WorkerOperation {
+            workspace_id: "another-workspace".into(),
+        };
+    }
+    for mut permissions in [
+        CONTROL_OPERATIONS
+            .iter()
+            .map(|(operation, _)| workspace_permission(operation))
+            .collect(),
+        vec![activation("worker")],
+        wrong_workspace,
+    ] {
+        permissions.push(workspace_permission("worker.create"));
+        f.assign(permissions, vec![]).await;
+        assert_backend_denied(&f, &f.source, &target, Some(StatusCode::FORBIDDEN)).await;
+    }
+    f.assign(vec![operation_permission("worker.messaging")], vec![])
+        .await;
+    assert_eq!(
+        backend_operation(&f, &f.source, &target, "notify").await.0,
+        StatusCode::OK
+    );
+    for (_, permission) in CONTROL_OPERATIONS
+        .iter()
+        .filter(|(_, permission)| *permission != "notify")
+    {
+        let before = effects(&f);
+        let (status, body) = backend_operation(&f, &f.source, &target, permission).await;
+        assert_eq!(
+            status,
+            StatusCode::FORBIDDEN,
+            "messaging is not {permission}: {body}"
+        );
+        assert_eq!(effects(&f), before);
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn peer_operation_use_requires_explicit_directional_target_grant_and_revocation_is_live() {
+    let f = OperationFixture::new().await;
+    let permissions = control_permissions();
+    f.assign(
+        vec![workspace_permission("worker.create")],
+        permissions.clone(),
+    )
+    .await;
+    let sender = f.child("peer-sender", permissions.clone()).await;
+    let recipient = f.child("peer-recipient", permissions).await;
+    assert_eq!(
+        f.check(&sender, &operation_permission("worker.messaging"))
+            .await
+            .0,
+        StatusCode::OK
+    );
+    assert_backend_denied(&f, &sender, &recipient, None).await;
+    let grant = peer_grant(&f, &sender, &recipient, &["notify"]);
+    let before = effects(&f);
+    let (status, body) = backend_operation(&f, &sender, &recipient, "notify").await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    assert_eq!(body["state"], "accepted");
+    assert_eq!(effects(&f).inputs, before.inputs + 1);
+    assert_backend_denied(&f, &recipient, &sender, None).await;
+    for (_, permission) in CONTROL_OPERATIONS
+        .iter()
+        .filter(|(_, permission)| *permission != "notify")
+    {
+        let before = effects(&f);
+        let (status, body) = backend_operation(&f, &sender, &recipient, permission).await;
+        assert!(
+            status.is_client_error(),
+            "notify-only peer {permission}: {body}"
+        );
+        assert_eq!(effects(&f), before);
+    }
+    f.api
+        .store
+        .revoke_worker_control_grant(TEST_WORKSPACE_ID, &grant, &now_registry_timestamp())
+        .unwrap();
+    assert_eq!(
+        f.check(&sender, &operation_permission("worker.messaging"))
+            .await
+            .0,
+        StatusCode::OK,
+        "revoking target control does not revoke operation use"
+    );
+    assert_backend_denied(&f, &sender, &recipient, None).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn saved_targeted_message_permissions_preserve_target_direction_and_reject_alias_requests() {
+    for legacy in ["worker.notify", "worker.result"] {
+        let f = OperationFixture::new().await;
+        f.assign(vec![workspace_permission("worker.create")], vec![])
+            .await;
+        let target = f.child("saved-target", vec![]).await;
+        let other = f.child("saved-other-target", vec![]).await;
+        let saved = target_permission(legacy, &target);
+        let policy = f.assign(vec![saved.clone()], vec![]).await;
+        assert_eq!(
+            policy.execution,
+            vec![saved.clone()],
+            "saved target must not be rewritten into an operation-wide capability"
+        );
+        assert_eq!(
+            f.check(&f.source, &operation_permission("worker.messaging"))
+                .await
+                .0,
+            StatusCode::OK
+        );
+        let (status, body) = backend_operation(&f, &f.source, &target, "notify").await;
+        assert_eq!(status, StatusCode::OK, "{legacy}: {body}");
+        let before = effects(&f);
+        let (status, body) = backend_operation(&f, &f.source, &other, "notify").await;
+        assert_eq!(
+            status,
+            StatusCode::FORBIDDEN,
+            "{legacy} must retain exact target despite another real control grant: {body}"
+        );
+        assert_eq!(effects(&f), before);
+        for request in [
+            saved,
+            ScopedPermission {
+                operation: legacy.into(),
+                resource: operation_permission("worker.messaging").resource,
+            },
+            target_permission("worker.messaging", &target),
+        ] {
+            let (status, body) = f.check(&f.source, &request).await;
+            assert_eq!(
+                status,
+                StatusCode::FORBIDDEN,
+                "storage-only alias/target must not become a public authorization request: {body}"
+            );
+        }
+        assert_backend_denied(&f, &target, &other, None).await;
+        assert_eq!(
+            open_subjektiv_store(&f.api)
+                .unwrap()
+                .subject_permissions(&f.subject)
+                .unwrap()
+                .unwrap(),
+            policy
+        );
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn installed_worker_tools_register_without_authority_and_enforce_operation_matrix_before_effects()
+ {
+    let f = OperationFixture::new().await;
+    let client = f.client(&f.source);
+    let host = worker_tool_host(&f._dir.path().join("unset-worker-tools"), client.clone()).await;
+    assert!(
+        client.requests.lock().unwrap().is_empty(),
+        "registration is local"
+    );
+    let spawn = json!({"runtime_id":f.source.runtime_id,"operation_id":"public-spawn-denied","profile":"builtin:coder","execution_grant":{"permissions":[],"redelegation":[]}});
+    for (name, args) in [("WorkerList", json!({})), ("WorkerSpawn", spawn)] {
+        let before = effects(&f);
+        assert!(
+            execute_tool_result(&host, name, args).is_err(),
+            "unset {name}"
+        );
+        assert_eq!(effects(&f), before);
+    }
+    f.assign(vec![workspace_permission("worker.create")], vec![])
+        .await;
+    let target = f.child("public-denied-target", vec![]).await;
+    for name in [
+        "WorkerNotify",
+        "WorkerSendInput",
+        "WorkerCancel",
+        "WorkerStop",
+        "WorkerRestore",
+        "WorkerRemove",
+    ] {
+        let before = effects(&f);
+        client.requests.lock().unwrap().clear();
+        assert!(
+            execute_tool_result(&host, name, tool_arguments(name, &target)).is_err(),
+            "missing operation {name}"
+        );
+        assert_eq!(effects(&f), before);
+        assert!(
+            client
+                .requests
+                .lock()
+                .unwrap()
+                .iter()
+                .all(|(_, path)| path == &authorization_path()),
+            "{name} must refuse before the control backend"
+        );
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn installed_public_spawn_notify_stop_and_child_reply_use_real_backend_grants() {
+    let f = OperationFixture::new().await;
+    let reply = operation_permission("worker.messaging");
+    let mut permissions = control_permissions();
+    permissions.extend([
+        workspace_permission("worker.create"),
+        workspace_permission("worker.list"),
+    ]);
+    f.assign(permissions, vec![reply.clone()]).await;
+    let client = f.client(&f.source);
+    let host = worker_tool_host(&f._dir.path().join("parent-tools"), client).await;
+    let created = execute_tool(&host, "WorkerSpawn", json!({
+        "runtime_id":f.source.runtime_id,"operation_id":"public-worker-child","profile":"builtin:coder",
+        "execution_grant":{"permissions":[reply],"redelegation":[]}
+    })).await;
+    let child = RuntimeWorkerRef::new(
+        created["runtime_id"].as_str().unwrap(),
+        created["worker_id"].as_str().unwrap(),
+    );
+    assert!(
+        f.api
+            .store
+            .get_worker_registry(TEST_WORKSPACE_ID, &child)
+            .unwrap()
+            .is_some()
+    );
+    let listed = execute_tool(&host, "WorkerList", json!({})).await;
+    assert!(
+        listed["items"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|item| item["subject"] == subject_value(&child))
+    );
+    let notified = execute_tool(
+        &host,
+        "WorkerNotify",
+        tool_arguments("WorkerNotify", &child),
+    )
+    .await;
+    assert_eq!(notified["state"], "accepted");
+    let child_host = worker_tool_host(&f._dir.path().join("reply-tools"), f.client(&child)).await;
+    let before = effects(&f);
+    let reply = execute_tool(
+        &child_host,
+        "WorkerNotify",
+        json!({"subject":subject_value(&f.source),"content":"completed child result"}),
+    )
+    .await;
+    assert_eq!(reply["state"], "accepted");
+    assert_eq!(effects(&f).inputs, before.inputs + 1);
+    for name in [
+        "WorkerSendInput",
+        "WorkerCancel",
+        "WorkerStop",
+        "WorkerRestore",
+        "WorkerRemove",
+    ] {
+        let before = effects(&f);
+        assert!(
+            execute_tool_result(&child_host, name, tool_arguments(name, &f.source)).is_err(),
+            "reply cannot authorize {name}"
+        );
+        assert_eq!(effects(&f), before);
+    }
+    for (_, permission) in CONTROL_OPERATIONS
+        .iter()
+        .filter(|(_, permission)| *permission != "notify")
+    {
+        let before = effects(&f);
+        let (status, body) = backend_operation(&f, &child, &f.source, permission).await;
+        assert_eq!(
+            status,
+            StatusCode::FORBIDDEN,
+            "direct reply escalation {permission}: {body}"
+        );
+        assert_eq!(effects(&f), before);
+    }
+    let reverse = f
+        .api
+        .store
+        .get_active_worker_control_grant(TEST_WORKSPACE_ID, &child, &f.source)
+        .unwrap()
+        .unwrap();
+    f.api
+        .store
+        .revoke_worker_control_grant(
+            TEST_WORKSPACE_ID,
+            &reverse.grant_id,
+            &now_registry_timestamp(),
+        )
+        .unwrap();
+    let replayed = execute_tool(&host, "WorkerSpawn", json!({
+        "runtime_id":f.source.runtime_id,"operation_id":"public-worker-child","profile":"builtin:coder",
+        "execution_grant":{"permissions":[operation_permission("worker.messaging")],"redelegation":[]}
+    })).await;
+    assert_eq!(replayed["worker_id"], created["worker_id"]);
+    assert!(
+        f.api
+            .store
+            .get_active_worker_control_grant(TEST_WORKSPACE_ID, &child, &f.source)
+            .unwrap()
+            .is_none(),
+        "exact Spawn replay must not revive revoked reply control"
+    );
+    let before = effects(&f);
+    assert!(
+        execute_tool_result(
+            &child_host,
+            "WorkerNotify",
+            json!({"subject":subject_value(&f.source),"content":"revoked reply"})
+        )
+        .is_err()
+    );
+    assert_eq!(
+        effects(&f),
+        before,
+        "revoked public Notify had no Runtime effect"
+    );
+    let stopped = execute_tool(&host, "WorkerStop", tool_arguments("WorkerStop", &child)).await;
+    assert_eq!(stopped["state"], "accepted");
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn replaced_policy_and_body_never_revive_delegated_worker_control() {
+    let f = OperationFixture::new().await;
+    let permissions = control_permissions();
+    f.assign(
+        vec![workspace_permission("worker.create")],
+        permissions.clone(),
+    )
+    .await;
+    let child = f.child("old-policy-controller", permissions.clone()).await;
+    let target = f.child("policy-control-target", vec![]).await;
+    peer_grant(
+        &f,
+        &child,
+        &target,
+        &[
+            "notify",
+            "send_input",
+            "cancel",
+            "stop",
+            "restore",
+            "remove",
+            "observe",
+        ],
+    );
+    let grant = open_subjektiv_store(&f.api)
+        .unwrap()
+        .worker_execution_grant(&child.runtime_id, &child.worker_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        backend_operation(&f, &child, &target, "notify").await.0,
+        StatusCode::OK
+    );
+    f.assign(
+        vec![workspace_permission("worker.create")],
+        permissions.clone(),
+    )
+    .await;
+    assert_backend_denied(&f, &child, &target, Some(StatusCode::FORBIDDEN)).await;
+    // Identical replacement permissions do not refresh the captured policy ID.
+    f.assign(
+        vec![workspace_permission("worker.create")],
+        permissions.clone(),
+    )
+    .await;
+    assert_backend_denied(&f, &child, &target, Some(StatusCode::FORBIDDEN)).await;
+    assert_eq!(
+        open_subjektiv_store(&f.api)
+            .unwrap()
+            .worker_execution_grant(&child.runtime_id, &child.worker_id)
+            .unwrap(),
+        Some(grant)
+    );
+    let current = f.child("old-body-controller", permissions).await;
+    peer_grant(
+        &f,
+        &current,
+        &target,
+        &[
+            "notify",
+            "send_input",
+            "cancel",
+            "stop",
+            "restore",
+            "remove",
+            "observe",
+        ],
+    );
+    assert_eq!(
+        backend_operation(&f, &current, &target, "notify").await.0,
+        StatusCode::OK
+    );
+    let store = open_subjektiv_store(&f.api).unwrap();
+    let attribution = store
+        .subject_session_attributions(&f.subject)
+        .unwrap()
+        .into_iter()
+        .find(|attr| attr.runtime_id == f.source.runtime_id && attr.worker_id == f.source.worker_id)
+        .unwrap();
+    let writer =
+        subjektiv::journal::JournalHostContext::validated_session(&store, attribution).unwrap();
+    store
+        .release_journal_writer(&subjektiv::journal::JournalWriterBinding {
+            subject_id: f.subject.clone(),
+            runtime_id: f.source.runtime_id.clone(),
+            worker_id: f.source.worker_id.clone(),
+            binding_id: writer.writer_binding_id().unwrap().into(),
+        })
+        .unwrap();
+    f.release_source_lease();
+    let Json(replacement) = create_workspace_worker(
+        State(f.api.clone()),
+        HeaderMap::new(),
+        Json(CreateWorkspaceWorkerRequest {
+            runtime_id: f.source.runtime_id.clone(),
+            display_name: "Replacement control origin".into(),
+            singleton_key: None,
+            profile: Some("project:authority-source".into()),
+            ticket_assignment: None,
+            initial_submit: vec![],
+            workdir_attachments: vec![],
+            feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
+                subjektiv: Some(server_api::SubjektivWorkerConnectionRequest {
+                    subject_id: f.subject.clone(),
+                }),
+            },
+            control_operation_id: None,
+            execution_grant: None,
+        }),
+    )
+    .await
+    .unwrap();
+    assert_ne!(replacement.worker_id, f.source.worker_id);
+    assert_backend_denied(&f, &current, &target, None).await;
+    assert_backend_denied(&f, &f.source, &target, Some(StatusCode::UNAUTHORIZED)).await;
+}

openapi/server-api.json

@@ -7668,6 +7668,24 @@
             ],
             "type": "object"
           },
+          {
+            "additionalProperties": false,
+            "description": "Authorizes operation use, not a target Worker or a Workspace collection.\nThe Worker Feature must still enforce its directional target control grant.",
+            "properties": {
+              "kind": {
+                "const": "worker_operation",
+                "type": "string"
+              },
+              "workspace_id": {
+                "type": "string"
+              }
+            },
+            "required": [
+              "kind",
+              "workspace_id"
+            ],
+            "type": "object"
+          },
           {
             "additionalProperties": false,
             "properties": {
@@ -10854,6 +10872,7 @@
       },
       "ScopedPermission": {
         "additionalProperties": false,
+        "description": "Stored scopes retain their original resources and exact operation names.\n\nA legacy Worker read adapter may use [`canonical_worker_operation`] to project\noperation use, but must retain the original target separately for the Worker\nFeature's control-grant check, including direction, revocation and delegation\nlimits. Replacing an old Worker resource with WorkerOperation and discarding\nthat target would broaden authority. Neither validation nor `covers` performs\nthat conversion.",
         "properties": {
           "operation": {
             "type": "string"
@@ -19401,7 +19420,7 @@
   "info": {
     "title": "Yoi Server API",
     "version": "0.1.0",
-    "x-yoi-source-digest": "sha256:5d2b02b3083d04376358a69766e8fd31fc5c8ccee4d727d3d2b493d663ee2a4a"
+    "x-yoi-source-digest": "sha256:812a8f0a3b60df97ffe16e4905915b5c8d31b6795b8cd535e2adc10ed19078ca"
   },
   "openapi": "3.1.0",
   "paths": {

web/workspace/src/lib/generated/companion-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-companion-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:5d2b02b3083d04376358a69766e8fd31fc5c8ccee4d727d3d2b493d663ee2a4a)
+// Input contract: 0.1.0 (sha256:812a8f0a3b60df97ffe16e4905915b5c8d31b6795b8cd535e2adc10ed19078ca)
 // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/companion-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check

web/workspace/src/lib/generated/repository-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-repository-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:5d2b02b3083d04376358a69766e8fd31fc5c8ccee4d727d3d2b493d663ee2a4a)
+// Input contract: 0.1.0 (sha256:812a8f0a3b60df97ffe16e4905915b5c8d31b6795b8cd535e2adc10ed19078ca)
 // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output
 // Output: web/workspace/src/lib/generated/repository-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check

web/workspace/src/lib/generated/runtime-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-runtime-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:5d2b02b3083d04376358a69766e8fd31fc5c8ccee4d727d3d2b493d663ee2a4a)
+// Input contract: 0.1.0 (sha256:812a8f0a3b60df97ffe16e4905915b5c8d31b6795b8cd535e2adc10ed19078ca)
 // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/runtime-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check

web/workspace/src/lib/generated/ticket-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:5d2b02b3083d04376358a69766e8fd31fc5c8ccee4d727d3d2b493d663ee2a4a)
+// Input contract: 0.1.0 (sha256:812a8f0a3b60df97ffe16e4905915b5c8d31b6795b8cd535e2adc10ed19078ca)
 // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/ticket-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check

web/workspace/src/lib/generated/workdir-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-workdir-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:1f5ae1d92ac04d33a05b9d4a34252e017e668d5c33520490bdfff7a6cde1e3cb)
+// Input contract: 0.1.0 (sha256:812a8f0a3b60df97ffe16e4905915b5c8d31b6795b8cd535e2adc10ed19078ca)
 // Options: workdir-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/workdir-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_workdir_api_types -- --check

web/workspace/src/lib/generated/worker-launch-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-worker-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:5d2b02b3083d04376358a69766e8fd31fc5c8ccee4d727d3d2b493d663ee2a4a)
+// Input contract: 0.1.0 (sha256:812a8f0a3b60df97ffe16e4905915b5c8d31b6795b8cd535e2adc10ed19078ca)
 // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/worker-launch-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check
@@ -35,7 +35,7 @@ export type PasteArtifactMediaType = "text_plain_utf8";
 
 export type PasteArtifactRef = { artifact_id: string; availability: PasteArtifactAvailability; byte_len: number; char_count: number; created_at_ms: number; line_count: number; media_type: PasteArtifactMediaType; sha256: string; source_entry_id: string; };
 
-export type PermissionResource = { job_id: string; kind: "job"; workspace_id: string; } | { kind: "ticket"; ticket_id: string; workspace_id: string; } | { kind: "objective"; objective_id: string; workspace_id: string; } | { kind: "merge_request"; merge_request_id: string; workspace_id: string; } | { kind: "subject"; subject_id: string; workspace_id: string; } | { kind: "drive_entry"; node_id: string; workspace_id: string; } | { kind: "drive_request"; request_id: string; workspace_id: string; } | { kind: "repository"; repository_key: string; workspace_id: string; } | { connection_id: string; kind: "workdir_attachment"; workspace_id: string; } | { kind: "workspace_config"; path: string; recursive: boolean; workspace_id: string; } | { kind: "network"; url: string; } | { kind: "sub_worker"; name: string; runtime_id: string; worker_id: string; } | { kind: "feature"; } | { kind: "workspace"; workspace_id: string; } | { kind: "workdir"; path: string; recursive: boolean; working_directory_id: string; } | { kind: "worker"; runtime_id: string; worker_id: string; };
+export type PermissionResource = { job_id: string; kind: "job"; workspace_id: string; } | { kind: "ticket"; ticket_id: string; workspace_id: string; } | { kind: "objective"; objective_id: string; workspace_id: string; } | { kind: "merge_request"; merge_request_id: string; workspace_id: string; } | { kind: "subject"; subject_id: string; workspace_id: string; } | { kind: "drive_entry"; node_id: string; workspace_id: string; } | { kind: "drive_request"; request_id: string; workspace_id: string; } | { kind: "repository"; repository_key: string; workspace_id: string; } | { connection_id: string; kind: "workdir_attachment"; workspace_id: string; } | { kind: "workspace_config"; path: string; recursive: boolean; workspace_id: string; } | { kind: "network"; url: string; } | { kind: "sub_worker"; name: string; runtime_id: string; worker_id: string; } | { kind: "feature"; } | { kind: "workspace"; workspace_id: string; } | { kind: "workdir"; path: string; recursive: boolean; working_directory_id: string; } | { kind: "worker_operation"; workspace_id: string; } | { kind: "worker"; runtime_id: string; worker_id: string; };
 
 export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; message: string; };