@spec borrowed_run(keyword()) :: {:ok, ALLM.Pipeline.PipelineRun.t()} | :error

Resolve a borrowed pipeline run from an inner pipeline's opts.

The canonical borrowed-run boundary. An umbrella pipeline lends its run by putting it under the :pipeline_run opt; the inner pipeline reads it here and receives a NON-OWNING handle (PipelineRun.borrow/1), so an accidental PipelineRun.complete/2 on the inner side is a detectable {:error, :not_run_owner} rather than a silent mid-loop :success that clobbers the umbrella's aggregate metadata.

The strip happens on RECEIPT rather than at the lending call site on purpose: a future umbrella that forgets to mark its lend still cannot have its run completed out from under it. Returns :error when no run was lent, which is the self-owned (*_single / CLI) branch.

@spec complete_pipeline_run(ALLM.Pipeline.PipelineRun.t(), [run_step_result()], [
  run_step_result()
]) ::
  {:ok, ALLM.Pipeline.PipelineRun.t()}
  | {:error, Ecto.Changeset.t()}
  | {:error, :not_run_owner}

Mark pipeline run as completed with statistics.

Requires an owning handle — see PipelineRun.complete/2. The guard lives on the schema function, not here: this wrapper is only 2 of the 17 call sites that complete a run, so guarding it alone would leave the other 15 unprotected.

@spec create_pipeline_run(String.t(), map(), keyword()) ::
  {:ok, ALLM.Pipeline.PipelineRun.t()} | {:error, Ecto.Changeset.t()}

Create a new pipeline run record.

The returned run is the owning handle: it carries the completion token that PipelineRun.complete/2 requires. Lend it to an inner pipeline through the :pipeline_run opt and the inner side strips the token via borrowed_run/1, so only this caller can complete the run.

attrs carries first-class COLUMN values — :trigger (what fired the run) and :parent_run_id (a linking sub-pipeline parent). They ride as scalar changeset fields, NOT through metadata, so they stay SQL/GraphQL-filterable (Subphase 2).

When :trigger is absent from attrs, it is backfilled from the process dictionary via trigger_from_process/0 (cron-stamped "cron:<name>", else "cli"). This is the single default point so every cron-dispatchable pipeline records the right trigger without each call site re-supplying it, and ad-hoc CLI single runs (which pass no attrs) are consistently tagged "cli" rather than nil. A call site can still override by passing :trigger explicitly.

@spec fail_pipeline_run(ALLM.Pipeline.PipelineRun.t(), term()) ::
  {:ok, ALLM.Pipeline.PipelineRun.t()}
  | {:error, Ecto.Changeset.t()}
  | {:error, :not_run_owner}

Mark pipeline run as failed.

Requires an owning handle, exactly as complete_pipeline_run/3 does — see PipelineRun.fail/2. The guard lives on the schema function, not here, for the same reason: this wrapper is only some of the sites that fail a run.

@spec finish_run(ALLM.Pipeline.PipelineRun.t(), result) :: result when result: var

Terminate the run you own according to result, and return result unchanged.

The complete-or-fail tail every self-owned pipeline entry point needs. It lived hand-written in four places (three run_list_only/1 helpers plus VideoSummaryPipeline's own copy) before being lifted here, which is also where it belongs: "finish the run you own" is framework logic, not consumer logic, so it travels with Executor when the framework moves.

Matches both result shapes the codebase produces — the 2-tuple a pipeline returns ({:ok, stats} / {:error, reason}) and the 3-tuple run_step/5 returns ({:ok, step_log, output} / {:error, step_log, reason}). Callers whose own return shape differs from the result they pass in keep that translation at the call site; this helper never reshapes.

Requires an owning handle — both branches now refuse a borrowed or re-loaded run (PipelineRun.complete/2, PipelineRun.fail/2).

It does NOT guard. A raise, exit or throw between the caller's create_pipeline_run/3 and this call strands the run at :running; this function only writes the terminal status for a result it is handed. An entry point that creates its own run wants ALLM.Pipeline.Lifecycle.owned_run/4, which owns creation, the guard and the metadata argument as well — see that module's "Versus Executor.finish_run/2" table for the boundary.

@spec get_status(Ecto.UUID.t()) :: {:ok, map()} | {:error, :not_found}

Get pipeline execution status and statistics.

@spec log_section(ALLM.Pipeline.PipelineRun.t(), String.t(), Ecto.UUID.t() | nil) ::
  {:ok, ALLM.Pipeline.StepLog.t()} | {:error, Ecto.Changeset.t()}

Log a section marker for visual grouping in the admin pipeline review UI.

@spec log_skipped(
  ALLM.Pipeline.PipelineRun.t(),
  String.t(),
  term(),
  Ecto.UUID.t() | nil
) ::
  {:ok, ALLM.Pipeline.StepLog.t()} | {:error, Ecto.Changeset.t()}

Write a visible :skipped step log for a gate decision that declined to process an item — the record a fan_out body writes before it returns {{:skipped, payload}, acc}.

input_step_id is the fan-out's parent (Context.input_step_id(ctx)), so the skip lands in the lineage tree at the position the processed step would have occupied. reason is stored jsonb-safe (see StepLog.create_skipped/4). The metric increment stays in the body — this call adds the LOG, not the count.

@spec log_summary(
  ALLM.Pipeline.PipelineRun.t(),
  String.t(),
  map(),
  Ecto.UUID.t() | nil
) ::
  {:ok, ALLM.Pipeline.StepLog.t()} | {:error, Ecto.Changeset.t()}

Record a step carrying structured output_data (e.g. a decision-log summary) for inspection in the pipeline-review UI.

@spec resume(Ecto.UUID.t(), Ecto.UUID.t()) ::
  {:ok, ALLM.Pipeline.PipelineRun.t()} | {:error, term()}

Re-open a failed pipeline run, having checked that from_step_id belongs to it. Two limitations, both current behaviour rather than bugs to route around:

1. It does not replay anything. All it does is validate the step and put the run back to :running (PipelineRun.start/1); no step output is restored and no step is skipped, so a caller that re-drives the pipeline re-executes everything. Recorded in §2.5 of steering/2026-08-10_ALLM_PIPELINE_EXTRACTION.md; resume-from-log — step fingerprints, replaying a :success log instead of executing — is Phase 7 (§3.11) and is not attempted here.

2. The returned handle is NOT an owner. The run is loaded via PipelineRun.get/1, and :completion_token is virtual, so the handle comes back token-less by construction and PipelineRun.complete/2, fail/2 and cancel/1 all refuse it with {:error, :not_run_owner} — a caller that finishes the resumed run without noticing strands it at :running. A caller that means to finish it must say so explicitly:

   {:ok, run} = Executor.resume(run_id, step_id)
   run = PipelineRun.assume_ownership(run)

   Deliberately not re-minted inside this function: that would make resume/2 a second, implicit mint point and undercut the invariant the whole ownership design rests on (user decision, 2026-08-13 — see steering/2026-08-10_ALLM_PIPELINE_EXTRACTION_RECORDS.md). Pinned by pipeline_run_test.exs's "every terminal writer refuses a non-owning handle", which loops this provenance alongside borrowed and re-loaded ones.

There are no production callers today; the callers are tests.

@spec run_step(
  ALLM.Pipeline.PipelineRun.t(),
  module(),
  struct(),
  Ecto.UUID.t() | nil,
  keyword()
) ::
  run_step_result()

Execute a single step with typed I/O validation and logging.

Parameters

• pipeline_run - The parent pipeline run
• step_module - Module implementing the Step behavior
• input_struct - Validated input struct matching the step's input_schema
• input_step_id - Optional ID of the step that produced this input (for lineage)
• opts - Additional options passed to the step context

Returns

• {:ok, step_log, output_struct} on success
• {:error, step_log, reason} on failure
• {:error, nil, reason} when the failure happened before a step log existed (input-schema mismatch, or the step_logs insert itself failing)

Never raises for a failure it can name. Callers fan this out through Task.async_stream, which LINKS its children — so a raise here kills the whole fan-out (and, with trap_exit off, the caller process itself) rather than failing one item. Every failure path therefore returns a tuple.

@spec trigger_from_process() :: String.t()

The trigger value for a run, sourced from the process dictionary.

A host's cron entry point stamps :pipeline_trigger before dispatching synchronously in the same process, so a pipeline's own create_pipeline_run call reads it here. Absent (a dev CLI path that never goes through the cron runner, and any direct/iex call), it falls back to "cli". Best-effort: this never raises.