agent: add bounded subagent progress forwarding

Forward each sub-agent run's Reasoning/Text/Notice/Retrying as reserved
ToolProgress channel events (reasonix.subagent.status/reasoning/text/notice)
so local frontends can show live per-child progress without the child's
bodies entering the parent transcript or provider context. A shared merger
per parent task group paces previews: one pending slot per (child, channel),
a 250ms merge window, a 32 event/sec group budget round-robined across
children, and an 8 KiB per-child pending buffer with UTF-8-safe tails.
RunProfileSpec emits exactly one terminal status (completed/cancelled/failed)
after synchronously flushing pending previews; late events are ignored.
Background runs emit queued at registration and running on slot acquisition;
parallel_tasks and fleet share one merger per call via context. Real
sub-agent tool progress is namespaced and forwarded like ToolDispatch.

internal/agent/fleet.go

@@ -194,15 +194,16 @@ func (f *FleetTool) Execute(ctx context.Context, args json.RawMessage) (string,
 			jobCtx = WithParentSession(jobCtx, parentSession)
 			jobCtx = evidence.WithLedger(jobCtx, backgroundEvidence)
 			defer func() { jobs.PublishEvidence(jobCtx, backgroundEvidence.Summary()) }()
-			return f.runFleet(jobCtx, nested, specs)
+			return f.runFleet(jobCtx, nested, specs, parentID)
 		})
 		return fmt.Sprintf("Started background fleet %q (%s). Collect results with wait; you will be notified when it finishes.", job.ID, label), nil
 	}
 
-	return f.runFleet(ctx, subSink(ctx), specs)
+	fleetParentID, _, _, _ := CallContext(ctx)
+	return f.runFleet(ctx, subSink(ctx), specs, fleetParentID)
 }
 
-func (f *FleetTool) runFleet(ctx context.Context, sink event.Sink, specs []ProfileExecSpec) (string, error) {
+func (f *FleetTool) runFleet(ctx context.Context, sink event.Sink, specs []ProfileExecSpec, groupParentID string) (string, error) {
 	if sink == nil {
 		sink = event.Discard
 	}
@@ -210,6 +211,16 @@ func (f *FleetTool) runFleet(ctx context.Context, sink event.Sink, specs []Profi
 	if !ok {
 		parentID = "fleet"
 	}
+	if groupParentID == "" {
+		groupParentID = parentID
+	}
+	// One shared progress merger per fleet run; children find it via their
+	// context and it is closed after every child has finished. The merger
+	// emits through the same sink the child dispatch cards flow through, so
+	// preview IDs always match the cards.
+	merger := newSubagentProgressMerger(realProgressClock{}, sink, groupParentID)
+	defer merger.Close()
+	ctx = withSubagentProgressMerger(ctx, merger)
 
 	n := len(specs)
 	results := make([]fleetItemResult, n)

internal/agent/parallel_tasks.go

@@ -119,6 +119,15 @@ func (p *ParallelTasksTool) Execute(ctx context.Context, args json.RawMessage) (
 		sink = event.Discard
 	}
 
+	// One shared progress merger paces previews for the whole group: children
+	// find it via their context and the merger is closed after every child has
+	// finished (each child flushes its own slots before publishing its result).
+	// The merger emits through the same sink the child dispatch cards flow
+	// through, so preview IDs always match the cards.
+	merger := newSubagentProgressMerger(realProgressClock{}, sink, parentID)
+	defer merger.Close()
+	ctx = withSubagentProgressMerger(ctx, merger)
+
 	type subResult struct {
 		index  int
 		output string

internal/agent/subagent_progress.go

@@ -0,0 +1,731 @@
+package agent
+
+import (
+	"context"
+	"sync"
+	"time"
+	"unicode/utf8"
+
+	"reasonix/internal/event"
+)
+
+// Sub-agent progress previews. A tracker per child run converts the child's
+// Reasoning/Text/Notice/Retrying events into reserved ToolProgress channel
+// events (event.SubagentProgress*Name) that local frontends render as progress
+// cards. A shared merger per parent task group paces and bounds the previews:
+// one pending slot per (child, channel), a 250ms merge window, and a group
+// budget of 32 non-terminal preview events/sec round-robined across children
+// so one hot sub-agent cannot starve the rest. The child's Message, and the
+// child's own reasoning/text bodies, never leave the progress pipeline.
+
+// subagentProgressPhase is one of the fixed states the status channel carries.
+type subagentProgressPhase string
+
+const (
+	subagentPhaseQueued     subagentProgressPhase = "queued"
+	subagentPhaseRunning    subagentProgressPhase = "running"
+	subagentPhaseReasoning  subagentProgressPhase = "reasoning"
+	subagentPhaseResponding subagentProgressPhase = "responding"
+	subagentPhaseTool       subagentProgressPhase = "tool"
+	subagentPhaseRetrying   subagentProgressPhase = "retrying"
+	subagentPhaseCompleted  subagentProgressPhase = "completed"
+	subagentPhaseFailed     subagentProgressPhase = "failed"
+	subagentPhaseCancelled  subagentProgressPhase = "cancelled"
+)
+
+func (p subagentProgressPhase) terminal() bool {
+	switch p {
+	case subagentPhaseCompleted, subagentPhaseFailed, subagentPhaseCancelled:
+		return true
+	}
+	return false
+}
+
+// Progress pacing and memory bounds. Preview slots merge for up to
+// subagentProgressMergeWindow before one event per (child, channel) is emitted;
+// a parent task group caps non-terminal preview events at
+// subagentProgressGroupEventsPerSec, round-robined across children. Terminal
+// events and the pre-terminal synchronous flush bypass both limits — the flush
+// is inherently bounded by the per-child pending budget below.
+const (
+	subagentProgressMergeWindow       = 250 * time.Millisecond
+	subagentProgressGroupEventsPerSec = 32
+	subagentProgressGroupBurst        = subagentProgressGroupEventsPerSec
+
+	// Per-child pending-send budget: reasoning/text/notice slots share 8 KiB,
+	// with a per-channel cap so one channel cannot crowd out the response
+	// preview. When the shared budget overflows, the notice slot is dropped
+	// first, then reasoning, then text — each keeping a UTF-8-safe tail.
+	subagentProgressMaxPendingBytes = 8 << 10
+	subagentProgressReasoningCap    = 8 << 10
+	subagentProgressTextCap         = 8 << 10
+	subagentProgressNoticeCap       = 2 << 10
+)
+
+// progressClock isolates time so tests drive merge windows with a fake clock.
+type progressClock interface {
+	Now() time.Time
+	NewTimer(d time.Duration) progressTimer
+}
+
+// progressTimer mirrors the *time.Timer surface the merger needs.
+type progressTimer interface {
+	C() <-chan time.Time
+	Reset(d time.Duration) bool
+	Stop() bool
+}
+
+type realProgressClock struct{}
+
+func (realProgressClock) Now() time.Time { return time.Now() }
+
+func (realProgressClock) NewTimer(d time.Duration) progressTimer {
+	return realProgressTimer{t: time.NewTimer(d)}
+}
+
+type realProgressTimer struct{ t *time.Timer }
+
+func (r realProgressTimer) C() <-chan time.Time        { return r.t.C }
+func (r realProgressTimer) Reset(d time.Duration) bool { return r.t.Reset(d) }
+func (r realProgressTimer) Stop() bool                 { return r.t.Stop() }
+
+// subagentProgressChannel identifies one preview channel.
+type subagentProgressChannel int
+
+const (
+	subagentProgressChanReasoning subagentProgressChannel = iota
+	subagentProgressChanText
+	subagentProgressChanNotice
+)
+
+func (c subagentProgressChannel) name() string {
+	switch c {
+	case subagentProgressChanReasoning:
+		return event.SubagentProgressReasoningName
+	case subagentProgressChanText:
+		return event.SubagentProgressTextName
+	default:
+		return event.SubagentProgressNoticeName
+	}
+}
+
+func (c subagentProgressChannel) cap() int {
+	switch c {
+	case subagentProgressChanReasoning:
+		return subagentProgressReasoningCap
+	case subagentProgressChanText:
+		return subagentProgressTextCap
+	default:
+		return subagentProgressNoticeCap
+	}
+}
+
+// progressSlot is the single pending slot for one (child, channel): at most one
+// unsent merged slice per child+channel, so pending preview memory is bounded
+// by construction. dueAt is the earliest time the merged slice may be sent.
+type progressSlot struct {
+	buf       string
+	truncated bool
+	dirty     bool
+	dueAt     time.Time
+	lastSend  time.Time
+}
+
+// progressStatusSlot holds the latest unsent phase for one child. Status
+// events bypass the group preview budget (the phase machine has ≤9 states per
+// child) but still merge within the 250ms window after their first send.
+type progressStatusSlot struct {
+	phase    subagentProgressPhase
+	dirty    bool
+	dueAt    time.Time
+	lastSend time.Time
+}
+
+// subagentProgressMerger paces and bounds progress previews for one parent
+// task group (a single task, a parallel_tasks call, or a fleet). It owns one
+// flusher goroutine that emits due slots round-robin; every owner must Close it
+// after all children finish so no timer or goroutine outlives the group.
+type subagentProgressMerger struct {
+	mu            sync.Mutex
+	clock         progressClock
+	sink          event.Sink // the same sink the group's dispatch events flow through
+	groupParentID string     // the group's own call ID (progress events' ParentID)
+
+	slots  map[string]map[subagentProgressChannel]*progressSlot
+	status map[string]*progressStatusSlot
+	order  []string // child IDs in registration order, for round-robin
+	rr     int      // rotating scan start for fairness
+
+	tokens     float64 // preview budget: subagentProgressGroupEventsPerSec
+	lastRefill time.Time
+
+	timer  progressTimer
+	wake   chan struct{}
+	done   chan struct{}
+	wg     sync.WaitGroup
+	closed bool
+}
+
+func newSubagentProgressMerger(clock progressClock, sink event.Sink, groupParentID string) *subagentProgressMerger {
+	now := clock.Now()
+	m := &subagentProgressMerger{
+		clock:         clock,
+		sink:          sink,
+		groupParentID: groupParentID,
+		slots:         make(map[string]map[subagentProgressChannel]*progressSlot),
+		status:        make(map[string]*progressStatusSlot),
+		tokens:        subagentProgressGroupBurst,
+		lastRefill:    now,
+		wake:          make(chan struct{}, 1),
+		done:          make(chan struct{}),
+		timer:         clock.NewTimer(0),
+	}
+	m.wg.Add(1)
+	go m.run()
+	return m
+}
+
+// Close stops the flusher goroutine and drops any pending state. The owner
+// calls it only after every child has finished (each child's finish flushed
+// its own slots), so Close never discards a needed preview.
+func (m *subagentProgressMerger) Close() {
+	m.mu.Lock()
+	if m.closed {
+		m.mu.Unlock()
+		return
+	}
+	m.closed = true
+	m.mu.Unlock()
+	close(m.done)
+	m.wg.Wait()
+}
+
+// directStatus sends a status event immediately (bypassing the merge slot and
+// group budget) and records the send on the child's status slot so the next
+// transition still merges for the 250ms window after this send. Used for the
+// guaranteed-first states (queued/running); terminal events go through
+// flushChild instead.
+func (m *subagentProgressMerger) directStatus(childID string, phase subagentProgressPhase) {
+	m.mu.Lock()
+	st := m.status[childID]
+	if st == nil {
+		st = &progressStatusSlot{}
+		m.status[childID] = st
+		m.ensureOrderLocked(childID)
+	}
+	st.lastSend = m.clock.Now()
+	st.dirty = false
+	st.phase = phase
+	m.mu.Unlock()
+	parentID := m.groupParentID
+	if parentID == childID {
+		parentID = ""
+	}
+	m.sink.Emit(event.Event{
+		Kind: event.ToolProgress,
+		Tool: event.Tool{
+			ID: childID, Name: event.SubagentProgressStatusName,
+			ParentID: parentID, Output: string(phase),
+		},
+	})
+}
+
+// statusEvent queues a phase transition for a child. The first transition per
+// child sends immediately; later transitions merge into the status slot.
+func (m *subagentProgressMerger) statusEvent(childID string, phase subagentProgressPhase) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.closed {
+		return
+	}
+	st := m.status[childID]
+	if st == nil {
+		st = &progressStatusSlot{}
+		m.status[childID] = st
+		m.ensureOrderLocked(childID)
+	}
+	if !st.dirty {
+		st.dirty = true
+		// The first status send is immediate; later transitions merge for the
+		// 250ms window after the previous send.
+		dueAt := m.clock.Now()
+		if !st.lastSend.IsZero() {
+			if after := st.lastSend.Add(subagentProgressMergeWindow); after.After(dueAt) {
+				dueAt = after
+			}
+		}
+		st.dueAt = dueAt
+	}
+	st.phase = phase
+	m.wakeLocked()
+}
+
+// deltaEvent appends a text delta to a child's preview slot. The slot is the
+// only pending slice for that (child, channel); overflow keeps a UTF-8-safe
+// tail and marks the round truncated.
+func (m *subagentProgressMerger) deltaEvent(childID string, ch subagentProgressChannel, delta string) {
+	if delta == "" {
+		return
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.closed {
+		return
+	}
+	if _, ok := m.slots[childID]; !ok {
+		m.slots[childID] = make(map[subagentProgressChannel]*progressSlot)
+		m.ensureOrderLocked(childID)
+	}
+	sl := m.slots[childID][ch]
+	if sl == nil {
+		sl = &progressSlot{}
+		m.slots[childID][ch] = sl
+	}
+	if !sl.dirty {
+		sl.dirty = true
+		sl.dueAt = m.clock.Now().Add(subagentProgressMergeWindow)
+	}
+	sl.buf += delta
+	if len(sl.buf) > ch.cap() {
+		sl.buf = utf8SafeTail(sl.buf, ch.cap())
+		sl.truncated = true
+	}
+	m.trimToBudgetLocked(childID)
+	m.wakeLocked()
+}
+
+// flushChild synchronously emits everything pending for the child and then the
+// terminal status event. Terminal events bypass merge windows and the group
+// budget; the flush is bounded by the per-child pending budget. Called by the
+// tracker's finish before any terminal is delivered, and only once per child.
+func (m *subagentProgressMerger) flushChild(childID string, terminal subagentProgressPhase, durationMs int64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.closed {
+		return
+	}
+	st := m.status[childID]
+	if st != nil && st.dirty {
+		phase := st.phase
+		st.dirty = false
+		m.emitStatusLocked(childID, phase, 0)
+	}
+	for c := subagentProgressChanReasoning; c <= subagentProgressChanNotice; c++ {
+		if sl := m.slots[childID][c]; sl != nil && sl.dirty {
+			m.emitDeltaLocked(childID, c, sl)
+		}
+	}
+	m.emitStatusLocked(childID, terminal, durationMs)
+	// Release per-child state; later events for this child are ignored by the
+	// tracker's own done flag, and the flusher has nothing left to wake for.
+	delete(m.status, childID)
+	delete(m.slots, childID)
+	m.removeOrderLocked(childID)
+}
+
+// run is the merger's flusher loop: drain due slots, then sleep until the
+// earliest deadline, a wake, or Close. The loop never holds the mutex while
+// sleeping, so queueing trackers never block on it.
+func (m *subagentProgressMerger) run() {
+	defer m.wg.Done()
+	defer m.timer.Stop()
+	for {
+		m.mu.Lock()
+		for m.stepLocked() {
+		}
+		closed := m.closed
+		clean := m.allCleanLocked()
+		if !clean && !closed {
+			d := m.nextDeadlineLocked()
+			m.mu.Unlock()
+			m.timer.Reset(d)
+			select {
+			case <-m.done:
+				return
+			case <-m.timer.C():
+			case <-m.wake:
+			}
+			continue
+		}
+		m.mu.Unlock()
+		if closed {
+			return
+		}
+		select {
+		case <-m.done:
+			return
+		case <-m.wake:
+		}
+	}
+}
+
+// stepLocked emits at most one non-terminal progress event, round-robining
+// across children. Status events bypass the group budget; preview events
+// consume it. Returns false when nothing can be emitted right now.
+func (m *subagentProgressMerger) stepLocked() bool {
+	m.refillLocked()
+	n := len(m.order)
+	if n == 0 {
+		return false
+	}
+	now := m.clock.Now()
+	for i := 0; i < n; i++ {
+		idx := (m.rr + i) % n
+		childID := m.order[idx]
+		if st := m.status[childID]; st != nil && st.dirty && !now.Before(st.dueAt) {
+			m.rr = (idx + 1) % n
+			phase := st.phase
+			st.dirty = false
+			st.lastSend = now
+			m.emitStatusLocked(childID, phase, 0)
+			return true
+		}
+		if m.tokens >= 1 {
+			for c := subagentProgressChanReasoning; c <= subagentProgressChanNotice; c++ {
+				if sl := m.slots[childID][c]; sl != nil && sl.dirty && !now.Before(sl.dueAt) {
+					m.rr = (idx + 1) % n
+					m.tokens--
+					m.emitDeltaLocked(childID, c, sl)
+					return true
+				}
+			}
+		}
+	}
+	return false
+}
+
+func (m *subagentProgressMerger) allCleanLocked() bool {
+	for _, st := range m.status {
+		if st.dirty {
+			return false
+		}
+	}
+	for _, chs := range m.slots {
+		for _, sl := range chs {
+			if sl.dirty {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+// nextDeadlineLocked returns the wait until the earliest due slot or the next
+// preview budget token. A zero result means "wake immediately".
+func (m *subagentProgressMerger) nextDeadlineLocked() time.Duration {
+	now := m.clock.Now()
+	var next time.Time
+	consider := func(t time.Time) {
+		if next.IsZero() || t.Before(next) {
+			next = t
+		}
+	}
+	for _, st := range m.status {
+		if st.dirty {
+			consider(st.dueAt)
+		}
+	}
+	for _, chs := range m.slots {
+		for _, sl := range chs {
+			if sl.dirty {
+				consider(sl.dueAt)
+			}
+		}
+	}
+	if m.tokens < 1 {
+		refillAt := m.lastRefill.Add(time.Duration((1 - m.tokens) * float64(time.Second) / subagentProgressGroupEventsPerSec))
+		consider(refillAt)
+	}
+	if next.IsZero() {
+		return 0
+	}
+	if d := next.Sub(now); d > 0 {
+		return d
+	}
+	return 0
+}
+
+func (m *subagentProgressMerger) refillLocked() {
+	now := m.clock.Now()
+	if now.After(m.lastRefill) {
+		elapsed := now.Sub(m.lastRefill).Seconds()
+		m.tokens += elapsed * subagentProgressGroupEventsPerSec
+		if m.tokens > subagentProgressGroupBurst {
+			m.tokens = subagentProgressGroupBurst
+		}
+		m.lastRefill = now
+	}
+}
+
+func (m *subagentProgressMerger) emitStatusLocked(childID string, phase subagentProgressPhase, durationMs int64) {
+	m.emitToolProgressLocked(childID, event.SubagentProgressStatusName, string(phase), false, durationMs)
+}
+
+func (m *subagentProgressMerger) emitDeltaLocked(childID string, ch subagentProgressChannel, sl *progressSlot) {
+	if sl.buf == "" {
+		sl.dirty = false
+		return
+	}
+	buf, truncated := sl.buf, sl.truncated
+	sl.buf, sl.truncated, sl.dirty = "", false, false
+	sl.lastSend = m.clock.Now()
+	m.emitToolProgressLocked(childID, ch.name(), buf, truncated, 0)
+}
+
+func (m *subagentProgressMerger) emitToolProgressLocked(childID, name, output string, truncated bool, durationMs int64) {
+	parentID := m.groupParentID
+	if parentID == childID {
+		parentID = ""
+	}
+	m.sink.Emit(event.Event{
+		Kind: event.ToolProgress,
+		Tool: event.Tool{
+			ID: childID, Name: name, ParentID: parentID,
+			Output: output, Truncated: truncated, DurationMs: durationMs,
+		},
+	})
+}
+
+// trimToBudgetLocked keeps the child's pending total at or under
+// subagentProgressMaxPendingBytes, dropping the lowest-priority channel's
+// content first (notice < reasoning < text) so the response preview survives.
+func (m *subagentProgressMerger) trimToBudgetLocked(childID string) {
+	if m.pendingBytesLocked(childID) <= subagentProgressMaxPendingBytes {
+		return
+	}
+	if sl := m.slots[childID][subagentProgressChanNotice]; sl != nil && sl.dirty && sl.buf != "" {
+		sl.buf = ""
+		sl.truncated = true
+	}
+	for _, ch := range []subagentProgressChannel{subagentProgressChanReasoning, subagentProgressChanText} {
+		over := m.pendingBytesLocked(childID) - subagentProgressMaxPendingBytes
+		if over <= 0 {
+			return
+		}
+		sl := m.slots[childID][ch]
+		if sl == nil || !sl.dirty || sl.buf == "" {
+			continue
+		}
+		keep := len(sl.buf) - over
+		if keep <= 0 {
+			sl.buf = ""
+		} else {
+			sl.buf = utf8SafeTail(sl.buf, keep)
+		}
+		sl.truncated = true
+	}
+}
+
+func (m *subagentProgressMerger) pendingBytesLocked(childID string) int {
+	total := 0
+	for _, sl := range m.slots[childID] {
+		if sl.dirty {
+			total += len(sl.buf)
+		}
+	}
+	return total
+}
+
+func (m *subagentProgressMerger) ensureOrderLocked(childID string) {
+	for _, id := range m.order {
+		if id == childID {
+			return
+		}
+	}
+	m.order = append(m.order, childID)
+}
+
+func (m *subagentProgressMerger) removeOrderLocked(childID string) {
+	for i, id := range m.order {
+		if id == childID {
+			m.order = append(m.order[:i], m.order[i+1:]...)
+			return
+		}
+	}
+}
+
+func (m *subagentProgressMerger) wakeLocked() {
+	select {
+	case m.wake <- struct{}{}:
+	default:
+	}
+}
+
+// utf8SafeTail returns the last maxBytes bytes of s, trimmed to a rune
+// boundary so a multi-byte character is never split.
+func utf8SafeTail(s string, maxBytes int) string {
+	if len(s) <= maxBytes {
+		return s
+	}
+	s = s[len(s)-maxBytes:]
+	for len(s) > 0 && !utf8.RuneStart(s[0]) {
+		s = s[1:]
+	}
+	return s
+}
+
+// subagentProgressTracker is the per-child state machine installed between a
+// sub-agent run and its parent sink. It converts the child's reasoning/text/
+// notice/retrying into preview slots on the group merger, forwards tool
+// activity unchanged, and guarantees exactly one terminal status event.
+type subagentProgressTracker struct {
+	mu         sync.Mutex
+	merger     *subagentProgressMerger
+	childID    string
+	sink       event.Sink // forwards real tool events (the subSinkFor wrapper)
+	phase      subagentProgressPhase
+	started    time.Time
+	ownsMerger bool
+	done       bool
+}
+
+// newSubagentProgressTracker creates (or joins) the group merger and returns a
+// tracker for one child run. wrapSink is the sink the child's real tool events
+// already flow through; the tracker's own preview events are emitted through
+// the merger's sink — the same sink the child's dispatch card flowed through —
+// so preview IDs always match the card IDs the frontend sees.
+func newSubagentProgressTracker(ctx context.Context, wrapSink event.Sink) *subagentProgressTracker {
+	parentID, parent, _, ok := CallContext(ctx)
+	merger := subagentProgressMergerFromContext(ctx)
+	owns := false
+	if merger == nil {
+		// Not part of a parent task group: own a merger that emits through
+		// the same sink the dispatch event flowed through (the call context's
+		// raw sink; Discard for headless/direct-execute runs).
+		sink := event.Discard
+		if ok && parent != nil {
+			sink = parent
+		}
+		merger = newSubagentProgressMerger(realProgressClock{}, sink, parentID)
+		owns = true
+	}
+	return &subagentProgressTracker{
+		merger:     merger,
+		childID:    parentID,
+		sink:       wrapSink,
+		started:    merger.clock.Now(),
+		ownsMerger: owns,
+	}
+}
+
+// queued marks the background registration state; running marks execution
+// start (or the moment a background job acquires its execution slot).
+// queued marks the background registration state; running marks execution
+// start (or the moment a background job acquires its slot). Both are emitted
+// synchronously — not through the merging status slot — so the first visible
+// states can never be merged away by a faster follow-up transition: a
+// background job that grabs its slot microseconds after registration must not
+// hide the queued state.
+func (t *subagentProgressTracker) queued() {
+	t.emitStatusDirect(subagentPhaseQueued)
+}
+
+func (t *subagentProgressTracker) running() {
+	t.emitStatusDirect(subagentPhaseRunning)
+}
+
+func (t *subagentProgressTracker) emitStatusDirect(p subagentProgressPhase) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.done {
+		return
+	}
+	t.phase = p
+	t.merger.directStatus(t.childID, p)
+}
+
+func (t *subagentProgressTracker) setPhase(p subagentProgressPhase) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.setPhaseLocked(p)
+}
+
+// setPhaseLocked records a phase change and queues the status event; repeat
+// transitions of the same phase do not re-queue.
+func (t *subagentProgressTracker) setPhaseLocked(p subagentProgressPhase) {
+	if t.done || t.phase == p {
+		return
+	}
+	t.phase = p
+	t.merger.statusEvent(t.childID, p)
+}
+
+// wrap returns the sink the child agent emits into: reasoning/text/notice/
+// retrying become preview slots; tool activity and usage pass through
+// unchanged (the child's Message and anything else stay dropped, as before).
+// Events arriving after the terminal are ignored.
+func (t *subagentProgressTracker) wrap() event.Sink {
+	return event.FuncSink(func(e event.Event) {
+		t.mu.Lock()
+		if t.done {
+			t.mu.Unlock()
+			return
+		}
+		switch e.Kind {
+		case event.Reasoning:
+			t.setPhaseLocked(subagentPhaseReasoning)
+			t.merger.deltaEvent(t.childID, subagentProgressChanReasoning, e.Text)
+		case event.Text:
+			t.setPhaseLocked(subagentPhaseResponding)
+			t.merger.deltaEvent(t.childID, subagentProgressChanText, e.Text)
+		case event.Notice:
+			text := e.Text
+			if text == "" {
+				text = e.Detail
+			}
+			t.merger.deltaEvent(t.childID, subagentProgressChanNotice, text)
+		case event.Retrying:
+			t.setPhaseLocked(subagentPhaseRetrying)
+		case event.ToolDispatch, event.ToolResult, event.ToolProgress:
+			t.setPhaseLocked(subagentPhaseTool)
+		}
+		t.mu.Unlock()
+		switch e.Kind {
+		case event.ToolDispatch, event.ToolResult, event.ToolProgress:
+			t.sink.Emit(e)
+		case event.Usage:
+			if e.UsageSource == "" {
+				e.UsageSource = event.UsageSourceSubagent
+			}
+			t.sink.Emit(e)
+		}
+	})
+}
+
+// finish flushes pending previews, emits the single terminal status, and — if
+// the tracker owns its merger — closes it. ctxErr non-nil maps to cancelled,
+// other errors to failed, success to completed. Idempotent: late events and
+// repeated calls are ignored.
+func (t *subagentProgressTracker) finish(ctxErr, runErr error) {
+	t.mu.Lock()
+	if t.done {
+		t.mu.Unlock()
+		return
+	}
+	t.done = true
+	phase := subagentPhaseCompleted
+	if ctxErr != nil {
+		phase = subagentPhaseCancelled
+	} else if runErr != nil {
+		phase = subagentPhaseFailed
+	}
+	durationMs := t.merger.clock.Now().Sub(t.started).Milliseconds()
+	t.mu.Unlock()
+	t.merger.flushChild(t.childID, phase, durationMs)
+	if t.ownsMerger {
+		t.merger.Close()
+	}
+}
+
+// subagentProgressMergerKey carries the group merger in the child's context so
+// parallel_tasks/fleet children share one pacing budget per parent call.
+type subagentProgressMergerKey struct{}
+
+func withSubagentProgressMerger(ctx context.Context, m *subagentProgressMerger) context.Context {
+	return context.WithValue(ctx, subagentProgressMergerKey{}, m)
+}
+
+func subagentProgressMergerFromContext(ctx context.Context) *subagentProgressMerger {
+	m, _ := ctx.Value(subagentProgressMergerKey{}).(*subagentProgressMerger)
+	return m
+}

internal/agent/task.go

@@ -565,10 +565,22 @@ func (r *ReadOnlyTaskTool) ResolveProfile(args json.RawMessage) *event.Profile {
 	return r.task.ResolveProfile(args)
 }
 
-func (r *ReadOnlyTaskTool) Execute(ctx context.Context, args json.RawMessage) (string, error) {
+func (r *ReadOnlyTaskTool) Execute(ctx context.Context, args json.RawMessage) (result string, err error) {
 	if r == nil || r.task == nil {
 		return "", fmt.Errorf("read_only_task is not configured")
 	}
+	// read_only_task shares the same progress tracker as RunProfileSpec so
+	// every sub-agent entry point emits the same phase machine. It owns its
+	// merger (no parent task group) and finishes on every exit path.
+	trk := newSubagentProgressTracker(ctx, subSink(ctx))
+	trk.running()
+	defer func() {
+		if p := recover(); p != nil {
+			trk.finish(nil, fmt.Errorf("panic: %v", p))
+			panic(p)
+		}
+		trk.finish(ctx.Err(), err)
+	}()
 	var p struct {
 		Prompt      string   `json:"prompt"`
 		Description string   `json:"description"`
@@ -617,7 +629,7 @@ func (r *ReadOnlyTaskTool) Execute(ctx context.Context, args json.RawMessage) (s
 	if r.task.mutationObserver != nil {
 		mutationObserver = r.task.mutationObserver.CloneForSubagent(recoveryTaskID, r.task.mutationObserver.OwnershipTurn(), false)
 	}
-	answer, err := r.task.runReadOnlySubSession(ctx, p.Prompt, subReg, subSink(ctx), maxSteps, prov, pricing, ctxWin, NewSession(DefaultReadOnlyTaskSystemPrompt), childDepth, recoveryTaskID, usageModelRef, mutationObserver)
+	answer, err := r.task.runReadOnlySubSession(ctx, p.Prompt, subReg, trk.wrap(), maxSteps, prov, pricing, ctxWin, NewSession(DefaultReadOnlyTaskSystemPrompt), childDepth, recoveryTaskID, usageModelRef, mutationObserver)
 	if err != nil {
 		return "", err
 	}
@@ -774,10 +786,30 @@ func (t *TaskTool) resolveWriterClaims(writePaths []string, requireClaim bool) (
 // RunProfileSpec executes a unified profile/task specification. Shared by task,
 // fleet items, and boot-wired skill runners so prompt, tools, claims, and
 // scheduling cannot drift across entry points.
-func (t *TaskTool) RunProfileSpec(ctx context.Context, spec ProfileExecSpec) (string, error) {
+func (t *TaskTool) RunProfileSpec(ctx context.Context, spec ProfileExecSpec) (result string, err error) {
 	if t == nil {
 		return "", fmt.Errorf("task tool is not configured")
 	}
+	// Per-child progress tracker: converts the child's reasoning/text/notice/
+	// retrying into reserved ToolProgress previews and guarantees exactly one
+	// terminal status (completed/cancelled/failed). The background job owns
+	// finish after handoff; every other exit finishes here, including
+	// validation errors and panics.
+	trk := newSubagentProgressTracker(ctx, subSink(ctx))
+	backgroundHandoff := false
+	defer func() {
+		if backgroundHandoff {
+			return
+		}
+		if p := recover(); p != nil {
+			trk.finish(nil, fmt.Errorf("panic: %v", p))
+			panic(p)
+		}
+		trk.finish(ctx.Err(), err)
+	}()
+	if !spec.RunInBackground {
+		trk.running()
+	}
 	if strings.TrimSpace(spec.Prompt) == "" {
 		return "", fmt.Errorf("prompt is required")
 	}
@@ -822,7 +854,7 @@ func (t *TaskTool) RunProfileSpec(ctx context.Context, spec ProfileExecSpec) (st
 
 	modelRef, effortRef := spec.Model, spec.Effort
 	usageModelRef := t.usageModelRef(modelRef, effortRef)
-	parentID, parent, _, _ := CallContext(ctx)
+	parentID, _, _, _ := CallContext(ctx)
 	run, err := t.prepareTranscriptRunWithPrompt(subReg, modelRef, effortRef, ParentSession(ctx), parentID, spec.ContinueFrom, spec.ForkFrom, spec.SystemPrompt, spec.Kind, spec.Name)
 	if err != nil {
 		return "", err
@@ -895,7 +927,6 @@ func (t *TaskTool) RunProfileSpec(ctx context.Context, spec ProfileExecSpec) (st
 		} else {
 			releaseStart = func() {}
 		}
-		nested := subSinkFor(parentID, parent)
 		label := firstNonEmpty(spec.Description, spec.Name, "task")
 		if t.transcripts != nil && run != nil && run.Ref != "" {
 			if err := t.transcripts.MarkRunning(run); err != nil {
@@ -918,6 +949,9 @@ func (t *TaskTool) RunProfileSpec(ctx context.Context, spec ProfileExecSpec) (st
 		backgroundEvidence := evidence.NewLedger()
 		// Capture acquire request by value for the job goroutine.
 		slotReq := acquireReq
+		// Emit queued before the job goroutine can start so the status slot
+		// never regresses to a stale queued after running.
+		trk.queued()
 		job := jm.StartForSession(jobs.SessionFromContext(ctx), "task", label, func(jobCtx context.Context, _ io.Writer) (result string, err error) {
 			if writerRegistered {
 				defer mutationObserver.UnregisterWriter(recoveryTaskID)
@@ -932,6 +966,9 @@ func (t *TaskTool) RunProfileSpec(ctx context.Context, spec ProfileExecSpec) (st
 					result = FormatSubagentRunResult("", run, true)
 					err = errors.Join(panicErr, t.transcripts.SaveFailed(run))
 				}
+				// The job owns the terminal status: the parent tool call has
+				// already returned its job id by now.
+				trk.finish(jobCtx.Err(), err)
 			}()
 			// Queue for a concurrency/write slot here — not before Start —
 			// so the parent tool call returns a job id immediately.
@@ -940,7 +977,8 @@ func (t *TaskTool) RunProfileSpec(ctx context.Context, spec ProfileExecSpec) (st
 				return FormatSubagentRunResult("", run, true), errors.Join(slotErr, t.transcripts.SaveFailed(run))
 			}
 			defer releaseSlot()
-			answer, err := runSession(jobCtx, nested, writerRegistered)
+			trk.running()
+			answer, err := runSession(jobCtx, trk.wrap(), writerRegistered)
 			if err != nil {
 				return FormatSubagentRunResult("", run, true), errors.Join(err, t.transcripts.SaveFailed(run))
 			}
@@ -950,6 +988,9 @@ func (t *TaskTool) RunProfileSpec(ctx context.Context, spec ProfileExecSpec) (st
 			return FormatSubagentRunResult(answer, run, false), nil
 		})
 		releaseStart()
+		// Hand the tracker to the job goroutine: the outer defer must not
+		// finish (and close) it while the job still runs.
+		backgroundHandoff = true
 		queuedNote := ""
 		if t.scheduler != nil {
 			queuedNote = " It may wait in the session queue until a concurrency/write slot is free."
@@ -968,7 +1009,7 @@ func (t *TaskTool) RunProfileSpec(ctx context.Context, spec ProfileExecSpec) (st
 	}
 	defer releaseSlot()
 	defer run.Release()
-	answer, err := runSession(ctx, subSink(ctx), false)
+	answer, err := runSession(ctx, trk.wrap(), false)
 	if err != nil {
 		return "", errors.Join(err, t.transcripts.SaveFailed(run))
 	}
@@ -2088,9 +2129,11 @@ func NestedSink(ctx context.Context, fallback event.Sink) event.Sink {
 	return subSinkFor(parentID, parent)
 }
 
-// subSink forwards a sub-agent's tool dispatch/result events and billable usage
-// to the parent's event stream. Only tool activity is nested visually; the
-// sub-agent's text/reasoning stays isolated and only its final answer is returned.
+// subSink forwards a sub-agent's tool dispatch/result/progress events and
+// billable usage to the parent's event stream. Only tool activity is nested
+// visually; the sub-agent's text/reasoning stays isolated (progress previews
+// travel as reserved ToolProgress channels, not as parent Text/Reasoning) and
+// only its final answer is returned.
 //
 // The sub-agent's own turn/text/reasoning events are dropped — forwarding them
 // would make the parent transcript noisy and could imply they belong to the
@@ -2102,8 +2145,10 @@ func NestedSink(ctx context.Context, fallback event.Sink) event.Sink {
 // Tool events are tagged with the parent task call's ID so a frontend nests them
 // under it. The forwarded call IDs are namespaced with the parent ID so a
 // sub-agent call can never collide with a parent call in the frontend's
-// dispatch→result matching. Falls back to Discard when there's no parent stream
-// (the headless run loop, or a direct Execute in tests).
+// dispatch→result matching. ToolProgress covers both the sub-agent's real tool
+// output and nested sub-agent progress previews, which ride the same sink so
+// their IDs match the cards they belong to. Falls back to Discard when there's
+// no parent stream (the headless run loop, or a direct Execute in tests).
 func subSink(ctx context.Context) event.Sink {
 	parentID, parent, _, ok := CallContext(ctx)
 	if !ok || parent == nil {
@@ -2121,7 +2166,7 @@ func subSinkFor(parentID string, parent event.Sink) event.Sink {
 	}
 	return event.FuncSink(func(e event.Event) {
 		switch e.Kind {
-		case event.ToolDispatch, event.ToolResult:
+		case event.ToolDispatch, event.ToolResult, event.ToolProgress:
 			e.Tool.ParentID = parentID
 			e.Tool.ID = parentID + "/" + e.Tool.ID
 			parent.Emit(e)

internal/event/subagent_progress.go

@@ -0,0 +1,35 @@
+package event
+
+// Reserved ToolProgress channel names for local sub-agent progress previews.
+//
+// These names are an internal wire contract between the agent progress
+// tracker and local frontends (desktop and CLI). They must never be presented
+// as provider-visible tool names: ACP and bot consumers ignore ToolProgress
+// bodies entirely, and the names are filtered out of any transcript or
+// provider context by construction (they only ever ride ToolProgress events).
+//
+// The status channel carries exactly one of the phase values produced by the
+// tracker (queued/running/reasoning/responding/tool/retrying/completed/
+// failed/cancelled) in Tool.Output; the other channels carry bounded UTF-8
+// text deltas. Tool.Truncated marks a merged or truncated preview round,
+// Tool.DurationMs rides terminal status events, and progress lookup is keyed
+// by Tool.ID (the child task card ID).
+const (
+	SubagentProgressStatusName    = "reasonix.subagent.status"
+	SubagentProgressReasoningName = "reasonix.subagent.reasoning"
+	SubagentProgressTextName      = "reasonix.subagent.text"
+	SubagentProgressNoticeName    = "reasonix.subagent.notice"
+)
+
+// IsSubagentProgressName reports whether the ToolProgress Name carries a
+// reserved sub-agent progress channel. Frontends use it to route progress
+// previews away from ordinary tool output; ACP/bot rely on it only in tests
+// locking that the bodies stay ignored.
+func IsSubagentProgressName(name string) bool {
+	switch name {
+	case SubagentProgressStatusName, SubagentProgressReasoningName,
+		SubagentProgressTextName, SubagentProgressNoticeName:
+		return true
+	}
+	return false
+}