Persists step execution logs to PostgreSQL for observability and lineage tracking.
Key design decisions:
pipeline_run_idstored directly for efficient querying (no recursive joins needed)input_step_idenables lineage tree reconstruction when needed- Timing fields capture execution performance metrics
- Input/output schemas stored as JSONB for debugging and replay
Two-layer serialization
input_data / output_data are produced by serialize_struct/2, which keeps
heavy bodies out of Postgres using two layers:
- Layer 1 — per-field flags. A struct whose module exports
__allm_schema__/1(i.e. one built withALLM.Pipeline.Schema) contributes__allm_schema__(:dropped)—log: falseorartifact: true— and__allm_schema__(:redacted). - Layer 2 —
@fallback_drop, a package-level list of four generic field names that applies to every struct, DSL or not. It is what covers plaindefstructs and live Ecto structs reached through recursion, which carry no flags at all.
The two are additive: the drop set is
:dropped ∪ (@fallback_drop − :kept). Flags do not replace the fallback — an
unflagged field :content, … on a DSL struct is still dropped, which is the
point (a host's render-step Output relies on it). The one
escape in the other direction is an explicit log: true.
⚠️ The predicate is __allm_schema__/1, never __schema__/1: every
Ecto.Schema module exports the latter, __schema__(:fields) succeeds with
a colliding shape, and __schema__(:dropped) raises FunctionClauseError —
on this un-rescued write path. See ALLM.Pipeline.Schema's moduledoc.
Recursion carries a depth budget of 16 levels and truncates rather than
raising; see @max_depth.
An %Ecto.Changeset{} reaching the serializer flattens SILENTLY
Ecto.Changeset is the one declared divergence from ALLM.Pipeline.Encodable's
leaf rules (Encodable renders it as %{"changeset_errors" => …}; this module
treats it as an ordinary struct), and subphase 2.2's tuple clause turned that
divergence from a loud failure into a quiet write. A changeset now flattens to
all fifteen defstruct keys and persists params, data, types,
changes and errors in full — so a changeset built from user or LLM
input writes those raw params into input_data / output_data, where
redact: cannot reach them (no field declares a changeset type, so layer 1 has
no flags to read). This is the same "loud failure turned QUIET" shape
ALLM.Pipeline.Encodable's moduledoc documents for its own is_struct
widening, mirrored here because StepLog is the path that reaches a row.
The quiet write is not total, which is the trap: measured 2026-08-14, a
changeset that has been through Ecto.Changeset.prepare_changes/2 carries
anonymous functions in :prepare, which reach Jason unchanged and still
raise Protocol.UndefinedError on the un-rescued log_start/4 path. So the
same type both writes and raises depending on how it was built.
No field DECLARES a changeset type today, and that is the exact scope of the claim — the sweep is type-declaration-based while the hazard is not. Re-derive it NUL-safely and across both extensions (2026-08-14):
python3 scripts/refsweep.py 'field\(.*Changeset' apps scripts steering \
--include '*.ex' --include '*.exs' --format hits→ 1 hit, and it is this moduledoc, at the line just below quoting the
superseded command grep -rna "field(.*Changeset" apps/ --include=*.ex (which
could not see .exs at all). Real declarations: zero. Read the expected
count as "every hit is prose", not as a number — this paragraph supplies its
own match, so a bare → 1 would stop meaning anything the moment someone
rewords it.
What no such sweep can see is a field declared term(), map() or
[map()] that holds a changeset at runtime — which is exactly how one would
arrive, since Step Outputs routinely carry term()-typed result collections.
That half is not closed by any grep. It was closed once, by tracing, and the
trace is not re-derivable from this file: subphase 2.3's security review
(.work/security-reviews/2026-08-14-allm-p2c.md, Informational 4) walked the
five loaders and found a changeset surfacing only as a failure return (→
normalize_error/1, which since Phase 5.10 renders it params-free via
Encodable.encode/1's changeset_errors leaf) and via Encodable.encode/1
(→ changeset_errors only),
so none reaches this serializer. That is a dated observation about the
loaders, not a property of this module, and a new term()-typed Step Output
can falsify it without touching anything here.
Which is why the standing fix is not a wider sweep: give serialize_struct/2
an %Ecto.Changeset{} clause mirroring Encodable's changeset_errors leaf.
It closes the silent write, the prepare_changes/2 raise, and the declared
divergence at once, and it needs no dated evidence. Tracked as an open item in
.work/HANDOFF.md.
Summary
Functions
Build lineage tree from step logs using recursive CTE query.
Create a changeset for a step log.
Per-step_type row counts for a run, as %{step_type => %{status => count}}.
Create a zero-duration :skipped step from scratch — the visible record of a
gate decision that declined to process an item.
Get a step log by ID.
Get all downstream steps (children) of a given step.
Get failed steps for a pipeline run.
Get pipeline statistics.
Get all steps for a pipeline run (efficient direct query via pipeline_run_id).
Get all root steps (steps with no input_step_id).
Get steps by type for a pipeline run.
Log step failure with error details.
Log a section marker for visual grouping in the admin UI.
Log an already-started step as skipped (UPDATE path).
Log the start of a step execution.
Log successful completion with output and artifact.
Create a successful, zero-duration step that carries structured output_data.
Types
@type status() :: :pending | :running | :success | :failed | :skipped
@type t() :: %ALLM.Pipeline.StepLog{ __meta__: term(), artifact_checksum: String.t() | nil, artifact_size_bytes: non_neg_integer() | nil, artifact_url: String.t() | nil, completed_at: DateTime.t() | nil, downstream_steps: [t()] | Ecto.Association.NotLoaded.t(), duration_ms: non_neg_integer() | nil, error: map() | nil, id: Ecto.UUID.t() | nil, input_data: map() | nil, input_schema: String.t() | nil, input_step: t() | Ecto.Association.NotLoaded.t() | nil, input_step_id: Ecto.UUID.t() | nil, inserted_at: DateTime.t() | nil, llm_artifact_checksum: String.t() | nil, llm_artifact_size_bytes: non_neg_integer() | nil, llm_artifact_url: String.t() | nil, llm_call_count: non_neg_integer() | nil, llm_total_tokens: non_neg_integer() | nil, output_data: map() | nil, output_schema: String.t() | nil, pipeline_run: ALLM.Pipeline.PipelineRun.t() | Ecto.Association.NotLoaded.t(), pipeline_run_id: Ecto.UUID.t() | nil, queue_time_ms: non_neg_integer() | nil, retry_count: non_neg_integer(), started_at: DateTime.t() | nil, status: status() | nil, step_type: String.t() | nil, updated_at: DateTime.t() | nil }
Functions
@spec build_lineage_tree(Ecto.UUID.t()) :: {:ok, [map()]} | {:error, term()}
Build lineage tree from step logs using recursive CTE query.
Returns steps from the root (oldest ancestor) to the given step.
@spec changeset(t(), map()) :: Ecto.Changeset.t()
Create a changeset for a step log.
@spec count_by_step_type(Ecto.UUID.t()) :: %{ required(String.t()) => %{required(atom()) => non_neg_integer()} }
Per-step_type row counts for a run, as %{step_type => %{status => count}}.
The aggregate form of "how many rows of type X, and how many of those
succeeded". Use this — not get_pipeline_steps/1 + Enum.group_by/2 — when
the answer is a COUNT: get_pipeline_steps/1 is select * and materialises
every row's input_data/output_data jsonb, which is a few kilobytes for a
31-item run and megabytes for a large one (the biggest
meeting_agenda_scrape runs in dev are ~2600 rows / ~5 MB, measured
2026-08-21). A use ALLM.Pipeline pipeline folding run-level counters in a
stage :tally is the canonical caller.
Sections are INCLUDED (unlike get_pipeline_stats/1, which excludes them):
callers key on a specific Step.step_type(), so a "section" bucket is
simply never read, and excluding it here would make the function unusable for
anyone who wanted to count them.
@spec create_skipped(Ecto.UUID.t(), String.t(), term(), Ecto.UUID.t() | nil) :: {:ok, t()} | {:error, Ecto.Changeset.t()}
Create a zero-duration :skipped step from scratch — the visible record of a
gate decision that declined to process an item.
Unlike log_skipped/2 (which UPDATES a row a step already started), this is a
CREATE path: a *ProcessingDecision skip fires before any step log exists, so
there is no %StepLog{} and no started_at to diff against — started_at and
completed_at are both now, giving duration_ms: 0. Promoted to a Store
callback and wired to the three ProcessingDecision skip branches in Phase 7.4
so a skip is a queryable :skipped row (counted by get_pipeline_stats/1)
rather than the invisible {:skipped, …} return it was through Phase 6.
reason is an arbitrary term (the pipelines pass a {scraper_identifier, reason} payload). It is made jsonb-safe by Encodable.encode/1 — which
flattens the tuple to a list and scrubs binaries — and stored under
output_data["reason"] (the audit-artifact column, as log_summary/4 uses,
NOT error: a skip is a benign decision, not a failure). input_step_id
should be the same lineage parent the processed step would have carried, so the
skip appears in build_lineage_tree/1 at the position the work would occupy.
@spec get(Ecto.UUID.t()) :: t() | nil
Get a step log by ID.
@spec get_downstream_steps(Ecto.UUID.t()) :: [t()]
Get all downstream steps (children) of a given step.
@spec get_failed_steps(Ecto.UUID.t()) :: [t()]
Get failed steps for a pipeline run.
@spec get_pipeline_stats(Ecto.UUID.t()) :: map()
Get pipeline statistics.
@spec get_pipeline_steps(Ecto.UUID.t()) :: [t()]
Get all steps for a pipeline run (efficient direct query via pipeline_run_id).
@spec get_root_steps(Ecto.UUID.t()) :: [t()]
Get all root steps (steps with no input_step_id).
@spec get_steps_by_type(Ecto.UUID.t(), String.t()) :: [t()]
Get steps by type for a pipeline run.
@spec log_failure(t(), term(), keyword()) :: {:ok, t()} | {:error, Ecto.Changeset.t()}
Log step failure with error details.
@spec log_section(Ecto.UUID.t(), String.t(), Ecto.UUID.t() | nil) :: {:ok, t()} | {:error, Ecto.Changeset.t()}
Log a section marker for visual grouping in the admin UI.
Creates a step_log with step_type "section" that serves as a visual divider in the pipeline review interface. Sections are excluded from pipeline stats.
The title is ALLM.Pipeline.Text.scrub/1-ed. It is the one field on this row
that comes from OUTSIDE — the DSL's section: hook derives it from a scraped
or OCR'd item — and a NUL byte or an invalid UTF-8 sequence in it fails the
insert with Postgres 22P05, aborting a fan-out mid-run for a value that is
only ever displayed. Same treatment ALLM.Pipeline.Encodable.encode/1 gives
run metadata.
@spec log_skipped(t(), String.t()) :: {:ok, t()} | {:error, Ecto.Changeset.t()}
Log an already-started step as skipped (UPDATE path).
Takes a %StepLog{} whose started_at is set and closes it :skipped,
computing duration_ms from that timestamp. This is NOT the path a
*ProcessingDecision skip takes — that decision happens before any step log
exists (no struct, no started_at), so it uses create_skipped/4 instead.
@spec log_start(Ecto.UUID.t(), module(), struct(), Ecto.UUID.t() | nil) :: {:ok, t()} | {:error, Ecto.Changeset.t()}
Log the start of a step execution.
@spec log_success(t(), struct(), map()) :: {:ok, t()} | {:error, Ecto.Changeset.t()}
Log successful completion with output and artifact.
@spec log_summary(Ecto.UUID.t(), String.t(), map(), Ecto.UUID.t() | nil) :: {:ok, t()} | {:error, Ecto.Changeset.t()}
Create a successful, zero-duration step that carries structured output_data.
Unlike log_section/3 (which only holds a title for visual grouping), this
records a real audit artifact — e.g. the video↔meeting match-decision log —
that the pipeline-review UI renders. output_data must be JSON-serializable.