protocol

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Aug 7, 2026 License: Apache-2.0 Imports: 3 Imported by: 0

Documentation

Overview

Package protocol is the Duraton engine<->runner wire contract for the Go SDK. The types mirror sdk/src/protocol/wire.ts and engine/internal/protocol/wire.go field-for-field; the JSON tags are the canonical wire names. It is a standalone, exported copy because the engine's protocol package is internal/ and cannot be imported from outside that module (see DECISIONS.md, decision 1).

Index

Constants

View Source 
const (
	MinVersion = 1
	MaxVersion = 1
)

MinVersion and MaxVersion bound the closed set of body versions this SDK can speak, advertised as minProtocolVersion/maxProtocolVersion on the handshake. They are the single source of truth for the range; every derived site (Negotiate, hello/register payloads) reads them. Day-one MinVersion == MaxVersion == Version == 1, so the range [1,1] negotiates to exactly 1 and nothing observable changes.

View Source 
const MaxMessageBytes = 1 << 20

MaxMessageBytes caps a single engine<->runner wire message (1 MiB). Both transports enforce it; the Connect client sets it as the socket read limit.

View Source 
const MaxNameLen = 256

MaxNameLen bounds event names and routing identifiers (app, runner, targetApp).

View Source 
const RootScope = "@root"

RootScope is the log scope of a handler-level (non-step) log line.

View Source 
const Version = 1

Version is the protocol body version both sides advertise (hello.protocolVersion, runner.register.protocolVersion). v1 wraps every invoke-result body in an object so logs ride alongside the result. It is distinct from the frozen subprotocol string (connect.Subprotocol = "duraton.connect.v0"), which names the transport framing.

View Source 
const VersionHeader = "X-Duraton-Protocol"

VersionHeader carries Version on the HTTP invoke POST (engine -> runner serve transport). The Connect transport carries the version in the hello/register bodies, not a header.

Variables

View Source 
var BackoffStrategies = []BackoffStrategy{BackoffFixed, BackoffLinear, BackoffExponential}

BackoffStrategies is the closed set of retry backoff shapes.

View Source 
var LogLevels = []LogLevel{LogDebug, LogInfo, LogWarn, LogError}

LogLevels is the closed set of log levels.

View Source 
var Ops = []Op{
	OpStepRun, OpSkip, OpSleep, OpSleepUntil, OpWaitForEvent, OpRunWorkflow,
	OpEmit, OpWebhook, OpApproval, OpInfer, OpScore,
}

Ops is the closed set of structural opcodes, for exhaustive switching and validation.

Functions

func CompatibleVersion 

func CompatibleVersion(v *int) bool

CompatibleVersion reports whether a peer-advertised body version is compatible with ours. The rule is strict-equality, lenient-on-absence: a nil pointer means the peer advertised none (predates versioning) and is treated as compatible; any present value must equal Version. Mirrors the engine's CompatibleVersion(*int) and the TS compatibleVersion, and is the source of truth for golden vectors VV-01..VV-06.

A polyglot SDK MUST implement exactly this behavior; it MUST NOT invent a range or capability negotiation (that is deferred to a future protocol task).

func HashStepID 

func HashStepID(stepID string) string

HashStepID maps a human-readable (deduplicated) step id to the opaque memo key used in InvokeRequest.Steps and Opcode.ID. It MUST produce byte-identical output to the TS sdk/src/protocol/hash.ts and the engine's hash.go: lowercase hex of the SHA-256 of the id's raw UTF-8 bytes, with no salt, separators, length prefix, or Unicode normalization. Go strings are already UTF-8, so []byte(stepID) is the UTF-8 encoding; do not transcode or normalize before hashing.

func Negotiate 

func Negotiate(selfMin, selfMax, peerLo, peerHi int) (int, bool)

Negotiate intersects our [selfMin, selfMax] with the peer's [peerLo, peerHi] and picks the highest common version. It returns (hi, true) on overlap, or (0, false) for NO_OVERLAP - the caller surfaces that as a typed error, never a silent drop.

func ResolvePeerRange 

func ResolvePeerRange(peerMin, peerMax, peerScalar *int) (lo, hi int)

ResolvePeerRange reduces a peer's handshake fields to a [lo, hi] body-version range. A present maxProtocolVersion wins (the range form); else a present protocolVersion is a scalar-only legacy peer treated as [v, v]; else an absent peer predates versioning and is assumed to speak exactly Version (the original v1). Source of truth for golden vectors VV-07..VV-13 (versionNegotiation.negotiate).

Types

type BackoffStrategy 

type BackoffStrategy string

BackoffStrategy is the closed set of retry backoff shapes. Fixed = constant delay; linear = delay*attempt; exponential = delay*2^(attempt-1). All capped at MaxDelayMs. 

const (
	BackoffFixed       BackoffStrategy = "fixed"
	BackoffLinear      BackoffStrategy = "linear"
	BackoffExponential BackoffStrategy = "exponential"
)

type BatchConfig 

type BatchConfig struct {
	MaxSize   int    `json:"maxSize"`
	TimeoutMs int64  `json:"timeoutMs"`
	Key       string `json:"key,omitempty"`
}

type BudgetConfig 

type BudgetConfig struct {
	MaxCost   *float64 `json:"maxCost,omitempty"`
	MaxTokens *int64   `json:"maxTokens,omitempty"`
	WindowMs  int64    `json:"windowMs"`
	WarnAtPct int      `json:"warnAtPct,omitempty"`
}

BudgetConfig is a rolling-window ceiling on a workflow scope's aggregate AI spend.

type CapConfig 

type CapConfig struct {
	MaxCost   *float64 `json:"maxCost,omitempty"`
	MaxTokens *int64   `json:"maxTokens,omitempty"`
}

CapConfig is a per-run hard ceiling on AI spend. At least one axis is set.

type ConcurrencyConfig 

type ConcurrencyConfig struct {
	Limit int    `json:"limit"`
	Key   string `json:"key,omitempty"`
}

type DebounceConfig 

type DebounceConfig struct {
	PeriodMs int64  `json:"periodMs"`
	Key      string `json:"key,omitempty"`
}

type EvalContext 

type EvalContext struct {
	SetID    string          `json:"setId"`
	ItemID   string          `json:"itemId"`
	Expected json.RawMessage `json:"expected,omitempty"`
}

EvalContext projects an eval run's identity and the source item's expected output.

type Event 

type Event struct {
	Name string          `json:"name"`
	Data json.RawMessage `json:"data"`
}

Event is the trigger payload carried into every invoke. Data is opaque JSON passthrough; the runner decodes it into its own type.

type ForkOverride 

type ForkOverride struct {
	Step   string          `json:"step"`
	Model  string          `json:"model,omitempty"`
	Prompt string          `json:"prompt,omitempty"`
	Params json.RawMessage `json:"params,omitempty"`
}

ForkOverride is the wire form of a fork-compare declared change.

type IdempotencyConfig 

type IdempotencyConfig struct {
	Key      string `json:"key,omitempty"`
	PeriodMs int64  `json:"periodMs,omitempty"`
}

type InvokeRequest 

type InvokeRequest struct {
	Event Event               `json:"event"`
	Steps map[string]StepMemo `json:"steps"`
	Ctx   RunCtx              `json:"ctx"`
}

InvokeRequest is engine -> runner: replay one pass.

type LogLevel 

type LogLevel string

LogLevel is the closed set of structured-log severities. 

const (
	LogDebug LogLevel = "debug"
	LogInfo  LogLevel = "info"
	LogWarn  LogLevel = "warn"
	LogError LogLevel = "error"
)

type LogLine 

type LogLine struct {
	Level   LogLevel        `json:"level"`
	Message string          `json:"message"`
	Fields  json.RawMessage `json:"fields,omitempty"`
	Scope   string          `json:"scope"`
	Index   int             `json:"index"`
	TsMs    int64           `json:"tsMs,omitempty"`
}

LogLine is one structured log entry a runner emitted during a pass. Scope is the enclosing step name (or RootScope for handler-level logs) and Index a per-scope counter; the engine derives a replay-stable dedupe id from (runId, attempt, scope, index) so a top-level log re-emitted each pass persists once. TsMs is the runner's wall clock, advisory only, and is an integer epoch-millis on the wire.

type Op 

type Op string

Op is a structural wire opcode: a directive the runner emits for a newly-discovered step. There are exactly eleven, all structural (each has a dedicated engine applier). The AI surface maps onto these - generate/wrap/embed/loop ride OpStepRun; only Infer and Score are dedicated - so there is no distinct "AI" opcode. 

const (
	OpStepRun Op = "StepRun"
	// OpSkip records a step the workflow deliberately bypassed (Skip): the runner makes
	// no call and carries an optional reason in Data. The engine records a terminal
	// skipped step, so a bypassed stage shows as skipped rather than vanishing.
	OpSkip         Op = "Skip"
	OpSleep        Op = "Sleep"
	OpSleepUntil   Op = "SleepUntil"
	OpWaitForEvent Op = "WaitForEvent"
	OpRunWorkflow  Op = "RunWorkflow"
	OpEmit         Op = "Emit"
	OpWebhook      Op = "Webhook"
	OpApproval     Op = "Approval"
	// OpInfer is an offloaded model call: the runner emits it (carrying the request in
	// Data) and is then free; the engine makes the call and completes the step.
	OpInfer Op = "Infer"
	// OpScore records an evaluation score on the run. The runner makes no external
	// call; it carries the score payload in Data and the engine writes the score row.
	OpScore Op = "Score"
)

type Opcode 

type Opcode struct {
	Op   Op              `json:"op"`
	ID   string          `json:"id"`
	Name string          `json:"name"`
	Data json.RawMessage `json:"data,omitempty"`
	// AI is the opaque model journal block a step.ai.* call reports; the engine journals
	// it verbatim. Input is the runner-supplied input of a StepRun (the opt-in form).
	AI    json.RawMessage `json:"ai,omitempty"`
	Input json.RawMessage `json:"input,omitempty"`
	Error *StepError      `json:"error,omitempty"`
	// Retriable/RetryAfterMs are retry control on a failed StepRun: Retriable=false
	// (NonRetriableError) fails the run now; RetryAfterMs (RetryAfterError) overrides the
	// policy backoff. Pointers so an absent value is omitted, not sent as false/0.
	Retriable    *bool  `json:"retriable,omitempty"`
	RetryAfterMs *int64 `json:"retryAfterMs,omitempty"`
	// Retry is a per-step retry policy declared via Run's retry option. When set it
	// governs this step's attempt budget and backoff instead of the workflow default, so
	// a polling step retries independently of the rest of the run.
	Retry        *RetryConfig `json:"retry,omitempty"`
	SleepMs      int64        `json:"sleepMs,omitempty"`
	SleepUntilMs int64        `json:"sleepUntilMs,omitempty"`
	EventName    string       `json:"eventName,omitempty"`
	TimeoutMs    int64        `json:"timeoutMs,omitempty"`
	// EventIf is an optional CEL predicate on a WaitForEvent: the run resumes only on an
	// event whose name matches AND whose payload satisfies EventIf, using the same CEL
	// dialect as workflow event triggers. Empty means name-match only.
	EventIf string `json:"eventIf,omitempty"`
	// ChildName/ChildData/ChildApp/ChildRunner address a runWorkflow; ChildTags is
	// customer-defined key/value metadata stamped on the child run.
	ChildName   string            `json:"childName,omitempty"`
	ChildData   json.RawMessage   `json:"childData,omitempty"`
	ChildApp    string            `json:"childApp,omitempty"`
	ChildRunner string            `json:"childRunner,omitempty"`
	ChildTags   map[string]string `json:"childTags,omitempty"`
	// TargetApp narrows an emit to one app's triggers; empty broadcasts namespace-wide.
	TargetApp string `json:"targetApp,omitempty"`
	// DedupeID drops a repeat of the same emitted event (per app) within the dedupe
	// window - the same idempotency key POST /events accepts.
	DedupeID string `json:"dedupeId,omitempty"`
	// WebhookURL is the destination of a Webhook opcode.
	WebhookURL string `json:"webhookUrl,omitempty"`
	// Approval fields: a run parks on a human decision for Tool with proposed args in
	// Data, at Risk, gated by Policy, with Summary/Context for the decider and
	// EscalatesTo the escalation target. TimeoutMs is the escalation deadline.
	Tool        string `json:"tool,omitempty"`
	Risk        string `json:"risk,omitempty"`
	Policy      string `json:"policy,omitempty"`
	Summary     string `json:"summary,omitempty"`
	Context     string `json:"context,omitempty"`
	EscalatesTo string `json:"escalatesTo,omitempty"`
	// Usage is the explicit metering axes a step.ai.* call reports; the engine meters
	// these directly rather than parsing the opaque AI journal.
	Usage *Usage `json:"usage,omitempty"`
}

Opcode is a single directive the runner emits for a newly-discovered step. Index and attempt are assigned by the engine, never the runner, so they are absent here.

type PriorityConfig 

type PriorityConfig struct {
	ShiftMs int64 `json:"shiftMs"`
}

type RateConfig 

type RateConfig struct {
	Limit int    `json:"limit"`
	PerMs int64  `json:"perMs"`
	Key   string `json:"key,omitempty"`
}

type RegisterRequest 

type RegisterRequest struct {
	App                string             `json:"app"`
	Runner             string             `json:"runner,omitempty"`
	URL                string             `json:"url"`
	Runtime            string             `json:"runtime,omitempty"`
	Language           string             `json:"language,omitempty"`
	SDKName            string             `json:"sdkName,omitempty"`
	Version            string             `json:"version,omitempty"`
	Framework          string             `json:"framework,omitempty"`
	Region             string             `json:"region,omitempty"`
	KeyFingerprint     string             `json:"keyFingerprint,omitempty"`
	ProtocolVersion    *int               `json:"protocolVersion,omitempty"`
	MinProtocolVersion *int               `json:"minProtocolVersion,omitempty"`
	MaxProtocolVersion *int               `json:"maxProtocolVersion,omitempty"`
	Workflows          []RegisterWorkflow `json:"workflows"`
}

RegisterRequest is runner -> engine on startup (the HTTP serve form). The Connect transport sends registerPayload, which omits URL and KeyFingerprint.

type RegisterWorkflow 

type RegisterWorkflow struct {
	Name          string               `json:"name"`
	Triggers      []Trigger            `json:"triggers,omitempty"`
	Retry         *RetryConfig         `json:"retry,omitempty"`
	Concurrency   *ConcurrencyConfig   `json:"concurrency,omitempty"`
	Throttle      *RateConfig          `json:"throttle,omitempty"`
	RateLimit     *RateConfig          `json:"rateLimit,omitempty"`
	Debounce      *DebounceConfig      `json:"debounce,omitempty"`
	Batch         *BatchConfig         `json:"batch,omitempty"`
	Priority      *PriorityConfig      `json:"priority,omitempty"`
	Singleton     *SingletonConfig     `json:"singleton,omitempty"`
	Idempotency   *IdempotencyConfig   `json:"idempotency,omitempty"`
	Cap           *CapConfig           `json:"cap,omitempty"`
	Budget        *BudgetConfig        `json:"budget,omitempty"`
	TokenThrottle *TokenThrottleConfig `json:"tokenThrottle,omitempty"`
	OnFailure     bool                 `json:"onFailure,omitempty"`
}

RegisterWorkflow is one workflow entry in a registration. Flow-control fields are flat and optional; durations are integer milliseconds on the wire.

type RetryConfig 

type RetryConfig struct {
	MaxAttempts    int             `json:"maxAttempts"`
	Backoff        BackoffStrategy `json:"backoff,omitempty"`
	InitialDelayMs *int64          `json:"initialDelayMs,omitempty"`
	MaxDelayMs     *int64          `json:"maxDelayMs,omitempty"`
}

RetryConfig is a retry policy. MaxAttempts is the total attempt budget (>= 1); Backoff shapes the inter-attempt delay (default fixed) and InitialDelayMs/MaxDelayMs bound it (nil -> engine defaults). Used workflow-wide (RegisterWorkflow) and per-step (a StepRun opcode via Run's retry option).

type RunCtx 

type RunCtx struct {
	RunID     string     `json:"runId"`
	Workflow  string     `json:"workflow,omitempty"`
	Attempt   int        `json:"attempt"`
	App       string     `json:"app"`
	Runner    string     `json:"runner,omitempty"`
	OnFailure bool       `json:"onFailure,omitempty"`
	Error     *StepError `json:"error,omitempty"`
	// Spent is the run's accumulated AI spend across already-committed steps, the
	// runner's only view of prior-pass spend (the memo carries outputs, not usage), used
	// to enforce a per-run Cap before the exceeding call. Nil when no model call yet.
	Spent *RunSpend `json:"spent,omitempty"`
	// Cap is the effective per-run cap resolved upstream, overriding the workflow-declared
	// cap. Honored if present; the engine does not populate it today (see DECISIONS.md D3).
	Cap *CapConfig `json:"cap,omitempty"`
	// Traceparent is the W3C trace context of this invoke pass's engine span, carried in
	// the body (not a header) so HTTP and WebSocket propagate identically.
	Traceparent string `json:"traceparent,omitempty"`
	// Fork is the one declared change a fork-compare shadow run applies. Eval is the
	// dataset item + expected output of an eval run. Both nil on live runs.
	Fork *ForkOverride `json:"fork,omitempty"`
	Eval *EvalContext  `json:"eval,omitempty"`
}

RunCtx is the per-invoke run context. Workflow is the dispatch key (the registered workflow name), distinct from Event.Name. A legacy run may omit it, in which case the event name is the dispatch key.

type RunSpend 

type RunSpend struct {
	Cost   *float64 `json:"cost,omitempty"`
	Tokens *int64   `json:"tokens,omitempty"`
}

RunSpend is a run's rolled-up AI spend. Cost is nil until a call supplied one (the engine never prices); Tokens is nil when no model call reported usage. Pointers keep int64/float64 precision and distinguish "not reported" from a real zero.

type SingletonConfig 

type SingletonConfig struct {
	Key  string `json:"key,omitempty"`
	Mode string `json:"mode,omitempty"`
}

type StepError 

type StepError struct {
	Message string `json:"message"`
	Stack   string `json:"stack,omitempty"`
}

StepError is a serialized failure from a step or workflow.

type StepMemo 

type StepMemo struct {
	Data    json.RawMessage `json:"data,omitempty"`
	Error   *StepError      `json:"error,omitempty"`
	Pending bool            `json:"pending,omitempty"`
}

StepMemo replays a step's state to the runner. A completed step carries its Data (or Error); Pending marks a step the engine has started but not finished (a parked sleep/waitForEvent/runWorkflow/approval), which the runner must neither re-execute nor re-emit - it stays unresolved this pass, keeping a parallel branch's timer alive across re-invokes driven by a sibling.

type TokenThrottleConfig 

type TokenThrottleConfig struct {
	Tokens int64  `json:"tokens"`
	PerMs  int64  `json:"perMs"`
	Key    string `json:"key,omitempty"`
}

TokenThrottleConfig is a token-denominated AI throttle.

type Trigger 

type Trigger struct {
	Event string `json:"event,omitempty"`
	If    string `json:"if,omitempty"`
	Cron  string `json:"cron,omitempty"`
	// RunOnStart makes a cron trigger also fire once immediately on registration (each
	// runner startup/deploy), for a catch-up run; without it cron is skip-and-forward.
	RunOnStart bool `json:"runOnStart,omitempty"`
}

Trigger is one entry in a workflow's trigger array: an event trigger (Event set, optional CEL If filter, event may end in a trailing "*" wildcard) or a cron trigger (Cron set, optional "TZ=Area/City" prefix). Exactly one of Event/Cron is non-empty.

type Usage 

type Usage struct {
	Model     string   `json:"model,omitempty"`
	TokensIn  *int64   `json:"tokensIn,omitempty"`
	TokensOut *int64   `json:"tokensOut,omitempty"`
	Cost      *float64 `json:"cost,omitempty"`
	LatencyMs *int64   `json:"latencyMs,omitempty"`
	CacheHit  *bool    `json:"cacheHit,omitempty"`
}

Usage is the abstract metering axes of a model call. Pointers distinguish "not reported" (nil, excluded from sums) from a real zero, and preserve int64 exactly.

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.