schedule.go

package exec

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/Agent-Field/codeaf/internal/ctxbudget"
	"github.com/Agent-Field/codeaf/internal/guard"
	"github.com/Agent-Field/codeaf/internal/plan"
	"github.com/Agent-Field/codeaf/internal/provider"
	"github.com/Agent-Field/codeaf/internal/store"
)

// Scheduler drives a graph to completion.
//
// It is dependency-driven rather than wave-locked. Waves are a way to describe
// a schedule, not a way to run one: waiting for the slowest node of a wave
// before starting anything in the next holds back work whose inputs are already
// finished. A node starts the moment the specific nodes it named are done, which
// is what the whole binding pass was for.
type Scheduler struct {
	registry    *Registry
	workspace   *Workspace
	concurrency int
	governor    *Governor
	usage       Usage
	usageMutex  sync.RWMutex

	// Budget bounds the whole run's token spend, in prompt+completion tokens.
	// Zero means unbounded — each leaf still has its own budget. When the
	// cumulative spend passes it, no new node is launched; whatever is in
	// flight lands (mirroring the per-leaf landing reserve), never-started
	// nodes are marked, and Run reports the stop reason.
	Budget int

	// Escalations is how many times a failed leaf may be re-run on a stronger
	// model. Zero — the default — is exactly today's behaviour: a leaf that
	// fails, fails. It is only worth setting when there is somewhere stronger to
	// go, so the caller sets it from the panel rather than the scheduler
	// assuming one exists.
	//
	// Only verdicts that a better model could plausibly fix count: running out
	// of budget, running out of turns, returning nothing at all. A provider
	// failure is weather and a rate limit is not cured by spending more.
	Escalations int

	// NodeTimeout is the watchdog on a single node. The executor has its own
	// deadline, so this only fires when an executor is wedged past every
	// deadline it was given — a hung pipe, a stuck transport. The ending is
	// recorded as an [Abandoned] rather than letting one stuck goroutine freeze
	// the run silently and forever, and what that ending MEANS is decided in
	// [Scheduler.apply]: a node with a record goes back on the queue, and a node
	// that reached nothing is failed. Zero disables it.
	NodeTimeout time.Duration

	// BeforeLaunch applies process policy immediately before a leaf starts.
	// Returning an error stops new launches while already-running leaves land.
	BeforeLaunch func(context.Context) error

	// OnEvent reports state changes as they happen. A run is long and mostly
	// invisible; without this the only feedback is silence followed by a graph.
	OnEvent func(Event)
}

// stallAfter is how long the run may go without a completion before the
// scheduler says which nodes it is still waiting on. A wedged provider call
// once froze the log for thirteen minutes mid-run; the only thing worse than
// the stall was that it was indistinguishable from the process having died.
const stallAfter = 3 * time.Minute

// drainGrace bounds how long a stopping run waits for in-flight nodes to
// land. Their contexts are already cancelled, so an honest executor returns
// in seconds; anything still out after this is recorded and abandoned.
const drainGrace = 30 * time.Second

// exhaustionRequeues is how many times one node may go back on the queue for
// having run out of the room it was given, rather than being settled failed.
// The reasoning is beside the rule, in [Scheduler.apply].
const exhaustionRequeues = 1

// Event is one thing happening to one node.
type Event struct {
	NodeID  int
	Title   string
	State   plan.State
	Detail  string
	Elapsed time.Duration
}

func NewScheduler(registry *Registry, workspace *Workspace, concurrency int) *Scheduler {
	if concurrency <= 0 {
		concurrency = 8
	}
	return &Scheduler{registry: registry, workspace: workspace, concurrency: concurrency,
		governor: HostGovernor()}
}

// WithGovernor replaces the shared host gate. Production uses the process-wide
// one so a headless run and a resident runner in the same process read the same
// machine rather than each discovering its own number.
func (s *Scheduler) WithGovernor(governor *Governor) *Scheduler {
	s.governor = governor
	return s
}

// Usage is the total cost of the run, summed as outcomes are applied. It is
// written on the scheduler goroutine. Media spend gates may read it from a
// worker immediately before generation, so the small value is copied under a
// read lock.
func (s *Scheduler) Usage() Usage {
	s.usageMutex.RLock()
	defer s.usageMutex.RUnlock()
	return s.usage
}

func (s *Scheduler) spentTokens() int {
	usage := s.Usage()
	return usage.PromptTokens + usage.CompletionTokens
}

func (s *Scheduler) addUsage(usage Usage) {
	s.usageMutex.Lock()
	defer s.usageMutex.Unlock()
	s.usage.merge(usage)
}

// Run executes every runnable node in the graph and records results onto it.
//
// The graph is mutated in place and never concurrently: workers return outcomes
// through a channel and the scheduler applies them on its own goroutine. That
// keeps the one shared structure single-writer, which matters more here than
// anywhere else — appending to g.Nodes while a worker held a *Node has already
// cost us a duplicated subtree once.
func (s *Scheduler) Run(ctx context.Context, graph *plan.Graph) error {
	started := time.Now()
	// Buffered to the whole graph so a worker can always deliver its result
	// and exit, even after the scheduler has stopped listening for it — an
	// abandoned worker blocked on an unbuffered send would leak forever.
	done := make(chan completion, len(graph.Nodes))
	// When a node started, keyed by id. Doubles as the in-flight set and keeps
	// the cancellation/teardown handles needed by watchdog abandonment.
	inFlight := map[int]leafFlight{}
	// How many times each node has already been given up on. It is also the
	// attempt number the leaf runs under, which is how a router is told to climb
	// without the scheduler knowing what it is climbing.
	retries := map[int]int{}
	// Why launching stopped. Once set, nothing new starts, in-flight work
	// lands, and Run reports it — a run may stop early, but it must never
	// stop silently.
	var stop string
	var stopCause error
	lastProgress := time.Now()

	ticker := time.NewTicker(s.tickEvery())
	defer ticker.Stop()

	for {
		// Anything whose inputs all failed can never run; retiring it before
		// looking for work stops the loop spinning on nodes that will never
		// become ready.
		s.propagateBlocked(graph)

		if stop == "" && s.Budget > 0 && s.spentTokens() >= s.Budget {
			stop = fmt.Sprintf("global budget exhausted: %d of %d tokens spent",
				s.spentTokens(), s.Budget)
		}
		if stop == "" {
			for _, id := range s.ready(graph) {
				if len(inFlight) >= s.concurrency {
					break
				}
				// The host-load gate, which the concurrency ceiling cannot
				// stand in for: workers are network-bound and park on sockets,
				// but a leaf's shell can pin every core, and `--concurrency 8`
				// on a machine already at 10.0 load makes eight leaves that all
				// run slower rather than eight leaves that run. A refusal is
				// not a hold — nothing in flight is delayed and nothing is
				// cancelled; the loop simply stops launching and the next tick
				// asks the machine again. With nothing in flight the governor
				// always admits, so back-pressure can never leave the run
				// doing nothing at all.
				if !s.governor.Admit(len(inFlight)) {
					break
				}
				if s.BeforeLaunch != nil {
					if err := s.BeforeLaunch(ctx); err != nil {
						stop, stopCause = err.Error(), err
						break
					}
				}
				node := graph.Node(id)
				node.State = plan.StateRunning
				s.emit(Event{NodeID: id, Title: node.Title, State: plan.StateRunning, Elapsed: time.Since(started)})
				task := s.taskFor(graph, node)
				// Within-node progress on the surface that has no thread to
				// post to: the run's own event stream, which is already how a
				// headless run learns that anything is happening at all. A leaf
				// that says nothing for forty minutes is indistinguishable from
				// a wedged one, and the stall reporter below can only say which
				// node it is still waiting on, never what that node is doing.
				task.Progress = func(phase string, done, total int, latest string) {
					s.emit(Event{NodeID: node.ID, Title: node.Title, State: plan.StateRunning,
						Detail: progressDetail(phase, done, total, latest), Elapsed: time.Since(started)})
				}
				leafCtx, cancel := context.WithCancel(ctx)
				control := &leafControl{}
				task.control = control
				inFlight[id] = leafFlight{started: time.Now(), cancel: cancel, control: control, timeout: s.NodeTimeout}
				go s.work(leafCtx, id, task, retries[id], leafShape(node), done)
			}
		}

		// Nothing running and nothing newly runnable means everything that can
		// be done has been — or that the run was told to stop and the last
		// in-flight node has landed. A cancellation that emptied the in-flight
		// set through the nodes' own child contexts still counts as a stop:
		// the race between a node landing cancelled and the scheduler seeing
		// ctx.Done must not decide whether the reason gets reported.
		if len(inFlight) == 0 {
			if stop == "" && ctx.Err() != nil {
				stop = fmt.Sprintf("run context cancelled (%v)", ctx.Err())
			}
			return s.finish(graph, stop, stopCause)
		}

		select {
		case finished := <-done:
			flight, active := inFlight[finished.nodeID]
			if !active {
				// A watchdog-abandoned executor may eventually return. Its late
				// completion must not overwrite the recorded abandonment.
				continue
			}
			flight.cancel()
			delete(inFlight, finished.nodeID)
			lastProgress = time.Now()
			s.apply(graph, finished.nodeID, finished.outcome, finished.err, started, retries)
		case <-ctx.Done():
			// The run context being cancelled — an interrupt, an operator
			// deadline — is a stop, not a vanishing act. In-flight nodes hold
			// child contexts that are already cancelled, so they are drained
			// with a bounded grace and their outcomes recorded before the
			// stop is reported.
			if stop == "" {
				stop = fmt.Sprintf("run context cancelled (%v)", ctx.Err())
			}
			s.drain(graph, done, inFlight, started)
			return s.finish(graph, stop, stopCause)
		case <-ticker.C:
			now := time.Now()
			for id, flight := range inFlight {
				if flight.timeout > 0 && now.Sub(flight.started) > flight.timeout {
					flight.cancel()
					// The ending is TYPED and its sentence is written once, by
					// [Abandoned.Error]. Spelling it out here as well was the
					// same sentence in two files, and the two were then read by
					// code that had to agree about what it meant: what happens
					// to an abandoned node is decided in apply below, from the
					// type, so the words are nobody's evidence.
					var abandoned error = &Abandoned{After: flight.timeout}
					if terminated := flight.control.terminate(); terminated > 0 {
						abandoned = fmt.Errorf("%w; %d background jobs terminated at leaf end", abandoned, terminated)
					}
					delete(inFlight, id)
					s.apply(graph, id, nil, abandoned, started, retries)
				}
			}
			// A silent run is indistinguishable from a dead one. When nothing
			// has completed for a while, say what is still out and for how
			// long, so a frozen log reads as waiting rather than as death.
			if len(inFlight) > 0 && now.Sub(lastProgress) >= stallAfter {
				lastProgress = now
				for id, flight := range inFlight {
					node := graph.Node(id)
					s.emit(Event{NodeID: id, Title: node.Title, State: plan.StateRunning,
						Detail:  fmt.Sprintf("still in flight after %s — the run is waiting, not dead", now.Sub(flight.started).Round(time.Second)),
						Elapsed: time.Since(started)})
				}
			}
		}
	}
}

// completion is one worker's report back to the scheduler's goroutine.
type completion struct {
	nodeID  int
	outcome *Outcome
	err     error
}

type leafFlight struct {
	started time.Time
	cancel  context.CancelFunc
	control *leafControl
	// timeout is this leaf's own watchdog, shaped to its worker. A flat
	// NodeTimeout sized for the generalist abandoned a coding pipeline at
	// seventeen minutes with a verification pass already in hand — and a leaf
	// killed from outside lands no terminal event, so its spend vanishes with
	// it. Zero disables, exactly as it does on the Scheduler field.
	timeout time.Duration
}

// work runs one node and always reports back, even when the executor panics —
// a panic that unwinds a worker silently would strand the scheduler waiting on
// a completion that can never come.
func (s *Scheduler) work(ctx context.Context, id int, task Task, attempt int, shape string, done chan<- completion) {
	defer func() {
		if recovered := recover(); recovered != nil {
			// The wording the scheduler already reports is kept; what is new is
			// that the stack reaches the log and the fault itself reaches the
			// journal, so a person watching the run learns that the work was
			// interrupted rather than inferring it from a silence. See
			// Task.Fault.
			task.Faulted(guard.Note("exec/scheduler leaf", recovered))
			failure := fmt.Sprintf("executor panicked: %v", recovered)
			if terminated := task.control.terminate(); terminated > 0 {
				failure += fmt.Sprintf("; %d background jobs terminated at leaf end", terminated)
			}
			done <- completion{nodeID: id, err: fmt.Errorf("%s", failure)}
		}
	}()
	// One leaf is one routable unit, opened here rather than inside the loop.
	// Every turn the executor takes belongs to this slot, so a router picks a
	// model once and the whole transcript stays on it — a leaf that changed
	// model mid-loop would rewrite its prefix cache every turn and splice two
	// lineages into one conversation.
	ctx = provider.WithCallShape(ctx, provider.ClassExecLeaf, attempt, shape)
	// The node's own choice, honoured. Registry.For serves an unknown name with
	// the generalist, so a graph that names a worker this process does not have
	// still gets its work done.
	outcome, err := s.registry.For(task.Subharness).Run(ctx, task)
	done <- completion{nodeID: id, outcome: outcome, err: err}
}

// leafShape says which population of leaves this node belongs to, so that what a
// router learns about one kind of leaf is not applied to every other kind.
//
// Three buckets, and the count is the design. Every leaf used to be one class,
// and arm B measured what that costs: five leaves that exhausted their budget on
// t1 — the one task whose leaves carry 2.2M prompt tokens — moved the single
// global rating far enough to reroute t2's document reading and t3's small
// repairs, where the demoted model had never once failed, and both collapsed
// from working to zero. But the opposite mistake is just as easy: a key so fine
// that no bucket ever accumulates enough graded outcomes to pass MinGraded is a
// ledger that has learned nothing at all, expensively.
//
// So it splits on the two things the harness already knows about a node before
// it runs, and nothing else. Kind separates the synthesis nodes the harness owns
// — many long inputs, a roll-up rather than a job — from the work the plan asked
// for. Size separates the rest along the axis the failures actually fell on:
// oversized and borderline are the leaves one agent may not fit, which is what a
// budget stop usually means, and atomic is the rest. Borderline sits with
// oversized rather than with atomic because the risk it names is the same risk,
// and because erring that way keeps a lesson learned on a doubtful leaf away
// from the leaves nobody doubted.
func leafShape(node *plan.Node) string {
	if node.Kind == plan.KindSynthesis {
		return "synthesis"
	}
	switch node.Size {
	case plan.SizeOversized, plan.SizeBorderline:
		return "oversized"
	default:
		return "atomic"
	}
}

// LeafShape exposes the scheduler's population key to resident execution. Both
// surfaces must write observations into the same ledger cells or neither has
// enough evidence to learn a useful ordering.
func LeafShape(node *plan.Node) string { return leafShape(node) }

// drain lets in-flight nodes land after the run has been told to stop. Their
// contexts are already cancelled, so each executor's own landing procedure is
// what runs here; the grace period only bounds a worker that is wedged past
// even that.
func (s *Scheduler) drain(graph *plan.Graph, done <-chan completion, inFlight map[int]leafFlight, started time.Time) {
	grace := time.NewTimer(drainGrace)
	defer grace.Stop()
	for len(inFlight) > 0 {
		select {
		case finished := <-done:
			flight, active := inFlight[finished.nodeID]
			if !active {
				continue
			}
			flight.cancel()
			delete(inFlight, finished.nodeID)
			// No retries while draining: the run has already been told to stop,
			// and putting a node back to pending here would leave it pending
			// forever with nothing left to launch it.
			s.apply(graph, finished.nodeID, finished.outcome, finished.err, started, nil)
		case <-grace.C:
			for id, flight := range inFlight {
				node := graph.Node(id)
				flight.cancel()
				terminated := flight.control.terminate()
				node.State = plan.StateFailed
				node.Failure = fmt.Sprintf("in flight when the run stopped and did not land within %s", drainGrace)
				if terminated > 0 {
					node.Failure += fmt.Sprintf("; %d background jobs terminated at leaf end", terminated)
				}
				s.emit(Event{NodeID: id, Title: node.Title, State: plan.StateFailed, Detail: node.Failure, Elapsed: time.Since(started)})
				delete(inFlight, id)
			}
		}
	}
}

// finish annotates whatever never ran and turns the stop reason into the run's
// error. Every early return in Run funnels through here, so no path can end
// the run without the graph saying what happened to each node.
func (s *Scheduler) finish(graph *plan.Graph, stop string, cause error) error {
	if stop == "" {
		return nil
	}
	for index := range graph.Nodes {
		node := &graph.Nodes[index]
		if node.State == plan.StatePending {
			node.State = plan.StateBlocked
			node.Failure = "never started: " + stop
		}
	}
	if cause != nil {
		return fmt.Errorf("run stopped: %w", cause)
	}
	return fmt.Errorf("run stopped: %s", stop)
}

// tickEvery sizes the housekeeping tick to the watchdog it drives; without a
// timeout the tick only feeds the stall heartbeat.
func (s *Scheduler) tickEvery() time.Duration {
	tick := 15 * time.Second
	if s.NodeTimeout > 0 && s.NodeTimeout/4 < tick {
		tick = s.NodeTimeout / 4
	}
	if tick < 10*time.Millisecond {
		tick = 10 * time.Millisecond
	}
	return tick
}

// ready lists pending nodes whose inputs are all done.
func (s *Scheduler) ready(graph *plan.Graph) []int {
	var ids []int
	for _, node := range graph.Nodes {
		if node.State != plan.StatePending {
			continue
		}
		runnable := true
		for _, need := range node.Needs {
			if source := graph.Node(need); source == nil || source.State != plan.StateDone {
				runnable = false
				break
			}
		}
		if runnable {
			ids = append(ids, node.ID)
		}
	}
	return ids
}

// propagateBlocked marks the descendants of a failure. One failed node should
// cost its own subtree and nothing else: everything independent still runs, and
// the report can say exactly what was lost rather than that the run died.
func (s *Scheduler) propagateBlocked(graph *plan.Graph) {
	for changed := true; changed; {
		changed = false
		for index := range graph.Nodes {
			node := &graph.Nodes[index]
			if node.State != plan.StatePending {
				continue
			}
			for _, need := range node.Needs {
				source := graph.Node(need)
				if source != nil && (source.State == plan.StateFailed || source.State == plan.StateBlocked) {
					node.State = plan.StateBlocked
					node.Failure = fmt.Sprintf("input %d (%s) did not complete", source.ID, source.Title)
					changed = true
					break
				}
			}
		}
	}
}

// taskFor assembles one node's assignment. The inputs are exactly its declared
// dependencies — this is the moment the edge list stops being a schedule and
// becomes a context router.
func (s *Scheduler) taskFor(graph *plan.Graph, node *plan.Node) Task {
	task := Task{
		NodeID:     node.ID,
		Title:      node.Title,
		Goal:       graph.Goal,
		Brief:      node.Brief,
		Contract:   node.Contract,
		Subharness: node.Subharness,
		Skills:     node.Skills,
		OutputHint: SuggestPath(node.ID, node.Title),
	}
	if strings.TrimSpace(task.Brief) == "" {
		task.Brief = fallbackBrief(node)
	}
	// Sized for the worker that will read it, which is the executor this node
	// was routed to and not the one that produced the material.
	inputBudget := s.inputBudget(node.Subharness)
	// One file is inlined once for one consumer, however many producers name it,
	// exactly as the resident surface's own fan-in does it: two upstream nodes
	// that both cite the shared spec must not hand the join two copies of it.
	inlined := make(map[string]bool)
	for _, need := range node.Needs {
		source := graph.Node(need)
		if source == nil {
			continue
		}
		task.Inputs = append(task.Inputs, s.input(source.Title, source.Result, source.Artifacts, inputBudget, inlined))
	}
	// A node put back to pending for escalation still carries what its last
	// attempt produced — the scheduler wrote it there and is about to overwrite
	// it. Handing it back is the difference between buying a stronger model and
	// buying a stronger model plus a second run of the work already done.
	if previous := strings.TrimSpace(node.Result); previous != "" {
		own := s.input("your own earlier attempt at this same task", previous, node.Artifacts, inputBudget, inlined)
		own.Result = "An earlier attempt on a weaker model ended as " + string(node.Verdict) +
			". What it had when it stopped:\n" + own.Result
		task.Inputs = append(task.Inputs, own)
	}
	return task
}

// input routes one producer's work into one consumer, and it is the headless
// half of the same policy the resident surface has been running: the edge
// carries the WORK, not an announcement of it.
//
// A leaf that answered by writing a file says so in one sentence — "the
// evaluation is complete, the file is at 07-vendors.md" — and that sentence is
// the honest final message for such a leaf. Handed on alone, under a header
// promising the consumer already holds this work, it made the header false: the
// consumer discovered it held a path and went and got the material. The resident
// path fixed that by reading the files back into the digest (store.readProduct);
// this path never did, so every headless fan-in leaf opened its inputs by hand
// and was invited to by the brief, which had no way to say otherwise because
// nothing here ever set Whole.
//
// Both halves are settled here from the same two structural facts: how much of
// the producer's own text fit, and whether every file it left is now in the
// block. Neither is a reading of what the text says.
func (s *Scheduler) input(title, result string, artifacts []string, budget int, inlined map[string]bool) Input {
	bounded := boundInput(result, artifacts, budget)
	// What the producer's own message did not spend is what its files may. A
	// message that already filled the budget buys no inlining, which is the
	// correct answer rather than a shortfall: the consumer is over its share
	// before a file is opened.
	block, held := "", len(artifacts) == 0
	if room := budget - len(bounded); room > 0 {
		// Recorded artifact paths are workspace-relative; the reader here is this
		// process rather than the leaf, whose cwd is the workspace. Locate is what
		// knows both spellings of the root.
		block, held = inlineProduct(s.workspace, artifacts, room, inlined)
	}
	return Input{
		Title:     title,
		Result:    bounded + block,
		Artifacts: artifacts,
		// True exactly when the block above IS the producer's material: its own
		// text uncut, and every file it left read back whole.
		Whole: len(bounded) == len(result) && held,
	}
}

// inlineProduct reads a producer's files back into the consumer's block, through
// the workspace so a recorded relative path resolves to the file it names.
//
// The reading itself is the store's (store.InlineProduct) rather than a second
// copy of it: the structural refusals — not a regular file, reads as binary,
// past the budget — are one judgment about what a consumer can be handed, and
// two spellings of it would be two answers to one question. A nil workspace
// reads the paths as given, which is what a caller holding absolute paths has.
func inlineProduct(space *Workspace, artifacts []string, budget int, inlined map[string]bool) (string, bool) {
	if len(artifacts) == 0 {
		return "", true
	}
	located := make([]string, 0, len(artifacts))
	found := true
	for _, path := range artifacts {
		if space == nil {
			located = append(located, path)
			continue
		}
		full, ok := space.Locate(path)
		if !ok {
			// A recorded path that is no longer on disk is not held, and saying
			// so is the difference between a consumer that opens a file and a
			// consumer that is told it need not.
			found = false
			continue
		}
		located = append(located, full)
	}
	block, whole := store.InlineProduct(located, budget, inlined)
	return block, whole && found
}

// maxInputBytes bounds one upstream result as it is routed downstream, on a
// machine that cannot say what the consumer holds.
//
// This is the roll-up half of the context problem and it multiplies worse than
// the loop's own: an input sits in the opening prompt, so it is resent on every
// turn the consumer takes. A node with five long inputs pays for all five, every
// turn, before it has done anything. The full text is never lost — it is in the
// artifact the producer wrote, one `sh` call away.
//
// Six kilobytes is that reasoning applied to the 32KB window an unrecognised
// model gets, so it stays as the named fallback and becomes the numerator of a
// share everywhere the window is known. See toolBudgets, which does the same
// thing to the four bounds inside the leaf.
const maxInputBytes = 6 << 10

// inputBudget is how much of one upstream result the consuming leaf may carry.
//
// It asks the executor rather than the plan, because the window belongs to the
// model the worker will actually run on and the registry is where that worker
// is known. An executor with nothing to say — a test double, a scheduler built
// without a registry — leaves the window unknown, and unknown is the fallback
// rather than a guess.
func (s *Scheduler) inputBudget(subharness string) int {
	tokens := 0
	if s.registry != nil {
		if sized, ok := s.registry.For(subharness).(contextSized); ok {
			tokens = sized.ContextLength()
		}
	}
	return ctxbudget.For(tokens).WithFloor(observationFixedFloorTokens).
		Share(maxInputBytes, toolBudgetReference, maxInputBytes)
}

// contextSized is an executor that knows how much its model holds. It is an
// optional interface rather than a method on Executor because the answer is a
// fact about a model, and a specialised worker that shells out to somebody
// else's agent genuinely does not have one.
type contextSized interface {
	ContextLength() int
}

func boundInput(result string, artifacts []string, limit int) string {
	if len(result) <= limit {
		return result
	}
	pointer := "the file it wrote"
	if len(artifacts) > 0 {
		pointer = strings.Join(artifacts, ", ")
	}
	kept := wholeRunesHead(result[:limit])
	return kept + fmt.Sprintf(
		"\n\n... [truncated at %d of %d bytes — the complete version is in %s]",
		len(kept), len(result), pointer)
}

// fallbackBrief covers nodes the planner never wrote an instruction for.
// Synthesis nodes are the normal case: the harness owns them, so it owns their
// instruction too rather than asking a model to invent one.
//
// The composition itself is the planner's, because the planner reaches the same
// question from the other side — a brief call that would not answer leaves a
// node to be composed for while the graph is still being built — and two
// compositions would be two answers to one question. See plan.ComposedBrief.
func fallbackBrief(node *plan.Node) string {
	return plan.ComposedBrief(*node)
}

func (s *Scheduler) apply(graph *plan.Graph, nodeID int, outcome *Outcome, err error, started time.Time, retries map[int]int) {
	node := graph.Node(nodeID)
	if node == nil {
		return
	}
	// A node the watchdog already retired may still report in late. Its state
	// has been decided; only the spend is real and must not be lost.
	if node.State != plan.StateRunning {
		if outcome != nil {
			s.addUsage(outcome.Usage)
		}
		return
	}
	if outcome != nil {
		s.addUsage(outcome.Usage)
	}
	// codeaf plan run (cmd/codeaf/run.go) is this scheduler's only caller. Its
	// nodes are minted under no store namespace, so there is nowhere durable to
	// journal into and nothing that would read one; that is why nil is passed
	// instead of carrying a hook no door sets. Reaching the seam anyway keeps the
	// ending's fields from drifting between the two doors, and a surface that
	// does have a namespace adds an argument here rather than a second copy.
	Settle(node, outcome, err, nil)
	// A leaf that failed in a way a stronger model might fix is worth one more
	// run. It is expressed by putting the node back to pending rather than by
	// launching from here: the scheduler's own ready-and-launch path is the only
	// place a node may start, and going through it keeps concurrency, budget and
	// blocking checks applying to a retry exactly as they do to a first attempt.
	//
	// The spend already made is kept. It was really spent, and a retry that
	// hid it would understate the run.
	if s.Escalations > 0 && retries != nil && outcome != nil &&
		retries[nodeID] < s.Escalations && outcome.Verdict.Escalates() {
		retries[nodeID]++
		node.State = plan.StatePending
		s.emit(Event{NodeID: nodeID, Title: node.Title, State: plan.StatePending,
			Detail:  fmt.Sprintf("%s — retrying on a stronger model", outcome.Verdict),
			Elapsed: time.Since(started)})
		return
	}
	// AN ENDING THAT IS EXHAUSTION IS NOT A VERDICT ON THE WORK, and this
	// scheduler used to be the last place that still answered it with a
	// failure. The rule is [Requeued] — the same function resident.Runner.runOne
	// asks — so the one-shot surface and the resident cannot disagree about what
	// a spent clock means; what is local here is only what the record IS. There
	// is no transcript bank on this side, so what a requeued node resumes from
	// is its own row: the turns, the tokens and the result an earlier attempt
	// left on it, all of which survive the requeue untouched.
	//
	// AND IT IS BOUNDED BY exhaustionRequeues, which the resident does not need.
	// Its bank grows with every attempt, so "is there something to resume from"
	// is a question whose answer moves and an attempt that recorded nothing ends
	// the chain. This node's record is written only when an attempt LANDS, so a
	// second abandonment reads exactly what the first one read, and the gate
	// alone would put the node back on the queue forever.
	if _, recorded, requeue := Requeued(err, func() int { return node.Turns }); requeue &&
		retries != nil && retries[nodeID] < exhaustionRequeues {
		retries[nodeID]++
		node.State = plan.StatePending
		s.emit(Event{NodeID: nodeID, Title: node.Title, State: plan.StatePending,
			Detail: fmt.Sprintf("ran out of the room it was given — back on the queue with %d turns recorded",
				recorded),
			Elapsed: time.Since(started)})
		return
	}
	if node.State == plan.StateFailed {
		node.Failure = "produced no result"
		if err != nil {
			node.Failure = err.Error()
		}
		s.emit(Event{NodeID: nodeID, Title: node.Title, State: plan.StateFailed, Detail: node.Failure, Elapsed: time.Since(started)})
		return
	}
	detail := fmt.Sprintf("%d turns, %dk tok", outcome.Turns, node.Tokens/1000)
	if outcome.Decayed > 0 {
		detail += fmt.Sprintf(", %d observations faded", outcome.Decayed)
	}
	if len(outcome.Artifacts) > 0 {
		detail += ", wrote " + strings.Join(outcome.Artifacts, ", ")
	}
	s.emit(Event{NodeID: nodeID, Title: node.Title, State: plan.StateDone, Detail: detail, Elapsed: time.Since(started)})
}

// progressDetail renders one within-node step for a line of terminal output.
// It is the same shape the chat surface's rows carry, said in one line, because
// the two surfaces are reporting the same fact and a reader moving between them
// should not have to learn it twice.
func progressDetail(phase string, done, total int, latest string) string {
	line := phase
	if total > 0 {
		line += fmt.Sprintf(" · %d of %d", done, total)
	} else if done > 0 {
		line += fmt.Sprintf(" · %d", done)
	}
	if latest = strings.TrimSpace(latest); latest != "" {
		line += " · " + latest
	}
	return line
}

func (s *Scheduler) emit(event Event) {
	if s.OnEvent != nil {
		s.OnEvent(event)
	}
}