server: minimize ticket notification context
This commit is contained in:
@@ -3131,17 +3131,6 @@ async fn scoped_queue_ticket(
|
|||||||
&api,
|
&api,
|
||||||
&path.workspace_id,
|
&path.workspace_id,
|
||||||
&path.id,
|
&path.id,
|
||||||
ticket
|
|
||||||
.events
|
|
||||||
.last()
|
|
||||||
.map(|event| event.sequence as i64)
|
|
||||||
.unwrap_or_default(),
|
|
||||||
ticket
|
|
||||||
.events
|
|
||||||
.last()
|
|
||||||
.map(|event| event.kind.as_str())
|
|
||||||
.unwrap_or("state_changed"),
|
|
||||||
"queue_ready",
|
|
||||||
TicketWorkflowState::Ready.as_str(),
|
TicketWorkflowState::Ready.as_str(),
|
||||||
ticket.state.as_str(),
|
ticket.state.as_str(),
|
||||||
None,
|
None,
|
||||||
@@ -3209,17 +3198,10 @@ async fn execute_worker_ticket_rest_operation(
|
|||||||
&& let Some(target) = target
|
&& let Some(target) = target
|
||||||
&& let Ok(ticket) = backend.show(target)
|
&& let Ok(ticket) = backend.show(target)
|
||||||
{
|
{
|
||||||
let event = ticket.events.last();
|
|
||||||
notify_ticket_recipients(
|
notify_ticket_recipients(
|
||||||
api,
|
api,
|
||||||
workspace_id,
|
workspace_id,
|
||||||
&ticket.meta.id,
|
&ticket.meta.id,
|
||||||
event
|
|
||||||
.and_then(|event| event.attributes.get("event_sequence"))
|
|
||||||
.and_then(|value| value.parse::<i64>().ok())
|
|
||||||
.unwrap_or(ticket.events.len() as i64),
|
|
||||||
event.map(|event| event.kind.as_str()).unwrap_or("mutation"),
|
|
||||||
operation_kind,
|
|
||||||
&previous_state,
|
&previous_state,
|
||||||
ticket.meta.workflow_state.as_str(),
|
ticket.meta.workflow_state.as_str(),
|
||||||
Some(source),
|
Some(source),
|
||||||
@@ -4247,13 +4229,16 @@ fn worker_ticket_source_context(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn ticket_notification_content(ticket_id: &str, current_state: &str) -> String {
|
||||||
|
format!(
|
||||||
|
"Ticket notification: ticket_id={ticket_id} current_state={current_state}. Reread the Ticket before acting."
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
fn notify_ticket_recipients(
|
fn notify_ticket_recipients(
|
||||||
api: &WorkspaceApi,
|
api: &WorkspaceApi,
|
||||||
workspace_id: &str,
|
workspace_id: &str,
|
||||||
ticket_id: &str,
|
ticket_id: &str,
|
||||||
event_sequence: i64,
|
|
||||||
event_kind: &str,
|
|
||||||
source_operation_kind: &str,
|
|
||||||
previous_state: &str,
|
previous_state: &str,
|
||||||
current_state: &str,
|
current_state: &str,
|
||||||
source: Option<RuntimeWorkerRef>,
|
source: Option<RuntimeWorkerRef>,
|
||||||
@@ -4276,15 +4261,7 @@ fn notify_ticket_recipients(
|
|||||||
recipients.sort();
|
recipients.sort();
|
||||||
recipients.dedup();
|
recipients.dedup();
|
||||||
|
|
||||||
let source_fields = source
|
let content = ticket_notification_content(ticket_id, current_state);
|
||||||
.as_ref()
|
|
||||||
.map(|source| {
|
|
||||||
format!(
|
|
||||||
" source_runtime_id={} source_worker_id={}",
|
|
||||||
source.runtime_id, source.worker_id
|
|
||||||
)
|
|
||||||
})
|
|
||||||
.unwrap_or_default();
|
|
||||||
for recipient in recipients {
|
for recipient in recipients {
|
||||||
if source.as_ref().is_some_and(|source| source == &recipient) {
|
if source.as_ref().is_some_and(|source| source == &recipient) {
|
||||||
continue;
|
continue;
|
||||||
@@ -4293,9 +4270,7 @@ fn notify_ticket_recipients(
|
|||||||
&recipient,
|
&recipient,
|
||||||
WorkerInputRequest {
|
WorkerInputRequest {
|
||||||
kind: WorkerInputKind::Notify,
|
kind: WorkerInputKind::Notify,
|
||||||
content: format!(
|
content: content.clone(),
|
||||||
"Ticket notification: workspace_id={workspace_id} ticket_id={ticket_id} event_sequence={event_sequence} event_kind={event_kind} source_operation_kind={source_operation_kind}.{source_fields} Reread the Ticket before acting.",
|
|
||||||
),
|
|
||||||
segments: None,
|
segments: None,
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
@@ -12730,6 +12705,7 @@ mod tests {
|
|||||||
>,
|
>,
|
||||||
materializer: worker_runtime::working_directory::LocalGitWorktreeMaterializer,
|
materializer: worker_runtime::working_directory::LocalGitWorktreeMaterializer,
|
||||||
spawn_failure: std::sync::Mutex<Option<String>>,
|
spawn_failure: std::sync::Mutex<Option<String>>,
|
||||||
|
inputs: std::sync::Mutex<Vec<(worker_runtime::identity::WorkerRef, String)>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for DeterministicExecutionBackend {
|
impl Default for DeterministicExecutionBackend {
|
||||||
@@ -12748,11 +12724,16 @@ mod tests {
|
|||||||
std::env::temp_dir().join(unique),
|
std::env::temp_dir().join(unique),
|
||||||
),
|
),
|
||||||
spawn_failure: std::sync::Mutex::new(None),
|
spawn_failure: std::sync::Mutex::new(None),
|
||||||
|
inputs: std::sync::Mutex::new(Vec::new()),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl DeterministicExecutionBackend {
|
impl DeterministicExecutionBackend {
|
||||||
|
fn take_inputs(&self) -> Vec<(worker_runtime::identity::WorkerRef, String)> {
|
||||||
|
std::mem::take(&mut *self.inputs.lock().expect("inputs lock"))
|
||||||
|
}
|
||||||
|
|
||||||
fn fail_first_spawn(message: impl Into<String>) -> Self {
|
fn fail_first_spawn(message: impl Into<String>) -> Self {
|
||||||
let backend = Self::default();
|
let backend = Self::default();
|
||||||
*backend.spawn_failure.lock().unwrap() = Some(message.into());
|
*backend.spawn_failure.lock().unwrap() = Some(message.into());
|
||||||
@@ -12871,6 +12852,10 @@ mod tests {
|
|||||||
handle: &worker_runtime::execution::WorkerExecutionHandle,
|
handle: &worker_runtime::execution::WorkerExecutionHandle,
|
||||||
input: worker_runtime::interaction::WorkerInput,
|
input: worker_runtime::interaction::WorkerInput,
|
||||||
) -> worker_runtime::execution::WorkerExecutionResult {
|
) -> worker_runtime::execution::WorkerExecutionResult {
|
||||||
|
self.inputs
|
||||||
|
.lock()
|
||||||
|
.expect("inputs lock")
|
||||||
|
.push((handle.worker_ref().clone(), input.content.clone()));
|
||||||
let context = self
|
let context = self
|
||||||
.contexts
|
.contexts
|
||||||
.lock()
|
.lock()
|
||||||
@@ -13115,6 +13100,173 @@ mod tests {
|
|||||||
assert_eq!(worker_source_actor_role(true, true), "coder");
|
assert_eq!(worker_source_actor_role(true, true), "coder");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ticket_notification_projection_exposes_only_ticket_and_current_state() {
|
||||||
|
for current_state in ["queued", "inprogress"] {
|
||||||
|
let content = ticket_notification_content("00001KZ9SR97B", current_state);
|
||||||
|
assert_eq!(
|
||||||
|
content,
|
||||||
|
format!(
|
||||||
|
"Ticket notification: ticket_id=00001KZ9SR97B current_state={current_state}. Reread the Ticket before acting."
|
||||||
|
)
|
||||||
|
);
|
||||||
|
for forbidden in [
|
||||||
|
"workspace_id",
|
||||||
|
"event_sequence",
|
||||||
|
"event_kind",
|
||||||
|
"source_operation_kind",
|
||||||
|
"source_runtime_id",
|
||||||
|
"source_worker_id",
|
||||||
|
"runtime_id",
|
||||||
|
"worker_id",
|
||||||
|
"operation_id",
|
||||||
|
"assignment_id",
|
||||||
|
] {
|
||||||
|
assert!(
|
||||||
|
!content.contains(forbidden),
|
||||||
|
"notification leaked forbidden field {forbidden}: {content}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn orchestrator_ticket_notifications_project_authoritative_post_mutation_state() {
|
||||||
|
let dir = tempfile::tempdir().unwrap();
|
||||||
|
let (api, execution) = test_api_with_recording_backend(dir.path()).await;
|
||||||
|
let source_worker = api
|
||||||
|
.runtime
|
||||||
|
.spawn_worker(
|
||||||
|
EMBEDDED_WORKER_RUNTIME_ID,
|
||||||
|
WorkerSpawnRequest {
|
||||||
|
requested_worker_name: Some("notification-source".to_string()),
|
||||||
|
intent: WorkerSpawnIntent::TicketRole {
|
||||||
|
ticket_id: "notification-source".to_string(),
|
||||||
|
role: TicketWorkerRole::Coder,
|
||||||
|
},
|
||||||
|
acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
|
||||||
|
expected_segments: 0,
|
||||||
|
},
|
||||||
|
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
|
||||||
|
ticket_assignment: None,
|
||||||
|
initial_submit: Vec::new(),
|
||||||
|
working_directory_request: None,
|
||||||
|
resolved_working_directory_request: None,
|
||||||
|
resolved_working_directory: None,
|
||||||
|
resolved_config_bundle: None,
|
||||||
|
resolved_worker_observation_enabled: false,
|
||||||
|
resolved_worker_observation_grants: Vec::new(),
|
||||||
|
resolved_workspace_api: Some(test_worker_workspace_api(
|
||||||
|
EMBEDDED_WORKER_RUNTIME_ID,
|
||||||
|
)),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.unwrap()
|
||||||
|
.worker
|
||||||
|
.unwrap();
|
||||||
|
let source =
|
||||||
|
RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, source_worker.worker.worker_id);
|
||||||
|
let source_headers = || {
|
||||||
|
let mut headers = HeaderMap::new();
|
||||||
|
headers.insert(
|
||||||
|
"x-yoi-runtime-id",
|
||||||
|
axum::http::HeaderValue::from_str(&source.runtime_id).unwrap(),
|
||||||
|
);
|
||||||
|
headers.insert(
|
||||||
|
"x-yoi-worker-id",
|
||||||
|
axum::http::HeaderValue::from_str(&source.worker_id).unwrap(),
|
||||||
|
);
|
||||||
|
headers
|
||||||
|
};
|
||||||
|
let Json(started) = scoped_start_workspace_orchestrator(
|
||||||
|
State(api.clone()),
|
||||||
|
AxumPath(ScopedWorkspacePath {
|
||||||
|
workspace_id: TEST_WORKSPACE_ID.to_string(),
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let orchestrator = started.worker.unwrap().worker;
|
||||||
|
execution.take_inputs();
|
||||||
|
|
||||||
|
let ticket = browser_ticket_backend(&api)
|
||||||
|
.unwrap()
|
||||||
|
.create(ticket::NewTicket::new("Bounded notification"))
|
||||||
|
.unwrap();
|
||||||
|
let ticket_id = TicketIdOrSlug::Id(ticket.id.clone());
|
||||||
|
let operations = [
|
||||||
|
TicketBackendOperation::SetWorkflowState {
|
||||||
|
id: ticket_id.clone(),
|
||||||
|
change: TicketStateChange::new(
|
||||||
|
"planning",
|
||||||
|
"ready",
|
||||||
|
"ready for implementation",
|
||||||
|
"test transition",
|
||||||
|
),
|
||||||
|
},
|
||||||
|
TicketBackendOperation::SetWorkflowState {
|
||||||
|
id: ticket_id.clone(),
|
||||||
|
change: TicketStateChange::new(
|
||||||
|
"ready",
|
||||||
|
"queued",
|
||||||
|
"queued for implementation",
|
||||||
|
"test transition",
|
||||||
|
),
|
||||||
|
},
|
||||||
|
TicketBackendOperation::SetWorkflowState {
|
||||||
|
id: ticket_id.clone(),
|
||||||
|
change: TicketStateChange::new(
|
||||||
|
"queued",
|
||||||
|
"inprogress",
|
||||||
|
"implementation accepted",
|
||||||
|
"test transition",
|
||||||
|
),
|
||||||
|
},
|
||||||
|
TicketBackendOperation::AddEvent {
|
||||||
|
id: ticket_id.clone(),
|
||||||
|
event: NewTicketEvent::new(TicketEventKind::Comment, "progress comment"),
|
||||||
|
},
|
||||||
|
TicketBackendOperation::AddEvent {
|
||||||
|
id: ticket_id.clone(),
|
||||||
|
event: NewTicketEvent::new(
|
||||||
|
TicketEventKind::ImplementationReport,
|
||||||
|
"implementation report",
|
||||||
|
),
|
||||||
|
},
|
||||||
|
TicketBackendOperation::AddEvent {
|
||||||
|
id: ticket_id,
|
||||||
|
event: NewTicketEvent::new(TicketEventKind::Decision, "review update"),
|
||||||
|
},
|
||||||
|
];
|
||||||
|
for operation in operations {
|
||||||
|
execute_worker_ticket_rest_operation(
|
||||||
|
&api,
|
||||||
|
TEST_WORKSPACE_ID,
|
||||||
|
source_headers(),
|
||||||
|
operation,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
let inputs = execution.take_inputs();
|
||||||
|
let expected_states = [
|
||||||
|
"queued",
|
||||||
|
"inprogress",
|
||||||
|
"inprogress",
|
||||||
|
"inprogress",
|
||||||
|
"inprogress",
|
||||||
|
];
|
||||||
|
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_eq!(
|
||||||
|
content,
|
||||||
|
&ticket_notification_content(&ticket.id, current_state)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn ticket_assignment_endpoints_read_and_clear_current_assignment() {
|
async fn ticket_assignment_endpoints_read_and_clear_current_assignment() {
|
||||||
let dir = tempfile::tempdir().unwrap();
|
let dir = tempfile::tempdir().unwrap();
|
||||||
@@ -14148,16 +14300,21 @@ mod tests {
|
|||||||
assert_eq!(detail.provenance.id, "workspace:triage-errors");
|
assert_eq!(detail.provenance.id, "workspace:triage-errors");
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn test_api(workspace_root: impl Into<PathBuf>) -> WorkspaceApi {
|
async fn test_api_with_recording_backend(
|
||||||
|
workspace_root: impl Into<PathBuf>,
|
||||||
|
) -> (WorkspaceApi, Arc<DeterministicExecutionBackend>) {
|
||||||
let config = test_server_config(workspace_root);
|
let config = test_server_config(workspace_root);
|
||||||
let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap();
|
let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap();
|
||||||
WorkspaceApi::new_with_execution_backend(
|
let execution = Arc::new(DeterministicExecutionBackend::default());
|
||||||
config,
|
let api =
|
||||||
Arc::new(store),
|
WorkspaceApi::new_with_execution_backend(config, Arc::new(store), execution.clone())
|
||||||
Arc::new(DeterministicExecutionBackend::default()),
|
.await
|
||||||
)
|
.unwrap();
|
||||||
.await
|
(api, execution)
|
||||||
.unwrap()
|
}
|
||||||
|
|
||||||
|
async fn test_api(workspace_root: impl Into<PathBuf>) -> WorkspaceApi {
|
||||||
|
test_api_with_recording_backend(workspace_root).await.0
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
|
|||||||
Reference in New Issue
Block a user