graph

type Annotations ¶

type Annotations struct {
	ReadOnlyHint    *bool
	IdempotentHint  *bool
	DestructiveHint *bool
	OpenWorldHint   *bool
}

Annotations are the protocol's own hints, "passed through unchanged. They are hints to a client, never a permission: what a tool may do is decided by the grants, not by a field in the file."

Each one is a pointer because a hint the file does not write is not the same as a hint written false, and what is published is what was written.

type Backoff ¶

type Backoff struct {
	Type BackoffType
	Base Duration
	Max  Duration
}

Backoff is how the wait between attempts grows. "exponential is the only type there is: the wait starts at base, lengthens after each failure and never goes past max."

type BackoffType ¶

type BackoffType int

BackoffType is the shape of the wait. There is one, and it is named rather than assumed so that a file saying type: exponential reads as what it is.

const BackoffExponential BackoffType = iota

func (b BackoffType) String() string

type Call ¶

type Call struct {
	Workflow string
	Ref      string
}

Call is a step that calls a sub-workflow instead of running a container: "workflow: <namespace>/<name>, with @<ref> appended to pin a tag or a commit, or in long form { workflow, ref }".

type Concurrency ¶

type Concurrency struct {
	Group            string
	CancelInProgress bool
}

Concurrency is the policy a trigger follows when a run of the same workflow is still going. "Concurrency groups are scoped to the namespace, so one namespace can never block or cancel another's runs."

type Defaults ¶

type Defaults struct {
	Timeout         *Duration
	Retain          *Retain
	Retry           *Retry
	Resources       *Resources
	Network         *Network
	EgressAllow     []string
	RunsOn          []string
	Secrets         []string
	Cache           *bool
	ContinueOnError *bool
	Idempotent      *bool
	Files           []FileSelector
	Shell           []string
	BeforeScript    []string
	AfterScript     []string
	When            []When
	// contains filtered or unexported fields
}

Defaults is the execution settings a workflow applies to every step that does not say otherwise, and it is exactly the set the documentation lists.

"What it may not carry: image, script, needs, inputs, outputs, params, if, merge, strategy, extends and workflow are refused there. Anything deciding what a step runs, or where it sits in the graph, belongs to the step: the cut is what keeps the graph readable from the steps block alone." The cut is this type: a step is these keywords and the ones that decide what it runs, and defaults is these alone.

Every field is optional. A pointer distinguishes a value the file wrote from one it did not, which matters for every boolean here: cache: false and no cache at all are different things to a step that extends a block setting it true.

type Driver ¶

type Driver interface {
	Run(ctx context.Context, t Task) (Result, error)
	Stop(ctx context.Context, s Stop) error
}

Driver is what runs a container. It is stated here and called by nothing here.

Stating it on this side is the point. The evaluator says which task should run and what it carries; something else runs it, and both halves agree about the shape of the handover because the shape is written once, next to the values that cross it. What this package must never grow is a call to either method: the loop that reads a Plan, hands each Task to a Driver and feeds each Result back belongs to cmd/agk and to the controller, and putting it here would be the evaluator executing.

Stop is a method rather than a cancelled context because a task in flight may be held by a runner in another process, where a context does not reach. Cancelling a context stops the caller waiting; it does not stop a container on another machine, and the three rules that call off running work need the container stopped and not the waiting.

A test fakes the whole of it in four lines, a map from step name to Result, which is what keeps every rule in this package testable with nothing behind it.

type Duration ¶

type Duration time.Duration

Duration is a duration as the workflow language writes it: "one number and one unit taken from ms, s, m, h and d". It is a time.Duration underneath, so that a caller computing a deadline converts and does not parse.

The grammar is narrow on purpose. "A compound such as 1h30m is refused, and so is a bare number", because a file that says 10m where a reader would otherwise have to work out what 600 means is the whole reason the unit is written at all. timeout, retain, the schedule jitter and both backoff values are all written that way.

func ParseDuration(s string) (Duration, error)

func (d Duration) String() string

type Edge ¶

type Edge struct {
	Step agk.Step
	Port agk.Port
	As   agk.Port
}

Edge is one inbound dependency: an output port of one step feeding an input port of this one. "That is the only form of dependency: there is no implicit stage, and the order in which steps appear in the file has no bearing on scheduling."

The short form of needs is a step name alone, which is "{ step: name, port: out, as: in }" written out.

type Evaluator ¶

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

Evaluator is a handle over a Graph and a State. It holds no progress of its own: the graph is fixed at validation and the state is the run, so a second process given the same two answers the same way.

func New(g *Graph, s *State, l agk.Limits, maxRequeues int, maxRunDuration time.Duration) (*Evaluator, error)

func Start(g *Graph, run agk.Run, o Options, at time.Time) (*Evaluator, error)

func (e *Evaluator) Cancel(now time.Time)

func (e *Evaluator) Graph() *Graph

func (e *Evaluator) Next(now time.Time) (Plan, error)

func (e *Evaluator) Outputs() (map[string]agk.Envelope, error)

func (e *Evaluator) Record(r Result, now time.Time) error

func (e *Evaluator) State() *State

type Event ¶

type Event struct {
	Type      string
	Source    string
	Filter    string
	Namespace string
	Map       map[string]any
}

Event is a CloudEvents 1.0 subscription, with a CEL filter over the attributes and over data.

Namespace is the namespace whose events it hears, empty for the workflow's own: another namespace's are heard "where it granted the workflow's namespace read access". Map fills the workflow inputs from the event, as a webhook's fills them from its request.

type FanOut ¶

type FanOut int

FanOut is how a step is divided into shards.

const (
	FanOutNone FanOut = iota
	FanOutItem
	FanOutBatch
	FanOutMatrix
)

func (f FanOut) String() string

type FileSelector ¶

type FileSelector struct {
	From string
	To   string
	Mode string
}

FileSelector narrows what a step receives from the repository tree, "which it otherwise gets whole under /agk/repo/".

The short form is a path or a glob and leaves To and Mode empty; the long form "relocates a path to wherever a tool insists on finding it". Narrowing is "an optimisation for large repositories, never a requirement, and never a permission boundary".

type Fired ¶ added in v0.5.0

type Fired struct {
	Commit      string
	Trigger     map[string]any
	Event       map[string]any
	TriggerKind string
	TriggeredBy string

	// Vars are the variables the workflow's namespace shows it, which the file's own are laid over
	// under vars, as the run the trigger starts reads them: the caller hands the run the same ones,
	// so that what its map and its filter read is what its steps read.
	Vars map[string]any
}

Fired is what a trigger's expressions read of the run it fires before the run exists: the request or the occurrence under trigger, the event under event, and the kind and the principal under run. The run's id and start are not yet, and an expression of the on block reading them is refused where it is evaluated.

type Fragment ¶

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

Fragment is an included file: "a fragment, not an entry point. It carries hidden blocks and shared settings; it has no apiVersion, no kind and no metadata."

It is opaque, and deliberately: a Fragment comes from ParseFragment and goes into Load, so there is nothing for a caller to build one out of and no way to hand the evaluator a fragment that was never read as one. "Validating a fragment against the workflow schema is a category error", and so would constructing one be.

func ParseFragment(doc []byte) (*Fragment, error)

type Graph ¶

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

Graph is the resolved graph: the steps of a workflow, the edges between their ports, and an order that no cycle can be read out of.

It is built once, when a workflow is registered or a run starts, and read many times while the run goes. Nothing in it changes as a run proceeds: what changes is the State, and the two are separate so that one graph can serve every run of a version.

func Build(wf *Workflow, manifests map[string]brick.Manifest) (*Graph, error)

func (g *Graph) Brick(name agk.Step) (brick, version string, ok bool)

func (g *Graph) Consumers(name agk.Step, port agk.Port) []agk.Step

func (g *Graph) Edges(name agk.Step) []Edge

func (g *Graph) Fill(m map[string]any, f Fired) (map[string]any, error)

func (g *Graph) FillFromEvent(m map[string]any, f Fired) (map[string]any, error)

func (g *Graph) Hears(e Event, f Fired) (bool, error)

func (g *Graph) Order() []agk.Step

func (g *Graph) Resolved(commit string) ([]byte, error)

func (g *Graph) Step(name agk.Step) (*Step, bool)

func (g *Graph) Steps() []agk.Step

func (g *Graph) Upstream(name agk.Step) []agk.Step

func (g *Graph) Workflow() *Workflow

type Include ¶

type Include struct {
	Path     string
	Workflow WorkflowRef
}

Include is one entry of the include block: either a file of this repository, resolved inside the same commit, or another repository at a ref.

"A workflow include must carry ref, a path include must not, and a ref beside a path is refused. Requiring it is not tidiness: an unpinned cross-repository include would let another repository change what this commit does."

type Included ¶ added in v0.4.0

type Included struct {
	Path     string
	Workflow WorkflowRef
	Commit   string
}

Included is one include as resolution applied it: a file of this tree by its path from the root, or another repository with the ref the include wrote and the commit that ref resolved to, which every run of a version records so that the other repository moving a tag afterwards changes nothing this one does.

func LoadLibrary(fsys fs.FS, remote Remote) ([]Included, error)

func LoadStoredLibrary(fsys fs.FS, remote Remote) ([]Included, error)

type Input ¶

type Input struct {
	Schema   json.RawMessage
	Required bool
	Default  any
}

Input is one declared workflow input: the boundary a trigger fills.

Schema is the JSON Schema document as it was written, uncompiled, because compiling it resolves a $ref against the commit's tree and the tree is not this package's to hold. Required and Default are the two keywords that are not JSON Schema's.

type Join ¶

type Join struct {
	On string
}

Join is the key join: "a JSON path, for example join: { on: "$.data.customer_id" }".

type MCP ¶

type MCP struct {
	Name        string
	Description string
	Tools       []Tool
}

MCP is the published surface: which inputs and outputs a model-driven client may reach, under what names.

A workflow carries a pointer to one rather than a value, because "an empty list publishes a server with nothing on it, which is not the same as declaring no mcp block at all: the first serves an empty tool list, the second serves nothing and returns 404".

type Merge ¶

type Merge int

Merge is how several edges arriving on one input port become the single envelope that port carries.

const (
	MergeWaitAll Merge = iota
	MergeZip
	MergeJoin
	MergeFirst
)

func (m Merge) String() string

type Metadata ¶

type Metadata struct {
	Name      string
	Namespace string
	Labels    map[string]string
}

Metadata is what the workflow is called and which namespace owns it. "Ownership is not only naming: a workflow reaches the secrets, the quotas and the runner pools of the namespace that owns it, and of no other."

type Network ¶

type Network int

Network is the posture a step's container gets. "Every task gets its own network, and there is no posture that puts a container on the host's."

const (
	NetworkNone Network = iota
	NetworkEgress
	NetworkInternal
)

func (n Network) String() string

type Options ¶

type Options struct {
	// Inputs are the workflow inputs the trigger supplied, already held to their
	// declared schemas with required and default applied. That validation is package
	// schema's and it happens before a run exists, so what arrives here is what the
	// run holds.
	Inputs map[string]any

	// Vars are the workflow and namespace variables, already merged, because which of
	// the two a name came from is the API's business and not an expression's.
	Vars map[string]any

	// Trigger is body, headers, query and scheduled_for. Event is the CloudEvents 1.0
	// document an event trigger matched.
	Trigger map[string]any
	Event   map[string]any

	// Limits are the namespace's four size rules, applied wherever an envelope is
	// built here: a merge, a publication and a port the inputs keyword feeds. The zero
	// value is the documented defaults, because a caller that says nothing about them
	// means the rules and not their absence.
	Limits agk.Limits

	// MaxRequeues is the installation's max_requeues: how many times one key is handed
	// out again after a loss. Nil is DefaultMaxRequeues, for the reason the zero Limits
	// are the defaults, and zero is an installation that requeues nothing, which is what
	// max_requeues: 0 says. A pointer, because the two are different settings, as a hint
	// a file does not write is not one written false: read as an int, the zero a caller
	// left unset and the zero an installation wrote would be one value, and one of them
	// would be read as the other.
	MaxRequeues *int

	// MaxRunDuration is the namespace's max_run_duration, which bounds the root timeout: a
	// workflow asking for more is held to it and one writing none is bounded by it. Zero
	// bounds nothing, as a namespace that sets none and agk run --local, which has no
	// namespace, bound nothing.
	MaxRunDuration time.Duration

	// Reuse are the steps a replay reuses, each in the state the run it replays left it,
	// made with Reused: they start over and are never handed out.
	Reuse map[agk.Step]StepState
}

Options is everything a run carries that the workflow file does not.

type Output ¶

type Output struct {
	From struct {
		Step agk.Step
		Port agk.Port
	}
	Schema json.RawMessage
	Retain Retain
}

Output is one declared workflow output: a view of one step port.

From names the step and the port, and never more than one of each: "an output that concatenated several would hide which step produced what".

type Placed ¶ added in v0.4.0

type Placed struct {
	Path string
	To   string
	Mode string
}

Placed is one file a step's files select: its path in the tree, where it goes when a selector relocates it, absolute, and the mode a selector gives it, empty where none does.

func Place(selectors []FileSelector, tree []string) ([]Placed, error)

type Plan ¶

type Plan struct {
	Start []Task    `json:"start,omitempty"`
	Stop  []Stop    `json:"stop,omitempty"`
	Wake  time.Time `json:"wake,omitzero"`
}

Plan is what should happen next: the tasks that have become ready, the tasks that should be stopped, and the moment to ask again.

Wake is zero when nothing waits on the clock. It is not zero when something does, and two things do: a retry backoff, which places an attempt at a moment in the future, and the root timeout, which places the end of the run at one. A caller that never comes back at Wake does not get a wrong Plan; it gets the right one late.

type Position ¶ added in v0.4.0

type Position struct {
	File   string
	Line   int
	Column int
}

Position is where a refusal points: the file of the tree, and the line and column its node begins at, each counted from 1, as a YAML 1.2 parser counts them. Line and Column are zero where the refusal is about the file as a whole, or about something no one place in it holds, and File is empty where it is about no file of the tree at all, a brick manifest's among them.

"The refusal is written on git's error stream as file:line:column: message (rule)", so that a person holding the file is sent to the line rather than to the key somewhere above it.

func (p Position) String() string

type Refusal ¶

type Refusal struct {
	Step   agk.Step
	Port   agk.Port
	Rule   Rule
	Detail string
	At     Position
}

Refusal is a workflow refused by a rule of the language, naming the step, the port and the rule, in the documentation's own words, and where in the tree it was written.

Step and Port are empty where the rule is not about one: a cycle is about a list of steps and a published tool is about the workflow's own boundary. Detail is the sentence, and it quotes the documentation rather than describing the code, because the person reading it is holding the file and not this package.

At is the file, line and column of the node refused, as far as the engine can know them: a hook writes a refusal as file:line:column: message (rule), and a person given a rule and a step still has to find which of four files wrote the value. A refusal about the tree or about a file as a whole names the file alone, and one about no file of the tree names none.

func (r *Refusal) Error() string

func (r *Refusal) Unwrap() error

type Remote ¶ added in v0.4.0

type Remote interface {
	Include(ref WorkflowRef) (fs.FS, string, error)
}

Remote reaches the repositories a workflow include names.

"A workflow include must carry ref, a tag or a commit, and needs workflow:read on that repository", and both the ref and the permission are the caller's to settle: which tag names which commit, who is asking and whether they may, are the installation's. What comes back is the other repository's tree at the commit the ref resolved to, and that commit, whole, which the resolved graph records so that the other repository moving a tag afterwards changes nothing this one does.

type Resources ¶

type Resources struct {
	CPU    string
	Memory string
	PIDs   int
}

Resources is what a step asks for, "capped by the runner policy and by the namespace quota".

CPU and Memory are text because the language writes them as text: "cpu and memory are written exactly as the brick manifest writes them, quotation marks included: cpu: 1 unquoted is refused", so that half a core reads as 0.5 wherever it travels.

type Result ¶

type Result struct {
	Task     agk.TaskID    `json:"task"`
	State    agk.TaskState `json:"state"`
	ExitCode int           `json:"exit_code,omitempty"`
	Requeue  int           `json:"requeue,omitempty"`
	Reason   string        `json:"reason,omitempty"`

	// NoExitCode says the result reported no exit code at all, which the 0 of ExitCode cannot
	// say. A stopped task whose container started may report none: a host answering a requeue
	// from a record it kept before stops kept their code has none to give. The evaluator reads
	// no code off a stopped task, so nothing is decided by it; it keeps the task's row from
	// reading 0, the code of success.
	NoExitCode bool `json:"no_exit_code,omitempty"`

	Outputs map[agk.Port]agk.Envelope

	// MemoisedFrom is the run a cache hit republished these outputs from, for a success no
	// container reported: "a hit republishes the same envelopes without starting a container".
	MemoisedFrom agk.RunID `json:"outputs,omitempty"`

	// Called is the run a call started, on the dispatch of a step that calls a workflow: what
	// the step waits on in place of a container, and whose outputs its ports become.
	Called agk.RunID `json:"called,omitempty"`

	DispatchedAt time.Time `json:"dispatched_at,omitzero"`
	StartedAt    time.Time `json:"started_at,omitzero"`
	FinishedAt   time.Time `json:"finished_at,omitzero"`
}

Result is what became of one task. It is the only thing that enters the evaluator after a run has started.

State may be non-terminal: recording the dispatch is what fixes the task's deadline, because the deadline runs from the moment the work became somebody's. A caller that records only terminal results gets a deadline computed from the moment the task was planned instead, which drifts forward with every plan, and that is the honest consequence of not telling the evaluator rather than a rule it can enforce.

ExitCode is read for a task that succeeded or failed and for no other state, through agk.Band, which is the exit code table and the only place a verdict comes from.

Outputs is one envelope per declared port, which is what brick.Collect returns, empty envelopes included: a port the container never wrote publishes an empty envelope, and that is success.

Requeue names the dispatch an ending is about, as ShardState counts them. The key names the unit of work and a requeue keeps it, so only this tells the requeue's ending from a late one of the dispatch it replaced, and only the requeue's is news. The dispatch before it was judged lost and handed out again: its loss has been heard, and whatever its runner reports afterwards is the report of a runner the attempt stopped waiting on, while the requeue is still out and still owed an answer.

Reason is why a failure that no container explains happened, in words for a person: the controller refusing to publish a task to a pool that does not accept its namespace, for one. It becomes the step's reason where the failure ends the step and nothing gave the step one first, since an exit code of 125 says whose account a failure is on and not what it was.

type Retain ¶

type Retain struct {
	For     Duration
	Fetches int
}

Retain is how long the artifacts of an output stay fetchable, and how many times they may be fetched.

Fetches is zero where the file wrote none, which is "for as many fetches as anyone makes". Only a workflow output may carry one: "an artifact travelling between two steps is read once per shard and again by a replay".

type Retry ¶

type Retry struct {
	Max     int
	On      []agk.Failure
	Backoff Backoff
}

Retry is the retry policy: how many further attempts, on which kinds of failure, and how long the wait between them grows.

type Rule ¶

type Rule string

Rule is what a document is refused by, spelled as the fixture corpus spells it.

The spelling is the point. The released corpus names, for every invalid document, the rule it holds a reader to, and a test that puts a fixture beside the rule it pins is then one lookup rather than a translation table somebody has to keep in step. The corpus marks each invalid workflow refused_by "schema" or "validator": everything marked schema is shape, which closed decoding gives for nothing and which is refused here by an ordinary error naming the key, and the twenty marked validator are, item for item, the rules below that carry a corpus name.

const (
	// RuleCycleInGraph is "cycles are rejected at validation, before registration.
	// Looping is done by calling a sub-workflow".
	RuleCycleInGraph Rule = "cycle-in-graph"
	// RuleNeedsUnknownStep is an edge from a step the workflow does not declare. "An
	// edge is the only form of dependency there is", so it has to name something that
	// exists.
	RuleNeedsUnknownStep Rule = "needs-unknown-step"
	// RuleEdgePortNotDeclared is an edge taking an output port nobody declares.
	RuleEdgePortNotDeclared Rule = "edge-port-not-declared"
	// RuleStepOutputNotInManifest is a step declaring an output port its brick manifest
	// does not carry: "outputs must be a subset of the ports in the brick manifest".
	RuleStepOutputNotInManifest Rule = "step-output-not-in-manifest"
	// RuleStepInputPortNotInManifest is a step feeding an input port nobody declares.
	// "The runner mounts one directory per port the brick declares, and there is no
	// directory for a port the manifest never named."
	RuleStepInputPortNotInManifest Rule = "step-input-port-not-in-manifest"
	// RuleOutputFromUnknownStep is a workflow output taken from a step the workflow
	// does not declare.
	RuleOutputFromUnknownStep Rule = "output-from-unknown-step"
	// RuleSecretNotDeclared is a step mounting a secret the workflow never names.
	// "Secrets are mounted by the name the secrets block lists them under, and only
	// secrets the owning namespace declares can be named there."
	RuleSecretNotDeclared Rule = "secret-not-declared"
	// RuleScriptWithoutOutputs is a script step that declares no output port. "The
	// image is a base image, no manifest is read and nothing about its ports is
	// inferred, so outputs must be declared."
	RuleScriptWithoutOutputs Rule = "script-without-outputs"
	// RuleMCPToolInputNotDeclared is a published tool naming something the workflow
	// does not declare. "The reference is to a workflow input or output, never to a
	// step or a port."
	RuleMCPToolInputNotDeclared Rule = "mcp-tool-input-not-declared"
	// RuleMCPToolInputWithoutSchema is a tool whose input is a workflow input carrying
	// no schema. "A tool has to publish an inputSchema, so the input it maps has to
	// have one."
	RuleMCPToolInputWithoutSchema Rule = "mcp-tool-input-without-schema"
	// RuleMCPDuplicateToolName is two published tools carrying the same name. "A tool
	// identifier is unique within the workflow, because clients hold it."
	RuleMCPDuplicateToolName Rule = "mcp-duplicate-tool-name"
	// RuleMCPInIncludedFile is an mcp block in an included file. "Reuse applies to
	// steps and defaults; the published surface is declared by the workflow that
	// publishes it, so that reading one file tells you everything that workflow
	// exposes."
	RuleMCPInIncludedFile Rule = "mcp-in-included-file"
	// RuleExpressionUnknownRoot is an expression reading a context root that is not
	// exposed. The roots are the ten of the table and there is no eleventh.
	RuleExpressionUnknownRoot Rule = "expression-unknown-root"
	// RuleExpressionItemOutsideFanOutItem is an expression reading item where the step
	// is not sharded per item, "which is what keeps the controller from loading a
	// namespace's data in order to schedule".
	RuleExpressionItemOutsideFanOutItem Rule = "expression-item-outside-fan-out-item"
	// RuleExpressionSecretInCondition is a condition reading a secret. "secrets is
	// exposed to params and to secrets and nowhere else."
	RuleExpressionSecretInCondition Rule = "expression-secret-in-condition"
)

const (
	// RuleManifestMissing is a brick step whose manifest was not handed in. Reading
	// /agk/brick.yaml means pulling an image and pulling an image is executing, so the
	// manifests arrive as an argument and the evaluator says which ones it needs.
	RuleManifestMissing Rule = "manifest-missing"
	// RuleParamsAgainstManifest is a step parameter the manifest refuses: one it does
	// not declare, one it declares required and the step does not supply, or one whose
	// value departs from the schema the manifest gives it.
	RuleParamsAgainstManifest Rule = "params-against-manifest"
	// RuleZipLength is merge: zip on envelopes of differing lengths, which "produce a
	// validation failure".
	RuleZipLength Rule = "zip-length"
	// RuleMCPSyncTimeoutAboveCeiling is a sync tool waiting longer than the ceiling.
	// "A sync call is capped at 120 seconds because clients time out, and a workflow
	// that cannot finish inside it has to be async."
	RuleMCPSyncTimeoutAboveCeiling Rule = "mcp-sync-timeout-above-ceiling"
	// RuleMCPTimeoutOnAsyncTool is a timeout on an async tool, "where it would mean
	// nothing: the call returns the run identifier at once". The value is the corpus's
	// own name for the fixture that pins it.
	RuleMCPTimeoutOnAsyncTool Rule = "mcp-async-tool-with-timeout"
	// RuleExpressionSecretInText is a secret in a position inside a string. "An
	// expression that fills the whole value keeps its type; an expression embedded in
	// a string is converted to text", and a secret has no text: it is an opaque
	// reference, so the expression that reads one fills the whole value or nothing.
	RuleExpressionSecretInText Rule = "expression-secret-in-text"
	// RuleExpressionDoesNotCompile is an expression that is not one: it never closes,
	// it does not parse, or it compares values whose types cannot be compared. CEL
	// "evaluates in linear time, is mutation free, and not Turing-complete", and what
	// it cannot check before a run is the one thing this engine cannot promise.
	RuleExpressionDoesNotCompile Rule = "expression-does-not-compile"
)

const (
	// RuleIncludeLeavesTree is a path include naming a file outside the repository tree. "A
	// path include resolves inside the same commit", and nothing outside the tree is readable
	// at all.
	RuleIncludeLeavesTree Rule = "include-leaves-tree"
	// RuleIncludeMissing is a path include naming a file the commit does not hold.
	RuleIncludeMissing Rule = "include-missing"
	// RuleIncludeCycle is an include coming back to a file still being resolved, which would
	// never finish resolving.
	RuleIncludeCycle Rule = "include-cycle"
	// The keys that make a document an entry point, refused in an included file one rule
	// each, as the fragment group files one fixture each and mcp-in-included-file already
	// was. "An included file is a fragment, not an entry point."
	RuleAPIVersionInIncludedFile  Rule = "api-version-in-included-file"
	RuleKindInIncludedFile        Rule = "kind-in-included-file"
	RuleMetadataInIncludedFile    Rule = "metadata-in-included-file"
	RuleInputsInIncludedFile      Rule = "inputs-in-included-file"
	RuleOutputsInIncludedFile     Rule = "outputs-in-included-file"
	RuleOnInIncludedFile          Rule = "on-in-included-file"
	RuleConcurrencyInIncludedFile Rule = "concurrency-in-included-file"
	RuleTimeoutInIncludedFile     Rule = "timeout-in-included-file"
)

const (
	// RuleCronValueOutOfRange is a schedule whose five fields are cron's shape and name a value
	// its field does not count, a minute of 60, a step of 0, a range written backwards, or no
	// day that comes, the 30th of February: an occurrence that never comes.
	RuleCronValueOutOfRange Rule = "cron-value-out-of-range"
	// RuleTimezoneUnknown is a schedule's zone written as a name the IANA time zone database
	// does not hold, so that no instant can be read from it.
	RuleTimezoneUnknown Rule = "timezone-unknown"
	// RuleWebhookOutputNotDeclared is a sync webhook answering with an output the workflow does
	// not declare: what a caller is answered with is one of the workflow's own outputs.
	RuleWebhookOutputNotDeclared Rule = "webhook-output-not-declared"
	// RuleWebhookMapInputNotDeclared is a webhook's map filling an input the workflow does not
	// declare: map builds the run's inputs, each validated against its schema before the run
	// exists, and an input nobody declared has none.
	RuleWebhookMapInputNotDeclared Rule = "webhook-map-input-not-declared"
	// RuleEventMapInputNotDeclared is an event trigger's map filling an input the workflow does
	// not declare, for the reason a webhook's is refused.
	RuleEventMapInputNotDeclared Rule = "event-map-input-not-declared"
	// RuleWebhookDuplicatePath is two webhooks answering one path with one method: "within a
	// namespace a path and a method answer one trigger".
	RuleWebhookDuplicatePath Rule = "webhook-duplicate-path"
	// RulePortPastBound is a port past agk.PortMaxBytes, wherever it is written: "a port
	// becomes the file <name>.json, which a filesystem holds to 255 characters with its
	// suffix". The value is the name the corpus files the fixture pinning it under.
	RulePortPastBound Rule = "port-name-251-characters"
	// RuleWorkflowOutputPastBound is a workflow output past the same bound, for the same
	// reason.
	RuleWorkflowOutputPastBound Rule = "workflow-output-251-characters"
)

type Schedule ¶

type Schedule struct {
	Cron     string
	Timezone string
	Jitter   Duration
	CatchUp  bool
}

Schedule is a five-field cron expression read in the timezone written beside it.

type SecretMount ¶

type SecretMount struct {
	Name  string
	Mount string
}

SecretMount is one secret a task is given: the name the workflow knows it by, and the path under /agk/secrets/ the brick manifest asks for it at.

It carries no value and never will. "The runner mounts secret values on tmpfs" and the evaluator decides which secrets a task gets, not what they are.

type SecretParam ¶

type SecretParam struct {
	Secret string `json:"secret"`
}

SecretParam is a step parameter whose value is a secret: a reference the runner redeems, and never the secret itself.

It exists because "sensitive is the only route by which a secret reaches a brick as a parameter rather than as a mounted file", and because a task message "names everything the runner needs and contains nothing sensitive". So the parameter travels as the name the workflow declared, and the value is put in /agk/params.json by the runner that has redeemed its grant, in the container and nowhere else.

type ShardState ¶

type ShardState struct {
	Shard  agk.Shard      `json:"shard,omitzero"`
	Matrix map[string]any `json:"matrix,omitempty"`

	// Attempt counts from 1, as AGK_ATTEMPT does.
	Attempt int `json:"attempt"`

	// Requeue counts the times this attempt was handed out again after its task was
	// lost, from 0 for the first time it was handed out. A requeue keeps the attempt
	// and so the idempotency key, which leaves this as the one thing that tells two
	// dispatches of one key apart: the controller keeps a row for each, and an ending is
	// news only about the dispatch the shard is on. It is also what max_requeues is
	// counted against, which is why the bound is per key.
	Requeue int `json:"requeue,omitempty"`

	// Task and ExitCode are what the driver reported, and they are read together:
	// the exit code table is the only place a verdict comes from, and the code is
	// read for a task that succeeded or failed and for no other state.
	Task     agk.TaskState `json:"task"`
	ExitCode int           `json:"exit_code,omitempty"`

	// NoExitCode is Result's: the ending reported no exit code, which ExitCode's 0 cannot say.
	NoExitCode bool `json:"no_exit_code,omitempty"`

	// Stopped says the evaluator ended this shard itself, cancelled, in the pass that
	// named its stop as superseded or sibling_failed, rather than hearing the ending from a
	// driver. The driver's report of it comes later, to a shard that is over, and all it
	// may still add is how the container exited: Record reads it for that and for nothing
	// else.
	Stopped bool `json:"stopped,omitempty"`

	// Ports is what this shard published, before the shards of the step are
	// concatenated port by port into what the step publishes.
	Ports map[agk.Port]agk.Envelope `json:"ports,omitempty"`

	// MemoisedFrom is the run whose task published what a cache hit republished for this
	// shard, which no container ran.
	MemoisedFrom agk.RunID `json:"memoised_from,omitempty"`

	// Called is the run this shard's call started, where its step calls a workflow.
	Called agk.RunID `json:"called,omitempty"`

	// CacheKey is the memoisation key the shard's current attempt was handed out with, where
	// its step is cached, so that whoever records its ending can remember what it published
	// under the key it was asked for by, without building the task again.
	CacheKey string `json:"cache_key,omitempty"`

	// The four moments a task passes through. DispatchedAt is what fixes the
	// deadline, because the deadline runs from the moment the work became somebody's
	// and not from the moment it was decided on. NextAttemptAt is when the shard may
	// be handed out again, and is zero when there is nothing to wait for.
	DispatchedAt  time.Time `json:"dispatched_at,omitzero"`
	StartedAt     time.Time `json:"started_at,omitzero"`
	FinishedAt    time.Time `json:"finished_at,omitzero"`
	NextAttemptAt time.Time `json:"next_attempt_at,omitzero"`
}

ShardState is what has happened to one shard of one step: which shard it is, which attempt it is on, and what the driver last said about the task running it.

A shard that is retried keeps its place here and moves to its next attempt rather than being added beside itself. One shard is one piece of the work, however many attempts it takes, which is what lets max_parallel count what holds a runner and fail_fast name what to stop.

type State ¶

type State struct {
	Version int            `json:"version"`
	Run     agk.Run        `json:"run"`
	Inputs  map[string]any `json:"inputs,omitempty"`
	Vars    map[string]any `json:"vars,omitempty"`

	// Trigger and Event are what started the run: "body, headers, query,
	// scheduled_for" and the CloudEvents document an event trigger matched. They sit
	// here rather than beside the evaluator because failover is a state resume: a
	// parameter written as ${{ trigger.body.cycle }} has to resolve to the same value
	// in the process that picks the run up, and a root the second process was never
	// given would resolve to nothing at all.
	Trigger map[string]any `json:"trigger,omitempty"`
	Event   map[string]any `json:"event,omitempty"`

	// Steps holds one entry per step of the graph, put there when the run starts,
	// so that a step nobody has reached is a pending entry rather than a missing
	// one and the reduction to a run verdict can tell the two apart.
	Steps map[agk.Step]StepState `json:"steps"`

	// Seq counts the decisions taken against this state. It is what a caller
	// persisting the state writes beside it to tell a stale copy from a current
	// one, and what makes two writers of one run detectable rather than silent.
	// A call that leaves the state as it was is no decision and does not move it,
	// so a caller has nothing to write when it has not moved.
	Seq int `json:"seq"`
}

State is one run in progress: what it was started with, and what has happened to each of its steps.

It restates nothing that sits in the agk.Run it holds. The run identifier, the workflow, the namespace, the commit and the state of the run itself are the run's, and a second copy here would be a second answer to what a run is.

func (s *State) UnmarshalJSON(b []byte) error

type Step ¶

type Step struct {
	Image string
	Call  *Call

	Needs   []Edge
	Inputs  map[agk.Port]any
	Outputs []agk.Port
	Params  map[string]any

	If       string
	When     []When
	Merge    Merge
	Join     Join
	Strategy Strategy

	Retry           Retry
	Timeout         Duration
	Retain          Retain
	ContinueOnError bool
	Resources       Resources
	Network         Network
	EgressAllow     []string
	RunsOn          []string
	Secrets         []string
	Cache           bool
	Idempotent      bool
	Files           []FileSelector

	Script       []string
	BeforeScript []string
	AfterScript  []string
	Shell        []string

	// Extends is the hidden blocks this step inherits from, the one it names first and each
	// block that one extends after it, kept after resolution because a run detail that says
	// where a setting came from reads better than one that shows the setting alone.
	Extends []string
}

Step is one step of the graph, with every keyword resolved: what arrived from an include, then from the blocks it extends, then from defaults, then what the step wrote itself.

It carries an image or a call and never both, and a call "runs no container of its own", so Script, BeforeScript, AfterScript and Shell are refused beside one.

type StepState ¶

type StepState struct {
	Verdict agk.Verdict  `json:"verdict"`
	Shards  []ShardState `json:"shards,omitempty"`

	// Ports is the publication: exactly one envelope per port the step declares,
	// written once, when the step ends. It is empty until then, which is what the
	// barrier below reads as a port that has not arrived.
	Ports map[agk.Port]agk.Envelope `json:"ports,omitempty"`

	// Since is when the step reached the verdict it carries. A published port is
	// stamped with it, and an empty envelope standing in for a step that published
	// nothing is stamped with it too, so that a batch of nothing still says when it
	// was not produced.
	Since time.Time `json:"since,omitzero"`

	// Reason says why, in the words of the rule that decided it, for the verdicts
	// where why is not obvious: a skip, a cancellation, a step whose when refused
	// every upstream state it saw. A run detail that says why reads better than one
	// that shows a verdict alone.
	Reason string `json:"reason,omitempty"`
}

StepState is what has happened to one step: the verdict it has reached, the shards it was split into, and the envelopes it published when it ended.

func Reused(from StepState, run agk.RunID) StepState

type Stop ¶

type Stop struct {
	Task   agk.TaskID `json:"task"`
	Reason StopReason `json:"reason"`
}

Stop is one task in flight that should be stopped, and why.

It exists because three rules in the documentation call off work that is already running, and a decision to stop is as much a decision as a decision to start. A controller that only ever heard about starts would leave a shard running for an hour after the run it belongs to was cancelled.

type StopReason ¶

type StopReason int

StopReason is why a task is being stopped. There are four, and each one is a rule the documentation states rather than a category invented here.

const (
	// StopSuperseded: a merge: first lifted the barrier on another edge, and no
	// other consumer needs what this task would publish. It names the tasks of the
	// cancelled step not started yet too, which never start.
	StopSuperseded StopReason = iota

	// StopSiblingFailed: fail_fast, where the first shard to fail stops the shards
	// still running beside it, and names the ones not started yet, which never start.
	StopSiblingFailed

	// StopDeadline: the deadline set by the root timeout of the entry point passed,
	// and the tasks still running are stopped.
	StopDeadline

	// StopCancelled: a principal holding workflow:run asked, or a concurrency group
	// did.
	StopCancelled
)

func (s StopReason) MarshalText() ([]byte, error)

func (s StopReason) String() string

func (s *StopReason) UnmarshalText(text []byte) error

type Strategy ¶

type Strategy struct {
	FanOut      FanOut
	Batch       int
	MaxParallel int
	Matrix      map[string][]any
	FailFast    bool
}

Strategy is the strategy block: how the step is divided, how much of it runs at once, what combinations it runs over, and whether one failing shard stops its siblings.

Batch is the size written inside batch(n), and it is zero unless FanOut is FanOutBatch.

type Task ¶

type Task struct {
	// ID is the idempotency key, run/step/attempt/shard, derived from what makes
	// this task this task and never minted. A driver that has already done the work
	// recognises it, which is what makes at-least-once delivery survivable.
	ID agk.TaskID `json:"id"`

	// Dispatch is the task_id of the dispatch a runner took, and empty where nothing was
	// dispatched, as on a laptop. The evaluator never sets it: a control plane mints one per
	// dispatch, and a requeue after loss keeps the key and takes a new one. The driver names the
	// container's span by it, falling back to the key, so that each dispatch is a span of its own
	// and agk run --local still has one to name.
	Dispatch string `json:"dispatch,omitempty"`

	Run       agk.RunID `json:"run"`
	Workflow  string    `json:"workflow"`
	Namespace string    `json:"namespace"`
	Commit    string    `json:"commit"`

	Step    agk.Step  `json:"step"`
	Attempt int       `json:"attempt"`
	Shard   agk.Shard `json:"shard,omitzero"`

	// Image is what the container is, or Call is the sub-workflow this step is
	// instead. A step carries one or the other and never both.
	Image string `json:"image,omitempty"`
	Call  *Call  `json:"call,omitempty"`

	// CallInputs are what a call hands the workflow it calls as that workflow's inputs: each
	// value the step's inputs keyword writes, of whatever type it evaluates to, and each port
	// another step feeds it, as the list of its items' data. Empty for a step that runs an image.
	CallInputs map[string]any `json:"call_inputs,omitempty"`

	// The script keywords, where the image is a base image rather than a brick.
	// after_script runs in the same container even when script failed, so that a
	// diagnostic dump survives a failure, and its own exit code does not change the
	// step's verdict: keeping that true is the driver's, because it is the driver
	// that watches the two and reports one.
	Script       []string `json:"script,omitempty"`
	BeforeScript []string `json:"before_script,omitempty"`
	AfterScript  []string `json:"after_script,omitempty"`
	Shell        []string `json:"shell,omitempty"`

	// Params are resolved: every expression evaluated, the matrix combination
	// injected, and the whole validated against the manifest schema. The driver
	// writes them to /agk/params.json and asks nothing about them.
	Params map[string]any `json:"params,omitempty"`

	// Secrets are names and mount points and never values. The task message names
	// the secrets it needs and carries none of them; the values exist only after the
	// runner redeems its grant.
	Secrets []SecretMount `json:"secrets,omitempty"`

	// Inputs are merged and sharded already, one envelope per input port.
	Inputs map[agk.Port]agk.Envelope `json:"inputs,omitempty"`

	// Outputs are the ports declared, which is what is collected from /agk/out/ and
	// what publishes an empty envelope where the container wrote nothing.
	Outputs []agk.Port `json:"outputs,omitempty"`

	Files       []FileSelector `json:"files,omitempty"`
	Resources   Resources      `json:"resources,omitzero"`
	Network     Network        `json:"network"`
	EgressAllow []string       `json:"egress_allow,omitempty"`
	RunsOn      []string       `json:"runs_on,omitempty"`

	// Timeout bounds this shard and Deadline is the moment it lands on, which is
	// what AGK_DEADLINE carries and what the runner stops the container at.
	Timeout  Duration  `json:"timeout,omitempty"`
	Deadline time.Time `json:"deadline,omitzero"`

	// Idempotent says whether running this again on the same inputs is safe, which
	// is what decides whether a lost task may be requeued and whether the result may
	// be cached. Cache and CacheKey are memoisation: the key combines the image
	// digest, the resolved parameters and the digests of the input envelopes,
	// prefixed by the namespace, and it never crosses a namespace boundary.
	Idempotent bool   `json:"idempotent"`
	Cache      bool   `json:"cache,omitempty"`
	CacheKey   string `json:"cache_key,omitempty"`
}

Task is one shard of one attempt of one step, carrying everything a driver needs and nothing it has to look up.

Inputs is exactly the argument brick.WriteInputs takes and Outputs is exactly the declared argument brick.Collect takes, so the driver is the thin thing between two calls that already exist. That is why the merge strategies are resolved on this side: wait_all, zip, join and first are rules the workflow file states, and a rule hidden inside the driver is a rule the workflow file cannot show.

The task message of the documentation carries digests rather than envelopes. That is one serialisation of this, made by a server driver that spills the envelope to the store and puts the digest on the bus, and it is not a second decision.

type Tool ¶

type Tool struct {
	Name        string
	Title       string
	Description string
	Input       ToolIO
	Output      *ToolIO
	Mode        ToolMode
	Timeout     Duration
	Annotations Annotations
}

Tool is one published tool. Name, Description and Input are the three the language requires: "a tool published without a description is exactly the tool a model has no way to decide to call".

type ToolIO ¶

type ToolIO struct {
	Input  string
	Output string
}

ToolIO names the workflow input or output a tool is a view of. Exactly one of the two is set, decided by the key it was written under: "the reference is to a workflow input or output, never to a step or a port".

type ToolMode ¶

type ToolMode int

ToolMode is how a call waits. "sync waits for the run and returns the output; async returns the run identifier at once."

const (
	ToolSync ToolMode = iota
	ToolAsync
)

func (m ToolMode) String() string

type Trigger ¶

type Trigger struct {
	Schedule []Schedule
	Webhook  []Webhook
	Event    []Event
}

Trigger is the on block: what starts a run of this workflow.

The block is spelled on:, which is why the file has to be read by a YAML 1.2 parser: a 1.1 parser reads that bare key as the boolean true.

type Vars ¶

type Vars map[string]any

Vars are the workflow variables, read by expressions under the vars root.

func (v Vars) Over(namespace map[string]any) map[string]any

type Webhook ¶

type Webhook struct {
	Path     string
	Method   string
	Auth     string
	Response string
	Output   string
	Map      map[string]any
}

Webhook is one HTTP entry point. "Paths are namespaced as /hooks/<namespace>/<path>", which is the API's doing and not written here.

Method, Auth and Response hold the value in force: POST, hmac and async where the file writes none. Output is the workflow output a sync response returns, and empty for an async one.

type When ¶

type When int

When is one upstream state that allows a step to start. "Defaults to [succeeded]."

Pending and running are not among them, and neither is cancelled: when reads the verdict of a step that has ended, and the four names are the four the language writes.

const (
	WhenSucceeded When = iota
	WhenFailed
	WhenSkipped
	WhenAlways
)

func (w When) String() string

type Workflow ¶

type Workflow struct {
	APIVersion  string
	Kind        string
	Metadata    Metadata
	Inputs      map[string]Input
	Outputs     map[string]Output
	On          Trigger
	MCP         *MCP
	Include     []Include
	Vars        Vars
	Secrets     []string
	Defaults    Defaults
	Concurrency Concurrency
	Timeout     Duration
	Steps       map[agk.Step]Step
	// contains filtered or unexported fields
}

Workflow is agentiik.yaml as a value: "the source of truth for what a workflow does".

Everything the file declares is here and nothing else is. It is not the source of truth for who may do what, so no grant appears; it is not a run, so no state appears; and it holds no compiled schema, because compiling one needs the repository tree and the tree is the caller's to provide.

A Workflow arrives from Parse or from Load and not from a literal. The unexported fields below are what resolution and a refusal need: the steps as they were written, the hidden blocks they extend, and the document itself, so that a position can name the line an expression was written on.

func Load(fsys fs.FS, entry string, remote Remote) (*Workflow, error)

func LoadStored(fsys fs.FS, entry string, remote Remote) (*Workflow, error)

func Parse(doc []byte) (*Workflow, error)

func (w *Workflow) Armable() error

func (wf *Workflow) DeclaredInputs(tree fs.FS) (map[string]schema.Input, error)

func (w *Workflow) Entry() string

func (w *Workflow) Included() []Included

func (wf *Workflow) InputFiles(tree fs.FS) (map[string]any, error)

func (w *Workflow) MetadataAt(key string) Position

func (w *Workflow) SecretAt(name string) Position

func (w *Workflow) StepAt(step agk.Step, keyword string, below ...any) Position

type WorkflowRef ¶

type WorkflowRef struct {
	Namespace string
	Name      string
	Ref       string
}

WorkflowRef names one workflow, and a ref of it where one is pinned. It is written <namespace>/<name> everywhere it appears, with @<ref> appended in the short form of a sub-workflow call.

It is comparable, and it is what a workflow include asks Remote for: the evaluator never reaches another repository, so a caller resolves the ref and hands over that repository's tree at the commit it resolved to.

func (r WorkflowRef) String() string