steering

type AppliedControl ¶

type AppliedControl struct {
	// Type is the control type that was applied.
	Type ControlType
	// RunID is the run the control targeted (a session may host multiple
	// concurrent runs — the run component disambiguates).
	RunID string
	// AppliedAt is the wall-clock time the side effect was applied,
	// stamped from the RunLoop's Clock.
	AppliedAt time.Time
	// Err is the non-nil error when the side effect failed (e.g. a
	// PRIORITIZE whose task does not exist). A failed apply is still
	// recorded — the history is the audit trail, and a silent drop would
	// violate CLAUDE.md §5 "fail loudly".
	Err error
}

AppliedControl is one entry in a session's applied-control history. It is the runtime's own bookkeeping record — what control was applied to which run, when, and whether the side effect succeeded. The rejected caller payload itself is NOT carried (mirroring ControlRejectedPayload): the history is a low-cardinality audit trail, not a payload archive.

type Clock ¶

type Clock interface {
	Now() time.Time
}

Clock is the minimal time source the inbox uses for the per-event `EnqueuedAt` stamp. Tests inject a controllable clock so no test sleeps for synchronisation (CLAUDE.md §11). Production code uses the real-time `systemClock`.

type ControlEvent ¶

type ControlEvent struct {
	// Type is one of the nine canonical control types. An invalid
	// Type is rejected by Inbox.Enqueue with ErrUnknownControlType.
	Type ControlType
	// Identity is the run quadruple this control targets. It MUST
	// match the Inbox's own quadruple — Enqueue rejects a mismatch
	// (an event for run A must never land on run B's inbox).
	Identity identity.Quadruple
	// CallerScope is the (trust-based at Phase 52) Scope the Protocol
	// edge derived from the submitting caller's JWT. Enqueue runs
	// CheckScope against it.
	CallerScope Scope
	// CallerTenant is the tenant the submitting caller authenticated
	// under. Used by the cross-tenant scope check (RFC §6.3).
	CallerTenant string
	// Payload is the sanitised, bounds-checked control payload. May
	// be nil — a bare CANCEL / PAUSE carries no payload. Enqueue runs
	// ValidatePayload against it.
	Payload map[string]any
	// EventID is the caller-supplied idempotency / correlation key
	// (ULID-shaped, mirrors events.EventID). Optional at Phase 52 —
	// Phase 53's control-history dedupe uses it. Empty is permitted.
	EventID string
	// EnqueuedAt is stamped by Inbox.Enqueue from the Inbox's Clock.
	// Callers MUST NOT pre-fill it; a non-zero value is rejected so
	// the Inbox owns the timeline (mirrors events.Event.Sequence).
	EnqueuedAt time.Time
}

ControlEvent is the canonical steering record (RFC §6.3, brief 02 §2 `ControlEvent`). It is what the Protocol edge constructs from an inbound control request, validates, scope-checks, and enqueues on the run's Inbox. Phase 53 drains these between planner steps and projects the result onto RunContext.Control.

type ControlLifecyclePayload ¶

type ControlLifecyclePayload struct {
	events.SafeSealed
	// Type is the control type that was received / applied.
	Type string
	// Outcome is a stable, low-cardinality classification of the apply
	// result — "received" for control.received, and one of "applied" /
	// "failed" for control.applied. Empty on a control.received event.
	Outcome string
	// Err is a short, redaction-safe description of why the side
	// effect failed, when Outcome == "failed". Empty otherwise. The
	// RunLoop derives this from a sentinel classification, never the
	// raw error message (which may quote caller data).
	Err string
}

ControlLifecyclePayload is the typed payload for control.received and control.applied events. SafePayload by construction: every field is the RunLoop's own bookkeeping — the control Type is one of nine canonical enum values, the Outcome / Err strings are low-cardinality runtime-derived classifications. The caller-controlled control payload itself is NOT carried (mirroring ControlRejectedPayload): these events are a low-cardinality audit trail, not a payload archive.

type ControlRejectedPayload ¶

type ControlRejectedPayload struct {
	events.SafeSealed
	// Type is the control type that was rejected (may be the empty
	// string when the rejection was an unknown / unparsable type).
	Type string
	// Reason is a stable, low-cardinality classification of why the
	// submission was rejected — one of "unknown_type",
	// "payload_invalid", "scope_mismatch", "identity_invalid".
	Reason string
	// CallerScope is the scope the rejected caller presented.
	CallerScope string
}

ControlRejectedPayload is the typed payload for a control.rejected event. SafePayload by construction: every field is the steering edge's own bookkeeping — the control Type is one of nine canonical enum values, the Reason is a fixed sentinel-derived string, the CallerScope is one of three canonical enum values. The rejected payload itself is NOT carried — it may hold caller-controlled data and is exactly what was rejected; persisting it would defeat the rejection.

type ControlType ¶

type ControlType string

ControlType is the string-typed enum of the nine canonical control event types (RFC §6.3 — Settled). The wire strings are the RFC's verbatim uppercase identifiers; the Protocol projection (Phase 54) accepts exactly these.

const (
	// ControlInjectContext — append operator-supplied context to the
	// run's trajectory; visible on the planner's next step.
	ControlInjectContext ControlType = "INJECT_CONTEXT"
	// ControlRedirect — rewrite the run's goal. Requires the user
	// (the agent's owner) — RFC §6.3.
	ControlRedirect ControlType = "REDIRECT"
	// ControlCancel — cancel the run (hard or soft; the soft/hard
	// distinction is a Phase 53 payload concern).
	ControlCancel ControlType = "CANCEL"
	// ControlPrioritize — change the run's task priority. Requires
	// admin — RFC §6.3.
	ControlPrioritize ControlType = "PRIORITIZE"
	// ControlPause — pause the run at the next planner-step boundary.
	// Phase 53 wires this onto the unified pause/resume primitive.
	ControlPause ControlType = "PAUSE"
	// ControlResume — resume a paused run. Phase 53 wires this onto
	// the unified pause/resume primitive.
	ControlResume ControlType = "RESUME"
	// ControlApprove — approve a HITL-gated step. Phase 53 wires this
	// onto the unified pause/resume primitive (advance a pause).
	ControlApprove ControlType = "APPROVE"
	// ControlReject — reject a HITL-gated step. Phase 53 wires this
	// onto the unified pause/resume primitive.
	ControlReject ControlType = "REJECT"
	// ControlUserMessage — inject a user-authored message into the
	// run; visible on the planner's next step.
	ControlUserMessage ControlType = "USER_MESSAGE"
)

func ControlTypes() []ControlType

type Inbox ¶

type Inbox struct {
	// contains filtered or unexported fields
}

Inbox is the per-run steering inbox. The Runtime owns it; planners never touch it (RFC §6.3 — "the Runtime owns the inbox"; planners observe RunContext.Control only). It is a FIFO enqueue + drain surface, identity-scoped to exactly one run quadruple.

An Inbox is concurrent-safe: N Protocol-edge goroutines may Enqueue while the run loop Drains, all against one Inbox — the queue is mutex-guarded (D-025). Per-run state never leaks across runs because each Inbox holds only its own run's events and is keyed by its own quadruple.

Construct an Inbox via Registry.Open; do not construct one directly.

func (in *Inbox) Drain() ([]ControlEvent, error)

func (in *Inbox) Enqueue(ev ControlEvent) error

func (in *Inbox) Identity() identity.Quadruple

func (in *Inbox) Len() int

func (in *Inbox) WaitForEvent(ctx context.Context) error

type Option ¶

type Option func(*Registry)

Option configures a Registry at construction time.

func WithClock(c Clock) Option

type Registry ¶

type Registry struct {
	// contains filtered or unexported fields
}

Registry is the process-wide owner of per-run steering inboxes. The Runtime builds ONE Registry per process and shares it across every run; it mints an Inbox when a run starts (Open), hands the Inbox to the Protocol edge (Lookup) and the run loop, and retires the Inbox when the run ends (Retire).

Registry is a compiled artifact (D-025): immutable after construction, with the run→inbox map behind a documented-invariant sync.Mutex. One Registry is safe to share across N concurrent goroutines — concurrent_test.go pins N≥100 under -race. Per-run state never leaks across runs: each run's events live on its own Inbox, keyed by its own identity quadruple.

func NewRegistry(opts ...Option) *Registry

func (r *Registry) Len() int

func (r *Registry) Lookup(q identity.Quadruple) (*Inbox, error)

func (r *Registry) Open(q identity.Quadruple) (*Inbox, error)

func (r *Registry) Retire(q identity.Quadruple) error

type RunLoop ¶

type RunLoop struct {
	// contains filtered or unexported fields
}

RunLoop is Harbor's per-run planner-step loop — the runtime component that drives a planner.Planner to a terminal planner.Finish decision, draining the per-run steering Inbox between steps and routing pause decisions through the unified pauseresume.Coordinator.

Why this is the steering wiring ¶

Phase 52 shipped the steering primitive (the inbox, the nine-type taxonomy, ValidatePayload, CheckScope, the Registry). Phase 50 shipped the pause/resume primitive (the Coordinator). Neither did anything by itself — there was no run loop to drain the inbox or to route a RequestPause decision through the Coordinator. RunLoop IS that loop. It is the §13 first consumer of BOTH primitives, landing in the same wave (Wave 9, Stage 3) per CLAUDE.md §13 + D-067 §4 + D-070 §5.

The loop (brief 02 §4) ¶

Per run, RunLoop owns a tight loop:

Open the run's Inbox on the Registry
for step := 0; step < MaxSteps; step++ {
    drain the Inbox            -- once, at the step boundary
    apply each control event   -- CANCEL / PAUSE / REDIRECT / ... side effects
    project onto RunContext.Control  -- the planner sees ONLY this
    if a RESUME/APPROVE advanced a pause: clear the pause, continue
    if a REJECT advanced a pause: terminate Finish{ConstraintsConflict}
    decision := Planner.Next(ctx, runContext)
    switch decision {
        RequestPause -> Coordinator.Request; block; re-enter on RESUME/APPROVE
        Finish       -> Retire the Inbox; return
        other        -> (decision execution is a later-phase concern;
                         Phase 53 records the step and re-enters)
    }
}
Retire the Inbox  -- always, even on error

The full applyEvent treatment happens exactly ONCE per step boundary. While a decision execution is in flight the loop ALSO drains the inbox (D-192), but consumes ONLY approval-bridge-eligible APPROVE / REJECT controls there (the D-097 gate bridge — without the mid-step drain an approval-gated tool deadlocks the run: RunGuarded parks until ResolveApproval, whose only production caller is this loop's drain). Every other control drained mid-step is deferred verbatim to the next boundary, preserving brief 02 §6's step-boundary semantics. The planner observes the result via RunContext.Control; it never touches the Inbox (brief 02 §5 sharp-edge #2).

Concurrent reuse (D-025) ¶

RunLoop is a compiled artifact: every field is set once at construction (the Registry, the Coordinator, the applier's dependencies, the control-history ring, the clock — all immutable after NewRunLoop returns). There is NO per-run state on the struct: Run reads its run-specific data from ctx + the RunSpec argument, and the per-step accumulator (stepControl) lives on the run's own goroutine stack. One RunLoop is safe to share across N concurrent goroutines; concurrent_test.go pins N≥100 under -race.

func NewRunLoop(reg *Registry, coord pauseresume.Coordinator, opts ...RunLoopOption) (*RunLoop, error)

func (rl *RunLoop) ControlHistory(sessionID string) []AppliedControl

func (rl *RunLoop) Run(ctx context.Context, spec RunSpec) (planner.Finish, error)

type RunLoopOption ¶

type RunLoopOption func(*runLoopConfig)

RunLoopOption configures a RunLoop at construction. Options are applied in order; later options override earlier ones for the same field.

func WithApprovalGates(gates map[string]*approval.ApprovalGate) RunLoopOption

func WithHardCancelHook(fn func(ctx context.Context, runID string) error) RunLoopOption

func WithMaxControlHistory(n int) RunLoopOption

func WithPauseStatusRecheckInterval(d time.Duration) RunLoopOption

func WithRunLoopBus(b events.EventBus) RunLoopOption

func WithRunLoopClock(c Clock) RunLoopOption

func WithRunLoopLogger(l *slog.Logger) RunLoopOption

func WithTaskRegistry(tr tasks.TaskRegistry) RunLoopOption

type RunSpec ¶

type RunSpec struct {
	// Planner is the swappable reasoning policy the loop drives. Nil
	// fails loud with ErrNoPlanner.
	Planner planner.Planner
	// Base is the run's RunContext template. RunLoop refreshes the
	// per-step fields (Control, Goal) on a copy each step; the planner
	// receives a fresh RunContext per Next call (Phase 42 contract).
	// Base.Quadruple is the run's identity — its triple is validated
	// identity-mandatory before the loop starts.
	Base planner.RunContext
	// TaskID is the run's task. Optional — when set, a PRIORITIZE
	// control event targets it; when empty, a PRIORITIZE fails loud.
	TaskID tasks.TaskID
	// MaxSteps caps the planner-step count. ≤ 0 ⇒ DefaultMaxSteps.
	MaxSteps int

	// ToolExecutor dispatches the planner's non-Finish, non-RequestPause
	// decisions (CallTool, CallParallel, SpawnTask, AwaitTask). Phase
	// 83i (D-152): when nil, the runloop's default case logs and
	// appends an empty-observation step (the Phase 53 behaviour) so
	// existing pause/steering tests still drive deterministic finishes.
	// In production the dev binary wires a real executor backed by the
	// tool catalog so the planner's CallTool decisions actually run.
	ToolExecutor ToolExecutor

	// Compression is the optional trajectory-compression runner
	// (Phase 111e — D-202; the §13 first call site of
	// planner.CompressionRunner.MaybeCompress). When non-nil AND the
	// run's Base.Budget.TokenBudget > 0, the runloop invokes
	// MaybeCompress at each step boundary (after the control drain +
	// projection, before Planner.Next) so an over-budget trajectory is
	// compacted into Trajectory.Summary BEFORE the next prompt build —
	// the React prompt builder's `Summary != nil` branch then renders
	// the five-field summary instead of the per-step history and the
	// prompt shrinks. Nil (or a zero TokenBudget) is byte-identical to
	// the pre-111e behaviour: no estimate, no summariser, no events.
	//
	// One compression per run at V1.1.x: the runner is idempotent on
	// `Trajectory.Summary != nil` (the documented scope fence — RFC
	// §6.5; re-compaction cadence is the recorded D-202 follow-up).
	// A MaybeCompress error fails the run LOUDLY (the runner already
	// emitted trajectory.compression_failed) — never a silent
	// fall-through that pretends compression happened.
	Compression *planner.CompressionRunner

	// OnToolDispatched is the optional per-run hook the runloop
	// invokes after the ToolExecutor returns WITHOUT ERROR (Phase 83m
	// item 7). The dev binary wires it to
	// `taskReg.IncrementToolCount(ctx, taskID)` so the Console Tasks
	// page reflects the per-task tool-dispatch count. A nil hook is
	// the legacy / test path (no counter wired); a hook that errors
	// fails the run loud — silent degradation of an observability
	// counter is forbidden per §13 (the counter is an integrity
	// surface, not a best-effort log line).
	//
	// The runloop calls the hook for every successful executor
	// dispatch — CallTool today, CallParallel + SpawnTask + AwaitTask
	// once those executor shapes land. A dispatch that the executor
	// reports as failed (the executor's own error path) does NOT
	// invoke the hook; the planner's repair / re-plan flow records
	// the failure on the trajectory and the counter stays put.
	OnToolDispatched func(ctx context.Context) error
}

RunSpec is the per-run input to RunLoop.Run. ALL run-specific state lives here + ctx — never on the RunLoop struct (D-025).

type Scope ¶

type Scope string

Scope is the privilege tier a steering caller presents. It is a trust-based claim in Phase 52 — cryptographic verification arrives with Protocol auth (Phase 61), mirroring the events package's Admin-claim posture (RFC §6.3 + events.Filter.Admin). The Protocol edge derives the Scope from the caller's JWT scope claim before calling CheckScope.

const (
	// ScopeSessionUser — a user authenticated into the session the
	// run belongs to. The weakest steering scope. Sufficient for
	// INJECT_CONTEXT and USER_MESSAGE.
	ScopeSessionUser Scope = "session_user"
	// ScopeOwnerUser — the user who owns the agent / run (the
	// "originating user"). Sufficient for REDIRECT and the
	// originating-user-or-admin controls (CANCEL, PAUSE, RESUME,
	// APPROVE, REJECT).
	ScopeOwnerUser Scope = "owner_user"
	// ScopeAdmin — an administrator. Sufficient for every control
	// type, including PRIORITIZE (admin-only) and any cross-tenant
	// steering.
	ScopeAdmin Scope = "admin"
)

func RequiredScope(t ControlType) (Scope, bool)

type ToolExecutor ¶

type ToolExecutor interface {
	// ExecuteDecision dispatches `decision` and returns BOTH the raw
	// observation (preserved for inspect-runs / audit) AND the
	// projection the next prompt sees (`llmObservation`, the D-026
	// heavy-content-discipline projection: a small summary +
	// ArtifactRef when the raw result is over the heavy threshold,
	// or just == raw when the result is small enough to inline).
	//
	// The runloop appends a trajectory.Step{Action: decision,
	// Observation: raw, LLMObservation: projection} so the planner's
	// renderer (Phase 46 / D-055) sees only the projection.
	//
	// `rc` is the per-step RunContext (identity, ToolContext, etc.).
	// ctx is the per-step ctx; the executor MUST honour cancellation.
	ExecuteDecision(ctx context.Context, rc planner.RunContext, decision planner.Decision) (observation, llmObservation any, err error)
}

ToolExecutor is the runtime-side dispatch surface the RunLoop calls when the planner returns a non-Finish, non-RequestPause decision (CallTool, CallParallel, SpawnTask, AwaitTask). The executor:

• Looks up the tool descriptor by name.
• Invokes it under the run's identity-scoped ctx.
• Returns a planner-readable observation (the runtime appends it onto trajectory.Step.Observation for the planner's next step).

Phase 83i (D-152) introduces this seam so the dev binary can wire a real `tools.ToolCatalog`-backed executor; before 83i the runloop's `default:` case dropped every CallTool on the floor (Phase 53's punted scope), which made multi-step ReAct structurally broken against real LLMs because the planner saw the same trajectory on every step.

An executor that does not support a given decision shape returns ErrDecisionShapeUnsupported with a message naming the unsupported shape — the runloop surfaces this as the step's observation so the planner can choose a different path (repair, finish, alternative tool).