ALLM.Pipeline.Query (allm_pipeline v0.1.0)

Copy Markdown View Source

The package-owned, read-only query facade for host UIs and host contexts.

Every function here delegates to an existing implementation on PipelineRun, StepLog or ArtifactStore. The value is a single, @spec'd, host-facing module instead of hosts reaching into three schemas directly — the host cutover (Phase 7.2) routes a host app's provenance/lineage reads through this module rather than hand-writing raw "step_logs" / "pipeline_runs" queries to avoid inverting the host->package dependency (a dependency that, since Phase 7.1, the host's core app may legitimately declare — an in-umbrella dep then, the Phase 8 path dep now).

Read-only by contract

Query exposes no write path — no create/update/delete, no run termination. Terminating a run is an ownership capability that lives on PipelineRun behind the completion token (see its moduledoc); a read facade must not offer a door around it. Adding a write function here is a design error, not a convenience.

No new Store callback

Query lives inside the package, so it calls the schema query functions directly. It is NOT a framework seam and needs no Store behaviour callback — those exist for the WRITE path the Executor drives (store.ex). A read that a host wants goes here as a delegating function, not onto the Store behaviour.

Summary

Functions

Count pipeline runs matching list_runs/1's filters (delegates to PipelineRun.count/1).

Fetch a stored artifact's decompressed bytes (delegates to ArtifactStore.fetch/1).

Get a run with its step logs preloaded (delegates to PipelineRun.get_with_steps/1).

Get a single step log (delegates to StepLog.get/1).

Build the lineage tree upward from a step log (delegates to StepLog.build_lineage_tree/1, which takes a step id).

List pipeline runs (delegates to PipelineRun.list/1).

Walk the input_step_id lineage upward from step_log_id and return the first (nearest) step's llm_artifact_url, or nil when none is found within max_depth hops.

Resolve a step log's identity to %{run_id, pipeline_name, step_type}, or nil when the id is nil, unknown, or its run is missing.

Aggregate step statistics for a run (delegates to StepLog.get_pipeline_stats/1).

Functions

count_runs(opts \\ [])

@spec count_runs(keyword()) :: non_neg_integer()

Count pipeline runs matching list_runs/1's filters (delegates to PipelineRun.count/1).

fetch_artifact(url)

@spec fetch_artifact(String.t()) :: {:ok, binary()} | {:error, term()}

Fetch a stored artifact's decompressed bytes (delegates to ArtifactStore.fetch/1).

get_run(id)

@spec get_run(Ecto.UUID.t()) :: ALLM.Pipeline.PipelineRun.t() | nil

Get a run with its step logs preloaded (delegates to PipelineRun.get_with_steps/1).

get_step(id)

@spec get_step(Ecto.UUID.t()) :: ALLM.Pipeline.StepLog.t() | nil

Get a single step log (delegates to StepLog.get/1).

lineage_tree(step_log_id)

@spec lineage_tree(Ecto.UUID.t()) :: {:ok, [map()]} | {:error, term()}

Build the lineage tree upward from a step log (delegates to StepLog.build_lineage_tree/1, which takes a step id).

list_runs(opts \\ [])

@spec list_runs(keyword()) :: [ALLM.Pipeline.PipelineRun.t()]

List pipeline runs (delegates to PipelineRun.list/1).

llm_artifact_url(step_log_id, max_depth)

@spec llm_artifact_url(Ecto.UUID.t() | nil, non_neg_integer()) :: String.t() | nil

Walk the input_step_id lineage upward from step_log_id and return the first (nearest) step's llm_artifact_url, or nil when none is found within max_depth hops.

The depth cap is a parameter, not a package constant — the host's @llm_lineage_max_depth stays host policy and is passed in (7.2). max_depth of 0 returns nil without examining any step; max_depth of n examines the steps at distance 0..n-1 from step_log_id, matching the host's original recursive walk exactly.

The upward traversal reuses StepLog.build_lineage_tree/1's recursive CTE rather than carrying a second input_step_id walker in the package (one traversal, one place).

resolve_step_log(step_log_id)

@spec resolve_step_log(Ecto.UUID.t() | nil) ::
  %{run_id: Ecto.UUID.t(), pipeline_name: String.t(), step_type: String.t()}
  | nil

Resolve a step log's identity to %{run_id, pipeline_name, step_type}, or nil when the id is nil, unknown, or its run is missing.

Returns a map, deliberately not a {run_id, name, step_type} tuple: a map lets a future field be added without breaking callers. This is the host provenance-attribution read path (Government.resolve_step_log/1, retired onto this in 7.2).

run_stats(run_id)

@spec run_stats(Ecto.UUID.t()) :: map()

Aggregate step statistics for a run (delegates to StepLog.get_pipeline_stats/1).