diff --git a/Cargo.lock b/Cargo.lock index d9ec4b4f5..1df49c9e2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6604,6 +6604,7 @@ dependencies = [ "session-store", "sha2 0.11.0", "subjektiv", + "syn 2.0.117", "tempfile", "thiserror 2.0.18", "ticket", diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index c956d4a3b..ea0a56011 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -144,9 +144,11 @@ use workdir::{ use worker::WorkerController; use worker::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession; use worker::feature::builtin::{ - CompositeWorkerObservationProvider, WorkerObservationError, WorkerObservationProvider, - WorkerObservationSubject, WorkerObservationSubjectRef, WorkerSessionCapture, - WorkspaceClientWorkerObservationProvider, + CompositeWorkerObservationProvider, WorkspaceClientWorkerObservationProvider, +}; +use worker::feature::contracts::{ + WorkerObservationError, WorkerObservationProvider, WorkerObservationSubject, + WorkerObservationSubjectRef, WorkerSessionCapture, }; #[cfg(feature = "ws-server")] use worker::ipc::protocol_session::{WorkerProtocolSessionStreams, live_log_entry_event}; diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index 368ec7662..f8b4f26f6 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -64,3 +64,4 @@ dotenv = "0.15.0" futures = { workspace = true } serial_test = "3.4.0" tempfile = { workspace = true } +syn = { version = "2", features = ["full", "visit"] } diff --git a/crates/worker/src/bootstrap.rs b/crates/worker/src/bootstrap.rs index 3c5fd8475..4e0a296b8 100644 --- a/crates/worker/src/bootstrap.rs +++ b/crates/worker/src/bootstrap.rs @@ -311,7 +311,7 @@ pub struct RestoredControllerBindings { pub backend_job: Option, pub workdir_sessions: std::sync::Arc, pub observation_provider: - Option>, + Option>, pub subjektiv_attached: bool, } diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index cce9352fe..cdae53d00 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -2083,7 +2083,7 @@ where } let mut observation_providers: Vec< - Arc, + Arc, > = Vec::new(); // Worker-orchestration tools derive child filesystem authority from the diff --git a/crates/worker/src/feature.rs b/crates/worker/src/feature.rs index f45ffc4c3..a59075e87 100644 --- a/crates/worker/src/feature.rs +++ b/crates/worker/src/feature.rs @@ -514,6 +514,11 @@ impl ServiceRequirement { } } + pub fn with_version(mut self, version: ServiceVersionReq) -> Self { + self.version = version; + self + } + pub fn optional(id: ServiceId, reason: impl Into) -> Self { Self { id, @@ -2372,6 +2377,7 @@ pub use invocation::{ FeatureInvocationHandlerError, FeatureInvocationRegistry, InvocationContributionRegistrar, }; pub mod builtin; +pub mod contracts; pub mod mcp; pub(crate) mod session; diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index c03d07983..817151949 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -47,7 +47,6 @@ pub mod workspace_worker_discovery; pub use task::{TaskFeature, task_tools_feature}; pub use ticket::{TicketFeature, TicketFeatureAccess, ticket_tools_feature}; pub use worker_observation::{ - CompositeWorkerObservationProvider, WorkerObservationError, WorkerObservationFeature, - WorkerObservationProvider, WorkerObservationSubject, WorkerObservationSubjectRef, - WorkerSessionCapture, WorkspaceClientWorkerObservationProvider, + CompositeWorkerObservationProvider, WorkerObservationFeature, + WorkspaceClientWorkerObservationProvider, }; diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 30230147c..078e91741 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -14,6 +14,12 @@ use server_api::{ CreateWorkspaceWorkerTicketAssignmentRequest, }; +use crate::feature::contracts::{ + WORKER_CONTROL_SERVICE_ID, WORKER_CONTROL_SERVICE_VERSION, WORKER_LIFECYCLE_SERVICE_ID, + WORKER_LIFECYCLE_SERVICE_VERSION, WorkerControlInputKind, WorkerControlService, + WorkerControlStopKind, WorkerControlSubject, WorkerLifecycleService, + WorkerLifecycleSpawnRequest, WorkerLifecycleWorkdirAttachment, WorkerObservationSubjectRef, +}; use crate::feature::{ FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ServiceDeclaration, ServiceId, ToolContribution, ToolDeclaration, @@ -25,78 +31,8 @@ const FEATURE_ID: &str = "worker"; const FEATURE_NAME: &str = "Worker"; const FEATURE_DESCRIPTION: &str = "Workspace-authority tools for managing Workdir-bound Backend/Runtime Worker sessions."; -pub const WORKER_LIFECYCLE_SERVICE_ID: &str = "worker.lifecycle"; -pub const WORKER_CONTROL_SERVICE_ID: &str = "worker.control"; -const WORKER_LIFECYCLE_SERVICE_VERSION: &str = "1"; const INITIAL_WORKDIR_ALIAS: &str = "workdir"; -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum WorkerControlSubject { - RuntimeWorker { - runtime_id: String, - worker_id: String, - }, - SubWorker { - name: String, - }, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum WorkerControlInputKind { - User, - Notify, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum WorkerControlStopKind { - Cancel, - Stop, -} - -#[async_trait] -pub trait WorkerControlService: Send + Sync { - fn workspace_id(&self) -> &str; - async fn list_workers(&self) -> Result; - async fn send_input( - &self, - subject: WorkerControlSubject, - content: String, - kind: WorkerControlInputKind, - ) -> Result; - async fn stop_worker( - &self, - subject: WorkerControlSubject, - kind: WorkerControlStopKind, - reason: Option, - ) -> Result; - async fn spawn_worker( - &self, - request: WorkerLifecycleSpawnRequest, - ) -> Result; - async fn restore_worker( - &self, - runtime_id: String, - worker_id: String, - request: server_api::WorkerRestoreRequest, - ) -> Result { - let _ = (runtime_id, worker_id, request); - Err(WorkspaceClientError::Unavailable( - "Runtime Worker restore is unavailable".to_string(), - )) - } - fn remove_runtime_worker( - &self, - runtime_id: &str, - worker_id: &str, - reason: &str, - ) -> Result; - async fn ensure_permission( - &self, - subject: &super::worker_observation::WorkerObservationSubjectRef, - permission: &str, - ) -> Result<(), WorkspaceClientError>; -} - struct WorkspaceWorkerControlService { client: Arc, workspace_id: String, @@ -296,9 +232,7 @@ impl WorkerControlService for WorkspaceWorkerControlService { )); } self.ensure_permission( - &super::worker_observation::WorkerObservationSubjectRef::SubWorker { - name: name.clone(), - }, + &WorkerObservationSubjectRef::SubWorker { name: name.clone() }, "send_input", ) .await?; @@ -377,9 +311,7 @@ impl WorkerControlService for WorkspaceWorkerControlService { )); } self.ensure_permission( - &super::worker_observation::WorkerObservationSubjectRef::SubWorker { - name: name.clone(), - }, + &WorkerObservationSubjectRef::SubWorker { name: name.clone() }, "stop", ) .await?; @@ -456,11 +388,11 @@ impl WorkerControlService for WorkspaceWorkerControlService { async fn ensure_permission( &self, - subject: &super::worker_observation::WorkerObservationSubjectRef, + subject: &WorkerObservationSubjectRef, permission: &str, ) -> Result<(), WorkspaceClientError> { match subject { - super::worker_observation::WorkerObservationSubjectRef::SubWorker { name } => { + WorkerObservationSubjectRef::SubWorker { name } => { let known = self .registry .as_ref() @@ -474,7 +406,7 @@ impl WorkerControlService for WorkspaceWorkerControlService { )) } } - super::worker_observation::WorkerObservationSubjectRef::RuntimeWorker { + WorkerObservationSubjectRef::RuntimeWorker { runtime_id, worker_id, } => { @@ -533,43 +465,6 @@ impl WorkerControlService for WorkspaceWorkerControlService { } } -#[async_trait] -pub trait WorkerLifecycleService: Send + Sync { - async fn spawn( - &self, - request: WorkerLifecycleSpawnRequest, - ) -> Result; -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct WorkerLifecycleWorkdirAttachment { - /// Stable Worker-local routing alias preserved into the canonical create request. - pub alias: String, - /// Workspace-authoritative Workdir id preserved into the canonical create request. - pub working_directory_id: String, - /// Optional normalized initial cwd relative to this Workdir. - pub relative_cwd: Option, -} - -#[derive(Debug, Clone)] -pub struct WorkerLifecycleSpawnRequest { - /// Observed delegation required for Orchestrator Ticket starts. - pub orchestrator_assignment_id: Option, - - pub runtime_id: String, - /// Ordered alias-keyed selections forwarded as canonical Workdir attachments. - pub workdir_attachments: Vec, - pub profile: String, - pub singleton_key: Option, - pub ticket_id: Option, - /// Required for ordinary and Ticket spawn; exact retries reuse this ID and full input. - pub operation_id: Option, - /// Explicit requested authority. Profile selection alone grants nothing. - pub execution_grant: Option, - pub display_name: String, - pub initial_submit: Vec, -} - struct WorkspaceWorkerLifecycleService { client: Arc, } @@ -702,7 +597,7 @@ impl FeatureModule for SubWorkerControlFeature { ) .with_provided_service(ServiceDeclaration::new( ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), - WORKER_LIFECYCLE_SERVICE_VERSION, + WORKER_CONTROL_SERVICE_VERSION, "Parent-owned SubWorker discovery and control operations", )); for operation in SUB_WORKER_CONTROL_OPERATIONS { @@ -721,7 +616,7 @@ impl FeatureModule for SubWorkerControlFeature { context.services().provide( ServiceDeclaration::new( ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), - WORKER_LIFECYCLE_SERVICE_VERSION, + WORKER_CONTROL_SERVICE_VERSION, "Parent-owned SubWorker discovery and control operations", ), control.clone(), @@ -755,7 +650,7 @@ impl FeatureModule for ManageWorkerFeature { )) .with_provided_service(ServiceDeclaration::new( ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), - WORKER_LIFECYCLE_SERVICE_VERSION, + WORKER_CONTROL_SERVICE_VERSION, "Known-Worker discovery and permission-fenced control operations", )); for operation in WorkerOperation::ALL { @@ -790,7 +685,7 @@ impl FeatureModule for ManageWorkerFeature { context.services().provide( ServiceDeclaration::new( ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), - WORKER_LIFECYCLE_SERVICE_VERSION, + WORKER_CONTROL_SERVICE_VERSION, "Known-Worker discovery and permission-fenced control operations", ), self.control.clone(), @@ -1570,7 +1465,7 @@ mod tests { async fn ensure_permission( &self, - _subject: &crate::feature::builtin::WorkerObservationSubjectRef, + _subject: &crate::feature::contracts::WorkerObservationSubjectRef, _permission: &str, ) -> Result<(), WorkspaceClientError> { Ok(()) @@ -1942,7 +1837,7 @@ mod tests { }); let permission_error = control .ensure_permission( - &crate::feature::builtin::WorkerObservationSubjectRef::RuntimeWorker { + &crate::feature::contracts::WorkerObservationSubjectRef::RuntimeWorker { runtime_id: "runtime-1".to_string(), worker_id: "worker-1".to_string(), }, diff --git a/crates/worker/src/feature/builtin/orchestration.rs b/crates/worker/src/feature/builtin/orchestration.rs index 6518b8c76..44b5b774b 100644 --- a/crates/worker/src/feature/builtin/orchestration.rs +++ b/crates/worker/src/feature/builtin/orchestration.rs @@ -9,14 +9,14 @@ use protocol::Segment; use schemars::JsonSchema; use serde::Deserialize; -use super::manage_worker::{ - WORKER_LIFECYCLE_SERVICE_ID, WorkerLifecycleService, WorkerLifecycleSpawnRequest, +use crate::feature::contracts::{ + TICKET_SERVICE_ID, TICKET_SERVICE_VERSION, TicketService, WORKER_LIFECYCLE_SERVICE_ID, + WORKER_LIFECYCLE_SERVICE_VERSION, WorkerLifecycleService, WorkerLifecycleSpawnRequest, WorkerLifecycleWorkdirAttachment, }; -use super::ticket::{TICKET_SERVICE_ID, TicketService}; use crate::feature::{ FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ServiceId, - ServiceRequirement, ToolContribution, ToolDeclaration, + ServiceRequirement, ServiceVersionReq, ToolContribution, ToolDeclaration, }; const FEATURE_ID: &str = "orchestration"; @@ -34,14 +34,20 @@ impl FeatureModule for OrchestrationFeature { fn descriptor(&self) -> FeatureDescriptor { FeatureDescriptor::builtin(FEATURE_ID, "Orchestration") .with_description("Semantic Ticket orchestration operations.") - .with_service_requirement(ServiceRequirement::required( - ServiceId::builtin(TICKET_SERVICE_ID), - "SpawnTicketWorker requires current typed Ticket authority", - )) - .with_service_requirement(ServiceRequirement::required( - ServiceId::builtin(WORKER_LIFECYCLE_SERVICE_ID), - "SpawnTicketWorker requires Workspace Worker lifecycle authority", - )) + .with_service_requirement( + ServiceRequirement::required( + ServiceId::builtin(TICKET_SERVICE_ID), + "SpawnTicketWorker requires current typed Ticket authority", + ) + .with_version(ServiceVersionReq::exact(TICKET_SERVICE_VERSION)), + ) + .with_service_requirement( + ServiceRequirement::required( + ServiceId::builtin(WORKER_LIFECYCLE_SERVICE_ID), + "SpawnTicketWorker requires Workspace Worker lifecycle authority", + ) + .with_version(ServiceVersionReq::exact(WORKER_LIFECYCLE_SERVICE_VERSION)), + ) .with_tool(ToolDeclaration::new(TOOL_NAME, TOOL_DESCRIPTION)) } @@ -275,7 +281,7 @@ mod tests { use ticket::{TicketError, TicketState}; use super::*; - use crate::feature::builtin::ticket::TicketHandoff; + use crate::feature::contracts::TicketHandoff; use crate::worker::{WorkspaceClientError, WorkspaceResponse}; struct FixedTicketService(TicketState); diff --git a/crates/worker/src/feature/builtin/ticket.rs b/crates/worker/src/feature/builtin/ticket.rs index 366b9d257..e45d9c243 100644 --- a/crates/worker/src/feature/builtin/ticket.rs +++ b/crates/worker/src/feature/builtin/ticket.rs @@ -46,6 +46,10 @@ mod native; mod notification_diagnostics; mod tags; +use crate::feature::contracts::{ + TICKET_SERVICE_ID, TICKET_SERVICE_VERSION, TicketHandoff, TicketService, +}; + use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::sync::{Arc, Mutex}; @@ -539,29 +543,6 @@ const FEATURE_NAME: &str = "Ticket tools"; const FEATURE_DESCRIPTION: &str = "Intent-led Ticket reads and authorized decisions through the Workspace API; no universal MR/approval gate, and no Backend judgment of natural-language satisfaction."; const TICKET_WORKFLOW_INSTRUCTION_ID: &str = "ticket.workflow"; const TICKET_WORKFLOW_PROMPT_REF: &str = "common.tickets"; -pub const TICKET_SERVICE_ID: &str = "ticket.authority"; -const TICKET_SERVICE_VERSION: &str = "1"; - -pub trait TicketService: Send + Sync { - fn ticket_handoff(&self, ticket_ref: &str) -> Result; - - /// Typed creation without a synthetic Tool context. The default keeps legacy - /// handoff-only fixture implementations fail-closed; production installs - /// provide creation through the shared authorized backend dispatch. - fn create(&self, _input: NewTicket) -> TicketResult { - Err(TicketError::Conflict( - "Ticket creation service is unavailable".into(), - )) - } -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct TicketHandoff { - pub id: String, - pub resource_key: String, - pub workflow_state: TicketState, -} - fn is_canonical_ticket_resource_key(resource_key: &str) -> bool { resource_key.strip_prefix("T-").is_some_and(|sequence| { !sequence.is_empty() && sequence.bytes().all(|byte| byte.is_ascii_digit()) diff --git a/crates/worker/src/feature/builtin/worker_observation.rs b/crates/worker/src/feature/builtin/worker_observation.rs index 59afdfaa1..bb740ca69 100644 --- a/crates/worker/src/feature/builtin/worker_observation.rs +++ b/crates/worker/src/feature/builtin/worker_observation.rs @@ -18,14 +18,17 @@ use agen::Item; use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; use async_trait::async_trait; use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; -use session_store::LogEntry; +use serde::Deserialize; -use super::manage_worker::{WORKER_CONTROL_SERVICE_ID, WorkerControlService}; +use crate::feature::contracts::{ + WORKER_CONTROL_SERVICE_ID, WORKER_CONTROL_SERVICE_VERSION, WorkerControlService, + WorkerObservationError, WorkerObservationProvider, WorkerObservationSubject, + WorkerObservationSubjectRef, WorkerSessionCapture, +}; use crate::feature::{ FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureInstructionContribution, FeatureInstructionDeclaration, FeatureInstructionId, FeatureModule, ServiceId, - ServiceRequirement, ToolContribution, ToolDeclaration, + ServiceRequirement, ServiceVersionReq, ToolContribution, ToolDeclaration, }; use crate::session_capture::{ ReadDetail, ReadOptions, ReadSelector, ReferenceKind, SearchOptions, SessionCapture, @@ -49,69 +52,6 @@ fn observation_instruction() -> FeatureInstructionDeclaration { .expect("static worker-observation instruction declaration is valid") } -#[derive( - Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, JsonSchema, -)] -#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] -pub enum WorkerObservationSubjectRef { - RuntimeWorker { - runtime_id: String, - worker_id: String, - }, - SubWorker { - name: String, - }, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct WorkerObservationSubject { - pub subject: WorkerObservationSubjectRef, - pub display_name: String, - pub relation: String, - pub status: String, -} - -#[derive(Debug, Clone)] -pub struct WorkerSessionCapture { - pub segment_id: String, - pub session: protocol::SessionSnapshot, -} - -impl WorkerSessionCapture { - pub fn from_log_entries( - segment_id: impl Into, - log_entries: &[LogEntry], - ) -> Result { - Ok(Self { - segment_id: segment_id.into(), - session: session_store::public_snapshot::project_current_session_snapshot(log_entries), - }) - } -} - -#[derive(Debug, thiserror::Error)] -pub enum WorkerObservationError { - #[error("worker session was not found or is not accessible")] - NotFound, - #[error("worker session observation failed: {0}")] - Unavailable(String), -} - -#[async_trait] -pub trait WorkerObservationProvider: Send + Sync { - /// Returns only subjects already authorized for the current Worker. - async fn list_worker_sessions( - &self, - ) -> Result, WorkerObservationError>; - - /// Reauthorizes and captures the latest committed session for one subject. - /// Unauthorized and missing subjects must both return `NotFound`. - async fn capture_worker_session( - &self, - subject: &WorkerObservationSubjectRef, - ) -> Result; -} - #[derive(Debug, Deserialize)] struct WorkspaceWorkerObservationListResponse { sessions: Vec, @@ -290,7 +230,7 @@ impl FeatureModule for WorkerObservationFeature { .with_service_requirement(ServiceRequirement::required( ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), "Worker observation extends the known-Worker control authority", - )) + ).with_version(ServiceVersionReq::exact(WORKER_CONTROL_SERVICE_VERSION))) .with_instruction(observation_instruction()) .with_tool(ToolDeclaration::new( "ViewSessionOverview", diff --git a/crates/worker/src/feature/contracts.rs b/crates/worker/src/feature/contracts.rs new file mode 100644 index 000000000..65bdb4ec4 --- /dev/null +++ b/crates/worker/src/feature/contracts.rs @@ -0,0 +1,223 @@ +//! Provider-independent contracts used by Worker Features and trusted host composition. +//! +//! Service IDs identify a capability's semantics, not a Feature, backend, Rust type name, or +//! TypeId. Versions are exact compatibility epochs checked by the existing install planner; +//! they are not crate versions or semver ranges. Compatible implementation changes retain the +//! ID/epoch. Changes to required methods, request/result meaning, authorization guarantees, or +//! subject routing that break consumers increment the affected epoch and update providers and +//! requirements together. A different capability needs a different ID, not provider priority. +//! Rust downcasts only validate an in-process binding and are not a public compatibility ABI. +//! +//! Trusted composition selects the exact-one control provider, including the explicit combined +//! Workspace/SubWorker implementation. Acquiring a handle does not grant authority: providers +//! must revalidate current authority, Workspace ownership and direct-child constraints at each +//! operation. Observation tools bound public projections; providers reauthorize every capture. +//! These Rust contracts are not a Plugin/guest ABI; an external boundary needs an explicit adapter. + +use async_trait::async_trait; +use protocol::Segment; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use session_store::LogEntry; +use ticket::{NewTicket, Result as TicketResult, TicketError, TicketRef, TicketState}; + +// Reuse the Host-neutral response/error vocabulary rather than redesigning transport DTOs. +use crate::worker::{WorkspaceClientError, WorkspaceResponse}; + +pub const WORKER_CONTROL_SERVICE_ID: &str = "worker.control"; +pub const WORKER_CONTROL_SERVICE_VERSION: &str = "1"; +pub const WORKER_LIFECYCLE_SERVICE_ID: &str = "worker.lifecycle"; +pub const WORKER_LIFECYCLE_SERVICE_VERSION: &str = "1"; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WorkerControlSubject { + RuntimeWorker { + runtime_id: String, + worker_id: String, + }, + SubWorker { + name: String, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkerControlInputKind { + User, + Notify, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkerControlStopKind { + Cancel, + Stop, +} + +#[async_trait] +pub trait WorkerControlService: Send + Sync { + fn workspace_id(&self) -> &str; + async fn list_workers(&self) -> Result; + async fn send_input( + &self, + subject: WorkerControlSubject, + content: String, + kind: WorkerControlInputKind, + ) -> Result; + async fn stop_worker( + &self, + subject: WorkerControlSubject, + kind: WorkerControlStopKind, + reason: Option, + ) -> Result; + async fn spawn_worker( + &self, + request: WorkerLifecycleSpawnRequest, + ) -> Result; + async fn restore_worker( + &self, + runtime_id: String, + worker_id: String, + request: server_api::WorkerRestoreRequest, + ) -> Result { + let _ = (runtime_id, worker_id, request); + Err(WorkspaceClientError::Unavailable( + "Runtime Worker restore is unavailable".to_string(), + )) + } + fn remove_runtime_worker( + &self, + runtime_id: &str, + worker_id: &str, + reason: &str, + ) -> Result; + async fn ensure_permission( + &self, + subject: &WorkerObservationSubjectRef, + permission: &str, + ) -> Result<(), WorkspaceClientError>; +} + +#[async_trait] +pub trait WorkerLifecycleService: Send + Sync { + async fn spawn( + &self, + request: WorkerLifecycleSpawnRequest, + ) -> Result; +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerLifecycleWorkdirAttachment { + /// Stable Worker-local routing alias preserved into the canonical create request. + pub alias: String, + /// Workspace-authoritative Workdir id preserved into the canonical create request. + pub working_directory_id: String, + /// Optional normalized initial cwd relative to this Workdir. + pub relative_cwd: Option, +} + +#[derive(Debug, Clone)] +pub struct WorkerLifecycleSpawnRequest { + /// Observed delegation required for Orchestrator Ticket starts. + pub orchestrator_assignment_id: Option, + + pub runtime_id: String, + /// Ordered alias-keyed selections forwarded as canonical Workdir attachments. + pub workdir_attachments: Vec, + pub profile: String, + pub singleton_key: Option, + pub ticket_id: Option, + /// Required for ordinary and Ticket spawn; exact retries reuse this ID and full input. + pub operation_id: Option, + /// Explicit requested authority. Profile selection alone grants nothing. + pub execution_grant: Option, + pub display_name: String, + pub initial_submit: Vec, +} + +#[derive( + Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, JsonSchema, +)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum WorkerObservationSubjectRef { + RuntimeWorker { + runtime_id: String, + worker_id: String, + }, + SubWorker { + name: String, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WorkerObservationSubject { + pub subject: WorkerObservationSubjectRef, + pub display_name: String, + pub relation: String, + pub status: String, +} + +#[derive(Debug, Clone)] +pub struct WorkerSessionCapture { + pub segment_id: String, + pub session: protocol::SessionSnapshot, +} + +impl WorkerSessionCapture { + pub fn from_log_entries( + segment_id: impl Into, + log_entries: &[LogEntry], + ) -> Result { + Ok(Self { + segment_id: segment_id.into(), + session: session_store::public_snapshot::project_current_session_snapshot(log_entries), + }) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum WorkerObservationError { + #[error("worker session was not found or is not accessible")] + NotFound, + #[error("worker session observation failed: {0}")] + Unavailable(String), +} + +#[async_trait] +pub trait WorkerObservationProvider: Send + Sync { + /// Returns only subjects already authorized for the current Worker. + async fn list_worker_sessions( + &self, + ) -> Result, WorkerObservationError>; + + /// Reauthorizes and captures the latest committed session for one subject. + /// Unauthorized and missing subjects must both return `NotFound`. + async fn capture_worker_session( + &self, + subject: &WorkerObservationSubjectRef, + ) -> Result; +} + +pub const TICKET_SERVICE_ID: &str = "ticket.authority"; +pub const TICKET_SERVICE_VERSION: &str = "1"; + +pub trait TicketService: Send + Sync { + fn ticket_handoff(&self, ticket_ref: &str) -> Result; + + /// Typed creation without a synthetic Tool context. The default keeps legacy + /// handoff-only fixture implementations fail-closed; production installs + /// provide creation through the shared authorized backend dispatch. + fn create(&self, _input: NewTicket) -> TicketResult { + Err(TicketError::Conflict( + "Ticket creation service is unavailable".into(), + )) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TicketHandoff { + pub id: String, + pub resource_key: String, + pub workflow_state: TicketState, +} +#[cfg(test)] +#[path = "contracts_tests.rs"] +mod tests; diff --git a/crates/worker/src/feature/contracts_tests.rs b/crates/worker/src/feature/contracts_tests.rs new file mode 100644 index 000000000..d09e8ba0b --- /dev/null +++ b/crates/worker/src/feature/contracts_tests.rs @@ -0,0 +1,378 @@ +//! Component coverage: real consumer installation, planner and Tool dispatch with contract-only +//! test providers. No concrete provider constructor, Workspace client or spawned-worker registry is +//! passed to a consumer. Production authority/routing is covered by the providers' own suites. + +use std::sync::{ + Arc, Mutex, + atomic::{AtomicBool, AtomicUsize, Ordering}, +}; + +use super::*; +use crate::feature::builtin::{ + orchestration::orchestration_feature, worker_observation::WorkerObservationFeature, +}; +use crate::feature::{ + FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, FeaturePlanError, + FeatureRegistryBuilder, ServiceDeclaration, ServiceId, ServiceRequirement, +}; +use crate::hook::HookRegistryBuilder; +use agen::tool::{Tool, ToolDefinition, ToolExecutionContext}; + +struct TestServices { + revoked: AtomicBool, + captures: AtomicUsize, + spawns: Mutex>, +} + +impl Default for TestServices { + fn default() -> Self { + Self { + revoked: AtomicBool::new(false), + captures: AtomicUsize::new(0), + spawns: Mutex::new(Vec::new()), + } + } +} + +fn response() -> WorkspaceResponse { + WorkspaceResponse { + status: 200, + body: r#"{"worker_id":"test-worker"}"#.into(), + } +} + +#[async_trait] +impl WorkerLifecycleService for TestServices { + async fn spawn( + &self, + request: WorkerLifecycleSpawnRequest, + ) -> Result { + self.spawns.lock().unwrap().push(request); + Ok(response()) + } +} + +impl TicketService for TestServices { + fn ticket_handoff(&self, ticket_ref: &str) -> Result { + assert_eq!(ticket_ref, "T-788"); + Ok(TicketHandoff { + id: "ticket-internal-id".into(), + resource_key: ticket_ref.into(), + workflow_state: TicketState::Open, + }) + } +} + +#[async_trait] +impl WorkerControlService for TestServices { + fn workspace_id(&self) -> &str { + "test-workspace" + } + async fn list_workers(&self) -> Result { + unreachable!("not an observation operation") + } + async fn send_input( + &self, + _: WorkerControlSubject, + _: String, + _: WorkerControlInputKind, + ) -> Result { + unreachable!("not a consumer operation") + } + async fn stop_worker( + &self, + _: WorkerControlSubject, + _: WorkerControlStopKind, + _: Option, + ) -> Result { + unreachable!("not a consumer operation") + } + async fn spawn_worker( + &self, + _: WorkerLifecycleSpawnRequest, + ) -> Result { + unreachable!("orchestration must use lifecycle") + } + fn remove_runtime_worker( + &self, + _: &str, + _: &str, + _: &str, + ) -> Result { + unreachable!("not a consumer operation") + } + async fn ensure_permission( + &self, + subject: &WorkerObservationSubjectRef, + permission: &str, + ) -> Result<(), WorkspaceClientError> { + assert_eq!(permission, "observe"); + if !self.revoked.load(Ordering::SeqCst) + && matches!(subject, WorkerObservationSubjectRef::SubWorker { name } if name == "child") + { + Ok(()) + } else { + Err(WorkspaceClientError::Request("not granted".into())) + } + } +} + +#[async_trait] +impl WorkerObservationProvider for TestServices { + async fn list_worker_sessions( + &self, + ) -> Result, WorkerObservationError> { + Ok(Vec::new()) + } + async fn capture_worker_session( + &self, + _: &WorkerObservationSubjectRef, + ) -> Result { + self.captures.fetch_add(1, Ordering::SeqCst); + WorkerSessionCapture::from_log_entries("test-segment", &[]) + .map_err(WorkerObservationError::Unavailable) + } +} + +struct TestProvider { + id: &'static str, + services: Arc, + declarations: Vec, + requirements: Vec, + undeclared: bool, + installs: Arc, +} + +impl TestProvider { + fn new(id: &'static str, services: Arc) -> Self { + Self { + id, + services, + declarations: vec![ + ServiceDeclaration::new( + ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), + WORKER_CONTROL_SERVICE_VERSION, + "test control", + ), + ServiceDeclaration::new( + ServiceId::builtin(WORKER_LIFECYCLE_SERVICE_ID), + WORKER_LIFECYCLE_SERVICE_VERSION, + "test lifecycle", + ), + ServiceDeclaration::new( + ServiceId::builtin(TICKET_SERVICE_ID), + TICKET_SERVICE_VERSION, + "test Ticket", + ), + ], + requirements: Vec::new(), + undeclared: false, + installs: Arc::new(AtomicUsize::new(0)), + } + } +} + +impl FeatureModule for TestProvider { + fn descriptor(&self) -> FeatureDescriptor { + let mut descriptor = FeatureDescriptor::builtin(self.id, "Contract test provider"); + descriptor.provides_services = self.declarations.clone(); + descriptor.requires_services = self.requirements.clone(); + descriptor + } + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + self.installs.fetch_add(1, Ordering::SeqCst); + for declaration in &self.declarations { + match declaration.id.as_str() { + "builtin:worker.control" => context.services().provide( + declaration.clone(), + self.services.clone() as Arc, + )?, + "builtin:worker.lifecycle" => context.services().provide( + declaration.clone(), + self.services.clone() as Arc, + )?, + "builtin:ticket.authority" => context.services().provide( + declaration.clone(), + self.services.clone() as Arc, + )?, + _ => unreachable!(), + } + } + if self.undeclared { + context.services().provide( + ServiceDeclaration::new(ServiceId::builtin("hidden"), "1", "undeclared"), + Arc::new(()), + )?; + } + Ok(()) + } +} + +fn consumers(services: Arc) -> FeatureRegistryBuilder { + FeatureRegistryBuilder::new() + .with_module(orchestration_feature()) + .with_module(WorkerObservationFeature::new(services)) +} + +fn tool(tools: &[ToolDefinition], name: &str) -> Arc { + tools + .iter() + .map(|factory| factory()) + .find(|(meta, _)| meta.name == name) + .unwrap() + .1 +} + +#[tokio::test] +async fn consumers_install_and_execute_with_contract_only_test_provider() { + let services = Arc::new(TestServices::default()); + let mut tools = Vec::new(); + let report = consumers(services.clone()) + .with_module(TestProvider::new("test-provider", services.clone())) + .install_into_pending(&mut tools, &mut HookRegistryBuilder::default()); + assert!(!report.has_errors(), "{report:?}"); + let result = tool(&tools, "SpawnTicketWorker").execute( + r#"{"ticket_id":"T-788","runtime_id":"runtime-test","profile":"builtin:coder","initial_request":"implement","orchestrator_assignment_id":"delegation-test"}"#, + ToolExecutionContext::new("spawn-call", "batch", 0), + ).await.unwrap(); + assert_eq!(result.summary, "Spawned Worker for Ticket T-788"); + { + let requests = services.spawns.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].ticket_id.as_deref(), Some("ticket-internal-id")); + assert_eq!( + requests[0].orchestrator_assignment_id.as_deref(), + Some("delegation-test") + ); + assert_eq!( + requests[0].operation_id.as_deref(), + Some("spawn-ticket-worker:ticket-internal-id:spawn-call") + ); + assert_eq!( + requests[0].initial_submit, + vec![ + Segment::Flow { + selector: "builtin:coder-review".into() + }, + Segment::text("implement") + ] + ); + } + + let overview = tool(&tools, "ViewSessionOverview"); + let granted = r#"{"subject":{"kind":"sub_worker","name":"child"}}"#; + overview + .execute(granted, ToolExecutionContext::direct()) + .await + .unwrap(); + assert_eq!(services.captures.load(Ordering::SeqCst), 1); + services.revoked.store(true, Ordering::SeqCst); + for denied in [ + granted, + r#"{"subject":{"kind":"sub_worker","name":"non-child"}}"#, + ] { + let error = overview + .execute(denied, ToolExecutionContext::direct()) + .await + .unwrap_err(); + assert!(error.to_string().contains("not found or is not accessible")); + } + assert_eq!( + services.captures.load(Ordering::SeqCst), + 1, + "revocation must fence the existing handle before capture" + ); +} + +#[test] +fn consumer_plan_rejections_leave_no_partial_installation() { + for condition in [ + "missing", + "duplicate", + "control-version", + "lifecycle-version", + "ticket-version", + "cycle", + "undeclared", + ] { + let services = Arc::new(TestServices::default()); + let mut provider = TestProvider::new("test-provider", services.clone()); + let installs = provider.installs.clone(); + let mut builder = consumers(services.clone()); + match condition { + "missing" => {} + "duplicate" => { + builder.add_module(TestProvider::new("duplicate-provider", services.clone())); + } + "control-version" => provider.declarations[0].version = "incompatible".into(), + "lifecycle-version" => provider.declarations[1].version = "incompatible".into(), + "ticket-version" => provider.declarations[2].version = "incompatible".into(), + "cycle" => { + provider.requirements.push(ServiceRequirement::required( + ServiceId::builtin("consumer-output"), + "cycle", + )); + struct CyclicConsumer; + impl FeatureModule for CyclicConsumer { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("cyclic-consumer", "Cycle") + .with_service_requirement(ServiceRequirement::required( + ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), + "control", + )) + .with_provided_service(ServiceDeclaration::new( + ServiceId::builtin("consumer-output"), + "1", + "output", + )) + } + fn install( + &self, + _: &mut FeatureInstallContext<'_>, + ) -> Result<(), FeatureInstallError> { + panic!("cycle must fail preflight") + } + } + builder.add_module(CyclicConsumer); + } + "undeclared" => provider.undeclared = true, + _ => unreachable!(), + } + if condition != "missing" { + builder.add_module(provider); + } + let mut tools = Vec::new(); + let report = builder.install_into_pending(&mut tools, &mut HookRegistryBuilder::default()); + assert!(report.has_errors(), "{condition}: {report:?}"); + assert!(tools.is_empty(), "{condition}"); + assert!(report.services.providers().is_empty(), "{condition}"); + assert!( + report.reports.iter().all(|entry| !entry.installed), + "{condition}" + ); + match (condition, report.plan_error) { + ("missing", Some(FeaturePlanError::MissingServiceProvider { .. })) + | ("duplicate", Some(FeaturePlanError::AmbiguousServiceProvider { .. })) + | ( + "control-version" | "lifecycle-version" | "ticket-version", + Some(FeaturePlanError::ServiceVersionMismatch { .. }), + ) + | ("cycle", Some(FeaturePlanError::ServiceDependencyCycle { .. })) => { + assert_eq!(installs.load(Ordering::SeqCst), 0) + } + ("undeclared", None) => { + assert_eq!(installs.load(Ordering::SeqCst), 1); + assert!( + report + .reports + .iter() + .flat_map(|entry| &entry.diagnostics) + .any(|diagnostic| diagnostic + .message + .contains("undeclared Service contribution")) + ); + } + (_, error) => panic!("{condition}: unexpected plan error {error:?}"), + } + } +} diff --git a/crates/worker/src/spawn/tool.rs b/crates/worker/src/spawn/tool.rs index 0b75266be..1b8f357ef 100644 --- a/crates/worker/src/spawn/tool.rs +++ b/crates/worker/src/spawn/tool.rs @@ -1528,16 +1528,16 @@ enabled = false crate::feature::builtin::worker_observation::SpawnedSubWorkerObservationProvider::new( registry.clone(), ); - let observed_child = - crate::feature::builtin::worker_observation::WorkerObservationSubjectRef::SubWorker { - name: "reviewer-child".to_string(), - }; - let first_capture = crate::feature::builtin::worker_observation::WorkerObservationProvider::capture_worker_session( - &observation, - &observed_child, - ) - .await - .unwrap(); + let observed_child = crate::feature::contracts::WorkerObservationSubjectRef::SubWorker { + name: "reviewer-child".to_string(), + }; + let first_capture = + crate::feature::contracts::WorkerObservationProvider::capture_worker_session( + &observation, + &observed_child, + ) + .await + .unwrap(); assert!( serde_json::to_string(&first_capture.session) .unwrap() @@ -1554,12 +1554,13 @@ enabled = false crate::internal_worker::InternalWorkerSessionStatus::Idle ); assert_eq!(calls.load(Ordering::SeqCst), 2); - let latest_capture = crate::feature::builtin::worker_observation::WorkerObservationProvider::capture_worker_session( - &observation, - &observed_child, - ) - .await - .unwrap(); + let latest_capture = + crate::feature::contracts::WorkerObservationProvider::capture_worker_session( + &observation, + &observed_child, + ) + .await + .unwrap(); assert!(latest_capture.session.entries.len() > first_capture.session.entries.len()); fail_requests.store(true, Ordering::SeqCst); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 45b002f12..b2073b8aa 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -53,8 +53,9 @@ use crate::compact::telemetry::{ }; use crate::compact::usage_tracker::{UsageTracker, persist_pending_usage}; use crate::feature::background::{BackgroundTaskRewriteGuard, FeatureBackgroundTaskRegistry}; +use crate::feature::builtin::TaskFeature; use crate::feature::builtin::memory::WorkspaceMemoryBackendError; -use crate::feature::builtin::{TaskFeature, WorkerObservationProvider}; +use crate::feature::contracts::WorkerObservationProvider; use crate::feature::session::{ CommittedRunExit, CommittedSessionCapture, CommittedSessionCaptureHandle, FeatureSessionError, SessionExtensionHandle, diff --git a/crates/worker/tests/service_contract_architecture_test.rs b/crates/worker/tests/service_contract_architecture_test.rs new file mode 100644 index 000000000..7d54586ae --- /dev/null +++ b/crates/worker/tests/service_contract_architecture_test.rs @@ -0,0 +1,108 @@ +//! Source dependency regression for the neutral contracts and representative consumers. Parse Rust +//! syntax so grouped/renamed imports and fully qualified type/expression paths are checked alike. + +use syn::visit::Visit; + +#[derive(Default)] +struct ProviderDependencies(Vec); + +impl ProviderDependencies { + fn check(&mut self, path: &[String]) { + // These modules are provider/composition namespaces. `super::*` inside a consumer's own + // unit tests is local access, but importing/aliasing the parent namespace is not. + if path.windows(2).any(|pair| pair == ["feature", "builtin"]) + || path.first().is_some_and(|segment| segment == "builtin") + || (path.first().is_some_and(|segment| segment == "super") + && path.get(1).is_none_or(|segment| segment != "*")) + { + self.0.push(path.join("::")); + } + } + + fn use_tree(&mut self, prefix: &mut Vec, tree: &syn::UseTree) { + match tree { + syn::UseTree::Path(path) => { + prefix.push(path.ident.to_string()); + self.use_tree(prefix, &path.tree); + prefix.pop(); + } + syn::UseTree::Group(group) => { + for item in &group.items { + self.use_tree(prefix, item); + } + } + syn::UseTree::Name(name) => { + prefix.push(name.ident.to_string()); + self.check(prefix); + prefix.pop(); + } + syn::UseTree::Rename(rename) => { + prefix.push(rename.ident.to_string()); + self.check(prefix); + prefix.pop(); + } + syn::UseTree::Glob(_) => { + prefix.push("*".into()); + self.check(prefix); + prefix.pop(); + } + } + } +} + +impl<'ast> Visit<'ast> for ProviderDependencies { + fn visit_item_use(&mut self, item: &'ast syn::ItemUse) { + self.use_tree(&mut Vec::new(), &item.tree); + } + + fn visit_path(&mut self, path: &'ast syn::Path) { + self.check( + &path + .segments + .iter() + .map(|segment| segment.ident.to_string()) + .collect::>(), + ); + syn::visit::visit_path(self, path); + } +} + +fn dependencies(source: &str) -> Vec { + let file = syn::parse_file(source).expect("valid Rust source"); + let mut visitor = ProviderDependencies::default(); + visitor.visit_file(&file); + visitor.0 +} + +#[test] +fn contracts_and_consumers_do_not_depend_on_provider_namespaces() { + for (name, source) in [ + ("contracts", include_str!("../src/feature/contracts.rs")), + ( + "orchestration", + include_str!("../src/feature/builtin/orchestration.rs"), + ), + ( + "worker_observation", + include_str!("../src/feature/builtin/worker_observation.rs"), + ), + ] { + assert_eq!(dependencies(source), Vec::::new(), "{name}"); + } +} + +#[test] +fn architecture_guard_detects_imports_reexports_aliases_and_qualified_provider_paths() { + for source in [ + "use super::manage_worker::WorkerControlService;", + "pub use super::ticket::TicketService;", + "use crate::feature::{builtin::{manage_worker as provider, ticket::*}};", + "use crate::feature::builtin as providers;", + "use super as providers;", + "type Contract = dyn crate::feature::builtin::manage_worker::WorkerControlService;", + "fn consumer() { crate::feature::builtin::manage_worker::manage_worker_feature(); }", + ] { + assert!(!dependencies(source).is_empty(), "guard missed {source}"); + } + assert!(dependencies("use ticket::TicketError; use crate::feature::contracts::TicketService; mod tests { use super::*; }").is_empty()); +} diff --git a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs index 6b80d8f9f..dfbf909c2 100644 --- a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs @@ -1015,7 +1015,7 @@ async fn ticket_worker_host_with_service( ) -> ( worker::Worker, bool, - Option>, + Option>, ) { let manifest = worker::WorkerManifest::from_toml( r#" @@ -1060,9 +1060,9 @@ async fn ticket_worker_host_with_service( Some( report .services - .service::( + .service::( &worker::feature::ServiceId::builtin( - worker::feature::builtin::ticket::TICKET_SERVICE_ID, + worker::feature::contracts::TICKET_SERVICE_ID, ), ) .unwrap(),