diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 93f6bfde4..80d872997 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -34,9 +34,9 @@ use fabro_store::{ ArtifactStore, Database, EventEnvelope, EventPayload, PendingInterviewRecord, StageId, }; use fabro_types::{ - EventBody, InterviewQuestionRecord, InterviewQuestionType, RunBlobId, RunClientProvenance, - RunControlAction, RunEvent, RunId, RunProvenance, RunServerProvenance, RunSubjectProvenance, - Settings, + ActorKind, ActorRef, EventBody, InterviewQuestionRecord, InterviewQuestionType, RunBlobId, + RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance, RunServerProvenance, + RunSubjectProvenance, Settings, }; use fabro_util::redact::redact_jsonl_line; use fabro_util::version::FABRO_VERSION; @@ -5143,16 +5143,26 @@ async fn append_control_request( state: &AppState, run_id: RunId, action: RunControlAction, + actor: Option, ) -> anyhow::Result<()> { let run_store = state.store.open_run(&run_id).await?; let event = match action { - RunControlAction::Cancel => workflow_event::Event::RunCancelRequested, - RunControlAction::Pause => workflow_event::Event::RunPauseRequested, - RunControlAction::Unpause => workflow_event::Event::RunUnpauseRequested, + RunControlAction::Cancel => workflow_event::Event::RunCancelRequested { actor }, + RunControlAction::Pause => workflow_event::Event::RunPauseRequested { actor }, + RunControlAction::Unpause => workflow_event::Event::RunUnpauseRequested { actor }, }; workflow_event::append_event(&run_store, &run_id, &event).await } +fn actor_from_subject(subject: &AuthenticatedSubject) -> Option { + let login = subject.login.clone()?; + Some(ActorRef { + kind: ActorKind::User, + id: Some(login.clone()), + display: Some(login), + }) +} + fn schedule_worker_kill(state: Arc, run_id: RunId, worker_pid: u32) { tokio::spawn(async move { sleep(WORKER_CANCEL_GRACE).await; @@ -5168,7 +5178,7 @@ fn schedule_worker_kill(state: Arc, run_id: RunId, worker_pid: u32) { } async fn cancel_run( - _auth: AuthenticatedService, + subject: AuthenticatedSubject, State(state): State>, Path(id): Path, ) -> Response { @@ -5234,7 +5244,13 @@ async fn cancel_run( }; if pending_control != Some(RunControlAction::Cancel) { - if let Err(err) = append_control_request(state.as_ref(), id, RunControlAction::Cancel).await + if let Err(err) = append_control_request( + state.as_ref(), + id, + RunControlAction::Cancel, + actor_from_subject(&subject), + ) + .await { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) .into_response(); @@ -5286,7 +5302,7 @@ async fn cancel_run( } async fn pause_run( - _auth: AuthenticatedService, + subject: AuthenticatedSubject, State(state): State>, Path(id): Path, ) -> Response { @@ -5324,7 +5340,14 @@ async fn pause_run( let Some(worker_pid) = worker_pid else { return ApiError::new(StatusCode::CONFLICT, "Run worker is not available.").into_response(); }; - if let Err(err) = append_control_request(state.as_ref(), id, RunControlAction::Pause).await { + if let Err(err) = append_control_request( + state.as_ref(), + id, + RunControlAction::Pause, + actor_from_subject(&subject), + ) + .await + { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(); } #[cfg(unix)] @@ -5347,7 +5370,7 @@ async fn pause_run( } async fn unpause_run( - _auth: AuthenticatedService, + subject: AuthenticatedSubject, State(state): State>, Path(id): Path, ) -> Response { @@ -5385,7 +5408,14 @@ async fn unpause_run( let Some(worker_pid) = worker_pid else { return ApiError::new(StatusCode::CONFLICT, "Run worker is not available.").into_response(); }; - if let Err(err) = append_control_request(state.as_ref(), id, RunControlAction::Unpause).await { + if let Err(err) = append_control_request( + state.as_ref(), + id, + RunControlAction::Unpause, + actor_from_subject(&subject), + ) + .await + { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(); } #[cfg(unix)] @@ -7448,7 +7478,7 @@ mod tests { managed_run.status = RunStatus::Running; managed_run.worker_pid = Some(u32::MAX); } - append_control_request(state.as_ref(), run_id, RunControlAction::Pause) + append_control_request(state.as_ref(), run_id, RunControlAction::Pause, None) .await .unwrap(); @@ -7479,7 +7509,7 @@ mod tests { managed_run.status = RunStatus::Running; managed_run.worker_pid = Some(u32::MAX); } - append_control_request(state.as_ref(), run_id, RunControlAction::Cancel) + append_control_request(state.as_ref(), run_id, RunControlAction::Cancel, None) .await .unwrap(); @@ -7613,7 +7643,7 @@ mod tests { workflow_event::Event::RunStarting { reason: None }, workflow_event::Event::RunRunning { reason: None }, workflow_event::Event::RunPaused, - workflow_event::Event::RunCancelRequested, + workflow_event::Event::RunCancelRequested { actor: None }, ], ) .await; diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 060f7d116..2f299a7e3 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -93,9 +93,18 @@ pub enum Event { #[serde(default, skip_serializing_if = "Option::is_none")] reason: Option, }, - RunCancelRequested, - RunPauseRequested, - RunUnpauseRequested, + RunCancelRequested { + #[serde(default, skip_serializing_if = "Option::is_none")] + actor: Option, + }, + RunPauseRequested { + #[serde(default, skip_serializing_if = "Option::is_none")] + actor: Option, + }, + RunUnpauseRequested { + #[serde(default, skip_serializing_if = "Option::is_none")] + actor: Option, + }, RunPaused, RunUnpaused, RunRewound { @@ -575,13 +584,13 @@ impl Event { Self::RunRemoving { reason } => { info!(?reason, "Run removing"); } - Self::RunCancelRequested => { + Self::RunCancelRequested { .. } => { info!("Run cancel requested"); } - Self::RunPauseRequested => { + Self::RunPauseRequested { .. } => { info!("Run pause requested"); } - Self::RunUnpauseRequested => { + Self::RunUnpauseRequested { .. } => { info!("Run unpause requested"); } Self::RunPaused => { @@ -1140,9 +1149,9 @@ pub fn event_name(event: &Event) -> &'static str { Event::RunStarting { .. } => "run.starting", Event::RunRunning { .. } => "run.running", Event::RunRemoving { .. } => "run.removing", - Event::RunCancelRequested => "run.cancel.requested", - Event::RunPauseRequested => "run.pause.requested", - Event::RunUnpauseRequested => "run.unpause.requested", + Event::RunCancelRequested { .. } => "run.cancel.requested", + Event::RunPauseRequested { .. } => "run.pause.requested", + Event::RunUnpauseRequested { .. } => "run.unpause.requested", Event::RunPaused => "run.paused", Event::RunUnpaused => "run.unpaused", Event::RunRewound { .. } => "run.rewound", @@ -1329,6 +1338,12 @@ fn stored_event_fields_for_variant(event: &Event) -> StoredEventFields { actor: provenance.as_ref().and_then(actor_from_provenance), ..StoredEventFields::default() }, + Event::RunCancelRequested { actor } + | Event::RunPauseRequested { actor } + | Event::RunUnpauseRequested { actor } => StoredEventFields { + actor: actor.clone(), + ..StoredEventFields::default() + }, Event::StageCompleted { node_id, name, .. } | Event::StageFailed { node_id, name, .. } | Event::StageStarted { node_id, name, .. } @@ -1513,17 +1528,17 @@ fn event_body_from_event(event: &Event) -> EventBody { Event::RunRemoving { reason } => { EventBody::RunRemoving(fabro_types::RunStatusTransitionProps { reason: *reason }) } - Event::RunCancelRequested => { + Event::RunCancelRequested { .. } => { EventBody::RunCancelRequested(fabro_types::RunControlRequestedProps { action: RunControlAction::Cancel, }) } - Event::RunPauseRequested => { + Event::RunPauseRequested { .. } => { EventBody::RunPauseRequested(fabro_types::RunControlRequestedProps { action: RunControlAction::Pause, }) } - Event::RunUnpauseRequested => { + Event::RunUnpauseRequested { .. } => { EventBody::RunUnpauseRequested(fabro_types::RunControlRequestedProps { action: RunControlAction::Unpause, }) @@ -3071,7 +3086,7 @@ mod tests { let (writer, reader) = tokio::io::duplex(4096); let sink = RunEventSink::json_lines(writer); - let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested); + let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested { actor: None }); sink.write_run_event(&event).await.unwrap(); @@ -3315,6 +3330,38 @@ mod tests { assert!(stored.parallel_branch_id.is_none()); } + #[test] + fn control_action_events_carry_actor_in_envelope() { + let actor = ActorRef { + kind: ActorKind::User, + id: Some("alice".to_string()), + display: Some("alice".to_string()), + }; + + let cancel = to_run_event( + &fixtures::RUN_1, + &Event::RunCancelRequested { + actor: Some(actor.clone()), + }, + ); + assert_eq!(cancel.event_name(), "run.cancel.requested"); + assert_eq!(cancel.actor.as_ref().expect("actor set"), &actor); + + let pause = to_run_event( + &fixtures::RUN_1, + &Event::RunPauseRequested { + actor: Some(actor.clone()), + }, + ); + assert_eq!(pause.actor.as_ref().expect("actor set"), &actor); + + let unpause = to_run_event( + &fixtures::RUN_1, + &Event::RunUnpauseRequested { actor: None }, + ); + assert!(unpause.actor.is_none()); + } + #[test] fn agent_assistant_message_populates_agent_actor() { let stored = to_run_event(