Compare commits
5
Commits
975b4fa700
...
2b33b9158d
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2b33b9158d | ||
|
|
3a7a3307ef | ||
|
|
0496cd907b | ||
|
|
060f280fdf | ||
|
|
7aaf189247 |
@@ -102,7 +102,7 @@ async fn run_and_persist(
|
||||
session_id: session_store::SessionId,
|
||||
segment_id: session_store::SegmentId,
|
||||
input: &str,
|
||||
) -> (Engine<MockLlmClient>, agen::EngineResult) {
|
||||
) -> (Engine<MockLlmClient>, agen::EngineRunExit) {
|
||||
// Mirror Worker's run-entry contract: log the user input as segments
|
||||
// before the worker pushes its flattened user_message; save_delta
|
||||
// skips the resulting user_message item to avoid double-write.
|
||||
@@ -125,31 +125,49 @@ async fn run_and_persist(
|
||||
session_store::save_turn_end(store, session_id, segment_id, worker.turn_count()).unwrap();
|
||||
|
||||
match &result {
|
||||
Ok(r) => {
|
||||
agen::EngineRunExit::Finished
|
||||
| agen::EngineRunExit::Paused
|
||||
| agen::EngineRunExit::Yielded => {
|
||||
let (legacy_result, interrupted) = match &result {
|
||||
agen::EngineRunExit::Finished => (agen::EngineResult::Finished, false),
|
||||
agen::EngineRunExit::Paused => (agen::EngineResult::Paused, true),
|
||||
agen::EngineRunExit::Yielded => (agen::EngineResult::Yielded, true),
|
||||
agen::EngineRunExit::Interrupted(_) => unreachable!(),
|
||||
};
|
||||
session_store::save_run_completed(
|
||||
store,
|
||||
session_id,
|
||||
segment_id,
|
||||
r.clone(),
|
||||
worker.last_run_interrupted(),
|
||||
legacy_result,
|
||||
interrupted,
|
||||
worker.active_run_turn_count(),
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
Err(e) => {
|
||||
agen::EngineRunExit::Interrupted(agen::StopReason::LimitReached) => {
|
||||
session_store::save_run_completed(
|
||||
store,
|
||||
session_id,
|
||||
segment_id,
|
||||
agen::EngineResult::LimitReached,
|
||||
false,
|
||||
worker.active_run_turn_count(),
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
agen::EngineRunExit::Interrupted(reason) => {
|
||||
session_store::save_run_errored(
|
||||
store,
|
||||
session_id,
|
||||
segment_id,
|
||||
e.to_string(),
|
||||
worker.last_run_interrupted(),
|
||||
format!("{reason:?}"),
|
||||
true,
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
let r = result.unwrap();
|
||||
(worker, r)
|
||||
(worker, result)
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
@@ -292,7 +310,7 @@ async fn session_resume_after_pause() {
|
||||
.unwrap();
|
||||
|
||||
let (_worker, result) = run_and_persist(worker, &store, sid, segid, "Weather?").await;
|
||||
assert!(matches!(result, agen::EngineResult::Paused));
|
||||
assert!(matches!(result, agen::EngineRunExit::Paused));
|
||||
|
||||
// Check RunCompleted is Paused
|
||||
let entries = store.read_all(sid, segid).unwrap();
|
||||
|
||||
@@ -581,6 +581,7 @@ pub(crate) enum IntakeRegistryUpdate {
|
||||
pub(crate) struct ReadyTicketPlanningReturnRequest {
|
||||
workspace_root: PathBuf,
|
||||
ticket_id: String,
|
||||
ticket_key: String,
|
||||
user_instruction: String,
|
||||
followup: ReadyTicketPlanningReturnFollowup,
|
||||
}
|
||||
@@ -2042,11 +2043,18 @@ impl DashboardApp {
|
||||
return None;
|
||||
};
|
||||
let ticket_id = ticket.id.clone();
|
||||
let ticket_key = match required_ticket_handoff_key(ticket.resource_key.as_deref()) {
|
||||
Ok(ticket_key) => ticket_key.to_string(),
|
||||
Err(error) => {
|
||||
self.notice = Some(error);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
let mut context =
|
||||
TicketRoleLaunchContext::new(current_workspace_root(), TicketRole::Intake);
|
||||
context.ticket = Some(TicketRef::id(ticket_id.clone()));
|
||||
context.user_instruction = Some(format!(
|
||||
"Continue Intake for existing Ticket {ticket_id}. Do not create a duplicate Ticket unless the user explicitly requests one. Read ShowTicket body/thread/artifacts before making routing or requirements decisions."
|
||||
"Continue Intake for existing Ticket {ticket_key}. Do not create a duplicate Ticket unless the user explicitly requests one. Read ShowTicket body/thread/artifacts before making routing or requirements decisions."
|
||||
));
|
||||
let store = match PanelRegistryStore::default_for_workspace(&context.workspace_root) {
|
||||
Ok(store) => store,
|
||||
@@ -2059,7 +2067,7 @@ impl DashboardApp {
|
||||
Ok(Some(claim)) => {
|
||||
let status = local_claim_status_for_pod(&claim.worker_name, &self.list);
|
||||
self.notice = Some(existing_ticket_claim_notice(
|
||||
&ticket_id,
|
||||
&ticket_key,
|
||||
&claim.worker_name,
|
||||
status,
|
||||
));
|
||||
@@ -2087,7 +2095,7 @@ impl DashboardApp {
|
||||
self.sending = true;
|
||||
self.notice = Some(format!(
|
||||
"Launching Ticket Intake for {} as {}…",
|
||||
ticket_id, planned.worker_name
|
||||
ticket_key, planned.worker_name
|
||||
));
|
||||
Some(IntakeLaunchRequest {
|
||||
context,
|
||||
@@ -2158,10 +2166,17 @@ impl DashboardApp {
|
||||
return None;
|
||||
};
|
||||
let ticket_id = ticket.id.clone();
|
||||
let ticket_key = match required_ticket_handoff_key(ticket.resource_key.as_deref()) {
|
||||
Ok(ticket_key) => ticket_key.to_string(),
|
||||
Err(error) => {
|
||||
self.notice = Some(error);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
if ticket.workflow_state != TicketWorkflowState::Ready {
|
||||
self.notice = Some(format!(
|
||||
"Ticket {} is {}; expected ready before returning to planning.",
|
||||
ticket_id,
|
||||
ticket_key,
|
||||
ticket.workflow_state.as_str()
|
||||
));
|
||||
return None;
|
||||
@@ -2213,7 +2228,7 @@ impl DashboardApp {
|
||||
TicketRoleLaunchContext::new(workspace_root.clone(), TicketRole::Intake);
|
||||
context.ticket = Some(TicketRef::id(ticket_id.clone()));
|
||||
context.user_instruction = Some(build_ready_ticket_refinement_launch_instruction(
|
||||
&ticket_id,
|
||||
&ticket_key,
|
||||
&user_instruction,
|
||||
));
|
||||
let peer_registration = self.prepare_intake_peer_registration(&mut context);
|
||||
@@ -2237,11 +2252,12 @@ impl DashboardApp {
|
||||
self.sending = true;
|
||||
self.notice = Some(format!(
|
||||
"Returning ready Ticket {} to planning for refinement…",
|
||||
ticket_id
|
||||
ticket_key
|
||||
));
|
||||
Some(ReadyTicketPlanningReturnRequest {
|
||||
workspace_root,
|
||||
ticket_id,
|
||||
ticket_key,
|
||||
user_instruction,
|
||||
followup,
|
||||
})
|
||||
@@ -3918,21 +3934,35 @@ fn bounded_refinement_instruction(input: &str) -> String {
|
||||
.to_string()
|
||||
}
|
||||
|
||||
fn build_ready_ticket_refinement_thread_body(ticket_id: &str, instruction: &str) -> String {
|
||||
fn required_ticket_handoff_key(resource_key: Option<&str>) -> Result<&str, String> {
|
||||
let resource_key = resource_key.ok_or_else(|| {
|
||||
"Ticket handoff is unavailable because the canonical T-* resource key is missing. Refresh the panel and retry."
|
||||
.to_string()
|
||||
})?;
|
||||
let sequence = resource_key.strip_prefix("T-").filter(|sequence| {
|
||||
!sequence.is_empty() && sequence.bytes().all(|byte| byte.is_ascii_digit())
|
||||
});
|
||||
sequence.map(|_| resource_key).ok_or_else(|| {
|
||||
"Ticket handoff is unavailable because the canonical T-* resource key is invalid. Refresh the panel and retry."
|
||||
.to_string()
|
||||
})
|
||||
}
|
||||
|
||||
fn build_ready_ticket_refinement_thread_body(ticket_key: &str, instruction: &str) -> String {
|
||||
format!(
|
||||
"Panel returned ready Ticket {ticket_id} to planning for requirements sync. This is not Queue routing and must not start implementation.\n\n## User refinement instruction\n\n{instruction}\n"
|
||||
"Panel returned ready Ticket {ticket_key} to planning for requirements sync. This is not Queue routing and must not start implementation.\n\n## User refinement instruction\n\n{instruction}\n"
|
||||
)
|
||||
}
|
||||
|
||||
fn build_ready_ticket_refinement_launch_instruction(ticket_id: &str, instruction: &str) -> String {
|
||||
fn build_ready_ticket_refinement_launch_instruction(ticket_key: &str, instruction: &str) -> String {
|
||||
format!(
|
||||
"Continue Ticket Intake / requirements sync for existing Ticket {ticket_id}. The Panel has returned the Ticket from ready to planning; do not queue the Ticket, do not route implementation, and do not create a duplicate unless the user explicitly asks for one. Read ShowTicket body/thread/artifacts before making requirements or readiness decisions.\n\nUser refinement instruction:\n\n{instruction}"
|
||||
"Continue Ticket Intake / requirements sync for existing Ticket {ticket_key}. The Panel has returned the Ticket from ready to planning; do not queue the Ticket, do not route implementation, and do not create a duplicate unless the user explicitly asks for one. Read ShowTicket body/thread/artifacts before making requirements or readiness decisions.\n\nUser refinement instruction:\n\n{instruction}"
|
||||
)
|
||||
}
|
||||
|
||||
fn build_ready_ticket_refinement_notify(ticket_id: &str, instruction: &str) -> String {
|
||||
fn build_ready_ticket_refinement_notify(ticket_key: &str, instruction: &str) -> String {
|
||||
format!(
|
||||
"Ticket {ticket_id} was returned from ready to planning from the Panel for requirements sync. Continue Intake/refinement only; do not Queue or route implementation. Read the Ticket thread for the recorded state change and user instruction.\n\nUser refinement instruction:\n\n{instruction}"
|
||||
"Ticket {ticket_key} was returned from ready to planning from the Panel for requirements sync. Continue Intake/refinement only; do not Queue or route implementation. Read the Ticket thread for the recorded state change and user instruction.\n\nUser refinement instruction:\n\n{instruction}"
|
||||
)
|
||||
}
|
||||
|
||||
@@ -3961,10 +3991,12 @@ async fn dispatch_ready_ticket_planning_return(
|
||||
let ticket = backend
|
||||
.show(id.clone())
|
||||
.map_err(|error| TicketActionError::Ticket(error.to_string()))?;
|
||||
let ticket_key =
|
||||
required_ticket_handoff_key(Some(&request.ticket_key)).map_err(TicketActionError::Stale)?;
|
||||
if ticket.meta.workflow_state != TicketWorkflowState::Ready {
|
||||
return Err(TicketActionError::Stale(format!(
|
||||
"Ticket {} is {}; expected ready before returning it to planning. Refresh the panel and retry if appropriate.",
|
||||
ticket.meta.id,
|
||||
ticket_key,
|
||||
ticket.meta.workflow_state.as_str()
|
||||
)));
|
||||
}
|
||||
@@ -3973,7 +4005,7 @@ async fn dispatch_ready_ticket_planning_return(
|
||||
TicketWorkflowState::Planning.as_str(),
|
||||
"panel_return_to_planning",
|
||||
MarkdownText::from(build_ready_ticket_refinement_thread_body(
|
||||
&ticket.meta.id,
|
||||
ticket_key,
|
||||
&request.user_instruction,
|
||||
)),
|
||||
);
|
||||
@@ -3987,7 +4019,7 @@ async fn dispatch_ready_ticket_planning_return(
|
||||
ReadyTicketPlanningReturnOutcome {
|
||||
notice: format!(
|
||||
"Ticket {} returned to planning for refinement; launching Ticket Intake…",
|
||||
ticket.meta.id
|
||||
ticket_key
|
||||
),
|
||||
followup: ReadyTicketPlanningReturnAfterMutation::LaunchIntake(request),
|
||||
}
|
||||
@@ -3997,19 +4029,19 @@ async fn dispatch_ready_ticket_planning_return(
|
||||
socket_path,
|
||||
} => {
|
||||
let message =
|
||||
build_ready_ticket_refinement_notify(&ticket.meta.id, &request.user_instruction);
|
||||
build_ready_ticket_refinement_notify(ticket_key, &request.user_instruction);
|
||||
match send_notify_only(&socket_path, message, true).await {
|
||||
Ok(()) => ReadyTicketPlanningReturnOutcome {
|
||||
notice: format!(
|
||||
"Ticket {} returned to planning for refinement; notified live Intake Worker {}.",
|
||||
ticket.meta.id, worker_name
|
||||
ticket_key, worker_name
|
||||
),
|
||||
followup: ReadyTicketPlanningReturnAfterMutation::None,
|
||||
},
|
||||
Err(error) => ReadyTicketPlanningReturnOutcome {
|
||||
notice: bounded_panel_diagnostic(format!(
|
||||
"Ticket {} returned to planning and instruction was recorded, but notifying Intake Worker {} failed: {}",
|
||||
ticket.meta.id, worker_name, error
|
||||
ticket_key, worker_name, error
|
||||
)),
|
||||
followup: ReadyTicketPlanningReturnAfterMutation::None,
|
||||
},
|
||||
@@ -4020,7 +4052,7 @@ async fn dispatch_ready_ticket_planning_return(
|
||||
ReadyTicketPlanningReturnOutcome {
|
||||
notice: format!(
|
||||
"Ticket {} returned to planning for refinement; opening/restoring claimed Intake Worker {}…",
|
||||
ticket.meta.id, worker_name
|
||||
ticket_key, worker_name
|
||||
),
|
||||
followup: ReadyTicketPlanningReturnAfterMutation::OpenClaim(request),
|
||||
}
|
||||
@@ -4029,7 +4061,7 @@ async fn dispatch_ready_ticket_planning_return(
|
||||
ReadyTicketPlanningReturnOutcome {
|
||||
notice: bounded_panel_diagnostic(format!(
|
||||
"Ticket {} returned to planning and instruction was recorded, but Intake launch was not attempted because existing Intake claim {} is stale; inspect or clear the local claim before launching another Intake Worker.",
|
||||
ticket.meta.id, worker_name
|
||||
ticket_key, worker_name
|
||||
)),
|
||||
followup: ReadyTicketPlanningReturnAfterMutation::None,
|
||||
}
|
||||
|
||||
@@ -390,6 +390,7 @@ fn planning_return_request(
|
||||
ReadyTicketPlanningReturnRequest {
|
||||
workspace_root: temp.path().to_path_buf(),
|
||||
ticket_id,
|
||||
ticket_key: "T-482".to_string(),
|
||||
user_instruction: instruction.to_string(),
|
||||
followup: ReadyTicketPlanningReturnFollowup::BlockedByStaleClaim {
|
||||
worker_name: "stale-intake".to_string(),
|
||||
@@ -494,6 +495,7 @@ fn ready_ticket_intake_enter_prepares_planning_return_not_queue_or_generic_launc
|
||||
};
|
||||
|
||||
assert_eq!(request.ticket_id, "20260608-000123-ready");
|
||||
assert_eq!(request.ticket_key, "T-1");
|
||||
assert_eq!(request.user_instruction, "clarify expected behavior");
|
||||
assert!(matches!(
|
||||
request.followup,
|
||||
@@ -515,6 +517,7 @@ async fn planning_return_with_launch_followup_changes_state_before_launch_follow
|
||||
let request = ReadyTicketPlanningReturnRequest {
|
||||
workspace_root: temp.path().to_path_buf(),
|
||||
ticket_id: ticket_id.clone(),
|
||||
ticket_key: "T-482".to_string(),
|
||||
user_instruction: "launch intake after state change".to_string(),
|
||||
followup: ReadyTicketPlanningReturnFollowup::LaunchIntake(IntakeLaunchRequest {
|
||||
context: TicketRoleLaunchContext::new(temp.path().to_path_buf(), TicketRole::Intake),
|
||||
@@ -3425,6 +3428,27 @@ fn ticket_action_error_records_f2_diagnostic_details() {
|
||||
assert!(!app.panel_diagnostic_open);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ready_ticket_refinement_projection_uses_only_canonical_resource_key() {
|
||||
const INTERNAL_ID: &str = "00001KZVNXFNK";
|
||||
let thread = build_ready_ticket_refinement_thread_body("T-482", "Clarify rollback.");
|
||||
let launch = build_ready_ticket_refinement_launch_instruction("T-482", "Clarify rollback.");
|
||||
let notify = build_ready_ticket_refinement_notify("T-482", "Clarify rollback.");
|
||||
|
||||
for projection in [&thread, &launch, ¬ify] {
|
||||
assert!(projection.contains("T-482"));
|
||||
assert!(!projection.contains(INTERNAL_ID));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ticket_handoff_fails_closed_without_canonical_resource_key() {
|
||||
assert_eq!(required_ticket_handoff_key(Some("T-482")), Ok("T-482"));
|
||||
for invalid in [None, Some(""), Some("00001KZVNXFNK"), Some("T-key")] {
|
||||
assert!(required_ticket_handoff_key(invalid).is_err());
|
||||
}
|
||||
}
|
||||
|
||||
fn plain_line(line: &Line<'_>) -> String {
|
||||
line.spans
|
||||
.iter()
|
||||
|
||||
@@ -251,14 +251,8 @@ impl DelegatingWorkdirSession {
|
||||
self.ensure_path(path, WorkdirDelegationPermission::Write)
|
||||
}
|
||||
|
||||
fn ensure_command(&self, starting: bool) -> Result<(), WorkdirError> {
|
||||
self.ensure_capability(WorkdirSessionCapability::Command, "command execution")?;
|
||||
if starting && self.has_active_write_lease() {
|
||||
return Err(WorkdirError::Denied(
|
||||
"command execution is denied while a child holds a write delegation".into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
fn ensure_command(&self) -> Result<(), WorkdirError> {
|
||||
self.ensure_capability(WorkdirSessionCapability::Command, "command execution")
|
||||
}
|
||||
|
||||
fn ensure_parent_write_available(&self, path: &FsPath) -> Result<(), WorkdirError> {
|
||||
@@ -281,20 +275,6 @@ impl DelegatingWorkdirSession {
|
||||
}
|
||||
}
|
||||
|
||||
fn has_active_write_lease(&self) -> bool {
|
||||
let mut leases = self
|
||||
.child_write_leases
|
||||
.lock()
|
||||
.expect("workdir delegation lease mutex poisoned");
|
||||
leases.retain(|_, lease| lease.validity.upgrade().is_some_and(|v| v.is_active()));
|
||||
leases.values().any(|lease| {
|
||||
lease
|
||||
.rules
|
||||
.iter()
|
||||
.any(|rule| rule.permission == WorkdirDelegationPermission::Write)
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_delegation_rules(
|
||||
&self,
|
||||
rules: &[WorkdirDelegationRule],
|
||||
@@ -503,12 +483,12 @@ impl WorkdirSession for DelegatingWorkdirSession {
|
||||
}
|
||||
|
||||
async fn start_command(&self, request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
|
||||
self.ensure_command(true)?;
|
||||
self.ensure_command()?;
|
||||
self.source.start_command(request).await
|
||||
}
|
||||
|
||||
async fn command_status(&self, handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
|
||||
self.ensure_command(false)?;
|
||||
self.ensure_command()?;
|
||||
self.source.command_status(handle).await
|
||||
}
|
||||
|
||||
@@ -516,12 +496,12 @@ impl WorkdirSession for DelegatingWorkdirSession {
|
||||
&self,
|
||||
request: CommandOutputRequest,
|
||||
) -> Result<CommandOutput, WorkdirError> {
|
||||
self.ensure_command(false)?;
|
||||
self.ensure_command()?;
|
||||
self.source.command_output(request).await
|
||||
}
|
||||
|
||||
async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> {
|
||||
self.ensure_command(false)?;
|
||||
self.ensure_command()?;
|
||||
self.source.cancel_command(handle).await
|
||||
}
|
||||
|
||||
@@ -753,6 +733,31 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_command(
|
||||
session: &WorkdirSessionHandle,
|
||||
command: impl Into<String>,
|
||||
tool_call_id: impl Into<String>,
|
||||
) -> CommandOutput {
|
||||
let handle = session
|
||||
.start_command(CommandRequest {
|
||||
command: command.into(),
|
||||
timeout_secs: 5,
|
||||
output_limit: 1024,
|
||||
tool_call_id: Some(tool_call_id.into()),
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
session
|
||||
.command_output(CommandOutputRequest {
|
||||
handle,
|
||||
cursor: 0,
|
||||
limit: 1024,
|
||||
wait: true,
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delegation_capable_session_forwards_command_telemetry() {
|
||||
let root = TempDir::new().unwrap();
|
||||
@@ -846,6 +851,18 @@ mod tests {
|
||||
);
|
||||
assert!(child.scoped_session.subscribe_command_events().is_none());
|
||||
assert!(child.scoped_session.command_snapshot().is_empty());
|
||||
assert!(matches!(
|
||||
child
|
||||
.scoped_session
|
||||
.start_command(CommandRequest {
|
||||
command: "printf denied".into(),
|
||||
timeout_secs: 5,
|
||||
output_limit: 1024,
|
||||
tool_call_id: Some("read-only-command".into()),
|
||||
})
|
||||
.await,
|
||||
Err(WorkdirError::Denied(_))
|
||||
));
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
@@ -924,7 +941,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn write_lease_blocks_parent_region_until_release() {
|
||||
async fn write_lease_keeps_typed_parent_writes_exclusive_without_blocking_commands() {
|
||||
let root = TempDir::new().unwrap();
|
||||
fs::create_dir_all(root.path().join("leased")).unwrap();
|
||||
fs::create_dir_all(root.path().join("other")).unwrap();
|
||||
@@ -938,37 +955,24 @@ mod tests {
|
||||
.capabilities
|
||||
.supports(WorkdirSessionCapability::Command)
|
||||
);
|
||||
let command = child
|
||||
.scoped_session
|
||||
.start_command(CommandRequest {
|
||||
command: "printf child-command".into(),
|
||||
timeout_secs: 5,
|
||||
output_limit: 1024,
|
||||
tool_call_id: Some("delegated-child-command".into()),
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let command_output = child
|
||||
.scoped_session
|
||||
.command_output(CommandOutputRequest {
|
||||
handle: command,
|
||||
cursor: 0,
|
||||
limit: 1024,
|
||||
wait: true,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(command_output.content, "child-command");
|
||||
assert!(
|
||||
parent
|
||||
.start_command(CommandRequest {
|
||||
command: "printf parent-command".into(),
|
||||
timeout_secs: 5,
|
||||
output_limit: 1024,
|
||||
tool_call_id: Some("blocked-parent-command".into()),
|
||||
})
|
||||
.await
|
||||
.is_err()
|
||||
let child_output = run_command(
|
||||
&child.scoped_session,
|
||||
"printf child-command",
|
||||
"delegated-child-command",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(child_output.content, "child-command");
|
||||
let parent_output = run_command(
|
||||
&parent,
|
||||
"printf parent-write > leased/from-command; printf parent-command",
|
||||
"parent-command-during-child-write",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(parent_output.status, CommandStatus::Completed);
|
||||
assert_eq!(parent_output.content, "parent-command");
|
||||
assert_eq!(
|
||||
fs::read_to_string(root.path().join("leased/from-command")).unwrap(),
|
||||
"parent-write"
|
||||
);
|
||||
|
||||
assert!(matches!(
|
||||
@@ -982,6 +986,18 @@ mod tests {
|
||||
.await
|
||||
.unwrap();
|
||||
child.release();
|
||||
assert!(matches!(
|
||||
child
|
||||
.scoped_session
|
||||
.start_command(CommandRequest {
|
||||
command: "printf revoked".into(),
|
||||
timeout_secs: 5,
|
||||
output_limit: 1024,
|
||||
tool_call_id: Some("revoked-child-command".into()),
|
||||
})
|
||||
.await,
|
||||
Err(WorkdirError::SessionClosed)
|
||||
));
|
||||
parent
|
||||
.write(write("leased/parent", "parent"))
|
||||
.await
|
||||
@@ -1033,6 +1049,78 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn nested_write_leases_do_not_block_command_capable_ancestors() {
|
||||
let root = TempDir::new().unwrap();
|
||||
fs::create_dir_all(root.path().join("docs/sub")).unwrap();
|
||||
let root_session = session(root.path());
|
||||
let child = root_session
|
||||
.delegate(request("docs", WorkdirDelegationPermission::Write))
|
||||
.await
|
||||
.unwrap();
|
||||
let nested = child
|
||||
.scoped_session
|
||||
.delegate(request("docs/sub", WorkdirDelegationPermission::Write))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
for (session, label) in [
|
||||
(&root_session, "root"),
|
||||
(&child.scoped_session, "child"),
|
||||
(&nested.scoped_session, "nested"),
|
||||
] {
|
||||
let output = run_command(
|
||||
session,
|
||||
format!("printf {label}"),
|
||||
format!("{label}-command-during-nested-write"),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(output.status, CommandStatus::Completed);
|
||||
assert_eq!(output.content, label);
|
||||
}
|
||||
|
||||
assert!(matches!(
|
||||
root_session.write(write("docs/root", "blocked")).await,
|
||||
Err(WorkdirError::Denied(_))
|
||||
));
|
||||
assert!(matches!(
|
||||
child
|
||||
.scoped_session
|
||||
.write(write("sub/child", "blocked"))
|
||||
.await,
|
||||
Err(WorkdirError::Denied(_))
|
||||
));
|
||||
nested
|
||||
.scoped_session
|
||||
.write(write("nested", "allowed"))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
nested.release();
|
||||
child.release();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reapplied_write_delegation_chain_forwards_command_lifecycle() {
|
||||
let root = TempDir::new().unwrap();
|
||||
fs::create_dir_all(root.path().join("delegated")).unwrap();
|
||||
let applied = apply_delegation_chain(
|
||||
session(root.path()),
|
||||
[request("delegated", WorkdirDelegationPermission::Write)],
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let output = run_command(
|
||||
&applied.scoped_session,
|
||||
"printf reapplied",
|
||||
"reapplied-command",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(output.status, CommandStatus::Completed);
|
||||
assert_eq!(output.content, "reapplied");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn applied_chain_cannot_replace_outer_provider_attenuation() {
|
||||
let root = TempDir::new().unwrap();
|
||||
@@ -1077,6 +1165,17 @@ mod tests {
|
||||
.unwrap();
|
||||
|
||||
parent.close().await.unwrap();
|
||||
assert!(matches!(
|
||||
parent
|
||||
.start_command(CommandRequest {
|
||||
command: "printf closed".into(),
|
||||
timeout_secs: 5,
|
||||
output_limit: 1024,
|
||||
tool_call_id: Some("closed-parent-command".into()),
|
||||
})
|
||||
.await,
|
||||
Err(WorkdirError::SessionClosed)
|
||||
));
|
||||
assert!(matches!(
|
||||
child.scoped_session.read(read("a")).await,
|
||||
Err(WorkdirError::SessionClosed)
|
||||
|
||||
@@ -66,6 +66,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
WorkerRunResult::Finished => println!("(finished)"),
|
||||
WorkerRunResult::Paused => println!("(paused)"),
|
||||
WorkerRunResult::LimitReached => println!("(turn limit reached)"),
|
||||
WorkerRunResult::Interrupted { message, .. } => println!("(interrupted: {message})"),
|
||||
WorkerRunResult::RolledBack => println!("(empty turn rolled back)"),
|
||||
}
|
||||
|
||||
|
||||
@@ -1650,13 +1650,13 @@ where
|
||||
WorkerRunResult::Paused => (WorkerStatus::Paused, RunResult::Paused),
|
||||
WorkerRunResult::LimitReached => (WorkerStatus::Idle, RunResult::LimitReached),
|
||||
WorkerRunResult::RolledBack => (WorkerStatus::Idle, RunResult::RolledBack),
|
||||
WorkerRunResult::Interrupted(_message) if pause_requested => {
|
||||
WorkerRunResult::Interrupted { .. } if pause_requested => {
|
||||
let _ = event_tx.send(Event::RunEnd { result: RunResult::Paused });
|
||||
return (WorkerStatus::Paused, shutdown_requested);
|
||||
}
|
||||
WorkerRunResult::Interrupted(message) => {
|
||||
WorkerRunResult::Interrupted { code, message } => {
|
||||
let _ = event_tx.send(Event::Error {
|
||||
code: ErrorCode::Internal,
|
||||
code,
|
||||
message: message.clone(),
|
||||
});
|
||||
if parent_originated {
|
||||
|
||||
@@ -89,18 +89,19 @@ impl Tool for SpawnTicketCoderTool {
|
||||
let input: SpawnTicketCoderInput = serde_json::from_str(input_json).map_err(|error| {
|
||||
ToolError::InvalidArgument(format!("invalid {TOOL_NAME} input: {error}"))
|
||||
})?;
|
||||
let ticket_id = authority_id(input.ticket_id, "ticket_id")?;
|
||||
let workflow_state = self
|
||||
let ticket_ref = authority_id(input.ticket_id, "ticket_id")?;
|
||||
let ticket = self
|
||||
.ticket_service
|
||||
.workflow_state(&ticket_id)
|
||||
.ticket_handoff(&ticket_ref)
|
||||
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
|
||||
if !matches!(
|
||||
workflow_state,
|
||||
ticket.workflow_state,
|
||||
ticket::TicketWorkflowState::Queued | ticket::TicketWorkflowState::InProgress
|
||||
) {
|
||||
return Err(ToolError::ExecutionFailed(format!(
|
||||
"Ticket {ticket_id} must be queued or inprogress before spawning its Coder; current state is {}",
|
||||
workflow_state.as_str()
|
||||
"Ticket {} must be queued or inprogress before spawning its Coder; current state is {}",
|
||||
ticket.resource_key,
|
||||
ticket.workflow_state.as_str()
|
||||
)));
|
||||
}
|
||||
let call_id = non_empty(ctx.call_id, "tool call_id")?;
|
||||
@@ -115,14 +116,14 @@ impl Tool for SpawnTicketCoderTool {
|
||||
)?,
|
||||
relative_cwd,
|
||||
profile: CODER_PROFILE.to_string(),
|
||||
ticket_id: Some(ticket_id.clone()),
|
||||
operation_id: Some(format!("spawn-ticket-coder:{ticket_id}:{call_id}")),
|
||||
display_name: format!("Coder · {ticket_id}"),
|
||||
ticket_id: Some(ticket.id.clone()),
|
||||
operation_id: Some(format!("spawn-ticket-coder:{}:{call_id}", ticket.id)),
|
||||
display_name: format!("Coder · {}", ticket.resource_key),
|
||||
initial_submit: vec![
|
||||
Segment::Flow {
|
||||
selector: CODER_FLOW.to_string(),
|
||||
},
|
||||
Segment::text(format!("Implement Ticket {ticket_id}.")),
|
||||
Segment::text(format!("Implement Ticket {}.", ticket.resource_key)),
|
||||
],
|
||||
})
|
||||
.await
|
||||
@@ -134,7 +135,7 @@ impl Tool for SpawnTicketCoderTool {
|
||||
)));
|
||||
}
|
||||
Ok(ToolOutput {
|
||||
summary: format!("Spawned Coder for Ticket {ticket_id}"),
|
||||
summary: format!("Spawned Coder for Ticket {}", ticket.resource_key),
|
||||
content: Some(response.body),
|
||||
attachments: Vec::new(),
|
||||
})
|
||||
@@ -201,21 +202,31 @@ mod tests {
|
||||
use crate::worker::{WorkspaceClientError, WorkspaceResponse};
|
||||
|
||||
use super::*;
|
||||
use crate::feature::builtin::ticket::TicketHandoff;
|
||||
|
||||
#[derive(Default)]
|
||||
struct RecordingTicketService;
|
||||
|
||||
impl TicketService for RecordingTicketService {
|
||||
fn workflow_state(&self, _ticket_id: &str) -> Result<TicketWorkflowState, TicketError> {
|
||||
Ok(TicketWorkflowState::Queued)
|
||||
fn ticket_handoff(&self, ticket_ref: &str) -> Result<TicketHandoff, TicketError> {
|
||||
assert_eq!(ticket_ref, "T-482");
|
||||
Ok(TicketHandoff {
|
||||
id: "00001KZXN51C7".to_string(),
|
||||
resource_key: "T-482".to_string(),
|
||||
workflow_state: TicketWorkflowState::Queued,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
struct FixedTicketService(TicketWorkflowState);
|
||||
|
||||
impl TicketService for FixedTicketService {
|
||||
fn workflow_state(&self, _ticket_id: &str) -> Result<TicketWorkflowState, TicketError> {
|
||||
Ok(self.0)
|
||||
fn ticket_handoff(&self, _ticket_ref: &str) -> Result<TicketHandoff, TicketError> {
|
||||
Ok(TicketHandoff {
|
||||
id: "00001KZXN51C7".to_string(),
|
||||
resource_key: "T-482".to_string(),
|
||||
workflow_state: self.0,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -247,7 +258,7 @@ mod tests {
|
||||
};
|
||||
tool.execute(
|
||||
&serde_json::json!({
|
||||
"ticket_id": "00001KZXN51C7",
|
||||
"ticket_id": "T-482",
|
||||
"runtime_id": "runtime-1",
|
||||
"working_directory_id": "workdir-1"
|
||||
})
|
||||
@@ -265,16 +276,20 @@ mod tests {
|
||||
request.operation_id.as_deref(),
|
||||
Some("spawn-ticket-coder:00001KZXN51C7:call-7")
|
||||
);
|
||||
assert_eq!(request.display_name, "Coder · 00001KZXN51C7");
|
||||
assert_eq!(request.display_name, "Coder · T-482");
|
||||
assert_eq!(
|
||||
request.initial_submit,
|
||||
vec![
|
||||
Segment::Flow {
|
||||
selector: CODER_FLOW.to_string()
|
||||
},
|
||||
Segment::text("Implement Ticket 00001KZXN51C7.")
|
||||
Segment::text("Implement Ticket T-482.")
|
||||
]
|
||||
);
|
||||
assert!(!request.display_name.contains("00001KZXN51C7"));
|
||||
assert!(request.initial_submit.iter().all(|segment| {
|
||||
!Segment::flatten_to_text(std::slice::from_ref(segment)).contains("00001KZXN51C7")
|
||||
}));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -267,7 +267,20 @@ pub const TICKET_SERVICE_ID: &str = "ticket.authority";
|
||||
const TICKET_SERVICE_VERSION: &str = "1";
|
||||
|
||||
pub trait TicketService: Send + Sync {
|
||||
fn workflow_state(&self, ticket_id: &str) -> Result<TicketWorkflowState, TicketError>;
|
||||
fn ticket_handoff(&self, ticket_ref: &str) -> Result<TicketHandoff, TicketError>;
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct TicketHandoff {
|
||||
pub id: String,
|
||||
pub resource_key: String,
|
||||
pub workflow_state: TicketWorkflowState,
|
||||
}
|
||||
|
||||
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())
|
||||
})
|
||||
}
|
||||
|
||||
struct BackendTicketService {
|
||||
@@ -275,10 +288,18 @@ struct BackendTicketService {
|
||||
}
|
||||
|
||||
impl TicketService for BackendTicketService {
|
||||
fn workflow_state(&self, ticket_id: &str) -> Result<TicketWorkflowState, TicketError> {
|
||||
self.backend
|
||||
.show(ticket_id.into())
|
||||
.map(|ticket| ticket.meta.workflow_state)
|
||||
fn ticket_handoff(&self, ticket_ref: &str) -> Result<TicketHandoff, TicketError> {
|
||||
let ticket = self.backend.show(ticket_ref.into())?;
|
||||
let resource_key = ticket
|
||||
.meta
|
||||
.resource_key
|
||||
.filter(|key| is_canonical_ticket_resource_key(key))
|
||||
.ok_or_else(|| TicketError::Conflict("ticket resource key is unavailable".into()))?;
|
||||
Ok(TicketHandoff {
|
||||
id: ticket.meta.id,
|
||||
resource_key,
|
||||
workflow_state: ticket.meta.workflow_state,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1770,6 +1791,14 @@ provider = "github"
|
||||
assert_eq!(removed.target, "01TARGET");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ticket_handoff_accepts_only_canonical_ticket_resource_keys() {
|
||||
assert!(is_canonical_ticket_resource_key("T-482"));
|
||||
for invalid in ["", "00001KZVNXFNK", "T-", "T-key", "O-482"] {
|
||||
assert!(!is_canonical_ticket_resource_key(invalid));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn workspace_http_backend_executes_ticket_create_operation() {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
|
||||
@@ -199,10 +199,12 @@ where
|
||||
on_cancel_sender(worker.engine_mut().cancel_sender());
|
||||
|
||||
match worker.run_text(&input).await {
|
||||
Ok(WorkerRunResult::Interrupted(message)) => Err(InternalWorkerError {
|
||||
source: WorkerError::Engine(EngineError::Aborted(message)),
|
||||
Ok(lifecycle @ WorkerRunResult::Finished)
|
||||
| Ok(lifecycle @ WorkerRunResult::Paused)
|
||||
| Ok(lifecycle @ WorkerRunResult::RolledBack) => Ok(InternalWorkerResult {
|
||||
usage: last_usage.lock().ok().and_then(|slot| slot.clone()),
|
||||
identity,
|
||||
lifecycle,
|
||||
history_entries: store.entries_count(session_id, segment_id),
|
||||
}),
|
||||
Ok(WorkerRunResult::LimitReached) => Err(InternalWorkerError {
|
||||
@@ -213,10 +215,10 @@ where
|
||||
identity,
|
||||
history_entries: store.entries_count(session_id, segment_id),
|
||||
}),
|
||||
Ok(lifecycle) => Ok(InternalWorkerResult {
|
||||
Ok(WorkerRunResult::Interrupted { message, .. }) => Err(InternalWorkerError {
|
||||
source: WorkerError::Engine(EngineError::Aborted(message)),
|
||||
usage: last_usage.lock().ok().and_then(|slot| slot.clone()),
|
||||
identity,
|
||||
lifecycle,
|
||||
history_entries: store.entries_count(session_id, segment_id),
|
||||
}),
|
||||
Err(source) => Err(InternalWorkerError {
|
||||
@@ -246,6 +248,7 @@ impl Default for InternalWorkerVisibility {
|
||||
pub(crate) enum InternalWorkerSessionStatus {
|
||||
Idle,
|
||||
Running,
|
||||
Paused,
|
||||
Stopping,
|
||||
Stopped,
|
||||
Failed,
|
||||
@@ -256,9 +259,10 @@ impl InternalWorkerSessionStatus {
|
||||
match self {
|
||||
Self::Idle => 0,
|
||||
Self::Running => 1,
|
||||
Self::Stopping => 2,
|
||||
Self::Stopped => 3,
|
||||
Self::Failed => 4,
|
||||
Self::Paused => 2,
|
||||
Self::Stopping => 3,
|
||||
Self::Stopped => 4,
|
||||
Self::Failed => 5,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -266,13 +270,35 @@ impl InternalWorkerSessionStatus {
|
||||
match value {
|
||||
0 => Self::Idle,
|
||||
1 => Self::Running,
|
||||
2 => Self::Stopping,
|
||||
3 => Self::Stopped,
|
||||
2 => Self::Paused,
|
||||
3 => Self::Stopping,
|
||||
4 => Self::Stopped,
|
||||
_ => Self::Failed,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn classify_internal_turn_result(
|
||||
result: Result<WorkerRunResult, WorkerError>,
|
||||
) -> (InternalWorkerSessionStatus, Option<String>) {
|
||||
match result {
|
||||
Ok(WorkerRunResult::Finished) => (InternalWorkerSessionStatus::Idle, None),
|
||||
Ok(WorkerRunResult::Paused) => (InternalWorkerSessionStatus::Paused, None),
|
||||
Ok(WorkerRunResult::LimitReached) => (
|
||||
InternalWorkerSessionStatus::Stopped,
|
||||
Some("internal Worker reached its turn limit".to_string()),
|
||||
),
|
||||
Ok(WorkerRunResult::Interrupted { message, .. }) => {
|
||||
(InternalWorkerSessionStatus::Stopped, Some(message))
|
||||
}
|
||||
Ok(WorkerRunResult::RolledBack) => (
|
||||
InternalWorkerSessionStatus::Stopped,
|
||||
Some("internal Worker run was cancelled before AI output".to_string()),
|
||||
),
|
||||
Err(error) => (InternalWorkerSessionStatus::Failed, Some(error.to_string())),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub(crate) enum InternalWorkerSessionError {
|
||||
#[error("failed to build internal Worker session: {message}")]
|
||||
@@ -367,6 +393,7 @@ impl InternalWorkerSessionHandle {
|
||||
entries,
|
||||
status: match self.status() {
|
||||
InternalWorkerSessionStatus::Running => WorkerStatus::Running,
|
||||
InternalWorkerSessionStatus::Paused => WorkerStatus::Paused,
|
||||
InternalWorkerSessionStatus::Idle => WorkerStatus::Idle,
|
||||
InternalWorkerSessionStatus::Stopping
|
||||
| InternalWorkerSessionStatus::Stopped
|
||||
@@ -402,6 +429,7 @@ impl InternalWorkerSessionHandle {
|
||||
.map_err(
|
||||
|current| match InternalWorkerSessionStatus::decode(current) {
|
||||
InternalWorkerSessionStatus::Running
|
||||
| InternalWorkerSessionStatus::Paused
|
||||
| InternalWorkerSessionStatus::Stopping => InternalWorkerSessionError::Busy,
|
||||
InternalWorkerSessionStatus::Stopped | InternalWorkerSessionStatus::Failed => {
|
||||
InternalWorkerSessionError::Stopped
|
||||
@@ -747,21 +775,7 @@ pub(crate) async fn prepare_internal_worker_session(
|
||||
loop {
|
||||
tokio::select! {
|
||||
result = &mut run => {
|
||||
let (turn_status, error) = match result {
|
||||
Ok(WorkerRunResult::Interrupted(message)) => (
|
||||
InternalWorkerSessionStatus::Stopped,
|
||||
Some(message),
|
||||
),
|
||||
Ok(WorkerRunResult::LimitReached) => (
|
||||
InternalWorkerSessionStatus::Stopped,
|
||||
Some("internal Worker reached its turn limit".to_string()),
|
||||
),
|
||||
Ok(_) => (InternalWorkerSessionStatus::Idle, None),
|
||||
Err(error) => (
|
||||
InternalWorkerSessionStatus::Failed,
|
||||
Some(error.to_string()),
|
||||
),
|
||||
};
|
||||
let (turn_status, error) = classify_internal_turn_result(result);
|
||||
actor_in_flight.clear();
|
||||
status.store(turn_status.encode(), std::sync::atomic::Ordering::Release);
|
||||
if let Some(message) = error {
|
||||
@@ -771,10 +785,15 @@ pub(crate) async fn prepare_internal_worker_session(
|
||||
message,
|
||||
});
|
||||
}
|
||||
let protocol_status = if turn_status == InternalWorkerSessionStatus::Idle {
|
||||
WorkerStatus::Idle
|
||||
} else {
|
||||
WorkerStatus::Stopped
|
||||
let protocol_status = match turn_status {
|
||||
InternalWorkerSessionStatus::Idle => WorkerStatus::Idle,
|
||||
InternalWorkerSessionStatus::Paused => WorkerStatus::Paused,
|
||||
InternalWorkerSessionStatus::Stopped
|
||||
| InternalWorkerSessionStatus::Failed => WorkerStatus::Stopped,
|
||||
InternalWorkerSessionStatus::Running
|
||||
| InternalWorkerSessionStatus::Stopping => {
|
||||
unreachable!("run completion cannot remain active")
|
||||
}
|
||||
};
|
||||
let _ = event_tx.send(Event::Status {
|
||||
status: protocol_status,
|
||||
@@ -1261,6 +1280,52 @@ permission = "write"
|
||||
assert_eq!(result.identity.kind, "test");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn internal_turn_result_mapping_is_exhaustive() {
|
||||
let cases = [
|
||||
(
|
||||
WorkerRunResult::Finished,
|
||||
InternalWorkerSessionStatus::Idle,
|
||||
false,
|
||||
),
|
||||
(
|
||||
WorkerRunResult::Paused,
|
||||
InternalWorkerSessionStatus::Paused,
|
||||
false,
|
||||
),
|
||||
(
|
||||
WorkerRunResult::LimitReached,
|
||||
InternalWorkerSessionStatus::Stopped,
|
||||
true,
|
||||
),
|
||||
(
|
||||
WorkerRunResult::Interrupted {
|
||||
code: protocol::ErrorCode::Internal,
|
||||
message: "cancelled".to_string(),
|
||||
},
|
||||
InternalWorkerSessionStatus::Stopped,
|
||||
true,
|
||||
),
|
||||
(
|
||||
WorkerRunResult::RolledBack,
|
||||
InternalWorkerSessionStatus::Stopped,
|
||||
true,
|
||||
),
|
||||
];
|
||||
|
||||
for (result, expected_status, expects_error) in cases {
|
||||
let (status, error) = classify_internal_turn_result(Ok(result));
|
||||
assert_eq!(status, expected_status);
|
||||
assert_eq!(error.is_some(), expects_error);
|
||||
}
|
||||
|
||||
let (status, error) = classify_internal_turn_result(Err(WorkerError::Engine(
|
||||
EngineError::Aborted("fatal".to_string()),
|
||||
)));
|
||||
assert_eq!(status, InternalWorkerSessionStatus::Failed);
|
||||
assert!(error.is_some_and(|message| message.contains("fatal")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fatal_internal_run_transitions_to_stopped_protocol_status() {
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
@@ -102,7 +102,6 @@ pub enum WorkerPrompt {
|
||||
AgentsMdSection,
|
||||
ResidentMemorySummarySection,
|
||||
WorkerOrchestrationGuidanceSection,
|
||||
TicketEventCompanionNotice,
|
||||
SubWorkerSpawnToolDescription,
|
||||
}
|
||||
|
||||
@@ -122,7 +121,6 @@ impl WorkerPrompt {
|
||||
Self::WorkerOrchestrationGuidanceSection => {
|
||||
"internal.worker_orchestration_guidance_section"
|
||||
}
|
||||
Self::TicketEventCompanionNotice => "worker.ticket_event_companion_notice",
|
||||
Self::SubWorkerSpawnToolDescription => "internal.sub_worker_spawn_tool_description",
|
||||
}
|
||||
}
|
||||
@@ -139,7 +137,6 @@ impl WorkerPrompt {
|
||||
WorkerPrompt::AgentsMdSection,
|
||||
WorkerPrompt::ResidentMemorySummarySection,
|
||||
WorkerPrompt::WorkerOrchestrationGuidanceSection,
|
||||
WorkerPrompt::TicketEventCompanionNotice,
|
||||
WorkerPrompt::SubWorkerSpawnToolDescription,
|
||||
];
|
||||
}
|
||||
@@ -593,6 +590,12 @@ mod tests {
|
||||
fn builtin_dcdl_catalog_loads() {
|
||||
let catalog = PromptCatalog::builtins_only().unwrap();
|
||||
assert!(!catalog.projection.templates.is_empty());
|
||||
assert!(
|
||||
!catalog
|
||||
.projection
|
||||
.templates
|
||||
.contains_key("worker.ticket_event_companion_notice")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -77,8 +77,8 @@ use crate::skill::{SkillActivationResponse, SkillClientError};
|
||||
#[cfg(test)]
|
||||
use async_trait::async_trait;
|
||||
use protocol::{
|
||||
AlertLevel, AlertSource, CompactionLifecycle, CompactionLifecycleState, Event, RewindSummary,
|
||||
RewindTarget, RewindTargetId, Segment,
|
||||
AlertLevel, AlertSource, CompactionLifecycle, CompactionLifecycleState, ErrorCode, Event,
|
||||
RewindSummary, RewindTarget, RewindTargetId, Segment,
|
||||
};
|
||||
use tokio::net::UnixStream;
|
||||
use tokio::sync::broadcast;
|
||||
@@ -2964,7 +2964,10 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
|
||||
}
|
||||
EngineRunExit::Interrupted(reason) => {
|
||||
self.last_run_interrupted = true;
|
||||
Ok(WorkerRunResult::Interrupted(stop_reason_message(&reason)))
|
||||
Ok(WorkerRunResult::Interrupted {
|
||||
code: stop_reason_error_code(&reason),
|
||||
message: stop_reason_message(&reason),
|
||||
})
|
||||
}
|
||||
EngineRunExit::Yielded => unreachable!("yielded handled above"),
|
||||
}
|
||||
@@ -4512,7 +4515,7 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
|
||||
fn extract_internal_worker_lifecycle_error(lifecycle: &WorkerRunResult) -> Option<WorkerError> {
|
||||
match lifecycle {
|
||||
WorkerRunResult::RolledBack => Some(WorkerError::Engine(EngineError::Cancelled)),
|
||||
WorkerRunResult::Interrupted(message) => {
|
||||
WorkerRunResult::Interrupted { message, .. } => {
|
||||
Some(WorkerError::Engine(EngineError::Aborted(message.clone())))
|
||||
}
|
||||
WorkerRunResult::Finished | WorkerRunResult::Paused | WorkerRunResult::LimitReached => None,
|
||||
@@ -5521,6 +5524,23 @@ fn restore_manifest_from_worker_metadata_snapshot(
|
||||
}
|
||||
}
|
||||
|
||||
fn stop_reason_error_code(reason: &StopReason) -> ErrorCode {
|
||||
match reason {
|
||||
StopReason::ContextWindowExceeded | StopReason::Unexpected(EngineError::Client(_)) => {
|
||||
ErrorCode::ProviderError
|
||||
}
|
||||
StopReason::Unexpected(EngineError::Tool(_)) => ErrorCode::ToolError,
|
||||
StopReason::LimitReached
|
||||
| StopReason::Cancelled
|
||||
| StopReason::Unexpected(
|
||||
EngineError::Aborted(_)
|
||||
| EngineError::Cancelled
|
||||
| EngineError::ConfigWarnings(_)
|
||||
| EngineError::HistoryAppend(_),
|
||||
) => ErrorCode::Internal,
|
||||
}
|
||||
}
|
||||
|
||||
fn stop_reason_message(reason: &StopReason) -> String {
|
||||
match reason {
|
||||
StopReason::LimitReached => "engine turn limit reached".to_string(),
|
||||
@@ -5540,7 +5560,7 @@ pub enum WorkerRunResult {
|
||||
/// The worker reached its configured max_turns limit.
|
||||
LimitReached,
|
||||
/// The run was interrupted by a known or unexpected terminal cause.
|
||||
Interrupted(String),
|
||||
Interrupted { code: ErrorCode, message: String },
|
||||
/// The submit-time user turn was rolled back because no AI output was materialized.
|
||||
RolledBack,
|
||||
}
|
||||
|
||||
@@ -6414,9 +6414,15 @@ fn worker_ticket_source_context(
|
||||
}
|
||||
}
|
||||
|
||||
fn ticket_notification_content(ticket_id: &str, current_state: &str) -> String {
|
||||
fn canonical_ticket_resource_key(resource_key: &str) -> Option<&str> {
|
||||
let sequence = resource_key.strip_prefix("T-")?;
|
||||
(!sequence.is_empty() && sequence.bytes().all(|byte| byte.is_ascii_digit()))
|
||||
.then_some(resource_key)
|
||||
}
|
||||
|
||||
fn ticket_notification_content(resource_key: &str, current_state: &str) -> String {
|
||||
format!(
|
||||
"Ticket notification: ticket_id={ticket_id} current_state={current_state}. Reread the Ticket before acting."
|
||||
"Ticket {resource_key} changed to {current_state}. Reread the current Ticket before acting."
|
||||
)
|
||||
}
|
||||
|
||||
@@ -6498,6 +6504,16 @@ fn notify_ticket_recipients(
|
||||
current_state: &str,
|
||||
source: Option<RuntimeWorkerRef>,
|
||||
) {
|
||||
let Ok(Some(resource_key)) =
|
||||
api.store
|
||||
.resource_key(workspace_id, WorkspaceResourceKind::Ticket, ticket_id)
|
||||
else {
|
||||
return;
|
||||
};
|
||||
let Some(resource_key) = canonical_ticket_resource_key(&resource_key) else {
|
||||
return;
|
||||
};
|
||||
|
||||
let mut recipients = Vec::new();
|
||||
if let Some(assignment) = api
|
||||
.store
|
||||
@@ -6518,7 +6534,7 @@ fn notify_ticket_recipients(
|
||||
recipients.sort();
|
||||
recipients.dedup();
|
||||
|
||||
let content = ticket_notification_content(ticket_id, current_state);
|
||||
let content = ticket_notification_content(resource_key, current_state);
|
||||
for recipient in recipients {
|
||||
if source.as_ref().is_some_and(|source| source == &recipient) {
|
||||
continue;
|
||||
@@ -18216,15 +18232,25 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ticket_notification_projection_exposes_only_ticket_and_current_state() {
|
||||
fn ticket_notification_requires_canonical_ticket_resource_key() {
|
||||
assert_eq!(canonical_ticket_resource_key("T-429"), Some("T-429"));
|
||||
for invalid in ["", "00001KZ9SR97B", "T-", "T-key", "O-429"] {
|
||||
assert_eq!(canonical_ticket_resource_key(invalid), None);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ticket_notification_projection_exposes_only_resource_key_and_current_state() {
|
||||
const INTERNAL_ID: &str = "00001KZ9SR97B";
|
||||
for current_state in ["queued", "inprogress"] {
|
||||
let content = ticket_notification_content("00001KZ9SR97B", current_state);
|
||||
let content = ticket_notification_content("T-429", current_state);
|
||||
assert_eq!(
|
||||
content,
|
||||
format!(
|
||||
"Ticket notification: ticket_id=00001KZ9SR97B current_state={current_state}. Reread the Ticket before acting."
|
||||
"Ticket T-429 changed to {current_state}. Reread the current Ticket before acting."
|
||||
)
|
||||
);
|
||||
assert!(!content.contains(INTERNAL_ID));
|
||||
for forbidden in [
|
||||
"workspace_id",
|
||||
"event_sequence",
|
||||
@@ -18402,9 +18428,13 @@ mod tests {
|
||||
assert_eq!(inputs.len(), expected_states.len());
|
||||
for ((recipient, content), current_state) in inputs.iter().zip(expected_states) {
|
||||
assert_eq!(recipient.worker_id.to_string(), orchestrator.worker_id);
|
||||
assert!(!content.contains(&ticket.id));
|
||||
assert_eq!(
|
||||
content,
|
||||
&ticket_notification_content(&ticket.id, current_state)
|
||||
&ticket_notification_content(
|
||||
ticket.resource_key.as_deref().unwrap(),
|
||||
current_state,
|
||||
)
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -19520,7 +19550,7 @@ mod tests {
|
||||
assert_eq!(
|
||||
notifications[0].1,
|
||||
ticket_notification_content(
|
||||
ticket_ref.id.as_str(),
|
||||
ticket_ref.resource_key.as_deref().unwrap(),
|
||||
TicketWorkflowState::Queued.as_str()
|
||||
)
|
||||
);
|
||||
|
||||
@@ -30,7 +30,6 @@ internalAgentsMdSection = import "./internal/agents_md_section.md";
|
||||
internalResidentMemorySummarySection = import "./internal/resident_memory_summary_section.md";
|
||||
internalSubWorkerSpawnToolDescription = import "./internal/sub_worker_spawn_tool_description.md";
|
||||
panelOrchestratorIdleQueueNotice = import "./panel/orchestrator_idle_queue_notice.md";
|
||||
workerTicketEventCompanionNotice = import "./worker/ticket_event_companion_notice.md";
|
||||
in
|
||||
{
|
||||
default_prompt = defaultDocument.content;
|
||||
@@ -69,7 +68,4 @@ in
|
||||
panel = {
|
||||
orchestrator_idle_queue_notice = panelOrchestratorIdleQueueNotice.content;
|
||||
};
|
||||
worker = {
|
||||
ticket_event_companion_notice = workerTicketEventCompanionNotice.content;
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1,7 +0,0 @@
|
||||
Ticket event notice (weak; auto_run=false)
|
||||
ticket: {{ ticket_id }}
|
||||
title: {{ title }}
|
||||
state: {{ state }}
|
||||
event: {{ event_kind }}
|
||||
summary: {{ summary }}
|
||||
ref: {{ ref_path }}
|
||||
@@ -5,7 +5,7 @@
|
||||
workspaceWorkersStore,
|
||||
type SidebarWorker,
|
||||
} from './worker-subscription';
|
||||
import { canShowWorkerInSidebar } from './workers';
|
||||
import { canShowWorkerInSidebar, sidebarWorkerActivity } from './workers';
|
||||
|
||||
const COLLAPSED_WORKER_COUNT = 6;
|
||||
|
||||
@@ -69,6 +69,7 @@
|
||||
<ul class="nav-list" aria-label="Workers">
|
||||
{#each visibleWorkers as worker (`${worker.runtime_id}:${worker.worker_id}`)}
|
||||
{@const href = workerConsoleHref(worker, workspaceId)}
|
||||
{@const activity = sidebarWorkerActivity(worker)}
|
||||
<li>
|
||||
<a
|
||||
href={href}
|
||||
@@ -77,11 +78,11 @@
|
||||
aria-current={currentPath === href ? 'page' : undefined}
|
||||
>
|
||||
<span class="worker-status-indicator">
|
||||
{#if worker.state === 'running'}
|
||||
{#if activity === 'worker-running'}
|
||||
<span class="worker-status-spinner"><Spinner label="Running" /></span>
|
||||
{:else if worker.has_running_internal_workers}
|
||||
{:else if activity === 'subworker-running'}
|
||||
<span class="worker-status-spinner is-subworker"><Spinner label="SubWorker running" /></span>
|
||||
{:else if worker.state === 'idle'}
|
||||
{:else if activity === 'idle'}
|
||||
<span class="worker-status-dot" aria-label="Idle"></span>
|
||||
{/if}
|
||||
</span>
|
||||
|
||||
@@ -14,13 +14,18 @@ declare const Deno: {
|
||||
test(name: string, fn: () => void | Promise<void>): void;
|
||||
};
|
||||
|
||||
function worker(runtimeId: string, workerId: string, revision: number): SubscriptionWorker {
|
||||
function worker(
|
||||
runtimeId: string,
|
||||
workerId: string,
|
||||
revision: number,
|
||||
hasRunningInternalWorkers = false,
|
||||
): SubscriptionWorker {
|
||||
return {
|
||||
worker_id: workerId,
|
||||
runtime_id: runtimeId,
|
||||
subject_revision: revision,
|
||||
state: 'idle',
|
||||
has_running_internal_workers: false,
|
||||
has_running_internal_workers: hasRunningInternalWorkers,
|
||||
workspace_id: 'workspace-test',
|
||||
display_name: null,
|
||||
profile: null,
|
||||
@@ -88,3 +93,28 @@ Deno.test('workspace Worker reducer ignores stale events and removes composite s
|
||||
assertEquals(projection.workers.size, 0);
|
||||
assertEquals(projection.revisions.get('runtime-a:1'), 4);
|
||||
});
|
||||
|
||||
Deno.test('fatal child stop replaces the running-child sidebar projection', () => {
|
||||
const projection = createWorkspaceWorkersProjection();
|
||||
projection.workers.set('runtime-a:1', worker('runtime-a', '1', 1, true));
|
||||
projection.revisions.set('runtime-a:1', 1);
|
||||
|
||||
applyWorkspaceWorkersFrame(projection, {
|
||||
protocol_version: 1,
|
||||
frame: 'event',
|
||||
message: {
|
||||
event: 'event',
|
||||
data: {
|
||||
subscription_id: 'subscription-1',
|
||||
subject_revision: 2,
|
||||
payload: {
|
||||
event: 'worker_upserted',
|
||||
data: { worker: worker('runtime-a', '1', 2, false) },
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
assertEquals(projection.workers.get('runtime-a:1')?.has_running_internal_workers, false);
|
||||
assertEquals(projection.revisions.get('runtime-a:1'), 2);
|
||||
});
|
||||
|
||||
@@ -2,6 +2,7 @@ import {
|
||||
canOpenWorkerConsole,
|
||||
canShowWorkerInSidebar,
|
||||
compareWorkersForSidebar,
|
||||
sidebarWorkerActivity,
|
||||
} from "./workers.ts";
|
||||
import type { Worker } from "./types.ts";
|
||||
|
||||
@@ -77,3 +78,21 @@ Deno.test("sidebar workers sort running then idle then stopped", () => {
|
||||
workers.sort(compareWorkersForSidebar);
|
||||
assertEquals(workers.map((candidate) => candidate.worker_id).join(","), "2,1,4,3");
|
||||
});
|
||||
|
||||
Deno.test("fatal child stop clears the sidebar SubWorker spinner activity", () => {
|
||||
const parent = { state: "idle", has_running_internal_workers: true };
|
||||
assertEquals(sidebarWorkerActivity(parent), "subworker-running");
|
||||
|
||||
parent.has_running_internal_workers = false;
|
||||
assertEquals(sidebarWorkerActivity(parent), "idle");
|
||||
});
|
||||
|
||||
Deno.test("stopped parents do not fall back to the idle indicator", () => {
|
||||
assertEquals(
|
||||
sidebarWorkerActivity({
|
||||
state: "stopped",
|
||||
has_running_internal_workers: false,
|
||||
}),
|
||||
"none",
|
||||
);
|
||||
});
|
||||
|
||||
@@ -1,5 +1,24 @@
|
||||
import type { Worker } from './types';
|
||||
|
||||
export type SidebarWorkerActivity =
|
||||
| 'worker-running'
|
||||
| 'subworker-running'
|
||||
| 'idle'
|
||||
| 'none';
|
||||
|
||||
type WorkerActivitySource = Pick<Worker, 'state'> & {
|
||||
has_running_internal_workers: boolean;
|
||||
};
|
||||
|
||||
export function sidebarWorkerActivity(
|
||||
worker: WorkerActivitySource,
|
||||
): SidebarWorkerActivity {
|
||||
if (worker.state === 'running') return 'worker-running';
|
||||
if (worker.has_running_internal_workers) return 'subworker-running';
|
||||
if (worker.state === 'idle') return 'idle';
|
||||
return 'none';
|
||||
}
|
||||
|
||||
export function canShowWorkerInSidebar(worker: Worker): boolean {
|
||||
return worker.implementation.kind !== 'backend_worker_registry';
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user