router

package
v0.2.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 20, 2026 License: Apache-2.0 Imports: 10 Imported by: 0

Documentation

Index

Constants

View Source 
const MetadataKeyResumeReverse = "zenflow-resume-reverse"

MetadataKeyResumeReverse is a sentinel Metadata key set on reverse Messages produced by Executor.runResume. When Router.Send encounters a closed target AND observes this key, it does NOT cascade into a second ResumeStep invocation - it emits DropReasonTargetTerminal instead.

Variables

View Source 
var (
	// ErrResumeShutdown signals the workflow ctx was cancelled mid-resume.
	// Router maps to DropReasonResumeShutdown.
	ErrResumeShutdown = errors.New("zenflow: resume cancelled by workflow shutdown")

	// ErrModelResolverMissing indicates the saved transcript references a
	// model identifier that does not match the Executor's default runner
	// model and no ModelResolver was configured to resolve it. Router
	// maps to DropReasonTargetTerminal (no dedicated reason - this is an
	// operator configuration error).
	ErrModelResolverMissing = errors.New("zenflow: resume: saved transcript model differs from executor model and no ModelResolver configured (use WithModelResolver)")

	// ErrModelResolverError indicates a ModelResolver WAS configured but
	// returned an error (or nil model with no error indicator) when
	// asked to resolve the saved transcript's model identifier. Distinct
	// from ErrModelResolverMissing so operators can tell "no resolver
	// installed" from "resolver ran and failed" (VA-6b). Router maps to
	// DropReasonTargetTerminal.
	ErrModelResolverError = errors.New("zenflow: resume: ModelResolver returned error")

	// ErrMailboxFullOnResume indicates a queued resume attempt was
	// rejected because the active resume's mailbox was already at its
	// configured cap. G2 - prevents false EventResumeQueued emission.
	// Router maps to DropReasonMailboxFull.
	ErrMailboxFullOnResume = errors.New("zenflow: resume: active resume mailbox full")
)

ErrResumeShutdown and the other sentinel errors below are returned by Resumer.ResumeStep for the Router hook to map into DropReason values. Kept here (rather than in transcript.go) so the Router can import/route them without pulling executor internals.

View Source 
var ErrMailboxFull = errors.New("zenflow: mailbox full")

ErrMailboxFull signals the bounded store rejected an Append because the per-step queue is at capacity. The Router maps this to DropReasonMailboxFull when emitting OnDrop.

Functions

func DropReasonStrings 

func DropReasonStrings() map[DropReason]string

DropReasonStrings returns a defensive copy of the canonical map of DropReason values to their wire-format strings. Production code should call DropReason.String instead; this accessor exists for tests that need to enumerate every reason without reaching into unexported state. Stable.

func MailboxLen 

func MailboxLen(store MailboxStore, stepID string) (unread, total int)

MailboxLen returns (unread, total) for the given stepID, falling back to len(Unread) when the underlying store does not implement LenAware. This is the canonical entry point used by the executor's invariant-check snapshot (stepLock + Len + state + pending-senders are all read in the same critical section).

func MessageIDs 

func MessageIDs(msgs []Message) []string

MessageIDs returns the MessageID of every message in msgs. Used to translate the []Message shape returned by MailboxStore.Unread into the []string ids accepted by MailboxStore.MarkRead (// F4 - CAS dedup contract).

Types

type BoundedInMemoryStore 

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

BoundedInMemoryStore wraps InMemoryMailboxStore with a hard per-step cap on queued unread messages. When Append would exceed the cap, the new message is rejected and Append returns ("", ErrMailboxFull) so the caller (Router.Send) can attribute the drop to "mailbox-full". Named field (not pointer embedding) avoids promoting the full InMemoryMailboxStore surface (Seal/Closed/Len) onto this type - only the four MailboxStore methods are delegated explicitly below.

func NewBoundedInMemoryStore 

func NewBoundedInMemoryStore(maxSize int) *BoundedInMemoryStore

NewBoundedInMemoryStore returns a BoundedInMemoryStore wrapping a fresh InMemoryMailboxStore. The maxSize parameter is the per-step queue cap: when an Append would exceed it, the store returns ErrMailboxFull and the Router maps that to DropReasonMailboxFull. FOOTGUN: A non-positive maxSize (<=0) disables the cap entirely, making this store effectively unbounded - equivalent to NewInMemoryMailboxStore but with a wrapper that adds no protection. Callers that want a true cap MUST pass a positive value. If you want unbounded behavior, prefer NewInMemoryMailboxStore directly so the intent is explicit at the call site. Stable.

func (*BoundedInMemoryStore) Append 

func (b *BoundedInMemoryStore) Append(stepID string, msg Message) (string, error)

Append enforces the max-size cap. On overflow returns ("", ErrMailboxFull) - the Router observes this and emits a DropReasonMailboxFull drop.

func (*BoundedInMemoryStore) Close 

func (b *BoundedInMemoryStore) Close(stepID string)

Close delegates to the inner store. Satisfies MailboxStore.

func (*BoundedInMemoryStore) Inner 

func (b *BoundedInMemoryStore) Inner() *InMemoryMailboxStore

Inner returns the wrapped InMemoryMailboxStore. Tests use this to assert that NewBoundedInMemoryStore initialised the inner store. Stable.

func (*BoundedInMemoryStore) MarkRead 

func (b *BoundedInMemoryStore) MarkRead(stepID string, ids []string) []string

MarkRead delegates to the inner store. Satisfies MailboxStore.

func (*BoundedInMemoryStore) MaxSize 

func (b *BoundedInMemoryStore) MaxSize() int

MaxSize returns the configured per-step queue cap. Tests use this to verify constructor wiring without reaching into unexported state. Stable.

func (*BoundedInMemoryStore) Unread 

func (b *BoundedInMemoryStore) Unread(stepID string) []Message

Unread delegates to the inner store. Satisfies MailboxStore.

type ChanWakeTarget 

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

ChanWakeTarget is the production EngineWakeTarget: it wraps a buffered chan struct{} of capacity 1 (matching the AgentRunner wake contract). Sends are non-blocking; if the channel already holds a pending wake, the duplicate is dropped - a single wake suffices to flush the entire mailbox.

func (*ChanWakeTarget) SignalWake 

func (t *ChanWakeTarget) SignalWake()

SignalWake implements EngineWakeTarget.

type ClosedAware 

type ClosedAware interface {
	Closed(stepID string) bool
}

ClosedAware is an optional MailboxStore extension implemented by stores that expose their per-step closed flag. The router uses it to surface "mailbox-closed-by-finalize" drops when a Close races a concurrent Send.

type DeliveryEngine 

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

DeliveryEngine is the per-run goroutine that drives mailbox-based message delivery. It periodically polls each active step's mailbox and signals the step's wake channel when the agent is StepIdle and has unread messages waiting. Engine never modifies the mailbox or the agent state - it is purely an observer + wake signaller. Lifecycle: one engine per workflow run. Started after the executor has wired up its router/mailbox/registry; stopped via context cancellation when Run completes or aborts. Tick cadence: 500ms by default, overridable via WithTickInterval for tests. The tradeoff is delivery latency vs. CPU overhead: 500ms is well below human-perceptible response time and far above any reasonable tick cost.

func NewDeliveryEngine 

func NewDeliveryEngine(source EngineActiveStepsSource, mailbox MailboxStore, registry EngineWakeRegistry, opts ...EngineOption) *DeliveryEngine

NewDeliveryEngine constructs a DeliveryEngine with the supplied source, mailbox, and wake registry. None may be nil - a nil dependency would silently no-op an entire role and mask wiring bugs.

func (*DeliveryEngine) PollOne 

func (e *DeliveryEngine) PollOne(stepID string)

PollOne triggers a single poll cycle for stepID, wiring through the usual mailbox / wake / step-locker pipeline. Exposed for tests that need to drive a specific stepID without spinning up Start. Stable.

func (*DeliveryEngine) Start 

func (e *DeliveryEngine) Start(ctx context.Context) (done <-chan struct{})

Start spawns the engine goroutine. Returns immediately; the goroutine runs until ctx is cancelled. Callers may rely on a separate synchronization signal (channel close, WaitGroup) to confirm shutdown when needed; for unit tests, the goroutine exits on the next tick or ctx-done select fire after cancellation. done returns a channel that closes once the engine goroutine has exited. This is the supported way to wait for shutdown - pollers, fakes, and tests should select on done rather than racing on internal state.

func (*DeliveryEngine) TickInterval 

func (e *DeliveryEngine) TickInterval() time.Duration

TickInterval reports the engine's configured tick cadence. Tests use this to verify WithEngineTickInterval / default-fallback wiring. Stable.

type DropError 

type DropError struct {
	Reason DropReason
}

DropError is the typed error returned by Router.Send when delivery is rejected. Callers that need to act on the specific drop reason (e.g. coord_tools' forward_to_agent appending a list of valid step IDs only on DropReasonUnknownStep) should use errors.As to extract the *DropError instead of substring-matching on err.Error. 

var de *zenflow.DropError
if errors.As(err, &de) && de.Reason == zenflow.DropReasonUnknownStep {

// ... 

}

Error returns the canonical "dropped: <reason>" string so existing consumers that pass err.Error through verbatim continue to work. Stable.

func (*DropError) Error 

func (e *DropError) Error() string

Error implements the error interface.

type DropEvent 

type DropEvent struct {
	StepID string
	Msg    Message
	Reason DropReason
}

DropEvent describes a single message that was discarded by the router without ever being appended to the target step's mailbox. Stable. It is the payload the router hands to its OnDrop callback so the executor can emit one EventMessageDropped per drop. Reason is the typed DropReason. The legacy string field is preserved via Reason.String for subscribers that read Event.Data["reason"]. The router does NOT emit EventMessageDropped itself (zenflow events require RunID + ProgressSink wiring); it instead hands a typed payload to the executor via OnDrop, which translates to the event.

type DropReason 

type DropReason int

DropReason is the typed enumeration of reasons a router message can be dropped without ever reaching the target agent. Stable. Per the "no silent drops" invariant: every drop emits exactly one EventMessageDropped carrying one of these reasons. String returns the canonical wire-format value used in Event.Data["reason"] for backward-compat with subscribers reading the pre-typed reasons. 

const (
	// DropReasonUnspecified is the zero value; never emitted in practice.
	DropReasonUnspecified DropReason = iota
	// DropReasonWorkflowCancelled - workflow ctx cancelled or abort fired
	// before the message could be delivered to the target's LLM context.
	DropReasonWorkflowCancelled
	// DropReasonTargetTerminal - Send to a step whose mailbox was closed
	// (target reached a terminal lifecycle state).
	DropReasonTargetTerminal
	// DropReasonUnknownStep - Send to a stepID that was never registered
	// and has no pending senders.
	DropReasonUnknownStep
	// DropReasonMailboxClosedByFinalize - mailbox raced with a concurrent
	// close; the closed flag won.
	DropReasonMailboxClosedByFinalize
	// DropReasonMaxWakeCycles - wake-loop hit the maxWakeCycles cap with
	// messages still pending; remainder drained as drops.
	DropReasonMaxWakeCycles
	// DropReasonHoldTimeout - the executor's hold-timeout
	// fired before the 3-invariant termination rule could converge.
	// Any messages still buffered in the mailbox are emitted with this
	// reason and the step is force-terminated.
	DropReasonHoldTimeout
	// DropReasonMailboxFull - the bounded in-memory mailbox is at the
	// MaxMailboxSize cap configured via WithMaxMailboxSize. The newest
	// message is rejected (oldest-wins fairness).
	DropReasonMailboxFull
	// resume-mechanism drop reasons. Emitted by
	// Router.Send when a resume attempt on a terminated step cannot
	// proceed.
	// DropReasonNoTranscript - target mailbox was closed AND the
	// executor's TranscriptStore has no saved transcript for the step.
	// Typically observed for steps that ran before, or for
	// steps whose transcript was explicitly deleted.
	DropReasonNoTranscript
	// DropReasonTranscriptTooLarge - the saved transcript exceeds the
	// configured cap (WithMaxTranscriptMessages / WithMaxTranscriptBytes),
	// so a resume would exceed the size bound. Natural cycle-detection
	// bound.
	DropReasonTranscriptTooLarge
	// DropReasonResumeShutdown - workflow ctx was cancelled mid-resume;
	// the in-flight resume goroutine exited early. Surfaced by
	// Executor.ResumeStep when ctx.Done fires between Load and the
	// AgentRunner's final assistant response.
	DropReasonResumeShutdown
	// DropReasonResolverError - a configured ModelResolver was consulted
	// for a saved-transcript model identifier and returned an error.
	// Distinct from DropReasonTargetTerminal so operators can tell
	// "resolver infrastructure failure" from generic terminal drops.
	DropReasonResolverError
)

func (DropReason) String 

func (r DropReason) String() string

String returns the canonical wire-format string for the reason. Stable. These values are stable and used as Event.Data["reason"] payloads. Unknown values (out-of-range integers) fall back to "unspecified" so a DropEvent emitted by an old binary against a new enum still has a useful payload.

type EngineActiveStepsSource 

type EngineActiveStepsSource interface {
	// ActiveSteps returns step IDs currently executing under this run.
	// The slice is a snapshot; subsequent ticks call again.
	ActiveSteps() []string
	// AgentState returns the *goai.AgentState for stepID, or nil if the
	// step has not been registered (or was unregistered after
	// completion).
	AgentState(stepID string) *goai.AgentState
}

EngineActiveStepsSource is the minimal subset of *Executor that the DeliveryEngine reads. Defined as an interface so tests can drop in a fake without standing up an Executor + Workflow + Storage.

type EngineClock 

type EngineClock interface {
	Tick(d time.Duration) <-chan time.Time
	Stop()
}

EngineClock abstracts time.Tick so tests can drive ticks deterministically. Production uses RealClock; tests use fakeClock. Tick returns a receive-only chan that yields once per tick boundary. The chan must remain valid for the engine's lifetime; the engine drains it once per loop iteration.

type EngineOption 

type EngineOption func(*DeliveryEngine)

EngineOption configures a DeliveryEngine.

func WithEngineClock 

func WithEngineClock(c EngineClock) EngineOption

WithEngineClock substitutes the engine's tick source. Tests pass a fakeClock to drive ticks deterministically; production code does not need this option.

func WithEngineTickInterval 

func WithEngineTickInterval(d time.Duration) EngineOption

WithEngineTickInterval overrides the default 500ms tick cadence. A non-positive value reverts to the default. Provided primarily for tests that want a fast tick - production callers should rely on the default.

func WithStepLocker 

func WithStepLocker(l EngineStepLocker) EngineOption

WithStepLocker wires the per-step RWMutex acquirer into the engine so the poll loop can satisfy the C5a "read-then-wake" atomicity invariant. When unset, the engine polls without stepLock - see EngineStepLocker godoc for why this is safe in test contexts.

type EngineStepLocker 

type EngineStepLocker interface {
	AcquireStepLock(stepID string) *sync.RWMutex
}

EngineStepLocker is the optional interface the engine uses to acquire a per-step RWMutex for the C5a "read-then-wake" atomicity invariant to enforce read-then-wake atomicity. The poller's Observe+SignalWake sequence MUST run under stepLock.RLock so that a concurrent Run-return defer (which takes the write-lock and calls SetTerminal) cannot transition the state between the read and the wake send. When the engine's source implements this interface, the poll loop wraps each step's Observe+wake sequence in RLock/RUnlock; when it does not, the engine falls back to lock-free polling (acceptable for tests and pre-stepLock callers - a spurious wake against a freshly-terminated step is harmless because the wake channel has cap 1 and the runner's defer already unregistered the wake target).

type EngineWakeRegistry 

type EngineWakeRegistry interface {
	WakeTarget(stepID string) EngineWakeTarget
}

EngineWakeRegistry is the lookup the engine uses to find the wake target for a given stepID. Returns nil when no target is registered (e.g. the step has not yet been admitted into the executor's mailbox path or was already unregistered at end-of-step).

type EngineWakeTarget 

type EngineWakeTarget interface {
	// SignalWake is non-blocking: if the wake channel already has a
	// pending signal, the call is a no-op (cap-1 buffer semantics).
	SignalWake()
}

EngineWakeTarget is the per-step wake handle the engine signals. In production each *AgentRunner exposes its Wake channel via a small adapter. The interface keeps tests independent of AgentRunner's surface.

func NewChanWakeTarget 

func NewChanWakeTarget(ch chan struct{}) EngineWakeTarget

NewChanWakeTarget wraps an existing wake channel. The channel MUST be buffered with cap >= 1; a cap-0 channel would cause SignalWake to drop signals deterministically when the agent is mid-LLM-call (no reader yet) and break the "wake at idle" guarantee.

type InMemoryMailboxStore 

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

InMemoryMailboxStore is the default MailboxStore: an in-process, per-stepID FIFO queue protected by a single mutex. Suitable for the process-local zenflow runtime. For multi-process workflows a persistent backend (sqlite, redis) would replace this. Stable. MessageIDs are monotonic counters formatted "m<N>", scoped per store instance (not per stepID). The format is opaque to consumers - only equality is required for MarkRead's CAS dedup.

func NewInMemoryMailboxStore 

func NewInMemoryMailboxStore() *InMemoryMailboxStore

NewInMemoryMailboxStore returns a ready-to-use in-memory mailbox. Stable.

func (*InMemoryMailboxStore) Append 

func (s *InMemoryMailboxStore) Append(stepID string, msg Message) (string, error)

Append implements MailboxStore. Generates a fresh MessageID, stamps it on the stored copy of msg, and returns it. On a closed step Append is a no-op and returns ("", nil).

func (*InMemoryMailboxStore) Close 

func (s *InMemoryMailboxStore) Close(stepID string)

Close implements MailboxStore. Hard-deletes pending messages and marks the step as closed (subsequent Appends are silently dropped, subsequent Unreads return nil). Equivalent to a hard delete when invoked after Seal. readCnt / totalCnt are also dropped because they are stats counters scoped to a step's lifetime; once Close marks the step's terminal state, no further Append/MarkRead can happen for that stepID, and the closed flag (the only post-Close-readable state) remains set. Without this cleanup the maps grew unbounded in long-running processes that ran many short-lived steps.

func (*InMemoryMailboxStore) Closed 

func (s *InMemoryMailboxStore) Closed(stepID string) bool

Closed reports whether stepID's mailbox has been closed. Implements the optional ClosedAware interface so the router can detect close-during-send races.

func (*InMemoryMailboxStore) Len 

func (s *InMemoryMailboxStore) Len(stepID string) (unread, total int)

Len reports (unread, total) for stepID. Implements LenAware (plan §4.2 #1). unread = len of the in-memory queue; total = lifetime Appends (whether or not later MarkRead'd) for stepID.

func (*InMemoryMailboxStore) MarkRead 

func (s *InMemoryMailboxStore) MarkRead(stepID string, ids []string) []string

MarkRead implements MailboxStore. Removes any queue entries whose MessageID matches one of ids. Returns the subset of ids that were already marked read on a prior call (CAS dedup): a normal drainer sees an empty slice; a duplicate-drain race surfaces here instead of silently consuming twice. Idempotent.

func (*InMemoryMailboxStore) Seal 

func (s *InMemoryMailboxStore) Seal(stepID string)

Seal is the soft-close variant: marks the step terminal so subsequent Appends are silently dropped, BUT existing unread messages remain readable until drained by the poller. Used when a step reaches terminal state and the poller is still flushing. After Seal, Append is a no-op (matching Close), but Unread continues to return any messages that were enqueued before Seal. The executor invokes Close (hard delete) only after the post-Seal flush completes. Idempotent: re-Sealing an already-Sealed (or Closed) step is a no-op.

func (*InMemoryMailboxStore) Unread 

func (s *InMemoryMailboxStore) Unread(stepID string) []Message

Unread implements MailboxStore. Returns a copy so callers may iterate without holding the lock.

type LenAware 

type LenAware interface {
	Len(stepID string) (unread, total int)
}

LenAware is an optional MailboxStore extension that exposes the queue length. Returns (unread, total) message counts for stepID. unread is the number of messages currently in the unread queue (i.e. visible to a subsequent Unread call); total is unread + already-MarkRead'd count. Stores that do not retain MarkRead'd messages report total == unread.

type MailboxStore 

type MailboxStore interface {
	// Append enqueues msg into the per-step mailbox identified by stepID.
	// The store assigns a fresh MessageID and returns it; the input msg
	// is not mutated. Append on a closed step is a no-op (drop silently)
	// and returns ("", nil) - callers (Router.Send) are strictly
	// non-blocking and never expect failure.
	// Returning the ID lets producers correlate individual messages
	// with later MarkRead CAS results.
	Append(stepID string, msg Message) (id string, err error)

	// Unread returns a snapshot of all messages currently buffered for
	// stepID, each carrying the MessageID assigned by Append. The store
	// does NOT mark them read; callers invoke MarkRead with the IDs
	// after they have been delivered to the agent. The two-step
	// (Unread → MarkRead) shape allows the poller to confirm successful
	// dispatch before clearing the buffer, preserving at-least-once
	// semantics if delivery fails mid-flight.
	Unread(stepID string) []Message

	// MarkRead removes the messages identified by ids from stepID's
	// buffer. The returned slice contains the subset of ids that had
	// already been marked read on a previous call (CAS dedup): a
	// well-behaved drainer always sees an empty slice; a duplicate-drain
	// race surfaces here instead of silently consuming twice.
	// MarkRead is idempotent: re-marking IDs that were already read
	// does not panic; they simply appear in alreadyRead.
	MarkRead(stepID string, ids []string) (alreadyRead []string)

	// Close drops all pending messages for stepID and marks the step as
	// closed. Subsequent Appends are silently dropped. Subsequent Unreads
	// return nil.
	Close(stepID string)
}

MailboxStore is a per-step message store with at-least-once delivery semantics. Messages persist until explicitly marked read or until Close is called for the step. Stable. the MailboxStore is the only delivery path on the Router. Wire a store via Router.SetMailbox before any Send. Implementations MUST be safe for concurrent Append + Unread + MarkRead + Close. The reference implementation is InMemoryMailboxStore.

type MapWakeRegistry 

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

MapWakeRegistry is a minimal in-memory EngineWakeRegistry. The Executor populates it as steps start and clears entries as they end.

func NewWakeRegistry 

func NewWakeRegistry() *MapWakeRegistry

NewWakeRegistry returns an empty registry safe for concurrent Register / Unregister / WakeTarget calls.

func (*MapWakeRegistry) Register 

func (r *MapWakeRegistry) Register(stepID string, t EngineWakeTarget)

Register stores the wake target for stepID. Overwrites any prior entry - useful for retried steps that reallocate their Wake channel.

func (*MapWakeRegistry) Unregister 

func (r *MapWakeRegistry) Unregister(stepID string)

Unregister removes stepID from the registry. Called from the runStep deferred cleanup so the engine stops trying to wake a completed step.

func (*MapWakeRegistry) WakeTarget 

func (r *MapWakeRegistry) WakeTarget(stepID string) EngineWakeTarget

WakeTarget implements EngineWakeRegistry.

type Message 

type Message struct {
	MessageID string
	From      string
	To        string
	Content   string
	Type      MessageType
	Metadata  map[string]string
	Timestamp time.Time
}

Message is a message between agents or coordinator. Stable. MessageID is assigned by the MailboxStore on Append (callers leave it empty). It is the stable identity used by MarkRead's CAS dedup contract: MarkRead(stepID, ids) returns the subset of ids that were already marked read on a prior call. This lets concurrent drainers detect double-consume without holding a lock across LLM calls.

type MessageType 

type MessageType int

MessageType classifies router messages. Stable. 

const (
	// MessageInfo is a general informational message.
	// Available for consumers to send informational messages between agents.
	MessageInfo MessageType = iota // info
	// MessageCancel requests the receiving agent to stop.
	MessageCancel // cancel
	// MessageContextUpdate injects new context into the agent's conversation.
	// Available for consumers to push context updates to running agents.
	MessageContextUpdate // context_update
	// MessageResumeReply is the reverse-routed reply produced by
	// Executor.runResume after a resumed step finishes. Tagged
	// distinctly from MessageInfo so observers can distinguish
	// resume responses from regular coordinator pushes. Drain logic
	// treats it the same as MessageInfo (appended as a user turn).
	MessageResumeReply // resume_reply
)

func (MessageType) String 

func (i MessageType) String() string

type RealClock 

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

RealClock is the production EngineClock. It wraps time.NewTicker.

func (*RealClock) Stop 

func (c *RealClock) Stop()

Stop stops the underlying ticker started by Tick. Safe to call when Tick has not been called (no-op) or after a previous Stop. After Stop the channel returned by Tick will not receive further ticks.

func (*RealClock) Tick 

func (c *RealClock) Tick(d time.Duration) <-chan time.Time

Tick starts a new ticker firing every d and returns its channel. If a prior ticker is still running on this RealClock (Tick called multiple times without an intervening Stop), the prior ticker is stopped so its goroutine and channel can be GC'd; otherwise repeated Tick calls would leak tickers. Not safe for concurrent calls on the same RealClock.

type ResumeHandle 

type ResumeHandle struct {
	// StepID is the terminated step that was resumed.
	StepID string
	// ResumeID is a per-invocation identifier used in event payloads
	// so operators can correlate EventResumeStarted /
	// EventResumeCompleted / EventResumeFailed for the same resume.
	ResumeID string
	// OriginalSender is the From field of the router message that
	// triggered the resume; the resumed AgentRunner's final assistant
	// response is routed back to this agent via reverse Message.
	// May be empty (e.g. external caller directly invoking ResumeStep);
	// the executor suppresses the reverse Send in that case.
	OriginalSender string
	// DoneCh is closed once the resume goroutine has finished - either
	// after a successful final assistant response or on Err.
	DoneCh chan struct{}
	// Result is the resumed agent's final assistant text. Populated
	// before DoneCh closes when Err == nil.
	Result string
	// Err is non-nil on failure (ctx-cancel, transcript cap,
	// AgentRunner error). Populated before DoneCh closes.
	Err error
}

ResumeHandle is the per-resume control block returned by Executor.ResumeStep.

type Resumer 

type Resumer interface {
	// CanResume returns true if stepID has a saved transcript and the
	// Executor is currently willing to spawn a resume (Run not cancelled).
	CanResume(stepID string) bool
	// ResumeStep loads the step's transcript, appends prompt as a
	// user turn, and spawns a fresh AgentRunner. Returns a
	// ResumeHandle the caller can block on via h.DoneCh. Errors on
	// transcript-missing / cap-exceeded / shutdown - the Router maps
	// those to DropReason* values.
	ResumeStep(ctx context.Context, stepID, prompt, fromAgent string) (*ResumeHandle, error)
}

Resumer is the Executor-side hook consulted by the Router when a message lands on a closed mailbox (target-terminal). When the hook reports CanResume=true, the Router calls ResumeStep instead of emitting the drop; the resume goroutine runs in the background and eventually routes a reverse Message back to the sender. Wiring: Executor implements Resumer (see executor.go) and installs itself via Router.SetResumer at the start of Run. When the Executor has no TranscriptStore (e.g. mailbox-delivery disabled), CanResume always returns false and the Router falls back to the pre-Phase-7.12 target-terminal drop path. R4.

type Router 

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

Router routes Messages from coordinator/agents into the per-step MailboxStore. Stable. The cutover removed the legacy buffered-chan delivery path; mailbox is now the only path. Lifecycle: - SetMailbox(store) MUST be called before any Send. Without a mailbox, Send becomes a no-op + emits a "unknown-step" drop event for every message (no silent loss). - RegisterInbox(stepID) marks a step as live so Send routes deliveries into the mailbox. The router rejects Sends to unregistered steps unless a sender slot is open (PendingSenders>0), which captures the pre-start race window where coordinator narration may target a step whose runStep goroutine has not yet executed RegisterInbox. - Close(stepID) marks the step terminal: subsequent Sends emit "target-terminal" drops. The mailbox is also closed via Append's no-op-on-closed contract. Trust model: the Router is internal to the library. All senders (coordinator, parent agents) are trusted - there is no sender authentication. The From field is informational only.

func NewRouter 

func NewRouter() *Router

NewRouter creates a new Router. Stable.

func (*Router) AcquireStepLock 

func (r *Router) AcquireStepLock(stepID string) *sync.RWMutex

AcquireStepLock returns the per-step RWMutex for stepID, creating it if necessary. Lifecycle code uses Lock for state transitions / Seal / Delete, the invariant-check uses RLock to take a coherent snapshot of (state, mailbox-len, pending-senders). Locks are reference-counted only by presence in the map: the executor is responsible for calling ReleaseStepLock when the step is fully retired.

func (*Router) Close 

func (r *Router) Close(stepID string)

Close marks stepID's mailbox as terminal. Subsequent Sends emit "target-terminal" drops via OnDrop. The underlying mailbox.Close is invoked so any future Append is a no-op at the store layer too. any messages that were Appended to the mailbox but never drained by the agent (e.g. coordinator messages that arrived during the agent's last LLM call, or that arrived after the agent reached idle but before the executor's flush ran) MUST emit a DropEvent so the "zero silent drops" contract holds. Without this, the abort path (`flushMailboxOnAbort`) is the only safety net - and it only fires when a step goroutine fails to send `done` before the workflow ctx cancels. Steps that respond quickly to ctx-cancel and signal done would otherwise lose pending messages silently. The drop reason for these messages is "target-terminal" - the same reason emitted for post-Close Sends - because semantically the target is no longer available to receive them. Idempotent: a second Close on the same stepID re-takes the lock, observes an empty mailbox (already drained), re-marks the step terminal (already true), and re-releases the lock. No drops fire twice. Production code paths only Close each stepID once; this guarantee is documented for embedders that wire their own retry loops.

func (*Router) CloseSender 

func (r *Router) CloseSender(targetStepID string)

CloseSender decrements the pending-senders counter for targetStepID. Defensive: clamps at zero - a Close without a matching Open is a no-op rather than an underflow into negative numbers, which would mask later sender leaks behind a phantom positive count. Tests must continue to pair Open/Close, but production paths get a safety net.

func (*Router) HasStepLock 

func (r *Router) HasStepLock(stepID string) bool

HasStepLock reports whether stepID currently has an entry in the per-step lock registry. Intended for tests that need to assert release-on-Close behaviour without reaching into unexported state. Stable.

func (*Router) Inboxes 

func (r *Router) Inboxes() []string

Inboxes returns a snapshot of every inbox the router is currently tracking (both open and already-closed). Used by tests that need to assert which step IDs were registered as inboxes; production code should use KnownSteps which also includes wrapper / pending-sender stepIDs. Stable.

func (*Router) IsWrapperStep 

func (r *Router) IsWrapperStep(stepID string) bool

IsWrapperStep returns true if stepID was registered via RegisterWrapperStep. - used by `forward_to_agent` to reject wrapper targets with a helpful error before Router.Send (so the silent misroute never happens). Concurrent-safe.

func (*Router) KnownSteps 

func (r *Router) KnownSteps() []string

KnownSteps returns a sorted snapshot of step IDs registered via RegisterStep. Used by coord prompt construction (inject "AVAILABLE STEPS" menu) and by forward_to_agent's unknown-step error feedback (return helpful message listing valid IDs). Concurrent-safe; returns a fresh slice copy so callers may iterate without holding the router's internal lock. Includes inner-DAG steps registered by nested executors (loop iterations, forEach items, include sub-workflows) - the namespaced IDs (e.g. "loopID.0.innerStep") materialise into the known set as each iteration starts. Coord prompt should refresh this list per wake to reflect the current snapshot. Stable.

func (*Router) Mailbox 

func (r *Router) Mailbox() MailboxStore

Mailbox returns the installed MailboxStore (or nil when unset). Callers use it to drain non-step inboxes such as "coordinator"

func (*Router) MarkWorkflowCancelled 

func (r *Router) MarkWorkflowCancelled()

MarkWorkflowCancelled flips the router into a cancelled state. After this call: - all subsequent Send calls return immediately with a DropReasonWorkflowCancelled drop event - subsequent Close calls attribute pending drops to DropReasonWorkflowCancelled (instead of the generic terminal reason) so operators can distinguish abort drops from natural end-of-step drops Idempotent: safe to call multiple times.

func (*Router) OpenSender 

func (r *Router) OpenSender(targetStepID string)

OpenSender increments the pending-senders counter for targetStepID. every entity that may call Send(targetStepID, ...) MUST take a sender slot via OpenSender + defer CloseSender so the executor's 3-invariant termination check (no senders + empty mailbox + stable idle) can determine when it is safe to close the target's mailbox. The counter is per-target (not per-sender-identity) because the invariant only cares whether any sender is still in flight, not who.

func (*Router) PendingSenders 

func (r *Router) PendingSenders(targetStepID string) int

PendingSenders returns the current pending-senders count for targetStepID. Used by the executor's 3-invariant termination check.

func (*Router) RegisterDelegate 

func (r *Router) RegisterDelegate(stepID string, delegate *Router)

RegisterDelegate adds a delegation entry: any future Send to stepID will be forwarded to delegate.Send instead of going through this router's own mailbox path. - used by nested-DAG executors (loop iterations, forEach items, includes) to expose their inner-step inboxes to the OUTER router so coord's forward_to_agent can reach inner steps. Replaces any prior delegation for the same stepID (sequential iterations of repeat-until rely on this - each iteration replaces the prior's delegation). Pass nil delegate to remove (equivalent to UnregisterDelegate).

func (*Router) RegisterInbox 

func (r *Router) RegisterInbox(stepID string)

RegisterInbox marks stepID as having a live mailbox. Call this before the step's AgentRunner.Run begins consuming the mailbox. Idempotent: re-registering an already-open step is a no-op; if the step was previously closed, RegisterInbox clears the closed flag and reopens it.

func (*Router) RegisterStep 

func (r *Router) RegisterStep(stepID string)

RegisterStep records stepID as a member of the current workflow's DAG. Idempotent. Called once per workflow step at Run start (before any runStep goroutine spawns) so that Send(siblingTarget, ...) issued from a sibling step can auto-open a sender slot just-in-time when the F7 DAG-aware matrix has not pre-opened one. Steps not registered here continue to drop with DropReasonUnknownStep. G7 - preserves "zero silent drops" under future sibling-direct Send paths added after F7.

func (*Router) RegisterWrapperStep 

func (r *Router) RegisterWrapperStep(stepID string)

RegisterWrapperStep marks stepID as a wrapper (loop / include container) that orchestrates sub-steps but does NOT host an agent of its own. Coord's `forward_to_agent` rejects messages targeting wrapper steps via IsWrapperStep - without this marker, Router.Send would silently accept the message into the wrapper's inbox where no agent ever reads it. explicit-marker approach. Earlier heuristic ("step X is wrapper if any other registered step starts with X.") didn't work because namespacing delegates inner-DAG registration to a child router; outer router never sees children, can't detect wrapper from prefix scan. Executor must call this explicitly when processing a step with Loop or Include definition. Idempotent. Concurrent-safe.

func (*Router) ReleaseStepLock 

func (r *Router) ReleaseStepLock(stepID string)

ReleaseStepLock removes the per-step RWMutex from the registry. Safe to call even when no lock was acquired (no-op).

func (*Router) Send 

func (r *Router) Send(stepID string, msg Message) error

Send delivers a message to stepID's mailbox. Non-blocking. Stable. Drops emit one DropEvent via OnDrop (when configured) so the executor can surface EventMessageDropped - never silent. Send now ALSO returns an error so per-call drops can surface to in-process callers (notably the send_message and forward_to_agent tools, which render the drop reason into their LLM-visible result string). The error is in addition to the OnDrop callback (which the executor uses to emit EventMessageDropped) - both fire on every drop. Returns nil on accept (mailbox.Append succeeded with no concurrent Close race). Returned errors are formatted as `"dropped: <reason>"` where `<reason>` is the canonical DropReason.String value, so callers can pass the error message directly through to LLM-visible tool results without reformatting. Drop matrix: - mailbox not configured → "unknown-step" - stepID never registered AND no pending senders → "unknown-step" - stepID was closed → "target-terminal" - mailbox.Append succeeds → no drop Drops while a sender slot is open (PendingSenders>0) for an unregistered step are NOT emitted as "unknown-step": the sender slot is the executor's promise that RegisterInbox is imminent. The message is still appended to the mailbox so the late-arriving step will see it on first Unread. (MailboxStore.Append never fails; a concurrent Close racing with Append is handled by the store's own closed-flag check and emits "mailbox-closed-by-finalize" via the Closed-after-Append fallback below.)

func (*Router) SetAfterSend 

func (r *Router) SetAfterSend(fn func(stepID string, msg Message))

SetAfterSend installs the post-Append hook fired after a successful Router.Send.: executor uses this to bridge Router.Send → wakeRegistry, giving every recipient (steps + coord) push-immediate wake instead of waiting for the DeliveryEngine's poll tick.: also passed the Message so the executor can bridge it into the coord runner's separate Mailbox. Drops (closed mailbox, unknown step, etc.) do NOT fire this hook - only successful Appends. Pass nil to disable.

func (*Router) SetMailbox 

func (r *Router) SetMailbox(store MailboxStore)

SetMailbox installs a MailboxStore as the delivery backend. Send observes nil mailbox until SetMailbox completes - call SetMailbox during setup before Send starts arriving.

func (*Router) SetOnDrop 

func (r *Router) SetOnDrop(fn func(DropEvent))

SetOnDrop installs a callback invoked once per dropped message. The executor wires this so router-side drops (target-terminal, unknown-step, mailbox-closed-by-finalize) translate to EventMessageDropped - preserving the "zero silent drops" contract

func (*Router) SetResumer 

func (r *Router) SetResumer(res Resumer)

SetResumer installs the Executor-side resume hook. May be called multiple times; the last setter wins. Passing nil disables resume (Send reverts to emitting DropReasonTargetTerminal for closed mailboxes). Safe to call before the Executor begins running - Send reads resumer under r.mu.RLock. R4.

func (*Router) SetRunCtxProvider 

func (r *Router) SetRunCtxProvider(ctxProvider func() context.Context)

SetRunCtxProvider installs a function that returns the Executor's run-lifetime context. When set, Send passes the result of ctxProvider to ResumeStep instead of context.Background, so cancellation of the workflow propagates correctly into resume goroutines. Called alongside SetResumer at the start of Run.

func (*Router) UnregisterDelegate 

func (r *Router) UnregisterDelegate(stepID string)

UnregisterDelegate removes the delegation entry for stepID. No-op if no entry exists. Called by nested-DAG executors when an iteration ends so the next iteration (or workflow termination) doesn't see stale routing.

func (*Router) WorkflowCancelled 

func (r *Router) WorkflowCancelled() bool

WorkflowCancelled reports whether MarkWorkflowCancelled has been called on this router. The poller and waitForStepTermination consult this flag to short-circuit waits when the workflow is being torn down.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.