Skip to content
Documentation
Reference
Go SDK
worker
API

worker

import "github.com/danmestas/dagnats/worker"

worker/cancel.go Cancellation reaching a running handler (#802). The engine publishes dagnats.cancel.{runID} on core NATS when a run is cancelled; the worker ends the context of every in-flight handler of that run. The signal is at-most-once, so Heartbeat also reads the run state.

worker/consumer_collision.go Registration-time precheck: refuse to start if any (taskType, group) pair in the worker’s configured handlers collides on the durable consumer name after sanitization. Catches cases like “render.gpu” + “render-gpu” before they corrupt NATS state via CreateOrUpdateConsumer.

worker/consumer_collision_xprocess.go Cross-process precheck (ADR-010): catches the case where two workers in different processes registered different task types whose sanitized durable names collide. Worker A registered first; this helper runs in Worker B’s subscribePullConsumer before CreateOrUpdateConsumer, sees the existing durable with our name but a different FilterSubject, and panics with both filters named so the operator knows which task types to rename.

Companion to assertNoConsumerNameCollisions (consumer_collision.go), which catches the same collision class in the same Worker. This file is the NATS-aware twin that catches the cross-process variant.

worker/consumer_naming.go Worker-side bindings for the shared consumer-naming convention.

The scheme itself lives in internal/consumername because the bridge’s poll path must produce byte-identical durables (issue #532), and TASK_QUEUES is a work-queue stream that rejects a second consumer on an overlapping filter. Sharing via internal/ keeps that coupling out of the worker package’s public SDK surface.

worker/controlplane.go ControlPlane is the worker-side handle a GATED task handler uses to author and launch workflows at runtime. It is a DEEP module: the two methods below hide def validation, server-side namespacing, run lineage, the maxNestingDepth cap, and every NATS round-trip. Handlers never see subjects, KV keys, or wire framing — they hold a small interface and get back a scoped name or a run ID, or a structured typed error they can branch on.

Why worker-side handle over the api micro.Service (not engine logic): the validated control-plane boundary already lives on the dagnats-api micro.Service (#456). Duplicating register/spawn logic in the worker would fork the namespace/lineage/depth rules. Instead the handle speaks NATS request/reply to two additive subjects and lets the server stay the single source of truth.

Every boundary failure returns *ControlPlaneError so the durable agent loop (ADR-002) can self-correct instead of crashing. Panics are reserved for programmer errors (nil ctx, nil receiver, empty ownerRunID) — never for agent-supplied data.

Public re-exports of the internal envelope types. Go’s internal/ rule blocks downstream workers from importing the canonical definitions directly, so without these aliases every consumer redeclares the struct shape and drifts over time. Aliases (not wrapper types) keep the engine and the worker on the same underlying type with zero conversion at the call site. See #235.

worker/identity.go Process identity helpers used to populate WorkerRegistration’s Pid / Hostname / Version fields (#289). Resolved once per process and cached, since none of these can change mid-run.

worker/inputs.go Agent inputs and durable waits on a taskContext (ADR-027). A handler places the inputs queued for its step at a turn boundary (steers after a tool round, follow-ups after a final answer); each placement is one input.placed journal record, the truth about placement, and the submission’s bucket status is a projection repaired from it. A Continue may ask the engine to wait, durably and with no worker claim held, for a delay, a deadline, or the next input.

worker/journal.go The agent journal on a taskContext (ADR-026). A handler opts in by calling Journal(): that claims the dispatch, and from then on every attempt-ending call commits one journal record before publishing its lifecycle event, Checkpoint and LoadCheckpoint read and write the journal’s state, and Pause writes a pause record. A task the journal says is stale is acknowledged untouched; a fenced one is left to its successor; an attempt whose commit already exists republishes it.

worker/log_writer.go logLane buffers and publishes one task ATTEMPT/ITERATION’s captured stdout/stderr to the BUILD_LOGS hot lane (#624, subject logs.{runID}.{stepID}.{attempt}.{iteration}). One lane per attempt/iteration, created lazily on first LogOut()/LogErr() Write so a handler that never logs never spins up a ticker goroutine.

Attempt is part of the subject (#624 review round 2), not just the payload: a retry re-dispatches the same runID/stepID, and BUILD_LOGS’s dedup window (2 min) is comfortably longer than the gap between two attempts, so a bare logs.{runID}.{stepID} subject would let attempt 2’s seq-0 chunk collide with attempt 1’s Nats-Msg-Id and silently vanish as a duplicate.

Iteration is a SECOND, independent dimension (#624 review round 3): an agent-loop step’s Continue re-dispatches the SAME attempt with iteration incremented — never bumping attempt, which would consume retry budget the step never spent — so without iteration in the subject/Msg-Id too, iteration 2+’s chunks would collide with iteration 0’s the exact same way un-scoped attempts collided before round 2. Scoping the subject by both makes every attempt+iteration’s chunk stream independent and gives from=failure an unambiguous target.

Complete, Fail, FailPermanent, FailRetryAfter, Continue, and Pause all drain the lane (flush pending bytes, emit the attempt-ending marker, stop the ticker) before publishing their resolution — the drain-before-resolve invariant — so a consumer that observes the terminal event can never race ahead of the log bytes that produced it, and the marker is always the true last message on the subject.

worker/restart_limit.go The crash-loop guard of ADR-028: a journal step whose deliveries keep claiming one iteration without progress is failed with reason restart_limit instead of being redelivered forever. The journal counts the claims (journal.Restarts); this file applies the limit and fails the step through the ordinary Fail path, so the failure is committed to the journal, published to the engine and retried by the step’s retry policy, and the retry resumes from the journal.

worker/services.go ServiceDef + Worker.RegisterService SDK method (ADR-017 / #321).

Services are a metadata namespace for grouping task types under a logical name. They are deliberately separated from the worker directory (worker/directory.go, #289): workers have a 60s TTL and a heartbeat loop; services have neither. A service entry persists across worker restarts and never expires automatically — it is a stable description, not a liveness signal.

This file does not import or extend Directory. Sharing machinery would conflate two different lifecycles. See ADR-017 §Alternatives for the rejected re-use option.

worker/stranded_check.go Defense-in-depth counterpart to the engine’s primary stranded-subject check (Orchestrator.checkStrandedGroupSubjects, internal/engine): #704 moved the grouped subject/durable encoding behind the group sentinel (consumername.GroupSentinel) with no compatibility consumer, so a message published to a legacy grouped subject before every process upgrades is never reclaimed by any other mechanism -- see the pinned design on issue #704. This narrows the check to the one (task, group) pair this worker is about to serve, checked right before subscribing that pair’s consumer.

worker/trigger_types.go Worker-side complement to the ExternalRegistrar ack micro endpoint (#327). RegisterTriggerType is the SDK call workers make once on boot to (a) publish their TriggerTypeDef into the `trigger_types` KV bucket and (b) ask the engine to allocate an externalRegistrar so subsequent `_TRIGGER.<kind>.{activate,deactivate}` requests get bridged to this worker.

KV-then-ack ordering is load-bearing — the engine’s handleAck reads the schema bytes straight from KV (audit-adjusted contract: “KV is the source of truth”). A worker that calls ack before its Put has landed will see a “trigger type %q not registered in KV” error and must retry.

Idempotency: re-registering with the same Name + OwnerWorkerID + ConfigSchema bytes returns nil — workers may call this on every boot without coordinating. Schema or owner drift surfaces as an error so silent fleet skew is impossible.

worker/watch_triggers.go Worker SDK for receiving External trigger activate/deactivate events (parent #273 Phase 2.4, #333). Bridges the engine’s `_TRIGGER.<kind>.{activate,deactivate}` request subjects (driven by the externalRegistrar in internal/trigger/registrar_external.go) into user-supplied callbacks.

Subscription lifecycle is internal — the audit (#333 comment) chose internal tracking over returning a Subscription handle because callers already own a Worker.Stop() lifecycle. A separate Unsubscribe() would add a third lifecycle they have to remember. Stop() drains every triggerSubs entry; see worker.go.

Catch-up contract: on subscribe, the worker scans the `triggers` KV bucket and fires onActivate for every Enabled entry whose External.Kind matches. Bounded at maxCatchupKeys = 10000 — over-cap the loop stops and logs a warning rather than silently truncating half-way through (TigerStyle “all loops must have fixed upper bounds”; CLAUDE.md).

Index

Constants

AdminTokenID is the reserved token_id value written to a registration created by the bridge’s admin bearer or dev mode (#650 round 3). Using "" for these entries made an admin takeover indistinguishable from a genuinely unowned entry (a pre-#627 record, or a native Go worker outside the bridge’s scope -- see Register/Deregister) and therefore claimable by the next bridge token to connect. workertoken.Mint asserts a minted id can never equal this value (ids are nuids, so the collision is not reachable in practice, but the assertion makes the invariant explicit), so an entry carrying AdminTokenID is unambiguously admin-owned: only an admin caller (which includes every dev-mode caller -- dev mode has no identity to enforce) may re-register or delete it. Defined in internal/workertoken (the token-identity package) and re- exported here so the dependency runs public -> internal, not the reverse.

const AdminTokenID = workertoken.AdminTokenID

ReasonRestartLimit prefixes the error of a step the guard failed.

const ReasonRestartLimit = "restart_limit"

RestartLimitDefault is the number of consecutive claims on one iteration without progress after which a journal step fails.

const RestartLimitDefault = 5

SpawnKeyMaxBytes is the longest accepted spawn key.

const SpawnKeyMaxBytes = 256

Variables

Sentinels for errors.Is. Each wraps its Kind so a freshly constructed *ControlPlaneError of the same Kind matches via the Is method below.

var (
    ErrPromotionUnsupported = &ControlPlaneError{
        Kind: KindPromotionUnsupported, Op: "RegisterWorkflow",
        Message: "promotion is not supported in Tier 1 (deferred to #378)",
    }
    ErrInvalidDef       = &ControlPlaneError{Kind: KindInvalidDef}
    ErrNamespace        = &ControlPlaneError{Kind: KindNamespace}
    ErrUnresolvableName = &ControlPlaneError{Kind: KindUnresolvableName}
    ErrTransport        = &ControlPlaneError{Kind: KindTransport}
    ErrDenied           = &ControlPlaneError{Kind: KindDenied}
    ErrDepthExceeded    = &ControlPlaneError{Kind: KindDepthExceeded}
    ErrQuotaExceeded    = &ControlPlaneError{Kind: KindQuotaExceeded}
    ErrRateLimited      = &ControlPlaneError{Kind: KindRateLimited}
)

Sentinel errors.

var (
    // ErrNoJournal: inputs are placed through the journal; the handler
    // must call Journal() first.
    ErrNoJournal = errors.New("worker: inputs need a journal; call Journal() first")
    // ErrInputsQueued: Continue(UntilInput()) was asked while inputs
    // are queued. The wait would never be woken for them (they arrived
    // before it), so the handler must place them and run another turn.
    ErrInputsQueued = errors.New(
        "worker: inputs are queued; place them before waiting for input")
    // ErrInputsNotProvisioned: the submissions bucket is missing.
    ErrInputsNotProvisioned = inputs.ErrNotProvisioned
)

ErrRunCancelled is returned by Heartbeat when the task’s run has been cancelled. The handler’s Context() is already done at that point; a handler should stop and return.

var ErrRunCancelled = errors.New("worker: run cancelled")

ErrWorkerIDContended is returned by RegisterOwned and DeregisterOwned when a revision-guarded write loses the race to a concurrent writer ownedWriteRetriesMax times in a row -- distinct from ErrWorkerIDOwned because contention is not an ownership decision: the fresh entry re-Get after each conflict may still belong to the caller, we simply never won the write. Callers should treat this as transient (e.g. the bridge maps it to 503 with Retry-After), never as a takeover.

var ErrWorkerIDContended = errors.New(
    "worker_id write lost the revision race too many times",
)

ErrWorkerIDOwned is returned by RegisterOwned (on register/ heartbeat) and DeregisterOwned (on disconnect) when workerID’s current entry is owned by a different token and the caller is neither that token nor an admin -- including when the caller lost a race to another writer between the ownership check and the revision-guarded write.

var ErrWorkerIDOwned = errors.New(
    "worker_id is registered to another token",
)

MaxWorkerStaleness is the read-time cutoff used by List(): entries whose last Put is older than this are treated as dead and filtered out. The workers KV bucket has a 60s TTL, but NATS may delay purging past the nominal TTL — this filter makes staleness deterministic for callers (e.g. `dagnats workers list`) so a SIGKILL’d worker stops appearing within MaxWorkerStaleness rather than waiting for the next NATS cleanup pass. Matches the bucket TTL so dead entries vanish promptly after the heartbeat would have refreshed them. Variable rather than const so tests can shrink the window.

var MaxWorkerStaleness = 60 * time.Second

func HandleCheckpoint 

func HandleCheckpoint(w *Worker, taskType string, fn func(CheckpointTask) error)

HandleCheckpoint registers a handler that receives CheckpointTask for save/restore across retries.

func HandleLoop 

func HandleLoop(w *Worker, taskType string, fn func(LoopTask) error)

HandleLoop registers a handler that receives LoopTask for agent-loop iteration with Continue and FailRetryAfter.

func HandleSignal 

func HandleSignal(w *Worker, taskType string, fn func(SignalTask) error)

HandleSignal registers a handler that receives SignalTask for inter-step coordination via WaitForSignal/SendSignal.

func HandleSimple 

func HandleSimple(w *Worker, taskType string, fn func(SimpleTask) error)

HandleSimple registers a handler that receives only SimpleTask. The handler sees a narrow interface — it cannot call Continue, Checkpoint, or other advanced methods.

func HandleStream 

func HandleStream(w *Worker, taskType string, fn func(StreamTask) error)

HandleStream registers a handler that receives StreamTask for streaming output and heartbeat keep-alive.

func HandleTyped 

func HandleTyped[I, O any](w *Worker, taskType string, fn TypedHandlerFunc[I, O], opts ...TypedOption)

HandleTyped registers a typed task handler that automatically marshals/unmarshals JSON. Combines Typed() and Handle() into a single call so workers don’t need to know about the wrapping. Optional TypedOption values (e.g. UnwrapTrigger) tune the wrapper.

type CheckpointTask

CheckpointTask adds checkpoint/resume capability for handlers that need to persist state across retries.

type CheckpointTask interface {
    SimpleTask
    Checkpoint(state []byte) error
    LoadCheckpoint() ([]byte, error)
    RetryCount() int
}

type ContinueOption

ContinueOption asks the engine to wait before the next iteration (ADR-027). Options combine: a time and UntilInput together wake on whichever comes first; Until wins over Delay.

type ContinueOption func(*protocol.ContinueWait)

func Delay 

func Delay(d time.Duration) ContinueOption

Delay wakes the step after d.

func Until 

func Until(t time.Time) ContinueOption

Until wakes the step at t.

func UntilInput 

func UntilInput() ContinueOption

UntilInput wakes the step when an input is queued for it. Refused with ErrInputsQueued while any input is already queued.

type ControlPlane

ControlPlane lets a gated handler author an ephemeral workflow def at runtime and launch a child run of it. nil unless the step declared the “control-plane” capability AND the deployment granted it — always nil-check before use.

Adding a method here EXTENDS the public interface: any external implementation (mock or alternate handle) must add it too. Budget() was added in #378; there are no external implementations in-repo, so this is source-compatible here, but downstream embedders must update their mocks.

type ControlPlane interface {
    // RegisterWorkflow validates def and persists it under a
    // server-computed scoped name, returning that name. opts.Promote is
    // WIRED (#377): true registers under the reaper-immune "promoted.*"
    // namespace. Promotion is GOVERNED (#380): the server authorizes it
    // against the grant policy's promote list (keyed on the author name) and
    // returns KindDenied when the workflow is not authorized. Promoted defs
    // still bypass the #378 root-scoped quota/rate limits (they have no
    // owning tree). The returned scopedName is what StartRun expects.
    RegisterWorkflow(
        ctx context.Context, def dag.WorkflowDef, opts RegisterOpts,
    ) (scopedName string, err error)

    // StartRun launches a child run of the named (scoped) workflow with
    // the given input, returning the child run ID. Lineage and the
    // nesting-depth cap are enforced server-side. Pass WithSpawnKey to make
    // the call idempotent: a repeat with the same key from the same step
    // returns the existing child instead of spawning another (#803).
    StartRun(
        ctx context.Context, name string, input []byte, opts ...StartOpt,
    ) (runID string, err error)

    // Budget reports the owning tree's current-vs-max for the two quota
    // dimensions (active runs, registered defs), computed by the same
    // server-side scan that enforces the quotas. A gated handler reads it
    // to self-throttle before hitting a KindQuotaExceeded reply (#378).
    Budget(ctx context.Context) (RuntimeBudget, error)
}

func NewControlPlane 

func NewControlPlane(nc *nats.Conn) ControlPlane

NewControlPlane constructs a ControlPlane bound to nc. The owning run and step are bound later, at grant time, via newControlPlaneFor — this public constructor is what a deployment wires through WithControlPlane. Panics if nc is nil (programmer error at startup).

type ControlPlaneError

ControlPlaneError is the single structured error type every boundary failure returns. Kind is the branch key; Op names the failing operation; Message is human-readable; wrapped carries any underlying cause for errors.Is/As against the sentinels below.

type ControlPlaneError struct {
    Kind    ControlPlaneErrorKind
    Op      string
    Message string
    // contains filtered or unexported fields
}

func (*ControlPlaneError) Error 

func (e *ControlPlaneError) Error() string

func (*ControlPlaneError) Is 

func (e *ControlPlaneError) Is(target error) bool

Is lets errors.Is match by Kind, so callers can write errors.Is(err, ErrPromotionUnsupported) regardless of the concrete instance returned.

func (*ControlPlaneError) Unwrap 

func (e *ControlPlaneError) Unwrap() error

type ControlPlaneErrorKind

ControlPlaneErrorKind is the small, closed set of failure categories a gated handler may branch on. Stable strings so they can also travel on the wire between the server endpoints and the worker handle.

type ControlPlaneErrorKind string

const (
    KindInvalidDef           ControlPlaneErrorKind = "invalid_def"
    KindNamespace            ControlPlaneErrorKind = "namespace"
    KindUnresolvableName     ControlPlaneErrorKind = "unresolvable_name"
    KindPromotionUnsupported ControlPlaneErrorKind = "promotion_unsupported"
    KindTransport            ControlPlaneErrorKind = "transport"
    KindDenied               ControlPlaneErrorKind = "denied"
    KindDepthExceeded        ControlPlaneErrorKind = "depth_exceeded"
    // Additive safety-limit kinds (#378). KindQuotaExceeded covers both the
    // active-run and the def quota; KindRateLimited covers the register
    // rate limit. A gated handler branches on these to back off.
    KindQuotaExceeded ControlPlaneErrorKind = "quota_exceeded"
    KindRateLimited   ControlPlaneErrorKind = "rate_limited"
)

type Directory

Directory provides worker visibility via NATS KV. Each worker writes its registration to the “workers” bucket; the bucket’s TTL ensures stale entries are purged automatically.

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

func NewDirectory 

func NewDirectory(js jetstream.JetStream) *Directory

NewDirectory creates a Directory backed by the “workers” KV bucket. Panics if js is nil or the bucket does not exist — both are programmer errors indicating missing setup.

func (*Directory) Deregister 

func (d *Directory) Deregister(workerID string) error

Deregister removes the worker’s entry from the directory. Panics if workerID is empty. Returns nil if the key does not exist.

func (*Directory) DeregisterOwned 

func (d *Directory) DeregisterOwned(workerID, callerTokenID string, callerIsAdmin bool) error

DeregisterOwned removes workerID’s entry, but only if the caller still owns it (#650, the delete-side counterpart to RegisterOwned): ownershipAllows must hold for the entry’s current token_id against the caller. A disconnect from a token that has since been superseded (e.g. an admin took the worker_id over while the original owner’s connection was still open) must not delete the current owner’s entry out from under it -- it returns ErrWorkerIDOwned instead and leaves the entry untouched. Uses the Get’s revision with jetstream.LastRevision on Delete so a concurrent write between the Get and the Delete aborts the delete instead of clobbering it, same TOCTOU window RegisterOwned closes on the write side -- but a bare revision conflict doesn’t prove who wrote in the window: it could be this same connection’s own heartbeat re-registering right as it disconnects, which must not skip a legitimate deregister. On conflict, deregisterOwnedAttempt re-Gets and this loop retries the decision against the fresh entry, up to ownedWriteRetriesMax times. Returns nil if the key does not exist.

func (*Directory) List 

func (d *Directory) List() ([]WorkerRegistration, error)

List returns all currently registered workers. Returns an empty slice when no workers are registered. Skips entries that fail to unmarshal (TTL expiry race).

Key enumeration is natsutil.ListKeys, not kv.ListKeys -- see its doc comment for the watcher-snapshot race this avoids (#699: List() reported a live, actively-heartbeating worker as absent roughly once per 2000 calls). Subjects whose last message is a delete marker still appear in that enumeration; the per-key Get below returns ErrKeyNotFound for those and skips them.

func (*Directory) Register 

func (d *Directory) Register(reg WorkerRegistration) error

Register writes the worker’s registration to the KV bucket with an unguarded Put -- no ownership check, no revision guard. Reserved for native Go workers, which never go through the bridge and so have no TokenID to enforce (#650’s ownership scope is the bridge’s HTTP connect/heartbeat path only; see RegisterOwned). The worker must call Register periodically (before the 60s TTL) to maintain its presence. Panics on empty WorkerID or TaskTypes.

func (*Directory) RegisterOwned 

func (d *Directory) RegisterOwned(reg WorkerRegistration, callerTokenID string, callerIsAdmin bool) error

RegisterOwned is the single, revision-guarded write path for worker_id registration used by the bridge for both the initial connect and the periodic heartbeat re-register (#650 round 3). A prior two-step “check ownership, then plain Put” (and the heartbeat’s unconditional Put) each raced a concurrent writer between the check and the write: two tokens racing an unclaimed id could both pass the check and last-writer-wins, and a heartbeat replaying its connect-time TokenID could resurrect a worker_id after an admin had taken it over. Get -> ownershipAllows -> Create (key absent) or Update with the Get’s revision (key present) closed both races, but a raw revision conflict on its own proves only that someone wrote in the window -- not who. A prior version mapped any conflict straight to ErrWorkerIDOwned, which wrongly rejected a reconnect racing its own heartbeat’s re-register (CI- only flake: TestConnectWorkerIDOwnershipDevMode’s second connect racing the first connection’s disconnect-time deregister). On conflict, registerOwnedAttempt re-Gets and this loop retries the decision against the fresh entry, up to ownedWriteRetriesMax times.

type HTTPEnvelope

HTTPEnvelope is the request shape lifted from inbound HTTP and webhook triggers. Bind it via worker.HandleTyped[HTTPEnvelope] when worker.UnwrapTrigger() is set.

type HTTPEnvelope = httpenvelope.Envelope

type HandlerFunc

HandlerFunc is the function signature for task handlers registered with a Worker.

type HandlerFunc func(ctx TaskContext) error

func Typed 

func Typed[I, O any](fn TypedHandlerFunc[I, O], opts ...TypedOption) HandlerFunc

Typed wraps a TypedHandlerFunc into a HandlerFunc by handling JSON serialization. Marshal/unmarshal failures are wrapped in NonRetryableError because bad serialization will not fix itself on retry. Optional TypedOption values tune the wrapper.

type HandlerOption

HandlerOption configures per-handler behavior at registration time. Distinct from WorkerOption (which configures the Worker itself): HandlerOptions bind a knob to a specific taskType. Variadic on Handle keeps existing callers source-compatible.

type HandlerOption func(w *Worker, taskType string)

func WithAckWait 

func WithAckWait(d time.Duration) HandlerOption

WithAckWait overrides the JetStream AckWait for the consumer that will be created for taskType. Sub-second tasks should use a short override so worker-crash redelivery latency is bounded; long-running agent loops can opt into a longer wait. Panics if d <= 0 — non-positive durations are programmer errors per TigerStyle.

type LoopTask

LoopTask adds agent-loop iteration capability for handlers that call Continue to request another execution cycle. Includes streaming and heartbeat for long-running iterations.

type LoopTask interface {
    CheckpointTask
    Continue(output []byte, opts ...ContinueOption) error
    FailRetryAfter(err error, after time.Duration) error
    PutStream(data []byte) error
    Heartbeat() error
}

type NonRetryableError

NonRetryableError wraps an error to signal that retrying will not help. The worker framework detects this via errors.As and calls ctx.Fail() instead of NakWithDelay, causing immediate permanent failure.

type NonRetryableError struct {
    Err error
}

func NewNonRetryableError 

func NewNonRetryableError(err error) *NonRetryableError

NewNonRetryableError wraps err so the worker framework skips retries. Panics if err is nil — a nil non-retryable error is a programmer mistake.

func (*NonRetryableError) Error 

func (e *NonRetryableError) Error() string

func (*NonRetryableError) Unwrap 

func (e *NonRetryableError) Unwrap() error

type RateLimitError

RateLimitError wraps an error to signal a rate limit was hit. The worker framework detects this via errors.As and calls ctx.FailRetryAfter with the specified delay instead of using the default NAK backoff.

type RateLimitError struct {
    Err        error
    RetryAfter time.Duration
}

func NewRateLimitError 

func NewRateLimitError(err error, retryAfter time.Duration) *RateLimitError

NewRateLimitError wraps err with a suggested retry delay. Panics if err is nil or retryAfter is not positive.

func (*RateLimitError) Error 

func (e *RateLimitError) Error() string

func (*RateLimitError) Unwrap 

func (e *RateLimitError) Unwrap() error

type RegisterOpts

RegisterOpts carries optional knobs for RegisterWorkflow. Promote requests the def be registered under the reaper-immune “promoted.*” namespace instead of the ephemeral “agent.<root>.*” namespace (#377). The worker forwards the flag; the server owns the namespace shape. Authorization for promotion is GOVERNED by the grant policy’s promote list (#380): an unauthorized caller gets KindDenied. Promoted defs remain outside the #378 root-scoped quota / rate limits (those bound only root-scoped ephemeral defs).

type RegisterOpts struct {
    Promote bool
}

type RuntimeBudget

RuntimeBudget is the server-computed snapshot of a spawn tree’s quota usage (#378). It carries real, scan-backed numbers — not a stub. Token / compute metering is deferred (#378 P3), so no such field exists here; a zero-valued field would lie about being tracked.

type RuntimeBudget struct {
    ActiveRuns        int `json:"active_runs"`
    MaxActiveRuns     int `json:"max_active_runs"`
    RegisteredDefs    int `json:"registered_defs"`
    MaxRegisteredDefs int `json:"max_registered_defs"`
}

type ServiceDef

ServiceDef is the metadata entry for a logical service in the `services` KV bucket. Pure descriptive surface — does NOT gate task invocation. The `service::task` convention in task-type names is just a naming hint; the engine never reads this bucket during dispatch.

Fields are intentionally minimal. A `ParentService` grouping field was considered and deferred: no consumer exists yet (#274 R11 may add one). Adding it later is additive and last-write-wins handles the migration.

type ServiceDef struct {
    Name         string    `json:"name"`
    Description  string    `json:"description"`
    RegisteredAt time.Time `json:"registered_at"`
}

func ListServices 

func ListServices(js jetstream.JetStream) ([]ServiceDef, error)

ListServices reads every entry from the `services` KV bucket. Returns an empty slice when no services are registered. Skips entries that fail to unmarshal so a single bad payload does not block the whole listing (defensive — the bucket is metadata only, never authoritative).

This is package-level rather than a method on Worker because the CLI reads the bucket without owning a Worker. It takes a jetstream.JetStream handle so callers can share their existing connection.

Key enumeration is natsutil.ListKeys (#698), not kv.ListKeys: RegisterService Puts by stable service name on every re-registration (a worker restart), which is exactly the write shape exposed to the watcher-snapshot race #699 fixed for the workers bucket. Called rarely (CLI/console reads), so the bucket’s backing stream is resolved inline rather than cached.

type SignalTask

SignalTask adds inter-step coordination for handlers that wait on or send signals to other steps in the workflow.

type SignalTask interface {
    SimpleTask
    WaitForSignal(
        name string, timeout time.Duration,
    ) ([]byte, error)
    SendSignal(runID, name string, data []byte) error
}

type SimpleTask

SimpleTask is the minimal interface for basic task handlers. Most handlers only need Input, Complete, and Fail.

type SimpleTask interface {
    Input() []byte
    RunID() string
    StepID() string
    Context() context.Context
    Complete(output []byte) error
    Fail(err error) error
    FailPermanent(err error) error
}

type StartOpt

StartOpt configures one StartRun call.

type StartOpt func(*startOpts)

func WithSpawnKey 

func WithSpawnKey(key string) StartOpt

WithSpawnKey makes StartRun idempotent per (owning run, step, key): a replayed tool call carrying the same key gets the original child run ID and no second child is spawned (#803). The key is scoped to the calling step and expires after 24h. Keys longer than SpawnKeyMaxBytes are rejected with KindInvalidDef; an empty key means no idempotency.

type StreamTask

StreamTask adds streaming output and heartbeat for handlers that produce incremental results or need keep-alive signals.

type StreamTask interface {
    SimpleTask
    PutStream(data []byte) error
    Heartbeat() error
}

type TaskContext

TaskContext is the interface workers use to interact with the DagNats engine. Includes step completion, checkpointing, signals, and streaming. Workers call exactly one of Complete, Fail, or Continue per execution.

Checkpoint and signal methods depend on optional KV buckets (“checkpoints” and “signals”). They return an error if the bucket was not provisioned at startup — check your natsutil.SetupAll call.

BREAKING CHANGE (#624): LogOut() io.Writer and LogErr() io.Writer were added to this interface. Any out-of-repo Go type that implements TaskContext directly (rather than embedding one dagnats already constructs, e.g. via a handler-wrapping test double) stops compiling against this version until it adds both methods. The simplest fix for a test double or thin wrapper that doesn’t care about log capture is to return io.Discard from both:

func (m *myFakeTaskContext) LogOut() io.Writer { return io.Discard }
func (m *myFakeTaskContext) LogErr() io.Writer { return io.Discard }

A type that embeds worker.TaskContext (or a *testing double this module already provides, e.g. dagnatstest.MockTaskContext) needs no change — it inherits the new methods automatically.

type TaskContext interface {
    // Step identity and input
    Input() []byte
    RunID() string
    StepID() string
    RetryCount() int
    // Metadata returns static per-step metadata from the workflow step
    // definition; nil if the step declared none.
    Metadata() map[string]string
    // Context ends when the run is cancelled (#802). A handler must
    // honour it: stop work and return, or the work is wasted.
    Context() context.Context

    // Step completion — call exactly one per execution
    Complete(output []byte) error
    Fail(err error) error
    FailPermanent(err error) error
    FailRetryAfter(err error, after time.Duration) error
    // Continue asks for another iteration. opts (Delay, Until,
    // UntilInput) make the engine wait first, durably, with no worker
    // claim held (ADR-027).
    Continue(output []byte, opts ...ContinueOption) error

    // PlaceInputs moves the queued inputs of mode into the
    // conversation, exactly once each, and returns them (ADR-027).
    // Call it at a turn boundary: steers after a tool round,
    // follow-ups after a final answer. Needs Journal() first.
    PlaceInputs(mode inputs.Mode) ([]journal.InputPlaced, error)

    // Streaming and heartbeat. PutStream is a non-durable preview
    // (ADR-026): a journal step that crashes mid-stream re-sends the
    // request; Journal().Model holds the durable response.
    PutStream(data []byte) error
    Heartbeat() error

    // LogOut and LogErr (#624) publish stdout/stderr-tagged chunks to
    // the BUILD_LOGS hot lane (logs.{runID}.{stepID}.{attempt}.{iteration}
    // — scoped to THIS attempt and iteration so a retry or agent-loop
    // Continue never collides with a prior dispatch's sequence numbers
    // within the stream's dedup window), buffered and flushed at
    // protocol.LogChunkBytesMax or after 250ms. Complete, Fail,
    // FailPermanent, FailRetryAfter, Continue, and Pause all drain any
    // buffered bytes AND emit an attempt-ending marker
    // (completed/failed/continued/paused) before publishing their own
    // resolution, so a consumer that observes the terminal event never
    // misses trailing log output and can treat the marker as a
    // reliable "this attempt is over" signal.
    //
    // The writers LogOut/LogErr return ARE safe to call concurrently —
    // each Write() holds the lane's own mutex — which matters because
    // the canonical use is handing them straight to an external
    // process: `cmd.Stdout = tc.LogOut(); cmd.Stderr = tc.LogErr()`
    // then `cmd.Run()`, where os/exec copies each stream on its own
    // goroutine.
    LogOut() io.Writer
    LogErr() io.Writer

    // Checkpointing — save/restore handler state across retries
    Checkpoint(state []byte) error
    LoadCheckpoint() ([]byte, error)
    Pause(name string, duration time.Duration) error

    // Journal claims this dispatch's agent journal (ADR-026) and
    // returns the handle for appending records. Call it first: from
    // then on Continue, Complete and Fail* commit a journal record
    // before their event, Checkpoint and LoadCheckpoint use the
    // journal's state, and Pause writes a pause record. The handle's
    // Tool and Model run tool calls and model requests durably, reused
    // on recovery. Return its error unchanged: journal.ErrStale,
    // ErrFenced, ErrCommitted and ErrRestartLimit mean this delivery
    // must do nothing more, and the Worker settles the task
    // accordingly.
    Journal() (*journal.Journal, error)

    // Signals — coordinate between steps
    WaitForSignal(
        name string, timeout time.Duration,
    ) ([]byte, error)
    SendSignal(runID, name string, data []byte) error

    // ControlPlane returns the runtime control-plane handle for this
    // step, or nil. It is nil unless the step declared the
    // "control-plane" capability AND the deployment granted one via
    // WithControlPlane — deny-by-default. Always nil-check before use.
    ControlPlane() ControlPlane
}

type TriggerEnvelope

TriggerEnvelope is the standard outer envelope every trigger publishes. Bind it when the worker needs the trigger metadata (kind, source, workflow_id, timestamp) alongside the inner data.

type TriggerEnvelope = trigger.TriggerEnvelope

type TypedHandlerFunc

TypedHandlerFunc is a task handler with typed input and output. The worker.Typed wrapper handles JSON marshal/unmarshal so handlers work with concrete Go types instead of raw []byte.

type TypedHandlerFunc[I, O any] func(ctx TaskContext, input I) (O, error)

type TypedOption

TypedOption configures the Typed/HandleTyped wrapper. Variadic on Typed and HandleTyped keeps existing call sites source-compatible. Distinct from HandlerOption (which mutates the Worker at registration time): TypedOption mutates only the in-memory wrapper config.

type TypedOption func(*typedConfig)

func UnwrapTrigger 

func UnwrapTrigger() TypedOption

UnwrapTrigger asks the Typed wrapper to auto-detect trigger envelopes in the task input and unmarshal the typed parameter from the envelope’s `data` field instead of the raw input. Auto-detect is structural: if the input is a JSON object with both a top-level `trigger` string AND a top-level `data` field, the input is treated as an envelope and `data` is extracted as the unmarshal source. Otherwise the input passes through unchanged — plain non-envelope inputs still work, e.g. during local unit tests or when the workflow is invoked directly without a trigger.

Metadata access (trigger kind, source, timestamp) is out of scope for v1. Workers that need those fields should drop to ctx.Input() and unmarshal the full envelope manually. See issue #229 for the path to first-class metadata access.

type Worker

Worker subscribes to task subjects and dispatches messages to registered handlers. Each task type gets its own JetStream subscription; messages are ack’d after the handler returns so failures are retried by JetStream’s MaxDeliver policy.

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

func NewWorker 

func NewWorker(nc *nats.Conn, opts ...WorkerOption) *Worker

NewWorker creates a Worker using the given connection. Panics if nc is nil or if JetStream cannot be initialised — both are programmer errors at startup. A W3C TraceContext+Baggage propagator is installed on the global OTel registry if the global is still the no-op default; an already-installed propagator (custom or otherwise) is never overwritten. Tracing and metrics use the global OTel providers (noop by default).

func (*Worker) Handle 

func (w *Worker) Handle(taskType string, handler HandlerFunc, opts ...HandlerOption)

Handle registers a HandlerFunc for the given task type. Optional HandlerOptions (e.g. WithAckWait) tune per-task knobs. Panics on empty taskType or nil handler — both are programmer errors.

func (*Worker) HandleSingleton 

func (w *Worker) HandleSingleton(taskType string, handler HandlerFunc)

HandleSingleton registers a handler that runs as a single- partition elastic consumer group. Only one consumer processes messages at a time across all worker instances. Implicitly enables partitioned mode if not already configured.

func (*Worker) RegisterService 

func (w *Worker) RegisterService(def ServiceDef) error

RegisterService publishes service metadata to the `services` KV bucket. Last-write-wins: re-calling with different Description (or any other field) silently replaces the prior entry without error. This is intentional — the bucket is a descriptive surface, not an authoritative registry, and worker restarts must be safe to repeat without conflict-handling boilerplate at every call site.

Idempotency contract:

  • Two calls with identical def → identical KV state.
  • Two calls with the same Name but different Description → second call’s Description wins, no error returned.
  • Concurrent calls race to the latest Put; no locking, no compare- and-swap. Callers needing strict serialization should layer it above.

Panics on empty Name (programmer error). Stamps RegisteredAt on every call so callers don’t have to.

func (*Worker) RegisterTriggerType 

func (w *Worker) RegisterTriggerType(ctx context.Context, def dagnatsext.TriggerTypeDef) error

RegisterTriggerType publishes def into the trigger_types KV bucket, then requests engine acknowledgement on `_REGISTRY.trigger_types.ack` (#330). Returns the engine’s error verbatim on failure.

Side effects:

  • Sets def.OwnerWorkerID to w.WorkerID() when empty so callers don’t have to look it up.
  • Sets def.RegisteredAt to time.Now().UTC() when zero.

Idempotent at the engine side per the ack contract (#327): same Name + same OwnerWorkerID + same ConfigSchema → nil. Schema drift → error. Owner drift → error.

func (*Worker) Start 

func (w *Worker) Start()

Start creates JetStream subscriptions for all registered task types. Panics if any subscription fails — stream misconfiguration is a startup error. Binds optional KV buckets for checkpoints and signals (nil if not present). When groups are configured, subscribes to group-specific subjects.

func (*Worker) Stop 

func (w *Worker) Stop()

func (*Worker) WatchTriggers 

func (w *Worker) WatchTriggers(ctx context.Context, kind string, onActivate triggerHandler, onDeactivate triggerHandler) error

WatchTriggers subscribes to `_TRIGGER.<kind>.activate` and `_TRIGGER.<kind>.deactivate`, decodes each request into a trigger.TriggerDef, and invokes the corresponding callback. The callback’s returned error is reported back to the engine via a `{“error”:"…"}` reply; a nil error replies with an empty body.

Catch-up: before returning, the worker scans the `triggers` KV bucket and fires onActivate for every Enabled entry whose External.Kind == kind. Bounded at maxCatchupKeys (10000). Over-cap → warning logged and scan stopped (no fires beyond the cap).

Lifecycle: both NATS subscriptions are appended to w.triggerSubs and drained by Worker.Stop(). Callers do not unsubscribe directly.

type WorkerOption

WorkerOption configures optional Worker behavior.

type WorkerOption func(*Worker)

func WithControlPlane 

func WithControlPlane(cp ControlPlane) WorkerOption

WithControlPlane grants this worker the runtime control plane: gated steps (those declaring the “control-plane” capability) will receive a per-step handle via TaskContext.ControlPlane(). Without this option the field stays nil and every step’s ControlPlane() returns nil — deny-by-default is structural, not a runtime check. Panics if cp is nil (a deployment that wires the grant must supply a real handle).

func WithGroups 

func WithGroups(groups ...string) WorkerOption

WithGroups configures the worker to subscribe only to specific worker groups. When provided, the worker subscribes to task.{taskType}.{group}.> instead of task.{taskType}.>.

func WithPartitions 

func WithPartitions(n int) WorkerOption

WithPartitions configures pcgroups elastic consumer groups with the given partition count. 0 = legacy consumer (default).

func WithRestartLimit 

func WithRestartLimit(n int) WorkerOption

WithRestartLimit sets how many deliveries may claim one iteration of a journal step without progress before the step fails with reason restart_limit (ADR-028). A pause counts as progress. Default 5.

type WorkerRegistration

WorkerRegistration is the directory entry for a running worker. The directory is observability-only — the engine never reads it. Workers register on startup and maintain their entry via periodic heartbeat writes (the KV bucket has a 60s TTL).

Identity & heartbeat fields (LastSeen, Pid, Hostname, Version) make the existing workers bucket double as a heartbeat surface — avoiding a parallel worker_heartbeats bucket (#289). LastSeen is stamped by Register on every write, so each periodic heartbeat tick advances it automatically. All four fields use omitempty so older payloads written before this struct grew (zero-valued) deserialise cleanly.

type WorkerRegistration struct {
    WorkerID  string            `json:"worker_id"`
    TaskTypes []string          `json:"task_types"`
    Language  string            `json:"language"`
    Transport string            `json:"transport"`
    MaxTasks  int               `json:"max_tasks"`
    Metadata  map[string]string `json:"metadata,omitempty"`
    // WorkerGroups names the worker groups this worker drains (#719),
    // so /v1/workers can answer "who serves group G". Plural because an
    // SDK worker may drain several (WithGroups). Empty means the
    // ungrouped queue. omitempty so a registration written before #719
    // reads back unchanged.
    WorkerGroups []string `json:"worker_groups,omitempty"`

    // Identity — populated once at worker boot, stable for the life
    // of the process.
    Pid      int    `json:"pid,omitempty"`
    Hostname string `json:"hostname,omitempty"`
    Version  string `json:"version,omitempty"`

    // LastSeen is the wall-clock timestamp of the most recent write
    // to the KV bucket. Register stamps this on every call, so the
    // periodic heartbeat naturally refreshes it. Readers compare it
    // to time.Now() to gauge worker liveness without depending on
    // NATS KV's TTL-eviction latency.
    LastSeen time.Time `json:"last_seen,omitempty"`

    // TokenID identifies the workertoken.Token the worker presented to
    // the HTTP bridge's /v1/workers/connect, if any (#627). Empty for
    // the env-token admin path, dev mode, or non-bridge (native NATS)
    // workers -- it exists purely so an operator can see, from the
    // directory, which minted token a given worker is using.
    TokenID string `json:"token_id,omitempty"`
}

Generated by gomarkdoc