engine

type Adjacency ¶

type Adjacency struct {
	From Node
	To   []Node
}

Adjacency is a (From Node, To []Node) pair the engine consumes to allocate channels. The full set of adjacencies forms the runtime DAG (with cycle opt-in per node). New validates the set:

• every To node must appear as a From in some adjacency OR be terminal (no children — i.e. an Outlet);
• no two adjacencies share a Node.Name;
• the graph contains no cycle unless the cycle's nodes have AllowCycle: true.

type EmitOption ¶

type EmitOption func(*emitOptions)

EmitOption is the Emit-time option type. Phase 12 added WithRunCapacity for per-run streaming backpressure overrides.

func WithRunCapacity(n int) EmitOption

type Engine ¶

type Engine interface {
	Emit(ctx context.Context, env messages.Envelope, opts ...EmitOption) error
	EmitTo(ctx context.Context, env messages.Envelope, target NodeRef) error
	Fetch(ctx context.Context, opts ...FetchOption) (messages.Envelope, error)
	FetchByRun(ctx context.Context, runID string) (messages.Envelope, error)
	Cancel(ctx context.Context, runID string) (bool, error)
	Run(ctx context.Context) error
	Stop(ctx context.Context) error
	// Topology builds a canonical types.TopologyProjection of the
	// engine's static node graph + live per-edge queue depth (Phase 74
	// / D-114). Identity-mandatory; pure read; safe for N concurrent
	// callers (D-025). See topology.go.
	Topology(ctx context.Context) (types.TopologyProjection, error)
}

Engine is the runtime container — the typed, async, queue-backed graph executor. One concrete implementation in V1 (the in-memory engine); future remote-engine drivers (post-V1) plug behind the same interface via the §4.4 seam pattern.

Lifecycle: New constructs the engine + validates the adjacency set (cycle detection, duplicate-name check, queue-size validation); Run starts one goroutine per node + the dispatcher; Stop joins them all under a deadline. Emit lands at the engine's inlet channel(s); Fetch drains the dispatcher's any-run subqueue.

func New(adjacencies []Adjacency, opts ...Option) (Engine, error)

type FetchOption ¶

type FetchOption func(*fetchOptions)

FetchOption is the Fetch-time option type. Phase 13 will add per-run filtering via FetchByRun (a dedicated method, not an option), but the type exists today for fetch-side knobs that may land later (e.g. FetchTimeout).

type Node ¶

type Node struct {
	// Name is the unique identifier for the node within an engine.
	// New rejects duplicates with ErrDuplicateNodeName.
	Name string
	// Func is invoked by the worker loop on each incoming envelope.
	// nil Func is rejected at New time.
	Func NodeFunc
	// Policy controls validation, timeout, retry, and backoff for
	// each invocation of Func. Zero value = "no policy" (single
	// invocation, no timeout, no retry — Phase 10's bare worker
	// behavior). See policy.go for fields.
	Policy NodePolicy
	// AllowCycle opts this node out of the cycle detector. Set true
	// for legitimate self-loop or controller-loop graphs (e.g. a
	// planner node that emits to itself for the next reasoning step).
	AllowCycle bool
}

Node wraps a typed async function with reliability policy + cycle opt-in. Phase 10 shipped the shape; Phase 11 fills NodePolicy with real fields (Validate / Timeout / Retry / Backoff) and the worker loop now applies them via the reliability shell.

func (n Node) Ref() NodeRef

type NodeContext ¶

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

NodeContext is the per-invocation handle the worker passes to the NodeFunc. Carries the engine reference so the function can Emit / EmitNoWait / EmitChunk / Fetch through the same channel mechanic as external callers. CallSubflow (Phase 14) hangs off this same type.

lastEnv records the incoming envelope for the current invocation so per-invocation operations (EmitChunk's identity propagation, future pause-resume hooks) can see the originating run's quadruple without requiring callers to thread it through manually. The worker loop sets this before invoking Func.

NodeContext is constructed by the worker; callers must not build one directly. The struct's internal fields are unexported.

func (nctx *NodeContext) CallSubflow(ctx context.Context, factory SubflowFactory, parentEnv messages.Envelope) (messages.Envelope, error)

func (nctx *NodeContext) Emit(ctx context.Context, env messages.Envelope) error

func (nctx *NodeContext) EmitChunk(ctx context.Context, frame StreamFrame) error

func (nctx *NodeContext) EmitNoWait(ctx context.Context, env messages.Envelope) error

type NodeFunc ¶

type NodeFunc func(ctx context.Context, in messages.Envelope, nctx *NodeContext) (messages.Envelope, error)

NodeFunc is the unit of computation a Node wraps. It receives the incoming envelope plus a per-invocation NodeContext (engine handle scoped to the running node) and returns the outgoing envelope.

Returning a nil envelope WITH nil error means "no emission on this hop" — the worker drops the result and waits for the next ingress envelope. Returning a non-nil error logs through the audit-redacted logger (Phase 04 wiring) and continues; Phase 11 promotes errors to the structured RunError envelope.

type NodePolicy ¶

type NodePolicy struct {
	// Validate selects which side(s) of the invocation the worker
	// runs ValidateFunc on. Zero value is ValidateNone (no validation
	// — matches the Phase 10 bare-worker behavior).
	Validate ValidateMode
	// TimeoutMS is the per-invocation deadline in milliseconds. 0
	// means "no timeout" — the worker invokes Func with the engine's
	// ctx unchanged. > 0 wraps each invocation in
	// context.WithTimeout(ctx, time.Duration(TimeoutMS) * Millisecond).
	TimeoutMS int
	// MaxRetries is the count of retry attempts AFTER the initial
	// invocation. 0 means "no retries" (the node runs exactly once on
	// failure). Total invocations = MaxRetries + 1.
	MaxRetries int
	// BackoffBase is the first-retry sleep before retry attempt 1.
	// Subsequent retries multiply by BackoffMult, capped at
	// MaxBackoff. 0 means no sleep between retries.
	BackoffBase time.Duration
	// BackoffMult is the multiplier between successive retry sleeps.
	// 0 or 1 means "no growth" (linear retries at BackoffBase). 2 is
	// the canonical exponential value.
	BackoffMult float64
	// MaxBackoff caps the per-retry sleep regardless of BackoffMult.
	// 0 means no cap.
	MaxBackoff time.Duration
	// ValidateFunc is the function pointer the worker calls on input
	// (Validate=Both/In) and/or output (Validate=Both/Out) envelopes.
	// nil with Validate != ValidateNone is a no-op (the worker treats
	// it as "no validator configured" — fail-loud at construction
	// time is the engine's responsibility, not the shell's; Phase 10
	// could harden this if needed).
	ValidateFunc func(messages.Envelope) error
	// RunCapacity (Phase 12) caps the number of pending stream frames
	// (EmitChunk-emitted) for a run originating at this node before
	// EmitChunk blocks. 0 means "use the engine's DefaultQueueSize"
	// (64). Per-run override via WithRunCapacity on the originating
	// Engine.Emit takes precedence over this field.
	//
	// Backpressure is per-run, not per-engine: one run's saturation
	// never pauses other runs (this is the deadlock-prevention
	// guarantee from brief 01 §4).
	RunCapacity int
}

NodePolicy controls per-node reliability semantics. Zero value is "no policy" — Phase 10's bare worker behavior (single invocation, no timeout, no retry, no validation). Construct explicitly for production nodes; the engine never silently applies defaults (per AGENTS.md §5 "Fail loudly").

Concurrent-reuse safe: NodePolicy is a value type. ValidateFunc is a function pointer; the function itself must be safe for concurrent invocation across the same Node.

type NodeRef ¶

type NodeRef struct {
	Name string
}

NodeRef identifies a node by name. Used for per-channel queue overrides and EmitTo's explicit-target form.

type Option ¶

type Option func(*engineConfig)

Option configures an engine at construction.

func WithCancelTTL(d time.Duration) Option

func WithChannelOverride(from, to NodeRef, n int) Option

func WithErrorEmissionToEgress(enabled bool) Option

func WithEventBus(b events.EventBus) Option

func WithQueueSize(n int) Option

func WithRunCancelledHandler(h RunCancelledHandler) Option

func WithRunErrorHandler(h RunErrorHandler) Option

type RunCancelledHandler ¶

type RunCancelledHandler func(ctx context.Context, n RunCancelledNotice)

RunCancelledHandler is the callback the engine fires after a Cancel observed an active run. Same hook pattern as RunErrorHandler: the seam exists for assemblies that host an engine graph to publish a runtime.run_cancelled event on the bus. No production assembly installs one today — `harbor dev` is planner/RunLoop-shaped and boots no engine graph (see docs/notes/sdk-friction-audit.md §3). Tests install recording callbacks.

Best-effort: a panic is recovered; bus errors must not block Cancel from returning.

type RunCancelledNotice ¶

type RunCancelledNotice struct {
	RunID                string
	CancelledAt          time.Time
	DroppedEnvelopeCount int64
}

RunCancelledNotice is the payload the engine hands to a RunCancelledHandler after Cancel(runID). The handler is the seam production wiring uses to publish runtime.run_cancelled on the bus without the engine importing the events package.

CancelledAt is wall-clock; DroppedEnvelopeCount counts the envelopes Cancel drained from channels (step 2 of the four-step propagation). Useful for operators measuring "how loaded was the cancelled run."

type RunError ¶

type RunError struct {
	// RunID identifies the run the failed envelope belonged to.
	// Empty when the envelope had no RunID (Phase 10 allows that;
	// Phase 13 will tighten when FetchByRun arrives).
	RunID string
	// TenantID, UserID, SessionID complete the identity triple from
	// the failing envelope. Used by audit subscribers + slog
	// attribute set.
	TenantID  string
	UserID    string
	SessionID string
	// NodeName is the unique node identifier within the engine.
	NodeName string
	// NodeID is reserved for future stable runtime identifiers; for
	// Phase 11 it mirrors NodeName (engines are single-process and
	// the Name is the stable id).
	NodeID string
	// Code categorises the failure. See RunErrorCode constants.
	Code RunErrorCode
	// Message is a short human-readable summary. The redactor sees
	// it via the bus emit path; do NOT include raw tool args or
	// secrets.
	Message string
	// Cause is the wrapped underlying error (if any). errors.Unwrap
	// follows this; deeper chains compose via the underlying error's
	// Unwrap method.
	Cause error
	// Metadata carries policy-relevant context (e.g. attempt count,
	// timeout in ms, validate side). Bounded; free-form.
	Metadata map[string]any
}

RunError is the structured error envelope emitted on terminal node failure. Carries the full identity quadruple via RunID + the per-invocation context (NodeName, NodeID). Cause carries one level of wrapping; deeper chains use errors.Unwrap on Cause.

Identity propagation: every RunError carries the failing envelope's (TenantID, UserID, SessionID, RunID) so audit logs and bus subscribers can scope by the multi-isolation triple.

func (e *RunError) Error() string

func (e *RunError) ToPayload() map[string]any

func (e *RunError) Unwrap() error

type RunErrorCode ¶

type RunErrorCode string

RunErrorCode categorises the failure mode. Stable across Harbor's runtime — downstream consumers (audit subscribers, Console) match against these constants.

const (
	// CodeNodeTimeout — the node's invocation exceeded NodePolicy.TimeoutMS.
	CodeNodeTimeout RunErrorCode = "node_timeout"
	// CodeNodeException — the node's Func returned a non-nil error
	// or panicked. The most common terminal-failure code.
	CodeNodeException RunErrorCode = "node_exception"
	// CodeRunCancelled — the engine's context (or the per-run cancel
	// flag, Phase 13) was triggered before invocation completed.
	CodeRunCancelled RunErrorCode = "run_cancelled"
	// CodeDeadlineExceeded — the envelope's wall-clock DeadlineAt
	// expired before the worker invoked the node. Distinct from
	// CodeNodeTimeout (per-invocation) — DeadlineExceeded is per-
	// envelope.
	CodeDeadlineExceeded RunErrorCode = "deadline_exceeded"
	// CodeValidationFailed — NodePolicy.ValidateFunc rejected the
	// input or output envelope.
	CodeValidationFailed RunErrorCode = "validation_failed"
)

type RunErrorHandler ¶

type RunErrorHandler func(ctx context.Context, re *RunError)

RunErrorHandler is the callback the engine fires on terminal node failure. The engine's default (when no handler is configured) is a no-op AFTER the engine's slog.Logger has logged the failure — the handler is the seam production wiring uses to route the structured RunError to Phase 04's *telemetry.Logger.Error so the wave-2 BusEmitter adapter publishes a runtime.error event.

Decoupled by design: the engine package does not import internal/telemetry. Callers who want bus-side runtime.error events pass a callback that invokes Logger.Error with the RunError's structured attrs.

type StreamFrame ¶

type StreamFrame struct {
	StreamID string
	Seq      int
	Text     string
	Done     bool
	Meta     map[string]any
}

StreamFrame is a chunked payload tied to a parent run. StreamID defaults to RunID; sub-streams within a run use a custom StreamID. Seq is monotonic per StreamID and is engine-assigned (callers must NOT pre-fill — the engine rejects with ErrSeqProvided).

Per-stream order is preserved as long as a single goroutine emits per StreamID — the dispatcher's per-run subqueue is FIFO. Two goroutines emitting on the same StreamID concurrently will interleave (the engine has no per-StreamID lock); operators who need cross-goroutine ordering must serialize the emit themselves.

type SubflowFactory ¶

type SubflowFactory func() (Engine, error)

SubflowFactory returns a fresh child Engine per call. Caller never reuses subflow engines; cheap construction is the contract per brief 01 §5 ("a subflow is a freshly-built engine that runs to completion for one parent envelope, then Stops").

type ValidateMode ¶

type ValidateMode string

ValidateMode selects which side of a node invocation the worker passes through ValidateFunc.

const (
	// ValidateNone disables validation. The perf escape hatch for hot
	// streaming paths (Phase 12 will lean on this for stream nodes).
	// Zero value of ValidateMode.
	ValidateNone ValidateMode = ""
	// ValidateBoth runs ValidateFunc on the input AND output envelope.
	// The default for production nodes — fail-loud per CLAUDE.md §5.
	ValidateBoth ValidateMode = "both"
	// ValidateIn runs ValidateFunc on the input only.
	ValidateIn ValidateMode = "in"
	// ValidateOut runs ValidateFunc on the output only.
	ValidateOut ValidateMode = "out"
)