refactor(events): dedupe schema v2 plumbing

Mostly consolidation of code added in the recent schema v2 work:

- Share a single ActorRef::user() constructor between server control
  actions and workflow provenance conversions.
- Share StageScope::from_context() between current_stage_scope and
  StageScope::for_handler so the 4-field construction lives in one place.
- Collapse RunEvent::to_value's if-let chain into an insert_opt helper.
- Use Value::String(id.to_string()) instead of serde_json::to_value for
  StageId/ParallelBranchId when seeding the parallel branch context.
- Share parse_event_envelopes via tests/it/support/mod.rs instead of
  duplicating the parsing block in two CLI run_events helpers.

Also fix parallel-branch git.commit to emit via emit_scoped with a
branch-specific StageScope so it carries stage_id / parallel_group_id /
parallel_branch_id alongside the other stage-scoped events.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

lib/crates/fabro-cli/tests/it/cmd/support.rs

@@ -661,15 +661,7 @@ pub(crate) fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
         run_dir,
         &format!("/api/v1/runs/{run_id}/events"),
     ));
-    let items = response["data"]
-        .as_array()
-        .cloned()
-        .expect("event list response should contain a data array");
-    items
-        .into_iter()
-        .map(serde_json::from_value)
-        .collect::<Result<Vec<_>, _>>()
-        .expect("wire event envelope list should parse")
+    crate::support::parse_event_envelopes(&response)
 }
 
 pub(crate) fn wait_for_event_names(run_dir: &Path, expected: &[&str]) {

lib/crates/fabro-cli/tests/it/support/mod.rs

@@ -1,6 +1,7 @@
 use std::path::{Path, PathBuf};
 
 use assert_cmd::Command;
+use fabro_store::EventEnvelope;
 use fabro_test::TestContext;
 use fabro_types::RunId;
 macro_rules! fabro_json_snapshot {
@@ -60,6 +61,17 @@ pub(crate) fn unique_run_id() -> String {
     RunId::new().to_string()
 }
 
+pub(crate) fn parse_event_envelopes(response: &serde_json::Value) -> Vec<EventEnvelope> {
+    response["data"]
+        .as_array()
+        .expect("event list response should contain a data array")
+        .iter()
+        .cloned()
+        .map(serde_json::from_value)
+        .collect::<Result<Vec<_>, _>>()
+        .expect("wire event envelope list should parse")
+}
+
 pub(crate) struct LightweightCli {
     home_dir: tempfile::TempDir,
 }

lib/crates/fabro-cli/tests/it/workflow/mod.rs

@@ -173,15 +173,7 @@ fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
         storage_dir,
         &format!("/api/v1/runs/{run_id}/events"),
     ));
-    let items = response["data"]
-        .as_array()
-        .cloned()
-        .expect("event list response should contain a data array");
-    items
-        .into_iter()
-        .map(serde_json::from_value)
-        .collect::<Result<Vec<_>, _>>()
-        .expect("wire event envelope list should parse")
+    crate::support::parse_event_envelopes(&response)
 }
 
 macro_rules! sandbox_tests {

lib/crates/fabro-server/src/server.rs

@@ -34,7 +34,7 @@ use fabro_store::{
     ArtifactStore, Database, EventEnvelope, EventPayload, PendingInterviewRecord, StageId,
 };
 use fabro_types::{
-    ActorKind, ActorRef, EventBody, InterviewQuestionRecord, InterviewQuestionType, RunBlobId,
+    ActorRef, EventBody, InterviewQuestionRecord, InterviewQuestionType, RunBlobId,
     RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance, RunServerProvenance,
     RunSubjectProvenance, Settings,
 };
@@ -5155,12 +5155,7 @@ async fn append_control_request(
 }
 
 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),
-    })
+    subject.login.clone().map(ActorRef::user)
 }
 
 fn schedule_worker_kill(state: Arc<AppState>, run_id: RunId, worker_pid: u32) {

lib/crates/fabro-types/src/run_event/mod.rs

@@ -44,6 +44,17 @@ pub struct ActorRef {
     pub display: Option<String>,
 }
 
+impl ActorRef {
+    #[must_use]
+    pub fn user(login: String) -> Self {
+        Self {
+            kind: ActorKind::User,
+            id: Some(login.clone()),
+            display: Some(login),
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq)]
 pub struct RunEvent {
     pub id: String,
@@ -677,50 +688,46 @@ impl RunEvent {
     }
 
     pub fn to_value(&self) -> serde_json::Result<Value> {
+        fn insert_opt<T: Serialize>(
+            map: &mut Map<String, Value>,
+            key: &str,
+            value: Option<&T>,
+        ) -> serde_json::Result<()> {
+            if let Some(v) = value {
+                map.insert(key.to_string(), serde_json::to_value(v)?);
+            }
+            Ok(())
+        }
+
         let mut map = Map::new();
-        map.insert("id".to_string(), serde_json::to_value(&self.id)?);
+        map.insert("id".to_string(), Value::String(self.id.clone()));
         map.insert("ts".to_string(), serde_json::to_value(self.ts)?);
         map.insert("run_id".to_string(), serde_json::to_value(self.run_id)?);
         map.insert(
             "event".to_string(),
             Value::String(self.body.event_name().to_string()),
         );
-        if let Some(value) = &self.session_id {
-            map.insert("session_id".to_string(), Value::String(value.clone()));
-        }
-        if let Some(value) = &self.parent_session_id {
-            map.insert(
-                "parent_session_id".to_string(),
-                Value::String(value.clone()),
-            );
-        }
-        if let Some(value) = &self.node_id {
-            map.insert("node_id".to_string(), Value::String(value.clone()));
-        }
-        if let Some(value) = &self.node_label {
-            map.insert("node_label".to_string(), Value::String(value.clone()));
-        }
-        if let Some(value) = &self.stage_id {
-            map.insert("stage_id".to_string(), serde_json::to_value(value)?);
-        }
-        if let Some(value) = &self.parallel_group_id {
-            map.insert(
-                "parallel_group_id".to_string(),
-                serde_json::to_value(value)?,
-            );
-        }
-        if let Some(value) = &self.parallel_branch_id {
-            map.insert(
-                "parallel_branch_id".to_string(),
-                serde_json::to_value(value)?,
-            );
-        }
-        if let Some(value) = &self.tool_call_id {
-            map.insert("tool_call_id".to_string(), Value::String(value.clone()));
-        }
-        if let Some(actor) = &self.actor {
-            map.insert("actor".to_string(), serde_json::to_value(actor)?);
-        }
+        insert_opt(&mut map, "session_id", self.session_id.as_ref())?;
+        insert_opt(
+            &mut map,
+            "parent_session_id",
+            self.parent_session_id.as_ref(),
+        )?;
+        insert_opt(&mut map, "node_id", self.node_id.as_ref())?;
+        insert_opt(&mut map, "node_label", self.node_label.as_ref())?;
+        insert_opt(&mut map, "stage_id", self.stage_id.as_ref())?;
+        insert_opt(
+            &mut map,
+            "parallel_group_id",
+            self.parallel_group_id.as_ref(),
+        )?;
+        insert_opt(
+            &mut map,
+            "parallel_branch_id",
+            self.parallel_branch_id.as_ref(),
+        )?;
+        insert_opt(&mut map, "tool_call_id", self.tool_call_id.as_ref())?;
+        insert_opt(&mut map, "actor", self.actor.as_ref())?;
         map.insert("properties".to_string(), self.body.properties_value()?);
         Ok(Value::Object(map))
     }

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

@@ -136,7 +136,6 @@ pub mod keys {
 pub use fabro_core::Context;
 
 use crate::event::StageScope;
-use crate::run_dir::visit_from_context;
 use fabro_graphviz::Fidelity;
 use fabro_types::{ParallelBranchId, StageId};
 
@@ -188,13 +187,7 @@ impl WorkflowContext for Context {
         let node_id = self
             .get(keys::CURRENT_NODE)
             .and_then(|value| value.as_str().map(String::from))?;
-        let visit = u32::try_from(visit_from_context(self)).unwrap_or(u32::MAX);
-        Some(StageScope {
-            node_id,
-            visit,
-            parallel_group_id: self.parallel_group_id(),
-            parallel_branch_id: self.parallel_branch_id(),
-        })
+        Some(StageScope::from_context(self, node_id))
     }
 }
 

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

@@ -1435,12 +1435,12 @@ fn stored_event_fields_for_variant(event: &Event) -> StoredEventFields {
 }
 
 fn actor_from_provenance(provenance: &RunProvenance) -> Option<ActorRef> {
-    let login = provenance.subject.as_ref()?.login.clone()?;
-    Some(ActorRef {
-        kind: ActorKind::User,
-        id: Some(login.clone()),
-        display: Some(login),
-    })
+    provenance
+        .subject
+        .as_ref()?
+        .login
+        .clone()
+        .map(ActorRef::user)
 }
 
 fn agent_tool_call_id(event: &AgentEvent) -> Option<&str> {
@@ -2475,18 +2475,25 @@ pub struct StageScope {
 }
 
 impl StageScope {
-    /// Build scope for a handler invocation. Prefers the current_stage_scope
-    /// set by the fidelity lifecycle before_attempt hook, but falls back to
-    /// a scope synthesized from the node id and the context's visit count
-    /// for tests and other direct-handler call sites that don't go through
-    /// the full lifecycle.
-    pub fn for_handler(context: &WfContext, node_id: impl Into<String>) -> Self {
-        context.current_stage_scope().unwrap_or_else(|| Self {
+    /// Build a scope from the given node id, sourcing visit count and parallel
+    /// ids from the current context.
+    pub fn from_context(context: &WfContext, node_id: impl Into<String>) -> Self {
+        Self {
             node_id: node_id.into(),
             visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX),
             parallel_group_id: context.parallel_group_id(),
             parallel_branch_id: context.parallel_branch_id(),
-        })
+        }
+    }
+
+    /// Build scope for a handler invocation. Prefers the `current_stage_scope`
+    /// seeded by the fidelity lifecycle before_attempt hook, and falls back to
+    /// synthesizing one from `node_id` for direct-handler call sites (tests,
+    /// etc.) that don't go through the full lifecycle.
+    pub fn for_handler(context: &WfContext, node_id: impl Into<String>) -> Self {
+        context
+            .current_stage_scope()
+            .unwrap_or_else(|| Self::from_context(context, node_id))
     }
 }
 

lib/crates/fabro-workflow/src/handler/parallel.rs

@@ -214,12 +214,11 @@ impl Handler for ParallelHandler {
             );
             branch_context.set(
                 keys::INTERNAL_PARALLEL_GROUP_ID,
-                serde_json::to_value(&parallel_group_id).expect("StageId serializes as string"),
+                serde_json::Value::String(parallel_group_id.to_string()),
             );
             branch_context.set(
                 keys::INTERNAL_PARALLEL_BRANCH_ID,
-                serde_json::to_value(&parallel_branch_id)
-                    .expect("ParallelBranchId serializes as string"),
+                serde_json::Value::String(parallel_branch_id.to_string()),
             );
 
             let (branch_sandbox, worktree_path): (Arc<dyn Sandbox>, Option<PathBuf>) = if let (
@@ -280,8 +279,6 @@ impl Handler for ParallelHandler {
             });
         }
 
-        let parent_scope = StageScope::for_handler(context, &node.id);
-
         // --- Fan out: concurrent execution ---
         let mut handles = Vec::new();
         for setup in branch_setups {
@@ -304,7 +301,12 @@ impl Handler for ParallelHandler {
                 .map(|gs| gs.git_author.clone())
                 .unwrap_or_default();
             let group_id = parallel_group_id.clone();
-            let branch_scope = parent_scope.clone();
+            let branch_scope = StageScope {
+                node_id: setup.target_id.clone(),
+                visit: 1,
+                parallel_group_id: Some(group_id.clone()),
+                parallel_branch_id: Some(setup.parallel_branch_id.clone()),
+            };
 
             let handle = tokio::spawn(async move {
                 let _permit = sem
@@ -405,10 +407,13 @@ impl Handler for ParallelHandler {
                     match sha_result {
                         Ok(r) if r.exit_code == 0 => {
                             let sha = r.stdout.trim().to_string();
-                            emitter.emit(&Event::GitCommit {
-                                node_id: Some(setup.target_id.clone()),
-                                sha: sha.clone(),
-                            });
+                            emitter.emit_scoped(
+                                &Event::GitCommit {
+                                    node_id: Some(setup.target_id.clone()),
+                                    sha: sha.clone(),
+                                },
+                                &branch_scope,
+                            );
                             Some(sha)
                         }
                         _ => None,