From 797c3747c9084dabf323dbd6420a7f739953ac1d Mon Sep 17 00:00:00 2001 From: wan9chi Date: Fri, 6 Nov 2026 15:42:35 +0800 Subject: [PATCH] refactor(reporter): pass reporters each leaf's position A JSON reporter for `vp run --dry-run --json` prints the run's structure: tasks, their dependencies, steps and nested plans, and places each step's cache status in it. Today a reporter learns about leaves one at a time, by their display text, which can't tell apart the same task in two nested plans. - `GraphExecutionReporter::new_leaf_execution` takes the leaf's `LeafPath`: its `(node, item index)` at each nesting level, outermost first. - The scheduler builds the path as it recurses into nodes, items and nested graphs. The existing reporters ignore it, and the summary reporter passes it to the reporter it wraps. The JSON reporter gets the graph itself when it's created. Co-Authored-By: Claude Opus 5.5 --- crates/vt/src/session/execute/scheduler.rs | 35 +++++++++++++------ crates/vt/src/session/reporter/grouped/mod.rs | 6 ++-- .../src/session/reporter/interleaved/mod.rs | 14 ++++---- crates/vt/src/session/reporter/labeled/mod.rs | 6 ++-- crates/vt/src/session/reporter/mod.rs | 17 ++++++--- .../src/session/reporter/summary_reporter.rs | 9 ++--- 6 files changed, 58 insertions(+), 29 deletions(-) diff --git a/crates/vt/src/session/execute/scheduler.rs b/crates/vt/src/session/execute/scheduler.rs index 4b4dfe680..f58a36c4b 100644 --- a/crates/vt/src/session/execute/scheduler.rs +++ b/crates/vt/src/session/execute/scheduler.rs @@ -24,7 +24,7 @@ use crate::{ event::{ CacheDisabledReason, CacheNotUpdatedReason, CacheStatus, CacheUpdateStatus, RunKind, }, - reporter::{ExitStatus, GraphExecutionReporter, GraphExecutionReporterBuilder}, + reporter::{ExitStatus, GraphExecutionReporter, GraphExecutionReporterBuilder, LeafPath}, }, }; @@ -93,12 +93,15 @@ impl ExecutionContext<'_> { /// (killing in-flight child processes). Ctrl-C cancels the `cancel_token`. /// Either cancellation causes this method to close the semaphore, drain /// remaining futures, and return. + /// + /// `path` is where `graph` is: empty for the root graph, and the path of + /// the item it expands for a nested one. #[tracing::instrument( level = "debug", skip_all, fields(tasks = graph.graph.node_count(), concurrency_limit = graph.concurrency_limit) )] - async fn execute_expanded_graph(&self, graph: &ExecutionGraph) { + async fn execute_expanded_graph(&self, graph: &ExecutionGraph, path: &LeafPath) { if graph.graph.node_count() == 0 { return; } @@ -118,7 +121,7 @@ impl ExecutionContext<'_> { // Schedule initially ready nodes (no dependencies). for (&node_ix, &count) in &dep_count { if count == 0 { - futures.push(self.spawn_node(graph, node_ix, &semaphore)); + futures.push(self.spawn_node(graph, path, node_ix, &semaphore)); } } @@ -139,7 +142,7 @@ impl ExecutionContext<'_> { let count = dep_count.get_mut(&dependent).expect("all nodes are in dep_count"); *count -= 1; if *count == 0 { - futures.push(self.spawn_node(graph, dependent, &semaphore)); + futures.push(self.spawn_node(graph, path, dependent, &semaphore)); } } } @@ -153,6 +156,7 @@ impl ExecutionContext<'_> { fn spawn_node<'a>( &'a self, graph: &'a ExecutionGraph, + path: &'a LeafPath, node_ix: ExecutionNodeIndex, semaphore: &Arc, ) -> LocalBoxFuture<'a, ExecutionNodeIndex> { @@ -167,7 +171,7 @@ impl ExecutionContext<'_> { waited = ?start.elapsed(), "acquired a concurrency permit" ); - self.execute_node(graph, node_ix).await; + self.execute_node(graph, path, node_ix).await; } node_ix } @@ -179,19 +183,27 @@ impl ExecutionContext<'_> { /// A node may have multiple items (from `&&`-split commands). Items are executed /// in order; if any item fails, `execute_leaf` cancels the `fast_fail_token` /// and remaining items are skipped (preserving `&&` semantics). - async fn execute_node(&self, graph: &ExecutionGraph, node_ix: ExecutionNodeIndex) { + async fn execute_node( + &self, + graph: &ExecutionGraph, + path: &LeafPath, + node_ix: ExecutionNodeIndex, + ) { let task_execution = &graph.graph[node_ix]; - for item in &task_execution.items { + for (item_index, item) in task_execution.items.iter().enumerate() { if self.cancelled() { return; } + let item_path: Vec<_> = path.iter().copied().chain([(node_ix, item_index)]).collect(); match &item.kind { ExecutionItemKind::Leaf(leaf_kind) => { - self.execute_leaf(&item.execution_item_display, leaf_kind).boxed_local().await; + self.execute_leaf(&item_path, &item.execution_item_display, leaf_kind) + .boxed_local() + .await; } ExecutionItemKind::Expanded(nested_graph) => { - self.execute_expanded_graph(nested_graph).boxed_local().await; + self.execute_expanded_graph(nested_graph, &item_path).boxed_local().await; } } } @@ -210,6 +222,7 @@ impl ExecutionContext<'_> { )] async fn execute_leaf( &self, + path: &LeafPath, item_display: &ExecutionItemDisplay, leaf_kind: &LeafExecutionKind, ) { @@ -217,7 +230,7 @@ impl ExecutionContext<'_> { // Borrow the reporter briefly to create the leaf reporter, then drop // the RefCell guard before any `.await` point. let mut leaf_reporter = - self.reporter.borrow_mut().new_leaf_execution(item_display, leaf_kind); + self.reporter.borrow_mut().new_leaf_execution(path, item_display, leaf_kind); let failed = match leaf_kind { LeafExecutionKind::InProcess(in_process_execution) => { @@ -331,7 +344,7 @@ impl Session<'_> { // Execute the graph with fast-fail: if any task fails, remaining tasks // are skipped. Leaf-level errors are reported through the reporter. - execution_context.execute_expanded_graph(&execution_graph).await; + execution_context.execute_expanded_graph(&execution_graph, &[]).await; // Nested graphs share the cache, so this waits for their uploads too. // After Ctrl-C, the uploads are cancelled without a message. diff --git a/crates/vt/src/session/reporter/grouped/mod.rs b/crates/vt/src/session/reporter/grouped/mod.rs index 276ab6cdd..737499a97 100644 --- a/crates/vt/src/session/reporter/grouped/mod.rs +++ b/crates/vt/src/session/reporter/grouped/mod.rs @@ -8,7 +8,7 @@ use vt_plan::{ExecutionItemDisplay, LeafExecutionKind}; use super::{ ColorSupport, ColorizeExt, ExitStatus, GraphExecutionReporter, GraphExecutionReporterBuilder, - LeafExecutionReporter, PipeWriters, StdioConfig, StdioSuggestion, + LeafExecutionReporter, LeafPath, PipeWriters, StdioConfig, StdioSuggestion, format_command_with_cache_status, format_task_label, maybe_strip_writer, write_leaf_trailing_output, }; @@ -58,6 +58,7 @@ struct GroupedGraphReporter { impl GraphExecutionReporter for GroupedGraphReporter { fn new_leaf_execution( &mut self, + _path: &LeafPath, display: &ExecutionItemDisplay, _leaf_kind: &LeafExecutionKind, ) -> Box { @@ -183,7 +184,8 @@ mod tests { ColorSupport::uniform(false), )); let mut reporter = builder.build(); - let mut leaf = reporter.new_leaf_execution(&item.execution_item_display, leaf_kind(item)); + let mut leaf = + reporter.new_leaf_execution(&[], &item.execution_item_display, leaf_kind(item)); let stdio_config = leaf.start(CacheStatus::Disabled(CacheDisabledReason::NoCacheMetadata)); assert_eq!(stdio_config.suggestion, StdioSuggestion::Piped); } diff --git a/crates/vt/src/session/reporter/interleaved/mod.rs b/crates/vt/src/session/reporter/interleaved/mod.rs index 14f2213c7..8358574b7 100644 --- a/crates/vt/src/session/reporter/interleaved/mod.rs +++ b/crates/vt/src/session/reporter/interleaved/mod.rs @@ -13,9 +13,9 @@ use vt_str::Str; use super::{ COMMAND_STYLE, ColorSupport, ColorizeExt as _, ExitStatus, GraphExecutionReporter, - GraphExecutionReporterBuilder, LeafExecutionReporter, PipeWriters, RetryingWriter, StdioConfig, - StdioSuggestion, format_command_display, format_command_with_cache_status, maybe_strip_writer, - write_leaf_trailing_output, + GraphExecutionReporterBuilder, LeafExecutionReporter, LeafPath, PipeWriters, RetryingWriter, + StdioConfig, StdioSuggestion, format_command_display, format_command_with_cache_status, + maybe_strip_writer, write_leaf_trailing_output, }; use crate::session::event::{CacheStatus, CacheUpdateStatus, ExecutionError, RunKind}; @@ -63,6 +63,7 @@ struct InterleavedGraphReporter { impl GraphExecutionReporter for InterleavedGraphReporter { fn new_leaf_execution( &mut self, + _path: &LeafPath, display: &ExecutionItemDisplay, leaf_kind: &LeafExecutionKind, ) -> Box { @@ -182,7 +183,7 @@ mod tests { RunKind::Real, )); let mut reporter = builder.build(); - let mut leaf = reporter.new_leaf_execution(display, leaf_kind); + let mut leaf = reporter.new_leaf_execution(&[], display, leaf_kind); leaf.start(CacheStatus::Disabled(CacheDisabledReason::NoCacheMetadata)).suggestion } @@ -220,11 +221,12 @@ mod tests { .build(); let not_updated = || CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::DryRun); - let mut leaf = reporter.new_leaf_execution(&item.execution_item_display, leaf_kind(item)); + let mut leaf = + reporter.new_leaf_execution(&[], &item.execution_item_display, leaf_kind(item)); leaf.start(CacheStatus::Miss(CacheMiss::NotFound)); leaf.finish(None, not_updated(), None); - let leaf = reporter.new_leaf_execution(&item.execution_item_display, leaf_kind(item)); + let leaf = reporter.new_leaf_execution(&[], &item.execution_item_display, leaf_kind(item)); let error = ExecutionError::Cache { kind: CacheErrorKind::Lookup, source: anyhow::anyhow!("database is locked"), diff --git a/crates/vt/src/session/reporter/labeled/mod.rs b/crates/vt/src/session/reporter/labeled/mod.rs index 1cf12c8b1..22f1c75a7 100644 --- a/crates/vt/src/session/reporter/labeled/mod.rs +++ b/crates/vt/src/session/reporter/labeled/mod.rs @@ -7,7 +7,7 @@ use vt_plan::{ExecutionItemDisplay, LeafExecutionKind}; use super::{ ColorSupport, ExitStatus, GraphExecutionReporter, GraphExecutionReporterBuilder, - LeafExecutionReporter, PipeWriters, RetryingWriter, StdioConfig, StdioSuggestion, + LeafExecutionReporter, LeafPath, PipeWriters, RetryingWriter, StdioConfig, StdioSuggestion, format_command_with_cache_status, format_task_label, maybe_strip_writer, write_leaf_trailing_output, }; @@ -55,6 +55,7 @@ struct LabeledGraphReporter { impl GraphExecutionReporter for LabeledGraphReporter { fn new_leaf_execution( &mut self, + _path: &LeafPath, display: &ExecutionItemDisplay, _leaf_kind: &LeafExecutionKind, ) -> Box { @@ -163,7 +164,8 @@ mod tests { ColorSupport::uniform(false), )); let mut reporter = builder.build(); - let mut leaf = reporter.new_leaf_execution(&item.execution_item_display, leaf_kind(item)); + let mut leaf = + reporter.new_leaf_execution(&[], &item.execution_item_display, leaf_kind(item)); let stdio_config = leaf.start(CacheStatus::Disabled(CacheDisabledReason::NoCacheMetadata)); assert_eq!(stdio_config.suggestion, StdioSuggestion::Piped); } diff --git a/crates/vt/src/session/reporter/mod.rs b/crates/vt/src/session/reporter/mod.rs index 1d49db265..6cab6b7ef 100644 --- a/crates/vt/src/session/reporter/mod.rs +++ b/crates/vt/src/session/reporter/mod.rs @@ -3,10 +3,12 @@ //! This module provides a typestate-based reporter system with three phases: //! //! 1. [`GraphExecutionReporterBuilder`] — created before execution begins. -//! Transitions to [`GraphExecutionReporter`] when `build()` is called. +//! Transitions to [`GraphExecutionReporter`] when `build()` is called with the graph to +//! execute. //! //! 2. [`GraphExecutionReporter`] — creates [`LeafExecutionReporter`] -//! instances for individual leaf executions via `new_leaf_execution()`. Finalized with `finish()`. +//! instances for individual leaf executions via `new_leaf_execution()`, each with its +//! [`LeafPath`] in the graph. Finalized with `finish()`. //! //! 3. [`LeafExecutionReporter`] — handles events for a single leaf execution (output streaming, //! cache status, errors). Finalized with `finish()`. @@ -42,7 +44,7 @@ pub use plain::PlainReporter; use retrying_writer::RetryingWriter; pub use summary_reporter::SummaryReporterBuilder; use vt_path::AbsolutePath; -use vt_plan::{ExecutionItemDisplay, LeafExecutionKind}; +use vt_plan::{ExecutionItemDisplay, LeafExecutionKind, execution_graph::ExecutionNodeIndex}; use vt_str::Str; use super::{ @@ -144,6 +146,11 @@ pub(super) fn maybe_strip_writer(writer: Box, color_support: bool) -> // Typestate traits // ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ +/// Where a leaf is in the execution graph: its node and item index at each +/// nesting level, outermost first. A nested `vp run` adds a level, so the same +/// task in two nested plans has two paths. +pub type LeafPath = [(ExecutionNodeIndex, usize)]; + /// Builder for creating a [`GraphExecutionReporter`]. /// /// This is the initial state before the execution graph is known. The `build` method @@ -158,9 +165,11 @@ pub trait GraphExecutionReporterBuilder { /// Creates [`LeafExecutionReporter`] instances for individual leaf executions /// and finalizes the session with `finish()`. pub trait GraphExecutionReporter { - /// Create a new leaf execution reporter for the given leaf. + /// Create a new leaf execution reporter for the leaf at `path` in the + /// graph being executed. fn new_leaf_execution( &mut self, + path: &LeafPath, display: &ExecutionItemDisplay, leaf_kind: &LeafExecutionKind, ) -> Box; diff --git a/crates/vt/src/session/reporter/summary_reporter.rs b/crates/vt/src/session/reporter/summary_reporter.rs index 96a77e0d0..ea7e44fb6 100644 --- a/crates/vt/src/session/reporter/summary_reporter.rs +++ b/crates/vt/src/session/reporter/summary_reporter.rs @@ -19,7 +19,7 @@ use vt_str::Str; use super::{ ExitStatus, GraphExecutionReporter, GraphExecutionReporterBuilder, LeafExecutionReporter, - StdioConfig, + LeafPath, StdioConfig, }; use crate::session::{ cache::remote::UploadError, @@ -108,10 +108,11 @@ struct SummaryGraphReporter { impl GraphExecutionReporter for SummaryGraphReporter { fn new_leaf_execution( &mut self, + path: &LeafPath, display: &ExecutionItemDisplay, leaf_kind: &LeafExecutionKind, ) -> Box { - let inner = self.inner.new_leaf_execution(display, leaf_kind); + let inner = self.inner.new_leaf_execution(path, display, leaf_kind); Box::new(SummaryLeafReporter { inner, tasks: Rc::clone(&self.tasks), @@ -308,7 +309,7 @@ mod tests { .build(); let upload_error = Arc::new(OnceLock::new()); - let mut leaf = reporter.new_leaf_execution(&item.execution_item_display, leaf_kind); + let mut leaf = reporter.new_leaf_execution(&[], &item.execution_item_display, leaf_kind); leaf.start(CacheStatus::Miss(CacheMiss::NotFound)); leaf.finish( Some(StdExitStatus::default()), @@ -353,7 +354,7 @@ mod tests { run_kind, )) .build(); - let leaf = reporter.new_leaf_execution(&item.execution_item_display, leaf_kind); + let leaf = reporter.new_leaf_execution(&[], &item.execution_item_display, leaf_kind); leaf.finish( None, CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::CacheDisabled),