invoker

type Cache ¶

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

Cache is an Invoker decorator that makes invocations idempotent under packtrail's at-least-once delivery. Every step is driven by a durable work message; if an engine crashes after invoking a node but before persisting the advance/ack, the work item is redelivered and the node would otherwise run twice — double side effects (a re-billed LLM call, a duplicate write, a second e-mail).

Cache keys a stored Result by (execution, node, attempt). A redelivery of the same attempt is served from the cache and never reaches the delegate, while a genuine retry (a new attempt number) gets a fresh key and does re-run, exactly as the node's retry policy intends. Transport errors are never cached, so a failed call is always retried.

Cache solves the engine-side double-dispatch window. It cannot make a non-deterministic task deterministic; an Invoker with external side effects it cannot see should still carry its own idempotency key where it can.

func NewCache(kv jetstream.KeyValue, delegate Invoker) *Cache

func (c *Cache) Invoke(ctx context.Context, req Request) (Result, error)

type Func ¶

type Func func(ctx context.Context, req Request) (Result, error)

Func adapts a plain function to the Invoker interface.

func (f Func) Invoke(ctx context.Context, req Request) (Result, error)

type Invoker ¶

type Invoker interface {
	Invoke(ctx context.Context, req Request) (Result, error)
}

Invoker executes a single node invocation. Implementations must be safe for concurrent use: the engine invokes many nodes in parallel.

type Registry ¶

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

Registry dispatches an invocation to a registered Invoker by kind (Request.Invoker). It is itself an Invoker, so the engine holds exactly one Invoker regardless of how many kinds are configured.

func NewRegistry() *Registry

func (r *Registry) Has(kind string) bool

func (r *Registry) Invoke(ctx context.Context, req Request) (Result, error)

func (r *Registry) Register(kind string, inv Invoker)

type Request ¶

type Request struct {
	Invoker     string          `json:"invoker"`      // invoker kind selected for this node
	Target      string          `json:"target"`       // invoker-specific target (resolved)
	ExecutionID string          `json:"execution_id"` // owning execution
	NodeID      string          `json:"node_id"`      // node being executed
	Payload     json.RawMessage `json:"payload"`      // shared execution context
	Attempt     int             `json:"attempt"`      // 0-based attempt number
	Deadline    time.Time       `json:"deadline"`     // hard deadline for this attempt
}

Request is everything an Invoker needs to execute one node invocation. It is transport-agnostic: Target is interpreted by the chosen Invoker (a subject, an agent name, a URL, …) and any {execution_id} placeholder is already resolved.

type Result ¶

type Result struct {
	Status  Status          `json:"status"`
	Payload json.RawMessage `json:"payload,omitempty"`
	Error   string          `json:"error,omitempty"`
}

Result is what an Invoker returns. A non-nil error from Invoke is treated as a transient transport failure (equivalent to StatusRetry) and is never cached.

type Status ¶

type Status string

Status is the outcome an Invoker reports for a single invocation. The string values intentionally match the built-in nats-task wire contract so transports can map onto them directly.

const (
	// StatusOK means the node succeeded; Result.Payload is the new shared context.
	StatusOK Status = "ok"
	// StatusError means the node failed permanently; the engine does not retry.
	StatusError Status = "error"
	// StatusRetry means the node asks to be retried per its node retry policy.
	StatusRetry Status = "retry"
	// StatusPending means the node was dispatched asynchronously and will be
	// settled later via Engine.CompleteActivity. The engine parks the execution
	// (waiting) and frees its work slot; Result.Payload is ignored. Use this for
	// long-running activities (e.g. an agent call) so the engine does not block.
	StatusPending Status = "pending"
)