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<ActorRef>` 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<ActorRef> 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) <noreply@anthropic.com>

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<ActorRef>,
 ) -> 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<ActorRef> {
+    let login = subject.login.clone()?;
+    Some(ActorRef {
+        kind: ActorKind::User,
+        id: Some(login.clone()),
+        display: Some(login),
+    })
+}
+
 fn schedule_worker_kill(state: Arc<AppState>, 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<AppState>, run_id: RunId, worker_pid: u32) {
 }
 
 async fn cancel_run(
-    _auth: AuthenticatedService,
+    subject: AuthenticatedSubject,
     State(state): State<Arc<AppState>>,
     Path(id): Path<String>,
 ) -> 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<Arc<AppState>>,
     Path(id): Path<String>,
 ) -> 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<Arc<AppState>>,
     Path(id): Path<String>,
 ) -> 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;

lib/crates/fabro-workflow/src/event.rs

@@ -93,9 +93,18 @@ pub enum Event {
         #[serde(default, skip_serializing_if = "Option::is_none")]
         reason: Option<StatusReason>,
     },
-    RunCancelRequested,
-    RunPauseRequested,
-    RunUnpauseRequested,
+    RunCancelRequested {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        actor: Option<ActorRef>,
+    },
+    RunPauseRequested {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        actor: Option<ActorRef>,
+    },
+    RunUnpauseRequested {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        actor: Option<ActorRef>,
+    },
     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(