Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: accept queued tickets after coder spawn

Browse Source
This commit is contained in:
Hare
2026-08-19 04:55:29 +09:00
parent d7e35ea9ee
commit 08daf782b9
5 changed files with 112 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/feature/builtin/orchestration.rs
+10 -7
View File
@@ -45,7 +45,7 @@ impl FeatureModule for OrchestrationFeature {
))
.with_tool(ToolDeclaration::new(
TOOL_NAME,
"Spawn and atomically assign a Coder Worker for an inprogress Ticket. The profile, Flow, display name, assignment operation, and initial message are fixed by orchestration policy.",
"Spawn and atomically assign a Coder Worker for a queued or already-inprogress Ticket. The guarded operation records queued acceptance only after spawn, initial input, assignment, and Workdir finalization. The profile, Flow, display name, assignment operation, and initial message are fixed by orchestration policy.",
))
}
@@ -96,9 +96,12 @@ impl Tool for SpawnTicketCoderTool {
.ticket_service
.workflow_state(&ticket_id)
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
if workflow_state != ticket::TicketWorkflowState::InProgress {
if !matches!(
workflow_state,
ticket::TicketWorkflowState::Queued | ticket::TicketWorkflowState::InProgress
) {
return Err(ToolError::ExecutionFailed(format!(
"Ticket {ticket_id} must be inprogress before spawning its Coder; current state is {}",
"Ticket {ticket_id} must be queued or inprogress before spawning its Coder; current state is {}",
workflow_state.as_str()
)));
}
@@ -206,7 +209,7 @@ mod tests {
impl TicketService for RecordingTicketService {
fn workflow_state(&self, _ticket_id: &str) -> Result<TicketWorkflowState, TicketError> {
Ok(TicketWorkflowState::InProgress)
Ok(TicketWorkflowState::Queued)
}
}
@@ -277,10 +280,10 @@ mod tests {
}
#[tokio::test]
async fn spawn_ticket_coder_rejects_ticket_before_worker_side_effect() {
async fn spawn_ticket_coder_rejects_ineligible_ticket_before_worker_side_effect() {
let worker_service = Arc::new(RecordingService::default());
let tool = SpawnTicketCoderTool {
ticket_service: Arc::new(FixedTicketService(TicketWorkflowState::Queued)),
ticket_service: Arc::new(FixedTicketService(TicketWorkflowState::Planning)),
worker_service: worker_service.clone(),
};
let error = tool
@@ -295,7 +298,7 @@ mod tests {
)
.await
.unwrap_err();
assert!(error.to_string().contains("must be inprogress"));
assert!(error.to_string().contains("must be queued or inprogress"));
assert!(worker_service.requests.lock().unwrap().is_empty());
}
crates/workspace-server/src/server.rs
+99 -4
View File
@@ -2902,9 +2902,13 @@ fn validate_ticket_assignment_state(
assignment: &WorkerTicketAssignmentRequest,
) -> Result<()> {
let ticket = api.authority.ticket(&assignment.ticket_id)?;
if ticket.state != TicketWorkflowState::InProgress.as_str() {
if !matches!(
ticket.state.as_str(),
state if state == TicketWorkflowState::Queued.as_str()
|| state == TicketWorkflowState::InProgress.as_str()
) {
return Err(Error::TicketAssignmentConflict(format!(
"Ticket {} must be inprogress before assigning an implementation Coder; current state is {}",
"Ticket {} must be queued or inprogress before assigning an implementation Coder; current state is {}",
ticket.id, ticket.state
)));
}
@@ -3072,6 +3076,33 @@ fn assign_ticket_worker_from_lifecycle(
.current)
}
fn accept_queued_ticket_after_worker_spawn(
api: &WorkspaceApi,
assignment: &crate::hosts::WorkerTicketAssignmentRequest,
) -> Result<()> {
let ticket = api.authority.ticket(&assignment.ticket_id)?;
if ticket.state == TicketWorkflowState::InProgress.as_str() {
return Ok(());
}
if ticket.state != TicketWorkflowState::Queued.as_str() {
return Err(Error::TicketAssignmentConflict(format!(
"Ticket {} left queued state before Coder spawn acceptance; current state is {}",
ticket.id, ticket.state
)));
}
let mut change = TicketStateChange::new(
TicketWorkflowState::Queued.as_str(),
TicketWorkflowState::InProgress.as_str(),
"Coder spawn, assignment, and initial input were durably accepted",
"",
);
change.author = Some("workspace-orchestrator".to_string());
browser_ticket_backend(api)?
.set_workflow_state(TicketIdOrSlug::Id(ticket.id), change)
.map_err(Error::from)?;
Ok(())
}
fn existing_lifecycle_assignment_worker(
api: &WorkspaceApi,
assignment: &crate::hosts::WorkerTicketAssignmentRequest,
@@ -9274,6 +9305,20 @@ fn browser_worker_response_from_summary(
link_worker_to_workdir(api, &worker_record, workdir_id, None)?;
}
}
if let Some(assignment) = assignment {
let context = WorkerSpawnCompensationContext {
assignment: Some(assignment),
prepared_workdir_id: selected_working_directory_id,
cleanup_spawned_workdir: false,
};
finalize_worker_spawn_stage(
api,
&worker,
&context,
WorkerSpawnFinalizeStage::TicketStateAccept,
accept_queued_ticket_after_worker_spawn(api, assignment).map_err(ApiError::from),
)?;
}
let runtime_id = worker.worker.runtime_id.clone();
let worker_id = worker.worker.worker_id.clone();
let workspace_id = api.workspace_id().to_string();
@@ -9488,6 +9533,7 @@ enum WorkerSpawnFinalizeStage {
WorkerRegistry,
TicketAssignmentBind,
TicketAssignmentCurrent,
TicketStateAccept,
WorkdirRegistry,
WorkdirAttachment,
}
@@ -9498,6 +9544,7 @@ impl WorkerSpawnFinalizeStage {
Self::WorkerRegistry => "worker_registry",
Self::TicketAssignmentBind => "ticket_assignment_bind",
Self::TicketAssignmentCurrent => "ticket_assignment_current",
Self::TicketStateAccept => "ticket_state_accept",
Self::WorkdirRegistry => "workdir_registry",
Self::WorkdirAttachment => "workdir_attachment",
}
@@ -12829,7 +12876,7 @@ mod tests {
}
#[tokio::test]
async fn ticket_assignment_spawn_requires_inprogress_before_runtime_side_effects() {
async fn ticket_assignment_spawn_requires_queued_or_inprogress_before_runtime_side_effects() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
@@ -12889,7 +12936,7 @@ mod tests {
let api = test_api(workspace.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
let mut input = ticket::NewTicket::new("Assigned Ticket");
input.workflow_state = Some(TicketWorkflowState::InProgress);
input.workflow_state = Some(TicketWorkflowState::Queued);
let ticket = backend.create(input).unwrap();
let response = create_workspace_worker(
State(api.clone()),
@@ -12914,6 +12961,10 @@ mod tests {
.unwrap()
.0;
assert_eq!(
api.authority.ticket(&ticket.id).unwrap().state,
TicketWorkflowState::InProgress.as_str()
);
let current = api
.store
.get_current_ticket_worker_assignment(&api.config.workspace_id, &ticket.id)
@@ -12932,6 +12983,50 @@ mod tests {
assert_eq!(operation.worker, Some(response.worker_ref));
}
#[tokio::test]
async fn failed_ticket_assignment_spawn_leaves_ticket_queued() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
let mut input = ticket::NewTicket::new("Queued Ticket");
input.workflow_state = Some(TicketWorkflowState::Queued);
let ticket = backend.create(input).unwrap();
let result = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(CreateWorkspaceWorkerRequest {
runtime_id: "missing-runtime".to_string(),
display_name: "Rejected Coder".to_string(),
profile: Some("builtin:coder".to_string()),
ticket_assignment: Some(CreateWorkspaceWorkerTicketAssignmentRequest {
ticket_id: ticket.id.clone(),
operation_id: "failed-queued-assignment".to_string(),
}),
initial_submit: vec![Segment::Flow {
selector: "builtin:coder-review".to_string(),
}],
working_directory: None,
control_operation_id: None,
resolved_control_operation: None,
}),
)
.await;
assert!(result.is_err());
assert_eq!(
api.authority.ticket(&ticket.id).unwrap().state,
TicketWorkflowState::Queued.as_str()
);
assert!(
api.store
.get_current_ticket_worker_assignment(&api.config.workspace_id, &ticket.id)
.unwrap()
.is_none()
);
}
#[tokio::test]
async fn worker_source_auth_rejects_cross_workspace_mutation() {
let workspace = tempfile::tempdir().unwrap();