operation

A graph operation has native input and output types and an explicit interrupted-effect contract. Adapters for tools, models and classifiers implement this same boundary; it has no workflow-runtime dependency.

A body’s typed errors are classified with the agent’s tool.Failure: tool.Explain is a definite failure (the run fails with graph.OperationFailed), tool.Uncertain an effect that may have happened (the run is graph.Blocked until it is reconciled).

Every wait (a signal, a job, a managed child or a fork) is bounded: 7 days after policy admission by default, with_deadline to change it, run.Infinity to wait without one. definition.build checks every setting of an operation and reports every problem at once.

Calling Relay tools from a graph

Copy the module relay_operation of consumers/relay_tools into the application; that consumer compiles and tests it. relay_operation.operation builds a graph operation identified by the tool name and a version. It calls the peer with the invocation’s correlation and idempotency key and maps failures with relay_tools.failure. See the README and the design layer for composition and failure handling.

Types

ConfigurationError

</>

A setting of an operation that definition.build refuses. This union may grow: match the variants you handle and keep a catch-all, or use definition.describe_build_error.

pub type ConfigurationError {
  InvalidLimit(
    limit: Limit,
    value: Int,
    minimum: Int,
    maximum: Int,
  )
  ReplayRequiresActivity
  DeadlineRequiresWait
}

Constructors

```
InvalidLimit(
  limit: Limit,
  value: Int,
  minimum: Int,
  maximum: Int,
)
```
A bound is outside minimum..maximum (both included), as for agent.InvalidLimit. Durations are in milliseconds.
```
ReplayRequiresActivity
```
with_replay on an operation that is not an activity.
```
DeadlineRequiresWait
```
with_deadline on an activity: its body is bounded by the runtime’s operation timeout instead (graph.with_operation_timeout).

Error

</>

pub type Error {
  NotExecutable
  ObservationFailed(String)
  InputEncodingFailed(codec.EncodeError)
  InputDecodingFailed(String)
  BodyFailed(tool.Failure)
  OutputEncodingFailed(
    evidence: String,
    error: codec.EncodeError,
  )
  OutputDecodingFailed(String)
}

Constructors

```
NotExecutable
```
```
ObservationFailed(String)
```
```
InputEncodingFailed(codec.EncodeError)
```
```
InputDecodingFailed(String)
```
```
BodyFailed(tool.Failure)
```
```
OutputEncodingFailed(evidence: String, error: codec.EncodeError)
```
A native value could not be encoded; its printed representation is diagnostic evidence only, never a result eligible for routing.
```
OutputDecodingFailed(String)
```

Invocation

</>

The call an operation body answers. Read it by label: Fabric may add fields.

run and activation identify one logical operation across attempts; a new visit to the same node always has a different activation. correlation is the run’s (see graph.start): pass it to the packages the body calls (http_gun.with_correlation, say), so that their events join the run’s.

pub type Invocation {
  Invocation(
    run: run.RunId,
    activation: Int,
    attempt: Int,
    correlation: correlation.Correlation,
  )
}

Constructors

Invocation(
  run: run.RunId,
  activation: Int,
  attempt: Int,
  correlation: correlation.Correlation,
)

Kind

</>

pub type Kind {
  Activity
  Signal
  Job(polling: job.Polling)
  OwnedJob(polling: job.Polling)
  Subgraph
  Agent
  Fork(max_members: Int, concurrency: Int, signature: String)
}

Constructors

```
Activity
```
```
Signal
```
```
Job(polling: job.Polling)
```
```
OwnedJob(polling: job.Polling)
```
```
Subgraph
```
```
Agent
```

Fork(max_members: Int, concurrency: Int, signature: String)

Limit

</>

A bound of an operation that definition.build checks, named after its setter. This union may grow.

pub type Limit {
  ReplayAttempts
  Deadline
  PollInterval
  MaxMembers
  Concurrency
}

Constructors

```
ReplayAttempts
```
with_replay’s attempts, 1 to 100.
```
Deadline
```
with_deadline, 1 ms to 2^32 - 1 ms.
```
PollInterval
```
job.with_poll_interval, 1 ms to 2^32 - 1 ms.
```
MaxMembers
```
graph.map’s max_members, at least 1.
```
Concurrency
```
graph.map’s concurrency, at least 1.

Operation

</>

A typed operation bound to its body or wait. Build one with new, await_signal, await_job or own_job.

pub type Operation(context, input, output) =
  @internal Operation(
    context,
    input,
    output,
    Kind,
    Recovery,
    Invocation,
    Error,
  )

Recovery

</>

pub type Recovery {
  RequireReconciliation
  ReplayInterrupted(max_attempts: Int)
}

Constructors

```
RequireReconciliation
```
A started body without a saved result requires reconciliation.
```
ReplayInterrupted(max_attempts: Int)
```
The application guarantees that repeating the body is safe, for example by using the stable run and activation as an external idempotency key. The bound includes the initial attempt. It is not a timeout classifier.

StopReason

</>

Why an admitted operation’s lifetime is being stopped. This does not imply that a remote effect stopped or that its cancellation was acknowledged.

pub type StopReason {
  CancellationRequested
  DeadlineReached(due: Int)
}

Constructors

```
CancellationRequested
```
```
DeadlineReached(due: Int)
```

Values

await_job

</>

pub fn await_job(
  job_observer: @internal Observer(
    context,
    receipt,
    output,
    job.Polling,
    job.Progress(String),
  ),
) -> @internal Operation(
  context,
  receipt,
  output,
  Kind,
  Recovery,
  Invocation,
  Error,
)

Retain a typed external receipt and wait for a checked business outcome. Observation is read-only; cancellation detaches without canceling remote work.

await_signal

</>

pub fn await_signal(
  input: codec.Codec(input),
  signal: signal.Signal(output),
) -> @internal Operation(
  context,
  input,
  output,
  Kind,
  Recovery,
  Invocation,
  Error,
)

A wait retains its typed request as input, then accepts a value with the signal’s output contract. Policy gates publishing the wait. No task body is started, and the waiting run holds no runner process.

idempotency_key

</>

pub fn idempotency_key(invocation: Invocation) -> String

An external idempotency key for this logical activation, stable across retries and independent of the attempt and correlation.

identity

</>

pub fn identity(
  operation: @internal Operation(
    context,
    input,
    output,
    Kind,
    Recovery,
    Invocation,
    Error,
  ),
) -> run.DefinitionId

input_codec

</>

pub fn input_codec(
  operation: @internal Operation(
    context,
    input,
    output,
    Kind,
    Recovery,
    Invocation,
    Error,
  ),
) -> codec.Codec(input)

The codec of the operation’s input, for reading graph.Receipt.input_json.

kind

</>

pub fn kind(
  operation: @internal Operation(
    context,
    input,
    output,
    Kind,
    Recovery,
    Invocation,
    Error,
  ),
) -> Kind

new

</>

pub fn new(
  identity: run.DefinitionId,
  input: codec.Codec(input),
  output: codec.Codec(output),
  perform: fn(context, Invocation, input) -> Result(output, error),
  classify: fn(error) -> tool.Failure,
) -> @internal Operation(
  context,
  input,
  output,
  Kind,
  Recovery,
  Invocation,
  Error,
)

An activity: perform runs once policy admits it, in its own task, bounded by the runtime’s operation timeout. classify says what each typed error means (tool.Explain or tool.Uncertain). A body crash, a timeout or a lost task is an uncertain effect, never a definite failure; a runner lost while the body ran replays it when the operation allows (with_replay).

output_codec

</>

pub fn output_codec(
  operation: @internal Operation(
    context,
    input,
    output,
    Kind,
    Recovery,
    Invocation,
    Error,
  ),
) -> codec.Codec(output)

The codec of the operation’s output, for reading graph.Receipt.output_json.

own_job

</>

pub fn own_job(
  job_observer: @internal Observer(
    context,
    receipt,
    output,
    job.Polling,
    job.Progress(String),
  ),
  request: fn(context, Invocation, receipt) -> Result(Nil, error),
  classify: fn(error) -> tool.Failure,
) -> @internal Operation(
  context,
  receipt,
  output,
  Kind,
  Recovery,
  Invocation,
  Error,
)

Admit ownership of an existing job, including cancellation authority. request acknowledges a stop request; the observer must confirm a terminal outcome separately. Interrupted requests are never automatically repeated. classify is as for new: tool.Explain refuses the stop, tool.Uncertain leaves it uncertain.

recovery

</>

pub fn recovery(
  operation: @internal Operation(
    context,
    input,
    output,
    Kind,
    Recovery,
    Invocation,
    Error,
  ),
) -> Recovery

with_deadline

</>

pub fn with_deadline(
  operation: @internal Operation(
    context,
    input,
    output,
    Kind,
    Recovery,
    Invocation,
    Error,
  ),
  within: run.Timeout,
) -> @internal Operation(
  context,
  input,
  output,
  Kind,
  Recovery,
  Invocation,
  Error,
)

Bounds an admitted wait (a signal, a job, a managed child or a fork): run.After(duration), from 1 ms to 2^32 - 1 ms, or run.Infinity to wait without a deadline. The default is 7 days. The store’s clock starts the duration once policy (and any approval) admits the wait; the due time is stored, so it survives restarts. A wait past its deadline fails (graph.DeadlineExpired); owned jobs, children and forks are stopped first, and their cleanup progress is kept.

The deadline is part of the operation’s stored contract: a stored run continues only under the deadline it was admitted with. A run stored without one (before waits had a default, or under run.Infinity) keeps none. Seven days set explicitly is the default. definition.build refuses a deadline on an activity (DeadlineRequiresWait) or out of range (InvalidLimit(Deadline, ..)).

with_replay

</>

pub fn with_replay(
  operation: @internal Operation(
    context,
    input,
    output,
    Kind,
    Recovery,
    Invocation,
    Error,
  ),
  max_attempts: Int,
) -> @internal Operation(
  context,
  input,
  output,
  Kind,
  Recovery,
  Invocation,
  Error,
)

Lets the runtime start an activity’s body again, up to max_attempts starts in all, when its runner was lost while it ran (a restart, a lost node). Use it only for a body whose effect is safe to repeat (keyed by Invocation.run and activation). definition.build refuses it on any other kind of operation (ReplayRequiresActivity) and outside 1 to 100 attempts (InvalidLimit(ReplayAttempts, ..)).