inspect: route RunMulti start failures through terminal markers

b8fa833→main·

dipree·4w ago·2 files·+61 added/-17 removed

Queue Start-failure terminal markers and let the dispatch loop write startErr and finishedAt, matching the terminal-marker protocol used for Wait results. The setup loop still initializes immutable state only; all terminal state now flows through fanIn. Start-error markers are sent from the closer goroutine so setup cannot block if an early agent fills fanIn before dispatch starts.

Adds coverage for the all-start-errors case to prove RunMulti still finishes and reports each start failure.

Sessions

17d97f13d931View transcript

[?
Checkout the hand off doc that I just added.Pi·Opus 4.8·1 step](/content/gh/entireio/cli/session/019eca64-8c2c-7b00-90c6-3aa49738c497#timeline-17d97f13d931/index.html)

Changes

2

• cmd/entire/cli/review

• Mrun_multi.go+28/-17

• Mrun_multi_test.go+33

34 unmodified lines

37 unmodified lines

// perAgentState tracks the accumulation for one agent during a multi-agent run. // // Concurrency: perAgentState has a single writer. The immutable fields // (name/agentName/model/startedAt) and startErr/finishedAt for agents whose // Start failed are set in the setup loop; every other field is written by the // dispatch loop as events — and a final terminal marker carrying // waitErr/finishedAt/timedOut — arrive over fanIn. Both run on the RunMulti // goroutine. The per-agent forwarding goroutines NEVER touch perAgentState; // they only send on fanIn. So there is no cross-goroutine field sharing, and // the post-loop accounting reads are safe by construction (the dispatch loop // has already returned). // Concurrency: perAgentState has a single writer after initialization. The // immutable fields (name/agentName/model/startedAt) are set before launch; all // terminal state (startErr/waitErr/finishedAt/timedOut) and event-derived state // are written by the dispatch loop as events and terminal markers arrive over // fanIn. The per-agent forwarding goroutines NEVER touch perAgentState; they // only send on fanIn. So there is no cross-goroutine field sharing, and the // post-loop accounting reads are safe by construction (the dispatch loop has // already returned).

type perAgentState struct { name string agentName string

terminal *agentTerminal }

// agentTerminal carries an agent's end-of-run results from its forwarding // goroutine to the dispatch loop, which writes them into perAgentState. // agentTerminal carries an agent's end-of-run results to the dispatch loop, // which writes them into perAgentState. Forwarding goroutines send this after // Wait; the setup loop queues the same marker shape for Start failures. type agentTerminal struct { startErr error waitErr error finishedAt time.Time timedOut bool }

// siblings and the judge proceed. timeout := inspectorTimeout(cfg) var wg sync.WaitGroup startTerminals := make([]taggedEvent, 0) for i, r := range reviewers { agentCtx := ctx var cancelAgent context.CancelFunc = func() {}

if err != nil { // No Process exists, so there is no Events/Wait lifecycle to preserve. // Cancel immediately to release the per-agent timeout timer while // siblings continue running. // siblings continue running. Queue the terminal marker so the dispatch // loop remains the single writer of perAgentState terminal fields. cancelAgent() states[i].startErr = err states[i].finishedAt = time.Now() startTerminals = append(startTerminals, taggedEvent{agentIdx: i, terminal: &agentTerminal{ startErr: err, finishedAt: time.Now(), }}) continue } states[i].proc = proc }

// Close fanIn after all forwarding goroutines finish. This goroutine // must be launched AFTER all wg.Add calls above so the WaitGroup // counter is correct before Wait is called. // Close fanIn after queued Start-failure markers are delivered and all // forwarding goroutines finish. This goroutine must be launched AFTER all // wg.Add calls above so the WaitGroup counter is correct before Wait is // called. Sending startTerminals here (instead of from the setup loop) avoids // blocking setup if an early-started agent fills fanIn before dispatch begins. go func() { for _, tagged := range startTerminals { fanIn <- tagged } wg.Wait() close(fanIn) }()

if tagged.terminal != nil { // End-of-run marker (internal): record terminal fields, don't forward // to sinks. st.startErr = tagged.terminal.startErr st.waitErr = tagged.terminal.waitErr st.finishedAt = tagged.terminal.finishedAt st.timedOut = tagged.terminal.timedOut }

func TestRunMulti_AllStartErrorsStillFinish(t *testing.T) {
    t.Parallel()
    firstErr := errors.New("first start failed")
    secondErr := errors.New("second start failed")
    rec := &stubSinkRecorder{}

summary, err := RunMulti(context.Background(), []reviewtypes.AgentReviewer{
        &stubReviewer{name: "first", startErr: firstErr},
        &stubReviewer{name: "second", startErr: secondErr},
    }, reviewtypes.RunConfig{}, []reviewtypes.Sink{rec})

if !errors.Is(err, firstErr) {
        t.Fatalf("RunMulti error = %v, want first start error", err)
    }
    if len(summary.AgentRuns) != 2 {
        t.Fatalf("AgentRuns = %d, want 2", len(summary.AgentRuns))
    }
    for _, run := range summary.AgentRuns {
        if run.Status != reviewtypes.AgentStatusFailed {
            t.Fatalf("%s status = %v, want Failed", run.Name, run.Status)
        }
        if run.Duration < 0 {
            t.Fatalf("%s duration = %v, want non-negative", run.Name, run.Duration)
        }
    }
    if !errors.Is(summary.AgentRuns[0].Err, firstErr) || !errors.Is(summary.AgentRuns[1].Err, secondErr) {
        t.Fatalf("AgentRun errors = %v / %v, want start errors", summary.AgentRuns[0].Err, summary.AgentRuns[1].Err)
    }
    if len(rec.finishedCalls) != 1 {
        t.Fatalf("RunFinished calls = %d, want 1", len(rec.finishedCalls))
    }
}

// TestRunMulti_ContextCancellation verifies that context cancellation causes
// summary.Cancelled=true and all AgentRuns to have status Cancelled.
func TestRunMulti_ContextCancellation(t *testing.T) {