scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor(workflow): split event module by responsibility

Browse source 
Keep fabro_workflow::event as the public facade while moving event conversion, names, redaction, sink, emitter, stored-field helpers, and StageScope into focused modules. Co-locate the existing event tests with the moved code and update the events strategy docs for the new module layout.
This commit is contained in:
Bryan Helmkamp 2026-05-02 14:52:10 -04:00
parent b1d560faf7
commit ae5ccb5ce2
No known key found for this signature in database
12 changed files with 4368 additions and 4057 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
docs/internal/events-strategy.md
View file

@ -19,7 +19,7 @@ Engine/Handler -> Event -> Emitter::emit()
`- CLI / tests / metrics listeners
```
The canonical `RunEvent` is built exactly once in `fabro-workflow/src/event.rs`.
The canonical `RunEvent` is built exactly once in the `fabro-workflow::event` module.
- `Event` (in `fabro-workflow`) is the internal typed event emitted by engine and handlers.
- `Emitter` owns an immutable `run_id` and converts `Event` into `RunEvent` via `to_run_event_at()`.
@ -94,7 +94,7 @@ The external event name is lowercase dot notation, for example:
- `sandbox.ready`
- `parallel.branch.completed`
`event_name()` in `event.rs` is exhaustive. Do not use wildcard fallthroughs when adding new variants.
`event_name()` in the `fabro-workflow::event` module is exhaustive. Do not use wildcard fallthroughs when adding new variants.
## Node And Session Metadata
@ -138,7 +138,7 @@ Add a variant to `EventBody` in `fabro-types/src/run_event/mod.rs` with a corres
### 5. Map envelope fields and construct `EventBody`
Update `stored_event_fields()` and `event_body_from_event()` in `fabro-workflow/src/event.rs`:
Update `stored_event_fields()` and `event_body_from_event()` in the `fabro-workflow::event` module:
- Move `node_id`, `node_label`, `session_id`, and `parent_session_id` into the envelope when appropriate.
- Construct the `EventBody` variant directly from the `Event` fields.

196
docs/superpowers/plans/2026-05-02-event-module-split.md Normal file
View file

@ -0,0 +1,196 @@
# Mechanical Event Module Split Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Split `fabro-workflow`'s oversized event module into focused child modules while preserving `fabro_workflow::event::{...}` as the public API.
**Architecture:** Keep `src/event.rs` as a facade that declares child modules and re-exports the existing public symbols. Move code mechanically by responsibility, co-locate tests with the modules they cover, and move `StageScope` to a crate-root module while preserving `fabro_workflow::event::StageScope`. `event/events.rs` intentionally remains the largest file because it keeps the `Event` enum and its exhaustive tracing behavior together.
**Tech Stack:** Rust, Tokio, serde/serde_json, chrono, uuid, fabro-types `RunEvent` / `EventBody`, fabro-store `EventPayload`, existing `cargo nextest` workflow tests.
---
## Files
- Modify: `lib/crates/fabro-workflow/src/event.rs`
- Create: `lib/crates/fabro-workflow/src/event/events.rs`
- Create: `lib/crates/fabro-workflow/src/event/names.rs`
- Create: `lib/crates/fabro-workflow/src/event/stored_fields.rs`
- Create: `lib/crates/fabro-workflow/src/event/convert.rs`
- Create: `lib/crates/fabro-workflow/src/event/redaction.rs`
- Create: `lib/crates/fabro-workflow/src/event/sink.rs`
- Create: `lib/crates/fabro-workflow/src/event/emitter.rs`
- Create: `lib/crates/fabro-workflow/src/stage_scope.rs`
- Modify: `lib/crates/fabro-workflow/src/lib.rs`
- Modify: `docs/internal/events-strategy.md`
## Task 1: Confirm Private Helpers and Build the Facade
- [x] Confirm these helpers are not used outside `event.rs` before moving them:
```bash
rg -n "StoredEventFields|event_body_from_event|normalized_event_value|redacted_event_value|RunEventCommand|RunEventSinkFuture|RunEventSinkCallback|RunEventTransform|agent_tool_call_id|agent_actor_for_event|default_node_label|node_stored_fields|billed_token_counts_from_llm|stage_status_from_string|epoch_millis" . --glob '!target' --glob '!apps/fabro-web/dist/**'
```
Expected: production hits are limited to `lib/crates/fabro-workflow/src/event.rs`; plan and historical docs may mention the names.
- [x] Replace `event.rs` with child module declarations and public re-exports:
```rust
mod convert;
mod emitter;
mod events;
mod names;
mod redaction;
mod sink;
mod stored_fields;
pub use fabro_types::{EventBody, RunNoticeLevel};
pub use self::convert::{to_run_event, to_run_event_at};
pub use self::emitter::Emitter;
pub use self::events::Event;
pub use self::names::event_name;
pub use self::redaction::{
build_redacted_event_payload, event_payload_from_redacted_json, redacted_event_json,
};
pub use self::sink::{
RunEventLogger, RunEventSink, StoreProgressLogger, append_event, append_event_to_sink,
};
pub use crate::stage_scope::StageScope;
```
- [x] Add `mod stage_scope;` to `lib.rs`. Do not expose a new `fabro_workflow::stage_scope` public module in this pass; preserve the existing public path through `pub use crate::stage_scope::StageScope` in `event.rs`.
## Task 2: Move Event, Names, Stage Scope, and Stored Fields
- [x] Move the `Event` enum, `Event::pull_request_created`, and `Event::trace` into `event/events.rs`. Keep all derives, serde attributes, clippy allowances, variant fields, tracing levels, tracing fields, and `PullRequestRecord` behavior unchanged.
- [x] Move `event_name` into `event/names.rs`. Keep the exhaustive match and all returned strings unchanged.
- [x] Move `StageScope` into `stage_scope.rs`. Keep constructors and `stage_id()` unchanged, including use of `visit_from_context`. Preserve `fabro_workflow::event::StageScope` by re-exporting it from `event.rs`.
- [x] In `stage_scope.rs`, import only the dependencies needed by `StageScope`: `fabro_types::{ParallelBranchId, StageId}`, `crate::context::{Context as WfContext, WorkflowContext}`, and `crate::run_dir::visit_from_context`. Do not import from `crate::event`.
- [x] Move stored-field helpers into `event/stored_fields.rs`:
- `StoredEventFields`
- `default_node_label`
- `node_stored_fields`
- `stored_event_fields`
- `stored_event_fields_for_variant`
- `agent_tool_call_id`
- `agent_actor_for_event`
- [x] Make both `StoredEventFields` and `stored_event_fields` `pub(super)` because `convert.rs` calls the function and reads fields from its return value. Keep `default_node_label`, `node_stored_fields`, `stored_event_fields_for_variant`, `agent_tool_call_id`, and `agent_actor_for_event` private to `stored_fields.rs`.
## Task 3: Move Conversion and Redaction
- [x] Move conversion helpers into `event/convert.rs`:
- `billed_token_counts_from_llm`
- `stage_status_from_string`
- `event_body_from_event`
- `to_run_event`
- `to_run_event_at`
- [x] Keep `event_body_from_event` private to `convert.rs`. Import `stored_event_fields` from `event/stored_fields.rs`. Keep `to_run_event` and `to_run_event_at` public through the facade re-export.
- [x] Move redaction helpers into `event/redaction.rs`:
- `build_redacted_event_payload`
- `redacted_event_json`
- `normalized_event_value`
- `redacted_event_value`
- `event_payload_from_redacted_json`
- [x] Keep `normalized_event_value` and `redacted_event_value` private. Keep redaction behavior exactly as `RunEvent::to_value() -> normalize_json_value -> redact_json_value`.
## Task 4: Move Sink, Logger, and Emitter Plumbing
- [x] Move sink and logger code into `event/sink.rs`:
- `append_event`
- `append_event_to_sink`
- `RunEventSink`
- `RunEventSinkFuture`
- `RunEventSinkCallback`
- `RunEventTransform`
- `RunEventCommand`
- `RunEventLogger`
- `StoreProgressLogger`
- [x] Keep `RunEventCommand` and callback type aliases private. Preserve the iterative `RunEventSink::write_run_event` stack logic and redacted JSONL output behavior.
- [x] Move emitter code into `event/emitter.rs`:
- `epoch_millis`
- `EventListener`
- `Emitter`
- `Debug`, `Default`, and inherent impls
- [x] Keep `dispatch_run_event` as `pub(crate)` and keep all public `Emitter` methods unchanged.
## Task 5: Co-locate Tests and Update Docs
- [x] Move each existing inline test into the module it characterizes. Keep test names, fixtures, assertions, and async test attributes unchanged.
- [x] Use this test placement:
- `emitter.rs`: `event_emitter_*`
- `sink.rs`: `run_event_sink_*`, `run_event_logger_*`, `append_event_writes_store_event_shape`
- `redaction.rs`: `build_redacted_event_payload_*`
- `names.rs`: `event_name_matches_new_dot_notation`, `run_archived_event_name_matches_dot_notation`
- `stored_fields.rs`: only direct helper tests introduced during the move, if needed; it is acceptable for this module to have no tests
- `convert.rs`: all `run_event_*` tests, actor envelope tests, tool call id tests, parallel id tests, stage id tests, metadata snapshot mapping tests, and `stage_scope_populates_stage_id_on_non_stage_events`
- `stage_scope.rs`: direct `StageScope` constructor tests introduced during the move, if needed; it is acceptable for this module to have no tests
- [x] Update `docs/internal/events-strategy.md` references that say the canonical conversion is in `fabro-workflow/src/event.rs` so they refer to the `fabro-workflow::event` module. Do not change event-strategy rules.
- [x] Do not update production call sites outside the event module unless compilation requires an import fix. The intended result is that existing imports such as `crate::event::{Emitter, Event}` and `fabro_workflow::event::{Event, to_run_event}` continue to work.
## Task 6: Verify the Mechanical Split
- [x] Run focused event tests:
```bash
cargo nextest run -p fabro-workflow event
```
Expected: all matching tests pass.
- [x] Run the full workflow crate test suite:
```bash
cargo nextest run -p fabro-workflow
```
Expected: all `fabro-workflow` tests pass.
- [x] Run format check:
```bash
cargo +nightly-2026-04-14 fmt --check --all
```
Expected: no formatting diffs.
- [x] Run clippy:
```bash
cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings
```
Expected: no warnings.
## Acceptance Criteria
- `lib/crates/fabro-workflow/src/event.rs` is a small facade module.
- Existing public API paths under `fabro_workflow::event::{...}` still compile.
- `StageScope` lives at crate root and remains available from `fabro_workflow::event::StageScope`.
- Tests are co-located with the module they cover; there is no catch-all `event/tests.rs`.
- Event wire names, envelope metadata, `EventBody` conversion, redaction, JSONL sink output, store payload shape, and emitter dispatch behavior are unchanged.
- No macro registry, generated event table, or domain-split `Event` enum is introduced in this pass.
- `docs/internal/events-strategy.md` remains accurate after the file split.
## Assumptions
- This is a strictly mechanical refactor; deeper cleanup like domain-specific event enums or shared event DTO extraction is out of scope.
- `event/events.rs` remains intentionally large because keeping `Event::trace` with `Event` avoids splitting a pure inherent `Event` behavior into a separate file.
- Keeping `convert.rs` focused on `Event -> EventBody` body conversion is acceptable even if it remains one of the larger event files.
- `RunEventSink`, `RunEventLogger`, and `StoreProgressLogger` remain together in `sink.rs` for this pass; split them later only if that file remains hard to navigate after tests are co-located.
- Existing tests are sufficient characterization coverage for this split; new behavior tests are not required unless a moved module exposes an accidental visibility issue.

4072
lib/crates/fabro-workflow/src/event.rs
View file

File diff suppressed because it is too large Load diff

1778
lib/crates/fabro-workflow/src/event/convert.rs Normal file
View file

File diff suppressed because it is too large Load diff

193
lib/crates/fabro-workflow/src/event/emitter.rs Normal file
View file

@ -0,0 +1,193 @@
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, Ordering};
use ::fabro_types::{RunEvent, RunId, RunNoticeLevel};
use chrono::Utc;
use fabro_agent::{WorktreeEvent, WorktreeEventCallback};
use super::Event;
use super::convert::to_run_event_at;
use crate::stage_scope::StageScope;
fn epoch_millis() -> i64 {
let millis = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis();
i64::try_from(millis).unwrap_or(i64::MAX)
}
/// Listener callback type for workflow run events.
type EventListener = Arc<dyn Fn(&RunEvent) + Send + Sync>;
/// Callback-based event emitter for workflow run events.
pub struct Emitter {
run_id: RunId,
listeners: std::sync::Mutex<Vec<EventListener>>,
/// Epoch milliseconds of the last `emit()` or `touch()` call. 0 until first
/// event.
last_event_at: AtomicI64,
}
impl std::fmt::Debug for Emitter {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let count = self.listeners.lock().map_or(0, |l| l.len());
f.debug_struct("Emitter")
.field("run_id", &self.run_id)
.field("listener_count", &count)
.field("last_event_at", &self.last_event_at.load(Ordering::Relaxed))
.finish()
}
}
impl Default for Emitter {
fn default() -> Self {
Self::new(RunId::new())
}
}
impl Emitter {
#[must_use]
pub fn new(run_id: RunId) -> Self {
Self {
run_id,
listeners: std::sync::Mutex::new(Vec::new()),
last_event_at: AtomicI64::new(0),
}
}
#[must_use]
pub fn run_id(&self) -> RunId {
self.run_id
}
pub fn on_event(&self, listener: impl Fn(&RunEvent) + Send + Sync + 'static) {
self.listeners
.lock()
.expect("listeners lock poisoned")
.push(Arc::new(listener));
}
pub fn emit(&self, event: &Event) {
self.emit_with_scope(event, None);
}
pub fn emit_scoped(&self, event: &Event, scope: &StageScope) {
self.emit_with_scope(event, Some(scope));
}
pub fn notice(
&self,
level: RunNoticeLevel,
code: impl Into<String>,
message: impl Into<String>,
) {
self.emit(&Event::RunNotice {
level,
code: code.into(),
message: message.into(),
});
}
fn emit_with_scope(&self, event: &Event, scope: Option<&StageScope>) {
self.last_event_at.store(epoch_millis(), Ordering::Relaxed);
event.trace();
if let Event::WorkflowRunStarted { run_id, .. } = event {
debug_assert_eq!(
*run_id, self.run_id,
"workflow run started event must match emitter run_id"
);
}
let stored = to_run_event_at(&self.run_id, event, Utc::now(), scope);
self.dispatch_run_event(&stored);
}
pub(crate) fn dispatch_run_event(&self, event: &RunEvent) {
self.last_event_at.store(epoch_millis(), Ordering::Relaxed);
// Clone the listener list so we don't hold the lock during dispatch.
// This prevents deadlocks if a listener calls emit() reentrantly.
// Note: listeners added during this emit() won't receive the current event.
let snapshot: Vec<EventListener> = self
.listeners
.lock()
.expect("listeners lock poisoned")
.clone();
for listener in &snapshot {
listener(event);
}
}
/// Returns the epoch milliseconds of the last `emit()` or `touch()` call.
/// Returns 0 if neither has been called.
pub fn last_event_at(&self) -> i64 {
self.last_event_at.load(Ordering::Relaxed)
}
/// Manually update the last-event timestamp (e.g. to seed the watchdog at
/// workflow run start).
pub fn touch(&self) {
self.last_event_at.store(epoch_millis(), Ordering::Relaxed);
}
/// Build a [`WorktreeEventCallback`] that forwards worktree lifecycle
/// events as [`Event`]s on this emitter.
pub fn worktree_callback(self: Arc<Self>) -> WorktreeEventCallback {
Arc::new(move |event| match event {
WorktreeEvent::BranchCreated { branch, sha } => {
self.emit(&Event::GitBranch { branch, sha });
}
WorktreeEvent::WorktreeAdded { path, branch } => {
self.emit(&Event::GitWorktreeAdd { path, branch });
}
WorktreeEvent::WorktreeRemoved { path } => {
self.emit(&Event::GitWorktreeRemove { path });
}
})
}
}
#[cfg(test)]
mod tests {
use std::sync::{Arc, Mutex};
use ::fabro_types::fixtures;
use super::*;
use crate::event::Event;
#[test]
fn event_emitter_new_has_no_listeners() {
let emitter = Emitter::new(fixtures::RUN_1);
assert_eq!(emitter.listeners.lock().unwrap().len(), 0);
}
#[test]
fn event_emitter_calls_listener_with_envelope() {
let emitter = Emitter::new(fixtures::RUN_1);
let received = Arc::new(Mutex::new(Vec::new()));
let received_clone = Arc::clone(&received);
emitter.on_event(move |event| {
received_clone.lock().unwrap().push(event.clone());
});
emitter.emit(&Event::WorkflowRunStarted {
name: "test".to_string(),
run_id: fixtures::RUN_1,
base_branch: None,
base_sha: None,
run_branch: None,
worktree_dir: None,
goal: None,
});
let events = received.lock().unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_name(), "run.started");
assert_eq!(events[0].run_id, fixtures::RUN_1);
assert!(events[0].id.len() >= 32);
}
#[test]
fn event_emitter_default() {
let emitter = Emitter::default();
assert_eq!(emitter.listeners.lock().unwrap().len(), 0);
}
}

1281
lib/crates/fabro-workflow/src/event/events.rs Normal file
View file

File diff suppressed because it is too large Load diff

193
lib/crates/fabro-workflow/src/event/names.rs Normal file
View file

@ -0,0 +1,193 @@
use fabro_agent::{AgentEvent, SandboxEvent};
use super::Event;
#[must_use]
pub fn event_name(event: &Event) -> &'static str {
match event {
Event::RunCreated { .. } => "run.created",
Event::WorkflowRunStarted { .. } => "run.started",
Event::RunSubmitted { .. } => "run.submitted",
Event::RunQueued => "run.queued",
Event::RunStarting => "run.starting",
Event::RunRunning => "run.running",
Event::RunBlocked { .. } => "run.blocked",
Event::RunUnblocked => "run.unblocked",
Event::RunRemoving => "run.removing",
Event::RunCancelRequested { .. } => "run.cancel.requested",
Event::RunPauseRequested { .. } => "run.pause.requested",
Event::RunUnpauseRequested { .. } => "run.unpause.requested",
Event::RunPaused => "run.paused",
Event::RunUnpaused => "run.unpaused",
Event::RunSupersededBy { .. } => "run.superseded_by",
Event::RunArchived { .. } => "run.archived",
Event::RunUnarchived { .. } => "run.unarchived",
Event::WorkflowRunCompleted { .. } => "run.completed",
Event::WorkflowRunFailed { .. } => "run.failed",
Event::RunNotice { .. } => "run.notice",
Event::MetadataSnapshotStarted { .. } => "metadata.snapshot.started",
Event::MetadataSnapshotCompleted { .. } => "metadata.snapshot.completed",
Event::MetadataSnapshotFailed { .. } => "metadata.snapshot.failed",
Event::StageStarted { .. } => "stage.started",
Event::StageCompleted { .. } => "stage.completed",
Event::StageFailed { .. } => "stage.failed",
Event::StageRetrying { .. } => "stage.retrying",
Event::ParallelStarted { .. } => "parallel.started",
Event::ParallelBranchStarted { .. } => "parallel.branch.started",
Event::ParallelBranchCompleted { .. } => "parallel.branch.completed",
Event::ParallelCompleted { .. } => "parallel.completed",
Event::InterviewStarted { .. } => "interview.started",
Event::InterviewCompleted { .. } => "interview.completed",
Event::InterviewTimeout { .. } => "interview.timeout",
Event::InterviewInterrupted { .. } => "interview.interrupted",
Event::CheckpointCompleted { .. } => "checkpoint.completed",
Event::CheckpointFailed { .. } => "checkpoint.failed",
Event::GitCommit { .. } => "git.commit",
Event::GitPush { .. } => "git.push",
Event::GitBranch { .. } => "git.branch",
Event::GitWorktreeAdd { .. } => "git.worktree.added",
Event::GitWorktreeRemove { .. } => "git.worktree.removed",
Event::GitFetch { .. } => "git.fetch",
Event::GitReset { .. } => "git.reset",
Event::EdgeSelected { .. } => "edge.selected",
Event::LoopRestart { .. } => "loop.restart",
Event::Prompt { .. } => "stage.prompt",
Event::PromptCompleted { .. } => "prompt.completed",
Event::Agent { event, .. } => match event {
AgentEvent::SessionStarted { .. } => "agent.session.started",
AgentEvent::SessionEnded => "agent.session.ended",
AgentEvent::ProcessingEnd => "agent.processing.end",
AgentEvent::UserInput { .. } => "agent.input",
AgentEvent::AssistantTextStart => "agent.output.start",
AgentEvent::AssistantOutputReplace { .. } => "agent.output.replace",
AgentEvent::AssistantMessage { .. } => "agent.message",
AgentEvent::TextDelta { .. } => "agent.text.delta",
AgentEvent::ReasoningDelta { .. } => "agent.reasoning.delta",
AgentEvent::ToolCallStarted { .. } => "agent.tool.started",
AgentEvent::ToolCallOutputDelta { .. } => "agent.tool.output.delta",
AgentEvent::ToolCallCompleted { .. } => "agent.tool.completed",
AgentEvent::Error { .. } => "agent.error",
AgentEvent::Warning { .. } => "agent.warning",
AgentEvent::LoopDetected => "agent.loop.detected",
AgentEvent::TurnLimitReached { .. } => "agent.turn.limit",
AgentEvent::SkillExpanded { .. } => "agent.skill.expanded",
AgentEvent::SteeringInjected { .. } => "agent.steering.injected",
AgentEvent::CompactionStarted { .. } => "agent.compaction.started",
AgentEvent::CompactionCompleted { .. } => "agent.compaction.completed",
AgentEvent::LlmRetry { .. } => "agent.llm.retry",
AgentEvent::SubAgentSpawned { .. } => "agent.sub.spawned",
AgentEvent::SubAgentCompleted { .. } => "agent.sub.completed",
AgentEvent::SubAgentFailed { .. } => "agent.sub.failed",
AgentEvent::SubAgentClosed { .. } => "agent.sub.closed",
AgentEvent::McpServerReady { .. } => "agent.mcp.ready",
AgentEvent::McpServerFailed { .. } => "agent.mcp.failed",
},
Event::SubgraphStarted { .. } => "subgraph.started",
Event::SubgraphCompleted { .. } => "subgraph.completed",
Event::Sandbox { event } => match event {
SandboxEvent::Initializing { .. } => "sandbox.initializing",
SandboxEvent::Ready { .. } => "sandbox.ready",
SandboxEvent::InitializeFailed { .. } => "sandbox.failed",
SandboxEvent::CleanupStarted { .. } => "sandbox.cleanup.started",
SandboxEvent::CleanupCompleted { .. } => "sandbox.cleanup.completed",
SandboxEvent::CleanupFailed { .. } => "sandbox.cleanup.failed",
SandboxEvent::SnapshotPulling { .. } => "sandbox.snapshot.pulling",
SandboxEvent::SnapshotPulled { .. } => "sandbox.snapshot.pulled",
SandboxEvent::SnapshotEnsuring { .. } => "sandbox.snapshot.ensuring",
SandboxEvent::SnapshotCreating { .. } => "sandbox.snapshot.creating",
SandboxEvent::SnapshotReady { .. } => "sandbox.snapshot.ready",
SandboxEvent::SnapshotFailed { .. } => "sandbox.snapshot.failed",
SandboxEvent::GitCloneStarted { .. } => "sandbox.git.started",
SandboxEvent::GitCloneCompleted { .. } => "sandbox.git.completed",
SandboxEvent::GitCloneFailed { .. } => "sandbox.git.failed",
},
Event::SandboxInitialized { .. } => "sandbox.initialized",
Event::SetupStarted { .. } => "setup.started",
Event::SetupCommandStarted { .. } => "setup.command.started",
Event::SetupCommandCompleted { .. } => "setup.command.completed",
Event::SetupCompleted { .. } => "setup.completed",
Event::SetupFailed { .. } => "setup.failed",
Event::StallWatchdogTimeout { .. } => "watchdog.timeout",
Event::ArtifactCaptured { .. } => "artifact.captured",
Event::SshAccessReady { .. } => "ssh.ready",
Event::Failover { .. } => "agent.failover",
Event::CliEnsureStarted { .. } => "cli.ensure.started",
Event::CliEnsureCompleted { .. } => "cli.ensure.completed",
Event::CliEnsureFailed { .. } => "cli.ensure.failed",
Event::CommandStarted { .. } => "command.started",
Event::CommandCompleted { .. } => "command.completed",
Event::AgentCliStarted { .. } => "agent.cli.started",
Event::AgentCliCompleted { .. } => "agent.cli.completed",
Event::PullRequestCreated { .. } => "pull_request.created",
Event::PullRequestFailed { .. } => "pull_request.failed",
Event::DevcontainerResolved { .. } => "devcontainer.resolved",
Event::DevcontainerLifecycleStarted { .. } => "devcontainer.lifecycle.started",
Event::DevcontainerLifecycleCommandStarted { .. } => {
"devcontainer.lifecycle.command.started"
}
Event::DevcontainerLifecycleCommandCompleted { .. } => {
"devcontainer.lifecycle.command.completed"
}
Event::DevcontainerLifecycleCompleted { .. } => "devcontainer.lifecycle.completed",
Event::DevcontainerLifecycleFailed { .. } => "devcontainer.lifecycle.failed",
Event::RetroStarted { .. } => "retro.started",
Event::RetroCompleted { .. } => "retro.completed",
Event::RetroFailed { .. } => "retro.failed",
}
}
#[cfg(test)]
mod tests {
use ::fabro_types::{ParallelBranchId, StageId};
use fabro_agent::AgentEvent;
use super::*;
use crate::event::Event;
#[test]
fn event_name_matches_new_dot_notation() {
assert_eq!(
event_name(&Event::RetroStarted {
prompt: None,
provider: None,
model: None,
}),
"retro.started"
);
assert_eq!(
event_name(&Event::ParallelBranchStarted {
parallel_group_id: StageId::new("plan", 1),
parallel_branch_id: ParallelBranchId::new(StageId::new("plan", 1), 0),
branch: "fork".to_string(),
index: 0,
}),
"parallel.branch.started"
);
assert_eq!(
event_name(&Event::Agent {
stage: "code".to_string(),
visit: 1,
event: AgentEvent::SubAgentSpawned {
agent_id: "a1".to_string(),
depth: 1,
task: "do it".to_string(),
},
session_id: None,
parent_session_id: None,
}),
"agent.sub.spawned"
);
}
#[test]
fn run_archived_event_name_matches_dot_notation() {
assert_eq!(
event_name(&Event::RunArchived { actor: None }),
"run.archived"
);
assert_eq!(
event_name(&Event::RunUnarchived { actor: None }),
"run.unarchived"
);
}
}

81
lib/crates/fabro-workflow/src/event/redaction.rs Normal file
View file

@ -0,0 +1,81 @@
use ::fabro_types::{RunEvent, RunId};
use anyhow::{Context, Result};
use fabro_redact::redact_json_value;
use fabro_store::EventPayload;
use fabro_util::json::normalize_json_value;
use serde_json::Value;
pub fn build_redacted_event_payload(event: &RunEvent, run_id: &RunId) -> Result<EventPayload> {
let value = redacted_event_value(event)?;
EventPayload::new(value, run_id).map_err(anyhow::Error::from)
}
pub fn redacted_event_json(event: &RunEvent) -> Result<String> {
serde_json::to_string(&redacted_event_value(event)?).map_err(anyhow::Error::from)
}
fn normalized_event_value(event: &RunEvent) -> Result<Value> {
let value = event.to_value()?;
Ok(normalize_json_value(value))
}
fn redacted_event_value(event: &RunEvent) -> Result<Value> {
Ok(redact_json_value(normalized_event_value(event)?))
}
pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result<EventPayload> {
let value = serde_json::from_str(line).context("Failed to parse redacted event payload")?;
EventPayload::new(value, run_id).map_err(anyhow::Error::from)
}
#[cfg(test)]
mod tests {
use ::fabro_types::{fixtures, run_event as fabro_types};
use super::*;
use crate::event::{Event, to_run_event};
#[test]
fn build_redacted_event_payload_requires_id() {
let stored = to_run_event(&fixtures::RUN_8, &Event::RetroStarted {
prompt: Some("Analyze the run".to_string()),
provider: None,
model: None,
});
let payload = build_redacted_event_payload(&stored, &fixtures::RUN_8).unwrap();
assert_eq!(payload.as_value()["id"], stored.id);
assert_eq!(payload.as_value()["event"], "retro.started");
assert_eq!(
payload.as_value()["properties"]["prompt"],
"Analyze the run"
);
}
#[test]
fn build_redacted_event_payload_redacts_exec_output_tail_values() {
let secret = "sk-ant-api03-xK9mZ2vL8nQ5rT1wY4bC7dF0gH3jE6pA";
let stored = to_run_event(&fixtures::RUN_8, &Event::SetupFailed {
command: "setup".to_string(),
index: 0,
exit_code: 1,
stderr: "compat stderr".to_string(),
exec_output_tail: Some(fabro_types::ExecOutputTail {
stdout: Some(format!("stdout {secret}")),
stderr: Some("plain stderr".to_string()),
stdout_truncated: false,
stderr_truncated: false,
}),
});
let payload = build_redacted_event_payload(&stored, &fixtures::RUN_8).unwrap();
let payload_text = serde_json::to_string(payload.as_value()).unwrap();
assert!(!payload_text.contains(secret));
assert!(payload_text.contains("REDACTED"));
assert_eq!(payload.as_value()["event"], "setup.failed");
assert_eq!(
payload.as_value()["properties"]["exec_output_tail"]["stderr"],
"plain stderr"
);
}
}

339
lib/crates/fabro-workflow/src/event/sink.rs Normal file
View file

@ -0,0 +1,339 @@
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use ::fabro_types::{RunEvent, RunId};
use anyhow::Result;
use fabro_store::RunDatabase;
use tokio::io::{AsyncWrite, AsyncWriteExt};
use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot};
use super::emitter::Emitter;
use super::redaction::{build_redacted_event_payload, redacted_event_json};
use super::{Event, to_run_event};
use crate::runtime_store::RunStoreHandle;
pub async fn append_event(run_store: &RunDatabase, run_id: &RunId, event: &Event) -> Result<()> {
let stored = to_run_event(run_id, event);
let payload = build_redacted_event_payload(&stored, run_id)?;
run_store
.append_event(&payload)
.await
.map(|_| ())
.map_err(anyhow::Error::from)
}
pub async fn append_event_to_sink(
sink: &RunEventSink,
run_id: &RunId,
event: &Event,
) -> Result<()> {
let stored = to_run_event(run_id, event);
sink.write_run_event(&stored).await
}
#[derive(Clone)]
pub enum RunEventSink {
Store(RunStoreHandle),
JsonLines(Arc<AsyncMutex<Pin<Box<dyn AsyncWrite + Send>>>>),
Callback(Arc<RunEventSinkCallback>),
Map {
transform: Arc<RunEventTransform>,
inner: Box<Self>,
},
Composite(Vec<Self>),
}
type RunEventSinkFuture = Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>;
type RunEventSinkCallback = dyn Fn(RunEvent) -> RunEventSinkFuture + Send + Sync + 'static;
type RunEventTransform = dyn Fn(RunEvent) -> RunEvent + Send + Sync + 'static;
impl RunEventSink {
#[must_use]
pub fn store(run_store: RunDatabase) -> Self {
Self::Store(RunStoreHandle::local(run_store))
}
#[must_use]
pub fn backend(run_store: RunStoreHandle) -> Self {
Self::Store(run_store)
}
#[must_use]
pub fn json_lines<W>(writer: W) -> Self
where
W: AsyncWrite + Send + 'static,
{
Self::JsonLines(Arc::new(AsyncMutex::new(Box::pin(writer))))
}
#[must_use]
pub fn callback<F, Fut>(callback: F) -> Self
where
F: Fn(RunEvent) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<()>> + Send + 'static,
{
Self::Callback(Arc::new(move |event| Box::pin(callback(event))))
}
#[must_use]
pub fn fanout(sinks: Vec<Self>) -> Self {
let mut flattened = Vec::new();
for sink in sinks {
match sink {
Self::Composite(inner) => flattened.extend(inner),
other => flattened.push(other),
}
}
Self::Composite(flattened)
}
#[must_use]
pub fn map<F>(transform: F, inner: Self) -> Self
where
F: Fn(RunEvent) -> RunEvent + Send + Sync + 'static,
{
Self::Map {
transform: Arc::new(transform),
inner: Box::new(inner),
}
}
pub async fn write_run_event(&self, event: &RunEvent) -> Result<()> {
let mut pending = vec![(self, event.clone())];
while let Some((sink, event)) = pending.pop() {
match sink {
Self::Store(run_store) => {
run_store.append_run_event(&event).await?;
}
Self::JsonLines(writer) => {
let line = redacted_event_json(&event)?;
let mut writer = writer.lock().await;
writer.write_all(line.as_bytes()).await?;
writer.write_all(b"\n").await?;
writer.flush().await?;
}
Self::Callback(callback) => callback(event).await?,
Self::Map { transform, inner } => {
pending.push((inner.as_ref(), transform(event)));
}
Self::Composite(sinks) => {
for sink in sinks.iter().rev() {
pending.push((sink, event.clone()));
}
}
}
}
Ok(())
}
}
#[allow(
clippy::large_enum_variant,
reason = "Logger queue messages stay inline to avoid boxing hot-path payloads."
)]
enum RunEventCommand {
Event(RunEvent),
Flush(oneshot::Sender<()>),
}
#[derive(Clone)]
pub struct RunEventLogger {
tx: mpsc::UnboundedSender<RunEventCommand>,
}
impl RunEventLogger {
#[must_use]
pub fn new(sink: RunEventSink) -> Self {
let (tx, mut rx) = mpsc::unbounded_channel();
tokio::spawn(async move {
while let Some(command) = rx.recv().await {
match command {
RunEventCommand::Event(event) => {
if let Err(err) = sink.write_run_event(&event).await {
tracing::warn!(error = %err, "Failed to write run event");
}
}
RunEventCommand::Flush(tx) => {
let _ = tx.send(());
}
}
}
});
Self { tx }
}
pub fn register(&self, emitter: &Emitter) {
let tx = self.tx.clone();
emitter.on_event(move |event| {
if tx.send(RunEventCommand::Event(event.clone())).is_err() {
tracing::warn!("Run event logger channel closed while forwarding event");
}
});
}
pub async fn flush(&self) {
let (tx, rx) = oneshot::channel();
if self.tx.send(RunEventCommand::Flush(tx)).is_err() {
tracing::warn!("Run event logger channel closed before flush");
return;
}
if rx.await.is_err() {
tracing::warn!("Run event logger flush dropped before completion");
}
}
}
#[derive(Clone)]
pub struct StoreProgressLogger {
inner: RunEventLogger,
}
impl StoreProgressLogger {
#[must_use]
pub fn new(run_store: impl Into<RunStoreHandle>) -> Self {
Self {
inner: RunEventLogger::new(RunEventSink::backend(run_store.into())),
}
}
pub fn register(&self, emitter: &Emitter) {
self.inner.register(emitter);
}
pub async fn flush(&self) {
self.inner.flush().await;
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use ::fabro_types::{AuthMethod, IdpIdentity, Principal, RunNoticeLevel, fixtures};
use tokio::sync::Mutex as AsyncMutex;
use super::*;
use crate::event::{
Emitter, Event, build_redacted_event_payload, event_payload_from_redacted_json,
to_run_event,
};
fn user_principal(login: &str) -> Principal {
Principal::user(
IdpIdentity::new("https://github.com", "12345").unwrap(),
login.to_string(),
AuthMethod::Github,
)
}
#[tokio::test]
async fn append_event_writes_store_event_shape() {
let store = fabro_store::Database::new(
std::sync::Arc::new(object_store::memory::InMemory::new()),
"",
std::time::Duration::from_millis(1),
None,
);
let run_store = store.create_run(&fixtures::RUN_7).await.unwrap();
let stored = to_run_event(&fixtures::RUN_7, &Event::RunNotice {
level: RunNoticeLevel::Warn,
code: "example".to_string(),
message: "notice".to_string(),
});
let payload = build_redacted_event_payload(&stored, &fixtures::RUN_7).unwrap();
run_store.append_event(&payload).await.unwrap();
let events = run_store.list_events().await.unwrap();
let line = events
.into_iter()
.next()
.map(|event| event.event.to_value().unwrap())
.unwrap();
assert!(line.get("id").is_some());
assert_eq!(line["event"], "run.notice");
assert_eq!(line["properties"]["code"], "example");
}
#[tokio::test]
async fn run_event_sink_json_lines_writes_canonical_event_lines() {
use tokio::io::{AsyncBufReadExt, BufReader};
let (writer, reader) = tokio::io::duplex(4096);
let sink = RunEventSink::json_lines(writer);
let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested { actor: None });
sink.write_run_event(&event).await.unwrap();
let mut reader = BufReader::new(reader);
let mut line = String::new();
reader.read_line(&mut line).await.unwrap();
let payload = event_payload_from_redacted_json(line.trim_end(), &fixtures::RUN_7).unwrap();
assert_eq!(payload.as_value()["event"], "run.pause.requested");
assert_eq!(payload.as_value()["properties"]["action"], "pause");
}
#[tokio::test]
async fn run_event_sink_map_applies_transform_before_fanout() {
let first = Arc::new(AsyncMutex::new(Vec::new()));
let second = Arc::new(AsyncMutex::new(Vec::new()));
let first_events = Arc::clone(&first);
let second_events = Arc::clone(&second);
let sink = RunEventSink::map(
|mut event| {
event.actor = Some(user_principal("alice"));
event
},
RunEventSink::fanout(vec![
RunEventSink::callback(move |event| {
let first_events = Arc::clone(&first_events);
async move {
first_events.lock().await.push(event);
Ok(())
}
}),
RunEventSink::callback(move |event| {
let second_events = Arc::clone(&second_events);
async move {
second_events.lock().await.push(event);
Ok(())
}
}),
]),
);
let event = to_run_event(&fixtures::RUN_7, &Event::RunPauseRequested { actor: None });
sink.write_run_event(&event).await.unwrap();
let first = first.lock().await;
let second = second.lock().await;
assert_eq!(first.len(), 1);
assert_eq!(second.len(), 1);
assert_eq!(first[0].actor, Some(user_principal("alice")));
assert_eq!(second[0].actor, Some(user_principal("alice")));
}
#[tokio::test]
async fn run_event_logger_registers_emitter_events_to_json_lines() {
use tokio::io::{AsyncBufReadExt, BufReader};
let (writer, reader) = tokio::io::duplex(4096);
let sink = RunEventSink::json_lines(writer);
let logger = RunEventLogger::new(sink);
let emitter = Emitter::new(fixtures::RUN_8);
logger.register(&emitter);
emitter.emit(&Event::RunPaused);
logger.flush().await;
let mut reader = BufReader::new(reader);
let mut line = String::new();
reader.read_line(&mut line).await.unwrap();
let payload = event_payload_from_redacted_json(line.trim_end(), &fixtures::RUN_8).unwrap();
assert_eq!(payload.as_value()["event"], "run.paused");
}
}

218
lib/crates/fabro-workflow/src/event/stored_fields.rs Normal file
View file

@ -0,0 +1,218 @@
use ::fabro_types::{ParallelBranchId, Principal, StageId, SystemActorKind};
use fabro_agent::AgentEvent;
use super::Event;
use crate::stage_scope::StageScope;
#[derive(Debug, Default)]
pub(super) struct StoredEventFields {
pub(super) session_id: Option<String>,
pub(super) parent_session_id: Option<String>,
pub(super) node_id: Option<String>,
pub(super) node_label: Option<String>,
pub(super) stage_id: Option<StageId>,
pub(super) parallel_group_id: Option<StageId>,
pub(super) parallel_branch_id: Option<ParallelBranchId>,
pub(super) tool_call_id: Option<String>,
pub(super) actor: Option<Principal>,
}
fn default_node_label(node_id: Option<&String>, node_label: Option<String>) -> Option<String> {
node_label.or_else(|| node_id.cloned())
}
fn node_stored_fields(node_id: Option<String>) -> StoredEventFields {
let node_label = default_node_label(node_id.as_ref(), None);
StoredEventFields {
node_id,
node_label,
..StoredEventFields::default()
}
}
pub(super) fn stored_event_fields(event: &Event, scope: Option<&StageScope>) -> StoredEventFields {
let mut fields = stored_event_fields_for_variant(event);
if let Some(scope) = scope {
if fields.node_id.is_none() {
fields.node_id = Some(scope.node_id.clone());
fields.node_label = default_node_label(Some(&scope.node_id), fields.node_label);
}
if fields.stage_id.is_none() {
fields.stage_id = Some(StageId::new(scope.node_id.clone(), scope.visit));
}
if fields.parallel_group_id.is_none() {
fields
.parallel_group_id
.clone_from(&scope.parallel_group_id);
}
if fields.parallel_branch_id.is_none() {
fields
.parallel_branch_id
.clone_from(&scope.parallel_branch_id);
}
}
fields
}
fn stored_event_fields_for_variant(event: &Event) -> StoredEventFields {
match event {
Event::RunCreated { provenance, .. } => StoredEventFields {
actor: provenance.as_ref().and_then(|p| p.subject.clone()),
..StoredEventFields::default()
},
Event::RunCancelRequested { actor }
| Event::RunPauseRequested { actor }
| Event::RunUnpauseRequested { actor }
| Event::RunArchived { actor }
| Event::RunUnarchived { actor, .. }
| Event::InterviewCompleted { actor, .. } => StoredEventFields {
actor: actor.clone(),
..StoredEventFields::default()
},
Event::StageCompleted { node_id, name, .. }
| Event::StageStarted { node_id, name, .. }
| Event::StageRetrying { node_id, name, .. } => {
let node_id_str = node_id.clone();
let node_label = default_node_label(Some(&node_id_str), Some(name.clone()));
StoredEventFields {
node_id: Some(node_id_str),
node_label,
..StoredEventFields::default()
}
}
Event::StageFailed {
node_id,
name,
actor,
..
} => {
let node_id_str = node_id.clone();
let node_label = default_node_label(Some(&node_id_str), Some(name.clone()));
StoredEventFields {
node_id: Some(node_id_str),
node_label,
actor: actor.clone(),
..StoredEventFields::default()
}
}
Event::ParallelStarted { node_id, visit, .. }
| Event::ParallelCompleted { node_id, visit, .. } => {
let node_id_str = node_id.clone();
let node_label = default_node_label(Some(&node_id_str), None);
let parallel_group_id = Some(StageId::new(node_id_str.clone(), *visit));
StoredEventFields {
node_id: Some(node_id_str),
node_label,
parallel_group_id,
..StoredEventFields::default()
}
}
Event::CheckpointCompleted { node_id, .. }
| Event::CheckpointFailed { node_id, .. }
| Event::SubgraphStarted { node_id, .. }
| Event::SubgraphCompleted { node_id, .. }
| Event::ArtifactCaptured { node_id, .. }
| Event::PromptCompleted { node_id, .. }
| Event::CommandStarted { node_id, .. }
| Event::CommandCompleted { node_id, .. }
| Event::AgentCliStarted { node_id, .. }
| Event::AgentCliCompleted { node_id, .. } => node_stored_fields(Some(node_id.clone())),
Event::Agent {
stage,
visit,
event: agent_event,
session_id,
parent_session_id,
} => {
let node_id = Some(stage.clone());
let node_label = default_node_label(node_id.as_ref(), None);
let stage_id = Some(StageId::new(stage.clone(), *visit));
let tool_call_id = agent_tool_call_id(agent_event).map(str::to_string);
let actor = agent_actor_for_event(
agent_event,
session_id.as_deref(),
parent_session_id.as_deref(),
);
StoredEventFields {
session_id: session_id.clone(),
parent_session_id: parent_session_id.clone(),
node_id,
node_label,
stage_id,
tool_call_id,
actor,
..StoredEventFields::default()
}
}
Event::GitCommit { node_id, .. } => node_stored_fields(node_id.clone()),
Event::ParallelBranchStarted {
parallel_group_id,
parallel_branch_id,
branch,
..
}
| Event::ParallelBranchCompleted {
parallel_group_id,
parallel_branch_id,
branch,
..
} => {
let node_id = Some(branch.clone());
let node_label = default_node_label(node_id.as_ref(), None);
StoredEventFields {
node_id,
node_label,
parallel_group_id: Some(parallel_group_id.clone()),
parallel_branch_id: Some(parallel_branch_id.clone()),
..StoredEventFields::default()
}
}
Event::Prompt { stage, .. }
| Event::InterviewStarted { stage, .. }
| Event::Failover { stage, .. } => node_stored_fields(Some(stage.clone())),
Event::InterviewTimeout { actor, stage, .. }
| Event::InterviewInterrupted { actor, stage, .. } => {
let mut fields = node_stored_fields(Some(stage.clone()));
fields.actor.clone_from(actor);
fields
}
Event::StallWatchdogTimeout { node, .. } => {
let mut fields = node_stored_fields(Some(node.clone()));
fields.actor = Some(Principal::System {
system_kind: SystemActorKind::Watchdog,
});
fields
}
_ => StoredEventFields::default(),
}
}
fn agent_tool_call_id(event: &AgentEvent) -> Option<&str> {
match event {
AgentEvent::ToolCallStarted { tool_call_id, .. }
| AgentEvent::ToolCallCompleted { tool_call_id, .. } => Some(tool_call_id.as_str()),
_ => None,
}
}
fn agent_actor_for_event(
event: &AgentEvent,
session_id: Option<&str>,
parent_session_id: Option<&str>,
) -> Option<Principal> {
match event {
AgentEvent::AssistantMessage { model, .. } => Some(Principal::Agent {
session_id: session_id.map(str::to_string),
parent_session_id: parent_session_id.map(str::to_string),
model: Some(model.clone()),
}),
AgentEvent::ToolCallStarted { .. }
| AgentEvent::ToolCallOutputDelta { .. }
| AgentEvent::ToolCallCompleted { .. } => Some(Principal::Agent {
session_id: session_id.map(str::to_string),
parent_session_id: parent_session_id.map(str::to_string),
model: None,
}),
_ => None,
}
}

1
lib/crates/fabro-workflow/src/lib.rs
View file

@ -152,6 +152,7 @@ pub mod runtime_store;
pub mod sandbox_git;
pub(crate) mod sandbox_git_runtime;
pub mod services;
mod stage_scope;
#[doc(hidden)]
pub mod test_support;
#[doc(hidden)]

67
lib/crates/fabro-workflow/src/stage_scope.rs Normal file
View file

@ -0,0 +1,67 @@
use fabro_types::{ParallelBranchId, StageId};
use crate::context::{Context as WfContext, WorkflowContext};
use crate::run_dir::visit_from_context;
/// Stage-level scope threaded through event emission to populate
/// `stage_id` / `parallel_group_id` / `parallel_branch_id` on events
/// that happen inside a concrete stage execution.
#[derive(Clone, Debug)]
pub struct StageScope {
pub node_id: String,
pub visit: u32,
pub parallel_group_id: Option<StageId>,
pub parallel_branch_id: Option<ParallelBranchId>,
}
impl StageScope {
/// 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_node` 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))
}
/// Build scope for the branch-lifecycle events emitted by the parallel
/// handler (`ParallelBranchStarted`, `ParallelBranchCompleted`, and the
/// pre-dispatch `GitCommit` for the branch worktree).
///
/// `target_visit` is the visit count of `target_node_id` for this
/// particular branch dispatch. The parallel handler currently passes
/// `1` because branches haven't been re-entered yet at the point of
/// scope construction; a future change that loops a parallel node
/// must pass the actual visit so envelope `stage_id`s stay accurate.
#[must_use]
pub fn for_parallel_branch(
target_node_id: impl Into<String>,
target_visit: u32,
parallel_group_id: StageId,
parallel_branch_id: ParallelBranchId,
) -> Self {
Self {
node_id: target_node_id.into(),
visit: target_visit,
parallel_group_id: Some(parallel_group_id),
parallel_branch_id: Some(parallel_branch_id),
}
}
#[must_use]
pub fn stage_id(&self) -> StageId {
StageId::new(self.node_id.clone(), self.visit)
}
}