executus/run/progress.go

package run

import "context"

// ProgressSink reports a one-line progress note for the current run upward to
// any ancestor run that is being watched by a run-critic. It exists to solve a
// specific false-positive: when an agent calls a long-running skill/agent as a
// single tool, the parent's agent loop is BLOCKED on that one tool call for the
// whole child run, so the parent's progress recorder sees "zero iterations,
// zero new tokens, no activity" and its critic concludes the tool "hung
// indefinitely" — even though the child is iterating happily. Forwarding the
// child's per-step activity up the chain keeps every blocked ancestor's
// last-activity fresh, so a healthy-but-slow child is no longer mistaken for a
// hang. A nil ProgressSink is safe to ignore (there is no ancestor to notify).
type ProgressSink func(note string)

type progressSinkKey struct{}

// WithProgressSink returns a context carrying sink for descendant runs to find
// via ProgressSinkFrom. A nil sink is stored as-is (ProgressSinkFrom returns
// nil), which callers treat as "no ancestor watching".
func WithProgressSink(ctx context.Context, sink ProgressSink) context.Context {
	return context.WithValue(ctx, progressSinkKey{}, sink)
}

// ProgressSinkFrom returns the ancestor progress sink carried on ctx, or nil
// if none is wired. The returned sink, when non-nil, forwards a note to the
// immediate parent run's recorder AND (transitively) to every further
// ancestor, because each level installs a sink that forwards upward.
func ProgressSinkFrom(ctx context.Context) ProgressSink {
	if v := ctx.Value(progressSinkKey{}); v != nil {
		if s, ok := v.(ProgressSink); ok {
			return s
		}
	}
	return nil
}

// InstallProgressBridge wires the current run into the ancestor progress chain.
//
//	report  — this run's own recorder hook (e.g. recorder.OnStatus). nil when
//	          the run has no critic recorder of its own (the common skill case);
//	          the bridge then purely forwards descendants' progress upward.
//
// It returns:
//
//	childCtx        — pass this to the agent loop / toolbox so descendant runs
//	                  (invoked as tools) forward their progress into this chain.
//	notifyAncestors — call this on each of THIS run's own loop steps to keep
//	                  every ancestor critic's last-activity fresh. nil when this
//	                  run has no ancestors (it is a top-level run); nil-safe to
//	                  call only via the returned value being checked, so callers
//	                  should guard `if notifyAncestors != nil`.
//
// The chain is built so that a note from any descendant bumps the recorders of
// ALL of its blocked ancestors, not just its immediate parent.
func InstallProgressBridge(ctx context.Context, report ProgressSink) (childCtx context.Context, notifyAncestors ProgressSink) {
	parent := ProgressSinkFrom(ctx)

	// The sink descendants will call. It must bump this run's own recorder
	// (report) AND forward to all ancestors (parent). Collapse to the minimal
	// closure so we don't stack a no-op wrapper for recorder-less runs.
	var child ProgressSink
	switch {
	case report == nil:
		// No recorder of our own: descendants forward straight to ancestors.
		child = parent
	case parent == nil:
		// Top-level run with a recorder: descendants feed only our recorder.
		child = report
	default:
		child = func(note string) {
			report(note)
			parent(note)
		}
	}

	childCtx = ctx
	if child != nil {
		childCtx = WithProgressSink(ctx, child)
	}
	// This run's own steps notify ancestors directly (its own recorder is fed
	// separately by its step observer, so we deliberately do not call report
	// here — only the ancestors need waking).
	return childCtx, parent
}