ALLM.Pipeline.Context (allm_pipeline v0.1.0)

Copy Markdown View Source

Pipeline execution context passed to each step — and, since Phase 4, to every ALLM.Pipeline escape-hatch body and FanOut.reduce/5 fold function as well.

Contains the current pipeline run and step log, allowing steps to access pipeline-level information and log additional data.

One context, not two (Phase 4 D5)

An escape-hatch body is not a step: it has no step log, and its lineage parent is the last successfully executed step's log id rather than its own row. Rather than introduce a second context type — two names for one idea, and a forced widening of ALLM.Pipeline.Step.context/0 anyway — this struct gained three fields and step_log became nilable:

FieldAccessorWho writes it
resourcesresource/2ALLM.Pipeline.Dsl.Runtime (resource is Phase 4.3; the field and reader ship in 4.1 so Runtime has one shape to build)
carrycarried/2Runtime, from a stage's carry: [...] declaration
input_step_idinput_step_id/1Runtime, or derived from step_log.input_step_id
accaccumulator/1Runtime, before each body invocation

acc is the read half of the accumulator channel. A body writes it by returning {item_result, acc} (ALLM.Pipeline's item-result contract) — which is only expressible if the body can also SEE the current value, since meeting_agenda's per-item body updates a nine-key stats map through fifteen nested call sites. It is a snapshot taken before the body runs, never a mutable handle: writing goes through the return value and nowhere else.

resources and carry are struct FIELDS, not opts keys. That is the whole point of §3.10's Context.resource(ctx, :browser) over Keyword.get(opts, :page): a resource is framework-managed state with a lifecycle, and burying it in the caller's option list makes it indistinguishable from a CLI flag. new/3 therefore pops :resources, :carry, :input_step_id and :acc out of opts — they never reach get_opt/3.

Summary

Types

Values captured by a carry: [...] declaration, keyed by field name. Carried values live HERE rather than on a step log, which is what makes a stage's survive any number of skipped stages between producer and consumer (Phase 4 D4).

Framework-managed handles acquired once per run (a browser, a connection pool), keyed by the name their resource declaration gave them.

t()

Functions

The pipeline accumulator as it stood when this unit was invoked.

A value captured by a carry: [...] declaration, or default.

A context for a Step invoked outside any pipeline run — a mix task, a backfill, an eval harness, an ad-hoc iex call.

Get an option from context.

The lineage parent this unit was invoked under.

Create a new context for a step execution.

Get the pipeline run ID from context.

A framework-managed resource acquired for this run, or nil.

Get the current step log ID from context.

Types

carry()

@type carry() :: %{optional(atom()) => term()}

Values captured by a carry: [...] declaration, keyed by field name. Carried values live HERE rather than on a step log, which is what makes a stage's survive any number of skipped stages between producer and consumer (Phase 4 D4).

The scope depends on the declaring stage's kind and the two are not the same: a stage captures from its own output and the values reach every later stage; a fan_out captures from each item into that item's own context only, and nothing is propagated past the stage. The authoritative statement is ALLM.Pipeline.Dsl.Stage's "carry:'s scope differs by kind".

resources()

@type resources() :: %{optional(atom()) => term()}

Framework-managed handles acquired once per run (a browser, a connection pool), keyed by the name their resource declaration gave them.

t()

@type t() :: %ALLM.Pipeline.Context{
  acc: term(),
  carry: carry(),
  input_step_id: Ecto.UUID.t() | nil,
  opts: keyword(),
  pipeline_run: ALLM.Pipeline.PipelineRun.t() | nil,
  resources: resources(),
  step_log: ALLM.Pipeline.StepLog.t() | nil
}

Functions

accumulator(context)

@spec accumulator(t()) :: term()

The pipeline accumulator as it stood when this unit was invoked.

A snapshot, not a handle: the only channel that writes the accumulator is a body returning {item_result, acc}. nil outside a DSL run and for a pipeline that declares no init: and never writes it.

carried(context, name, default \\ nil)

@spec carried(t(), atom(), term()) :: term()

A value captured by a carry: [...] declaration, or default.

From an earlier stage (its captures reach every later stage), or from this item's own fan_out (its captures are item-scoped and do not propagate). See carry/0's typedoc.

default here does not distinguish "declared and legitimately absent" from "declared a key the subject never had" — but the second is not silent: Runtime.capture/3 warns at capture time. See ALLM.Pipeline.Dsl.Stage's "A key the subject does not have".

detached(opts \\ [])

@spec detached(keyword()) :: t()

A context for a Step invoked outside any pipeline run — a mix task, a backfill, an eval harness, an ad-hoc iex call.

Carries no run and no step log, only opts. This is the named replacement for the SomeStep.execute(%{}, input) idiom, which stopped being in contract when Phase 4 widened ALLM.Pipeline.Step.context/0 from a bare map to this struct (D5). The bare map still WORKS — a struct is a map and every such caller ignores the context — so migrating the remaining sites is bookkeeping, tracked in .work/HANDOFF.md, not a correctness fix.

get_opt(context, key, default \\ nil)

@spec get_opt(t(), atom(), term()) :: term()

Get an option from context.

input_step_id(context)

@spec input_step_id(t()) :: Ecto.UUID.t() | nil

The lineage parent this unit was invoked under.

For a step, the step log's own input_step_id. For an escape-hatch body or FanOut.reduce/5 fold function, the id of the last successfully executed step's log — which the body threads to its own ALLM.Pipeline.Executor.run_step/5 calls exactly as hand-written orchestrator code does today. Lineage in is free; lineage out is by hand (Phase 4 D2).

new(pipeline_run, step_log, opts \\ [])

Create a new context for a step execution.

step_log may be nil — an escape-hatch body is not a step and has no row.

Four keys are read out of opts onto struct fields and removed from the option list: :resources, :carry, :acc and :input_step_id. When :input_step_id is absent it falls back to the step log's own input_step_id, so the accessor answers the same question ("what is this unit's lineage parent?") for a step and for a body.

pipeline_run_id(context)

@spec pipeline_run_id(t()) :: Ecto.UUID.t() | nil

Get the pipeline run ID from context.

nil for a detached/1 context, which has no run — mirroring step_log_id/1. A single-clause version raises FunctionClauseError on every detached caller, which is the wrong failure for an accessor whose whole purpose is to answer "am I inside a run?".

resource(context, name)

@spec resource(t(), atom()) :: term() | nil

A framework-managed resource acquired for this run, or nil.

Reads the resources struct field and nothing else. It deliberately does not fall back to opts: a lookup that degrades to the caller's option list answers plausibly for a resource that was never acquired, which is the failure this accessor exists to make visible.

step_log_id(context)

@spec step_log_id(t()) :: Ecto.UUID.t() | nil

Get the current step log ID from context.

nil for an escape-hatch body, which has no step log of its own.