From 8097c224ecc450ad0b6f8aa839267bb3daa62bfd Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 9 Apr 2026 16:42:24 -0400 Subject: [PATCH] feat(events): populate actor on control-action events Per the schema v2 spec (docs-internal/fabro-event-schema-v2-concrete-shape.md:208-229), `actor` is expected on control actions like `run.cancel.requested` to identify the user who initiated the request. Before this commit, the three Event::Run{Cancel,Pause,Unpause}Requested variants were bare unit variants and the cancel/pause/unpause HTTP handlers used the _auth: AuthenticatedService ZST extractor which discards user identity. - fabro-workflow/src/event.rs: add `actor: Option` to Event::RunCancelRequested, Event::RunPauseRequested, Event::RunUnpauseRequested. Add a stored_event_fields_for_variant match arm that copies the actor into the envelope. Update event_body_from_event, event_name, and the trace! debug arm to ignore the new field via `{ .. }`. - fabro-server/src/server.rs: switch cancel_run, pause_run, unpause_run from _auth: AuthenticatedService to subject: AuthenticatedSubject (which handles cookie/JWT/mTLS identity uniformly via lib/crates/fabro-server/src/jwt_auth.rs). Add an actor_from_subject helper that mirrors the existing actor_from_provenance in fabro-workflow -- both produce an ActorRef { kind: User, id: login, display: login }. append_control_request takes a new Option argument and constructs the variants with it. Test call sites pass None. Test: new unit test control_action_events_carry_actor_in_envelope in event.rs covering cancel/pause/unpause with Some(actor) and unpause with None. Run mode AuthMode::Disabled returns subject.login = None, so actor ends up None in that path -- matches the spec's "actor is optional" guidance. Wire format is backward compatible: actor uses #[serde(default, skip_serializing_if = "Option::is_none")] so old persisted events without the field still parse cleanly. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/crates/fabro-server/src/server.rs | 60 +++++++++++++++------ lib/crates/fabro-workflow/src/event.rs | 73 +++++++++++++++++++++----- 2 files changed, 105 insertions(+), 28 deletions(-) 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(