From b9dadb6a08070ef694c0dad45af1174801ca9d6b Mon Sep 17 00:00:00 2001 From: Hare Date: Sun, 9 Aug 2026 23:19:29 +0900 Subject: [PATCH] worker: support typed initial submit on spawn --- Cargo.lock | 1 + crates/protocol/Cargo.toml | 2 + crates/protocol/src/lib.rs | 1 + crates/worker-runtime/src/runtime.rs | 37 +++++ crates/worker/Cargo.toml | 2 +- .../src/feature/builtin/manage_worker.rs | 105 +++++++++++- crates/workspace-server/src/hosts.rs | 111 +++++++++---- crates/workspace-server/src/server.rs | 149 ++++++++++++------ docs/design/flow-state-graph.md | 2 + .../console/worker-console.ui.test.ts | 2 +- .../workspace/sidebar/worker-launch.test.ts | 29 +++- .../lib/workspace/sidebar/worker-launch.ts | 16 +- .../w/[workspaceId]/workers/new/+page.svelte | 4 +- 13 files changed, 367 insertions(+), 94 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f6f1c387..e2a882bc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3056,6 +3056,7 @@ version = "0.1.0" name = "protocol" version = "0.1.0" dependencies = [ + "schemars", "serde", "serde_json", "tokio", diff --git a/crates/protocol/Cargo.toml b/crates/protocol/Cargo.toml index 5b2ad758..7d6032ab 100644 --- a/crates/protocol/Cargo.toml +++ b/crates/protocol/Cargo.toml @@ -8,8 +8,10 @@ license.workspace = true default = ["stream"] stream = ["dep:tokio"] typescript = ["dep:ts-rs"] +json-schema = ["dep:schemars"] [dependencies] +schemars = { workspace = true, optional = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } tokio = { workspace = true, features = ["io-util"], optional = true } diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 8e9764ec..33e7f216 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -183,6 +183,7 @@ impl WorkerEvent { /// the dropped intent. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] #[serde(tag = "kind", rename_all = "snake_case")] pub enum Segment { /// Free-form text. The fallback every client can produce. diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index cc57999b..48475fbe 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -3427,6 +3427,43 @@ mod tests { ); } + #[test] + fn restore_does_not_redispatch_spawn_initial_submit() { + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = Runtime::with_execution_backend( + RuntimeOptions { + ..RuntimeOptions::default() + }, + backend.clone(), + ) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let mut request = task_request("flow restore"); + request.initial_input = Some(WorkerInput { + kind: WorkerInputKind::User, + content: String::new(), + segments: Some(vec![ + protocol::Segment::Flow { + selector: "builtin:coder-review".to_string(), + }, + protocol::Segment::text("Implement Ticket 00001"), + ]), + }); + let detail = runtime.create_worker(request).unwrap(); + assert_eq!(backend.dispatched_inputs.lock().unwrap().len(), 1); + + runtime + .stop_worker(&detail.worker_ref, Some("restore test".to_string())) + .unwrap(); + runtime.restore_worker(&detail.worker_ref).unwrap(); + + assert_eq!( + backend.dispatched_inputs.lock().unwrap().len(), + 1, + "restore must continue durable Worker state without replaying spawn initial input" + ); + } + #[test] fn send_input_dispatches_segment_only_flow_submission() { let backend = Arc::new(TestExecutionBackend::default()); diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index 35995728..c7c7785e 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -16,7 +16,7 @@ session-store = { workspace = true } secrets = { workspace = true } manifest = { workspace = true } mcp = { workspace = true } -protocol = { workspace = true } +protocol = { workspace = true, features = ["json-schema"] } client = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 7b93967e..f34723f9 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -9,6 +9,8 @@ use llm_engine::tool::{ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +use protocol::Segment; + use crate::feature::{ FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution, ToolDeclaration, @@ -96,8 +98,10 @@ struct WorkerSpawnInput { profile: String, #[serde(default)] display_name: Option, + /// Normal typed initial user submission delivered after spawn. An empty + /// vector starts the Worker without initial input. #[serde(default)] - initial_text: Option, + initial_submit: Vec, #[serde(default)] relative_cwd: Option, } @@ -107,7 +111,7 @@ struct WorkerSpawnRequest { runtime_id: String, display_name: String, profile: String, - initial_text: String, + initial_submit: Vec, working_directory: WorkerWorkingDirectorySelection, } @@ -197,7 +201,7 @@ impl Tool for WorkspaceWorkerTool { .filter(|value| !value.trim().is_empty()) .unwrap_or_else(|| "Workspace Worker".to_string()), profile: non_empty(input.profile, "profile")?, - initial_text: input.initial_text.unwrap_or_default(), + initial_submit: input.initial_submit, working_directory: WorkerWorkingDirectorySelection { working_directory_id: authority_id( &input.working_directory_id, @@ -320,7 +324,78 @@ fn validate_relative_cwd(value: &str) -> Result { #[cfg(test)] mod tests { + use std::sync::Mutex; + use super::*; + use crate::worker::{WorkspaceClientError, WorkspaceResponse}; + + #[derive(Debug, Default)] + struct RecordingWorkspaceClient { + requests: Mutex>, + } + + impl WorkspaceClient for RecordingWorkspaceClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace/test") + } + + fn kind(&self) -> &str { + "recording" + } + + fn is_available(&self) -> bool { + true + } + + fn execute( + &self, + request: WorkspaceRequest, + ) -> Result { + self.requests.lock().unwrap().push(request); + Ok(WorkspaceResponse { + status: 200, + body: "{}".to_string(), + }) + } + } + + #[tokio::test] + async fn worker_spawn_forwards_typed_initial_submit_to_workspace_api() { + let client = Arc::new(RecordingWorkspaceClient::default()); + let tool = WorkspaceWorkerTool { + operation: WorkerOperation::Spawn, + client: client.clone(), + workspace_id: "workspace%2Ftest".to_string(), + }; + tool.execute( + &serde_json::json!({ + "runtime_id": "runtime-1", + "working_directory_id": "workdir-1", + "profile": "builtin:coder", + "initial_submit": [ + { "kind": "flow", "selector": "builtin:coder-review" }, + { "kind": "text", "content": "Implement Ticket 00001" } + ] + }) + .to_string(), + ToolExecutionContext::direct(), + ) + .await + .unwrap(); + + let requests = client.requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].path, "/api/w/workspace%2Ftest/workers"); + let body: serde_json::Value = + serde_json::from_str(requests[0].body.as_deref().unwrap()).unwrap(); + assert_eq!(body["initial_submit"][0]["kind"], "flow"); + assert_eq!( + body["initial_submit"][0]["selector"], + "builtin:coder-review" + ); + assert_eq!(body["initial_submit"][1]["kind"], "text"); + assert!(body.get("initial_text").is_none()); + } #[test] fn worker_tool_family_is_distinct_from_sub_worker_tools() { @@ -330,13 +405,28 @@ mod tests { ); } + #[test] + fn worker_spawn_schema_exposes_normal_typed_segment_variants() { + let schema = serde_json::to_value(schemars::schema_for!(WorkerSpawnInput)).unwrap(); + let text = serde_json::to_string(&schema).unwrap(); + assert!(text.contains("initial_submit")); + assert!(text.contains("selector")); + assert!(text.contains("flow")); + assert!(!text.contains("initial_text")); + } + #[test] fn worker_spawn_request_uses_authority_ids_without_runtime_paths() { let request = WorkerSpawnRequest { runtime_id: "runtime-1".to_string(), display_name: "Coder".to_string(), profile: "builtin:coder".to_string(), - initial_text: "Implement the Ticket".to_string(), + initial_submit: vec![ + Segment::Flow { + selector: "builtin:coder-review".to_string(), + }, + Segment::text("Implement the Ticket"), + ], working_directory: WorkerWorkingDirectorySelection { working_directory_id: "wd-1".to_string(), relative_cwd: Some("repo".to_string()), @@ -348,6 +438,13 @@ mod tests { assert!(value.get("cwd").is_none()); assert!(value.get("runtime_url").is_none()); assert!(value["working_directory"].get("mode").is_none()); + assert_eq!(value["initial_submit"][0]["kind"], "flow"); + assert_eq!( + value["initial_submit"][0]["selector"], + "builtin:coder-review" + ); + assert_eq!(value["initial_submit"][1]["kind"], "text"); + assert!(value.get("initial_text").is_none()); } #[test] diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 13c720f8..44ff693d 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1,6 +1,7 @@ use crate::Error; use crate::resource_broker::{BackendResourceBroker, BackendResourceTarget}; use chrono::Utc; +use protocol::Segment; use reqwest::blocking::{Client as BlockingHttpClient, RequestBuilder}; use reqwest::header::{AUTHORIZATION, CONTENT_TYPE}; use reqwest::{Client as AsyncHttpClient, StatusCode, Url}; @@ -359,8 +360,8 @@ pub struct WorkerSpawnRequest { pub profile: ProfileSelector, #[serde(default, skip_serializing_if = "Option::is_none")] pub ticket_assignment: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub initial_input: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub initial_submit: Vec, /// Optional safe working-directory creation request. The Workspace server resolves /// this into a runtime-internal `WorkingDirectoryRequest` from configured /// repositories before calling a host. @@ -410,6 +411,17 @@ pub enum WorkerSpawnAcceptanceRequirement { RunAccepted { expected_segments: usize }, } +fn initial_worker_input(segments: &[Segment]) -> Option { + if segments.is_empty() { + return None; + } + Some(EmbeddedWorkerInput { + kind: EmbeddedWorkerInputKind::User, + content: Segment::flatten_to_text(segments), + segments: Some(segments.to_vec()), + }) +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkerSpawnResult { pub state: WorkerOperationState, @@ -1141,6 +1153,30 @@ impl RuntimeRegistry { request: WorkerSpawnRequest, ) -> Result { validate_backend_identifier("runtime_id", runtime_id)?; + match request.acceptance { + WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments } + if expected_segments != request.initial_submit.len() => + { + return Err(RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: runtime_id.to_string(), + code: "worker_initial_segment_count_mismatch".to_string(), + message: format!( + "spawn acceptance expects {expected_segments} initial segment(s), request carries {}", + request.initial_submit.len() + ), + }); + } + WorkerSpawnAcceptanceRequirement::SocketReady if !request.initial_submit.is_empty() => { + return Err(RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: runtime_id.to_string(), + code: "worker_initial_submit_require_run_acceptance".to_string(), + message: + "spawn requests with initial segments must require RunAccepted acceptance" + .to_string(), + }); + } + _ => {} + } let runtime = self.runtime(runtime_id)?; Ok(runtime.spawn_worker(request)) } @@ -1865,7 +1901,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { display_name: request.requested_worker_name.clone(), config_bundle: None, profile_source, - initial_input: request.initial_input.clone(), + initial_input: initial_worker_input(&request.initial_submit), working_directory_request: request.resolved_working_directory_request.clone(), working_directory: request.resolved_working_directory.clone(), worker_observation_enabled: request.resolved_worker_observation_enabled, @@ -2970,7 +3006,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { display_name: request.requested_worker_name.clone(), config_bundle: None, profile_source, - initial_input: request.initial_input.clone(), + initial_input: initial_worker_input(&request.initial_submit), working_directory_request: request.resolved_working_directory_request.clone(), working_directory: request.resolved_working_directory.clone(), worker_observation_enabled: request.resolved_worker_observation_enabled, @@ -4551,7 +4587,7 @@ mod tests { }, profile: ProfileSelector::Builtin("builtin:coder".to_string()), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -4601,29 +4637,48 @@ mod tests { } #[test] - fn embedded_runtime_rejects_system_initial_input_without_worker_projection() { - let runtime = EmbeddedWorkerRuntime::new_memory_with_execution_backend( - "local:test", - Arc::new(AcceptingExecutionBackend::default()), - ) - .expect("test backend should connect"); + fn worker_spawn_idempotency_fingerprint_covers_canonical_initial_submit() { let mut request = embedded_spawn_request(); - request.initial_input = Some(EmbeddedWorkerInput { - kind: EmbeddedWorkerInputKind::Notify, - content: "system/role instruction belongs in profile".to_string(), - segments: None, + request.ticket_assignment = Some(WorkerTicketAssignmentRequest { + ticket_id: "00001KVZSGT0Q".to_string(), + operation_id: "operation-1".to_string(), }); + request.initial_submit = vec![ + Segment::Flow { + selector: "builtin:coder-review".to_string(), + }, + Segment::text("Implement Ticket 00001KVZSGT0Q"), + ]; + request.acceptance = WorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: request.initial_submit.len(), + }; - let spawned = runtime.spawn_worker(request); - assert_eq!(spawned.state, WorkerOperationState::Rejected); - assert!(spawned.worker.is_none()); - assert!(spawned.diagnostics.iter().any(|diagnostic| { - diagnostic.code == "embedded_worker_initial_input_kind_invalid" - && diagnostic - .message - .contains("initial worker input must be user input") - })); - assert!(runtime.list_workers(10).items.is_empty()); + let first = worker_spawn_idempotency(&request).unwrap().unwrap(); + let repeated = worker_spawn_idempotency(&request).unwrap().unwrap(); + assert_eq!(first, repeated); + assert_eq!(first.0, "operation-1"); + + let mut changed = request.clone(); + changed.initial_submit[1] = Segment::text("Different instruction"); + let changed = worker_spawn_idempotency(&changed).unwrap().unwrap(); + assert_ne!(first.1, changed.1); + } + + #[test] + fn shared_spawn_projects_typed_initial_submit_to_runtime_user_input() { + let segments = vec![ + Segment::Flow { + selector: "builtin:coder-review".to_string(), + }, + Segment::text("Implement Ticket 00001"), + ]; + + let input = initial_worker_input(&segments).expect("typed initial input"); + + assert_eq!(input.kind, EmbeddedWorkerInputKind::User); + assert_eq!(input.content, Segment::flatten_to_text(&segments)); + assert_eq!(input.segments, Some(segments)); + assert!(initial_worker_input(&[]).is_none()); } #[test] @@ -4702,7 +4757,7 @@ mod tests { }, profile: ProfileSelector::Builtin("builtin:coder".to_string()), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -4799,7 +4854,7 @@ mod tests { }, profile: ProfileSelector::Builtin("builtin:coder".to_string()), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -4835,7 +4890,7 @@ mod tests { acceptance: WorkerSpawnAcceptanceRequirement::SocketReady, profile: ProfileSelector::Builtin("builtin:companion".to_string()), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index e70c4a51..844141a1 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -18,6 +18,7 @@ use memory::backend::{ MemoryBackendHttpResponse, MemoryBackendOperation, MemoryConsolidateStagingOperation, MemoryConsolidationOutput, }; +use protocol::Segment; use protocol::stream::{decode_method, encode_event}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -114,9 +115,6 @@ use worker_runtime::http_server::{ RuntimeHttpSummaryResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse, }; use worker_runtime::identity::RuntimeWorkerRef; -use worker_runtime::interaction::{ - WorkerInput as EmbeddedWorkerInput, WorkerInputKind as EmbeddedWorkerInputKind, -}; const EMBEDDED_WORKER_RUNTIME_ID: &str = "embedded-worker-runtime"; @@ -1558,12 +1556,13 @@ pub struct BrowserWorkspaceOrchestratorResponse { #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] -pub struct BrowserCreateWorkerRequest { +pub struct CreateWorkspaceWorkerRequest { pub runtime_id: String, pub display_name: String, #[serde(default)] pub profile: Option, - pub initial_text: String, + #[serde(default)] + pub initial_submit: Vec, #[serde(default)] pub working_directory: Option, } @@ -3949,11 +3948,7 @@ fn start_memory_staging_consolidation( &api.config.workspace_created_at, MEMORY_CONSOLIDATION_PROFILE, )?; - let input = EmbeddedWorkerInput { - kind: EmbeddedWorkerInputKind::User, - content: input_content, - segments: None, - }; + let initial_submit = vec![Segment::text(input_content)]; let result = api.spawn_workspace_worker( &runtime_id, WorkerSpawnRequest { @@ -3964,7 +3959,7 @@ fn start_memory_staging_consolidation( }, profile: profile_selector, ticket_assignment: None, - initial_input: Some(input), + initial_submit, working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -4510,7 +4505,7 @@ async fn scoped_start_workspace_orchestrator( }, profile: ProfileSelector::Builtin("builtin:orchestrator".to_string()), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -4574,7 +4569,7 @@ fn workspace_orchestrator_is_online(worker: &WorkerSummary) -> bool { async fn scoped_create_workspace_worker( State(api): State, AxumPath(path): AxumPath, - Json(request): Json, + Json(request): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; create_workspace_worker(State(api), Json(request)).await @@ -6832,9 +6827,38 @@ fn configured_working_directory_request( )) } +fn validate_worker_initial_submit(segments: &[Segment]) -> Result<()> { + let flow_selectors = segments + .iter() + .filter_map(|segment| match segment { + Segment::Flow { selector } => Some(selector), + _ => None, + }) + .collect::>(); + if flow_selectors.len() > 1 { + return Err(Error::InvalidInput( + "initial_submit may contain at most one Flow segment".to_string(), + )); + } + if let Some(selector) = flow_selectors.first() { + selector + .parse::() + .map_err(|error| Error::InvalidInput(error.to_string()))?; + } + if segments + .iter() + .any(|segment| matches!(segment, Segment::Unknown)) + { + return Err(Error::InvalidInput( + "initial_submit must not contain unknown segment variants".to_string(), + )); + } + Ok(()) +} + async fn create_workspace_worker( State(api): State, - Json(request): Json, + Json(request): Json, ) -> ApiResult> { let profile = request .profile @@ -6883,16 +6907,9 @@ async fn create_workspace_worker( if display_name == crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY { return Err(Error::ReservedWorkerName(display_name).into()); } - let initial_text = request.initial_text.trim().to_string(); - let initial_input = if initial_text.is_empty() { - None - } else { - Some(EmbeddedWorkerInput { - kind: EmbeddedWorkerInputKind::User, - content: initial_text.clone(), - segments: None, - }) - }; + let initial_submit = request.initial_submit; + validate_worker_initial_submit(&initial_submit)?; + let expected_segments = initial_submit.len(); let selected_working_directory_id = request .working_directory .as_ref() @@ -6914,12 +6931,10 @@ async fn create_workspace_worker( WorkerSpawnRequest { requested_worker_name: Some(display_name.clone()), intent: WorkerSpawnIntent::WorkspaceCoding, - acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { - expected_segments: if initial_input.is_some() { 1 } else { 0 }, - }, + acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments }, profile: profile_selector, ticket_assignment: None, - initial_input, + initial_submit, working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory, @@ -10212,6 +10227,50 @@ mod tests { ); } + #[test] + fn worker_initial_submit_validate_flow_shape_before_spawn() { + let valid = vec![ + Segment::Flow { + selector: "builtin:coder-review".to_string(), + }, + Segment::text("Implement Ticket 00001"), + ]; + assert!(validate_worker_initial_submit(&valid).is_ok()); + assert!(validate_worker_initial_submit(&[]).is_ok()); + + let duplicate = vec![ + Segment::Flow { + selector: "builtin:coder-review".to_string(), + }, + Segment::Flow { + selector: "workspace:coder-review".to_string(), + }, + ]; + assert!(matches!( + validate_worker_initial_submit(&duplicate), + Err(Error::InvalidInput(message)) if message.contains("at most one") + )); + assert!(matches!( + validate_worker_initial_submit(&[Segment::Flow { + selector: "coder-review".to_string(), + }]), + Err(Error::InvalidInput(_)) + )); + assert!(matches!( + validate_worker_initial_submit(&[Segment::Unknown]), + Err(Error::InvalidInput(message)) if message.contains("unknown") + )); + assert!( + serde_json::from_value::(serde_json::json!({ + "runtime_id": "runtime-1", + "display_name": "coder", + "initial_text": "legacy parallel authority", + "working_directory": { "kind": "without_workspace" } + })) + .is_err() + ); + } + #[tokio::test] async fn worker_source_auth_rejects_cross_workspace_mutation() { let workspace = tempfile::tempdir().unwrap(); @@ -10219,11 +10278,11 @@ mod tests { let api = test_api(workspace.path()).await; let Json(created) = create_workspace_worker( State(api.clone()), - Json(BrowserCreateWorkerRequest { + Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Scoped Worker".to_string(), profile: Some("builtin:coder".to_string()), - initial_text: String::new(), + initial_submit: Vec::new(), working_directory: None, }), ) @@ -10252,11 +10311,11 @@ mod tests { let Json(generic) = create_workspace_worker( State(api.clone()), - Json(BrowserCreateWorkerRequest { + Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Generic Orchestrator Profile Worker".to_string(), profile: Some("builtin:orchestrator".to_string()), - initial_text: String::new(), + initial_submit: Vec::new(), working_directory: None, }), ) @@ -10266,11 +10325,11 @@ mod tests { assert!(find_workspace_orchestrator(&api).is_none()); let reserved = create_workspace_worker( State(api.clone()), - Json(BrowserCreateWorkerRequest { + Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY.to_string(), profile: Some("builtin:orchestrator".to_string()), - initial_text: String::new(), + initial_submit: Vec::new(), working_directory: None, }), ) @@ -11123,7 +11182,7 @@ mod tests { }, profile: ProfileSelector::Builtin(MEMORY_CONSOLIDATION_PROFILE.to_string()), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -11305,7 +11364,7 @@ mod tests { }, profile: ProfileSelector::Builtin("builtin:coder".to_string()), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -11521,7 +11580,7 @@ mod tests { }, profile: ProfileSelector::Builtin("builtin:coder".to_string()), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -11663,7 +11722,7 @@ mod tests { ticket_id: first_ticket.id.clone(), operation_id: "spawn-assignment-operation".to_string(), }), - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -11895,7 +11954,7 @@ mod tests { }, profile: ProfileSelector::Builtin("builtin:coder".to_string()), ticket_assignment: Some(assignment.clone()), - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -12812,7 +12871,7 @@ mod tests { "runtime_id": "remote-runtime", "display_name": "Coding Worker", "profile": "builtin:coder", - "initial_text": "", + "initial_submit": [], "working_directory": { "working_directory_id": working_directory_id, "relative_cwd": "../escape" @@ -13173,7 +13232,7 @@ mod tests { serde_json::json!({ "runtime_id": "embedded-worker-runtime", "display_name": "", - "initial_text": "" + "initial_submit": [] }), ) .await; @@ -13239,7 +13298,7 @@ mod tests { "runtime_id": "remote-runtime", "display_name": "Remote Worker", "profile": "builtin:companion", - "initial_text": "" + "initial_submit": [] })), StatusCode::BAD_REQUEST, ) @@ -13352,7 +13411,7 @@ mod tests { "runtime_id": "embedded-worker-runtime", "display_name": "Coding Worker", "profile": "builtin:coder", - "initial_text": "", + "initial_submit": [], "kind": "internal" })), StatusCode::UNPROCESSABLE_ENTITY, @@ -13898,7 +13957,7 @@ mod tests { }, profile: ProfileSelector::Builtin("builtin:coder".to_string()), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, @@ -14418,7 +14477,7 @@ mod tests { "builtin:companion".to_string(), ), ticket_assignment: None, - initial_input: None, + initial_submit: Vec::new(), working_directory_request: None, resolved_working_directory_request: None, resolved_working_directory: None, diff --git a/docs/design/flow-state-graph.md b/docs/design/flow-state-graph.md index 0f4ec776..43d84b7b 100644 --- a/docs/design/flow-state-graph.md +++ b/docs/design/flow-state-graph.md @@ -69,6 +69,8 @@ Flow invocation uses the normal Submit/Run segment vector rather than a Worker-c Runtime accepts exactly one Flow segment only when the resolved Profile enables `feature.flow` and a Workspace client is available. The Worker asks Workspace authority only for an immutable source snapshot, creates the instance locally, replaces the Flow segment with the entered state's instructions, and commits that runtime state atomically with the remaining Submit segments before LLM execution. A Worker with an active Flow rejects the duplicate input without changing its local state or events. +The generic model-facing `WorkerSpawn` accepts `initial_submit: Vec` and routes them unchanged through the shared Workspace spawn request into Runtime `CreateWorkerRequest.initial_input`. It does not have a parallel `initial_text` or a role-specific `SpawnCoder` wrapper. Backend derives the flat content projection from the canonical segment vector, validates Flow shape before spawn, and includes the segment vector in lifecycle idempotency fingerprints. Restoring the same Worker never replays spawn initial segments. + `RequestFlowTransition` accepts only: ```json diff --git a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts index 90a08c10..2b3b71e7 100644 --- a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts +++ b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts @@ -509,7 +509,7 @@ Deno.test("workspace Worker sidebar links New to the dedicated create page", asy ); assert( newWorkerPage.includes("worker-launch-form") && - newWorkerPage.includes("buildBrowserCreateWorkerRequest") && + newWorkerPage.includes("buildCreateWorkspaceWorkerRequest") && newWorkerPage.includes("/workers/launch-options"), "New Worker page should own launch options and creation form behavior", ); diff --git a/web/workspace/src/lib/workspace/sidebar/worker-launch.test.ts b/web/workspace/src/lib/workspace/sidebar/worker-launch.test.ts index baaa8b88..c7dbcc95 100644 --- a/web/workspace/src/lib/workspace/sidebar/worker-launch.test.ts +++ b/web/workspace/src/lib/workspace/sidebar/worker-launch.test.ts @@ -1,5 +1,5 @@ import { - buildBrowserCreateWorkerRequest, + buildCreateWorkspaceWorkerRequest, defaultWorkerLaunchForm, } from "./worker-launch.ts"; import type { WorkerLaunchOptionsResponse } from "./types.ts"; @@ -182,8 +182,8 @@ Deno.test("defaultWorkerLaunchForm preserves a Ticket repository target", () => assertEquals(form.working_directory_selector, "work/ticket"); }); -Deno.test("buildBrowserCreateWorkerRequest sends working_directory id and relative cwd only", () => { - const request = buildBrowserCreateWorkerRequest({ +Deno.test("buildCreateWorkspaceWorkerRequest sends working_directory id and relative cwd only", () => { + const request = buildCreateWorkspaceWorkerRequest({ runtime_id: "embedded", display_name: "Worker", profile: "builtin:coder", @@ -198,7 +198,7 @@ Deno.test("buildBrowserCreateWorkerRequest sends working_directory id and relati runtime_id: "embedded", display_name: "Worker", profile: "builtin:coder", - initial_text: "go", + initial_submit: [{ kind: "text", content: "go" }], working_directory: { working_directory_id: "wd-1-repo", relative_cwd: "crates/yoi", @@ -206,8 +206,23 @@ Deno.test("buildBrowserCreateWorkerRequest sends working_directory id and relati }); }); -Deno.test("buildBrowserCreateWorkerRequest omits working_directory for embedded no-workdir launches", () => { - const request = buildBrowserCreateWorkerRequest({ +Deno.test("buildCreateWorkspaceWorkerRequest sends no initial segments for an empty draft", () => { + const request = buildCreateWorkspaceWorkerRequest({ + runtime_id: "embedded", + display_name: "Worker", + profile: "builtin:companion", + initial_text: " ", + working_directory_id: "", + working_directory_repository_id: "", + working_directory_selector: "", + relative_cwd: "", + }); + + assertEquals(request.initial_submit, []); +}); + +Deno.test("buildCreateWorkspaceWorkerRequest omits working_directory for embedded no-workdir launches", () => { + const request = buildCreateWorkspaceWorkerRequest({ runtime_id: "embedded", display_name: "Worker", profile: "builtin:companion", @@ -222,6 +237,6 @@ Deno.test("buildBrowserCreateWorkerRequest omits working_directory for embedded runtime_id: "embedded", display_name: "Worker", profile: "builtin:companion", - initial_text: "chat", + initial_submit: [{ kind: "text", content: "chat" }], }); }); diff --git a/web/workspace/src/lib/workspace/sidebar/worker-launch.ts b/web/workspace/src/lib/workspace/sidebar/worker-launch.ts index f1f50d13..e6068cb0 100644 --- a/web/workspace/src/lib/workspace/sidebar/worker-launch.ts +++ b/web/workspace/src/lib/workspace/sidebar/worker-launch.ts @@ -1,3 +1,5 @@ +import type { Segment } from "$lib/generated/protocol"; + import type { BrowserWorkerWorkingDirectorySelection, WorkerLaunchOptionsResponse, @@ -14,11 +16,11 @@ export type WorkerLaunchFormState = { relative_cwd: string; }; -export type BrowserCreateWorkerRequest = { +export type CreateWorkspaceWorkerRequest = { runtime_id: string; display_name: string; profile: string; - initial_text: string; + initial_submit: Segment[]; working_directory?: BrowserWorkerWorkingDirectorySelection; }; @@ -92,14 +94,16 @@ export function defaultWorkerLaunchForm( }; } -export function buildBrowserCreateWorkerRequest( +export function buildCreateWorkspaceWorkerRequest( form: WorkerLaunchFormState, -): BrowserCreateWorkerRequest { - const request: BrowserCreateWorkerRequest = { +): CreateWorkspaceWorkerRequest { + const request: CreateWorkspaceWorkerRequest = { runtime_id: form.runtime_id, display_name: form.display_name, profile: form.profile, - initial_text: form.initial_text, + initial_submit: form.initial_text.trim() + ? [{ kind: "text", content: form.initial_text }] + : [], }; if (form.working_directory_id) { request.working_directory = { diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/new/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/workers/new/+page.svelte index 06605624..87750c61 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/new/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/workers/new/+page.svelte @@ -3,7 +3,7 @@ import { untrack } from 'svelte'; import { workspaceApiPath } from '$lib/workspace/api/http'; import { formatCurrentWorkdirRevision } from '$lib/workspace/settings/workdir-revision'; - import { buildBrowserCreateWorkerRequest, defaultWorkerLaunchForm } from '$lib/workspace/sidebar/worker-launch'; + import { buildCreateWorkspaceWorkerRequest, defaultWorkerLaunchForm } from '$lib/workspace/sidebar/worker-launch'; import type { BrowserCreateWorkerResponse, BrowserWorkingDirectoryCreateResponse, @@ -210,7 +210,7 @@ const response = await fetch(workerApiPath('/workers'), { method: 'POST', headers: { 'content-type': 'application/json' }, - body: JSON.stringify(buildBrowserCreateWorkerRequest({ + body: JSON.stringify(buildCreateWorkspaceWorkerRequest({ runtime_id: runtimeId, display_name: displayName, profile,