defmodule OpenAgents.Stacks.CheckRun do

  @moduledoc """

  One check evaluation of one stacked pull request layer.

  Validity keys to the full context — `(pull_request_id, workflow_name,

  context, head_oid, effective_base_oid, workflow_definition_oid)` — never

  to the head OID alone (docs/stacked-prs.md section 11.3). `tested_oid` is

  the immutable snapshot the check actually runs against: the layer head

  when the stack is rebased onto the current trunk, otherwise a synthetic

  commit published under `refs/internal/checks/<run-id>` whose tree is the

  current trunk plus every layer through this position (section 11.4).

  A trunk advance changes the effective base OID, so existing runs no

  longer match the current identity and become `stale` rather than

  silently vouching for a state that no longer exists.

  """

  use Ecto.Schema

  import Ecto.Changeset

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  @timestamps_opts [type: :utc_datetime_usec]

  @states ~w(pending passed failed stale)

  @contexts ~w(layer merge_group)

  @run_on_policies ~w(every_layer top_layer_only bottom_layer_only changed_paths merge_group_only)

  schema "stack_check_runs" do

    belongs_to :repository, OpenAgents.Repositories.Repository

    belongs_to :stack, OpenAgents.Stacks.Stack

    belongs_to :pull_request, OpenAgents.PullRequests.PullRequest

    field :workflow_name, :string

    field :run_on, :string

    field :required, :boolean, default: false

    field :run_reason, :string

    field :context, :string, default: "layer"

    field :head_oid, OpenAgents.Stacks.OID

    field :effective_base_oid, OpenAgents.Stacks.OID

    field :workflow_definition_oid, OpenAgents.Stacks.OID

    field :tested_oid, OpenAgents.Stacks.OID

    field :synthetic_ref, :string

    field :state, :string, default: "pending"

    field :concluded_at, :utc_datetime_usec

    timestamps()

  end

  def states, do: @states

  def contexts, do: @contexts

  def run_on_policies, do: @run_on_policies

  def changeset(check_run, attrs) do

    check_run

    |> cast(attrs, [

      :workflow_name,

      :run_on,

      :required,

      :run_reason,

      :context,

      :head_oid,

      :effective_base_oid,

      :workflow_definition_oid,

      :tested_oid,

      :synthetic_ref,

      :state

    ])

    |> put_programmatic_change(attrs, :id)

    |> put_programmatic_change(attrs, :repository_id)

    |> put_programmatic_change(attrs, :stack_id)

    |> put_programmatic_change(attrs, :pull_request_id)

    |> validate_required([

      :repository_id,

      :stack_id,

      :pull_request_id,

      :workflow_name,

      :run_on,

      :run_reason,

      :context,

      :head_oid,

      :effective_base_oid,

      :workflow_definition_oid,

      :tested_oid,

      :state

    ])

    |> validate_inclusion(:state, @states)

    |> validate_inclusion(:context, @contexts)

    |> validate_inclusion(:run_on, @run_on_policies)

    |> unique_constraint(

      [

        :pull_request_id,

        :workflow_name,

        :context,

        :head_oid,

        :effective_base_oid,

        :workflow_definition_oid

      ],

      name: :stack_check_runs_identity_index

    )

    |> check_constraint(:state, name: :stack_check_runs_state_check)

    |> check_constraint(:context, name: :stack_check_runs_context_check)

    |> check_constraint(:run_on, name: :stack_check_runs_run_on_check)

    |> foreign_key_constraint(:repository_id)

    |> foreign_key_constraint(:stack_id)

    |> foreign_key_constraint(:pull_request_id)

  end

  @doc "Concludes a pending run as passed or failed."

  def conclusion_changeset(check_run, state, concluded_at) when state in ~w(passed failed) do

    change(check_run, state: state, concluded_at: concluded_at)

  end

  @doc "Marks a run stale: its effective base no longer matches the trunk."

  def stale_changeset(check_run) do

    change(check_run, state: "stale")

  end

  defp put_programmatic_change(changeset, attrs, field) do

    case Map.fetch(attrs, field) do

      {:ok, value} -> put_change(changeset, field, value)

      :error -> changeset

    end

  end

end

defmodule OpenAgents.Stacks.Checks do

  @moduledoc """

  Stack-aware check planning (docs/stacked-prs.md sections 11.3–11.5).

  Checks on a stacked pull request test an immutable snapshot that includes

  every lower layer. Each planned run keys its validity to the full context

  — `(pull_request_id, workflow_name, context, head_oid, effective_base_oid,

  workflow_definition_oid)` — so a trunk advance, a head move, or a workflow

  definition change produces a different identity instead of reusing a

  verdict for a state that no longer exists. Planning marks runs whose

  effective base no longer matches the trunk tip `stale` before creating

  runs for the current identity.

  When a layer head does not already sit on the snapshot base (the stack

  needs a rebase, or the plan speculates for a merge group), the snapshot

  is a boundary-scoped replay of the layer chain onto the current trunk,

  published once under `refs/internal/checks/<run-id>` with an

  expected-absent compare-and-swap. Nothing ever moves that ref afterward:

  the run's snapshot is immutable for its lifetime.

  Workflow definitions live in `#{inspect(".forge/workflows.json")}` in the

  effective-base tree — never an intermediate branch tree — and each

  workflow declares an explicit `run_on` policy: `every_layer`,

  `top_layer_only`, `bottom_layer_only`, `changed_paths` (with declared

  path prefixes), or `merge_group_only`. A plan skips a layer only through

  that declared policy, and every skip is returned explicitly in the plan

  rather than silently dropped. In the `merge_group` context every workflow

  runs: the merge group is the authoritative gate and never optimizes a

  required check away.

  """

  import Ecto.Query

  alias OpenAgents.Forge.Browse

  alias OpenAgents.Forge.GitPlane

  alias OpenAgents.Forge.Repos

  alias OpenAgents.Repo

  alias OpenAgents.Repositories.Repository

  alias OpenAgents.Stacks

  alias OpenAgents.Stacks.CheckRun

  alias OpenAgents.Stacks.Stack

  @workflows_path ".forge/workflows.json"

  @principal "stack-checks"

  @doc "The path of the workflow definitions file in the effective-base tree."

  def workflows_path, do: @workflows_path

  @doc """

  The workflow definitions governing checks at `base_oid`.

  Reads `#{inspect(".forge/workflows.json")}` from the given commit's tree and

  returns `{:ok, %{definition_oid: blob_oid, workflows: [workflow]}}` where

  each workflow has `name`, `run_on`, `required`, and `paths`. Returns

  `{:error, :no_workflows}` when the file is absent and

  `{:error, {:invalid_workflows, reason}}` when it does not parse into the

  declared policy shape.

  """

  def workflows(%Repository{} = repository, base_oid) do

    with {:ok, definition_oid} <- definition_oid(repository, base_oid),

         {:ok, blob} <- read_blob(repository, base_oid),

         {:ok, workflows} <- parse_workflows(blob.content) do

      {:ok, %{definition_oid: definition_oid, workflows: workflows}}

    end

  end

  @doc """

  Plan check runs for every active layer of `stack` against the current

  trunk tip.

  Marks runs whose effective base is no longer the trunk tip `stale`, then

  creates (or reuses, by identity) one run per layer and workflow that the

  declared `run_on` policy selects. Pass `context: "merge_group"` to plan

  the merge-group gate, where every workflow runs.

  Returns `{:ok, %{trunk_oid, definition_oid, runs, skipped, invalidated}}`

  where `skipped` lists each `(position, workflow)` the declared policy

  excluded — a skip is always explicit, never silent.

  """

  def plan(%Repository{} = repository, %Stack{} = stack, opts \\ []) do

    context = Keyword.get(opts, :context, "layer")

    with :ok <- check_context(context),

         {:ok, trunk_oid} <- resolve(repository, stack.trunk_ref),

         {:ok, %{definition_oid: definition_oid, workflows: workflows}} <-

           workflows(repository, trunk_oid) do

      entries = Stacks.active_entries(stack)

      invalidated = mark_stale(stack, trunk_oid)

      with {:ok, snapshots} <- snapshots(repository, entries, trunk_oid),

           {:ok, runs, skipped} <-

             materialize(repository, stack, entries, workflows, snapshots, %{

               context: context,

               trunk_oid: trunk_oid,

               definition_oid: definition_oid

             }) do

        {:ok,

         %{

           trunk_oid: trunk_oid,

           definition_oid: definition_oid,

           runs: runs,

           skipped: skipped,

           invalidated: invalidated

         }}

      end

    end

  end

  @doc """

  Mark every non-concluded run of `stack` whose effective base is not the

  current trunk tip as `stale`, so a trunk advance invalidates affected

  checks. Returns `{:ok, %{trunk_oid, invalidated}}`.

  """

  def refresh(%Repository{} = repository, %Stack{} = stack) do

    with {:ok, trunk_oid} <- resolve(repository, stack.trunk_ref) do

      {:ok, %{trunk_oid: trunk_oid, invalidated: mark_stale(stack, trunk_oid)}}

    end

  end

  @doc "Concludes a pending run as `passed` or `failed`."

  def report(%CheckRun{state: "pending"} = run, state) when state in ~w(passed failed) do

    Repo.update(CheckRun.conclusion_changeset(run, state, DateTime.utc_now()))

  end

  def report(%CheckRun{}, state) when state in ~w(passed failed),

    do: {:error, :not_pending}

  @doc "All runs of a stack, bottom layer first, newest identity last."

  def runs(%Stack{id: stack_id}) do

    Repo.all(

      from run in CheckRun,

        where: run.stack_id == ^stack_id,

        order_by: [asc: run.inserted_at, asc: run.id]

    )

  end

  ## Workflow definitions

  defp definition_oid(repository, base_oid) do

    path = Repos.bare_path(repository.storage_key)

    case Repos.git(path, ["rev-parse", "--verify", "--quiet", base_oid <> ":" <> @workflows_path]) do

      {output, 0} -> {:ok, String.trim(output)}

      _other -> {:error, :no_workflows}

    end

  end

  defp read_blob(repository, base_oid) do

    case Browse.blob(repository, base_oid, @workflows_path) do

      {:ok, blob} -> {:ok, blob}

      _other -> {:error, :no_workflows}

    end

  end

  defp parse_workflows(content) do

    case Jason.decode(content) do

      {:ok, %{"workflows" => workflows}} when is_list(workflows) ->

        parse_each(workflows, [])

      {:ok, _other} ->

        {:error, {:invalid_workflows, :missing_workflows_list}}

      {:error, _reason} ->

        {:error, {:invalid_workflows, :malformed_json}}

    end

  end

  defp parse_each([], parsed), do: {:ok, Enum.reverse(parsed)}

  defp parse_each([workflow | rest], parsed) do

    case parse_workflow(workflow) do

      {:ok, parsed_workflow} -> parse_each(rest, [parsed_workflow | parsed])

      {:error, reason} -> {:error, {:invalid_workflows, reason}}

    end

  end

  defp parse_workflow(%{"name" => name, "run_on" => run_on} = workflow)

       when is_binary(name) and name != "" do

    cond do

      run_on not in CheckRun.run_on_policies() ->

        {:error, {:unknown_run_on, name, run_on}}

      run_on == "changed_paths" and not valid_paths?(Map.get(workflow, "paths")) ->

        {:error, {:missing_paths, name}}

      not is_boolean(Map.get(workflow, "required", false)) ->

        {:error, {:invalid_required, name}}

      true ->

        {:ok,

         %{

           name: name,

           run_on: run_on,

           required: Map.get(workflow, "required", false),

           paths: Map.get(workflow, "paths", [])

         }}

    end

  end

  defp parse_workflow(_other), do: {:error, :invalid_workflow_entry}

  defp valid_paths?(paths) do

    is_list(paths) and paths != [] and Enum.all?(paths, &(is_binary(&1) and &1 != ""))

  end

  ## Run selection

  defp check_context(context) when context in ~w(layer merge_group), do: :ok

  defp check_context(_other), do: {:error, :invalid_context}

  # In the merge-group context every workflow runs: the merge group is the

  # authoritative gate. In the layer context the declared run_on policy

  # selects layers, and changed_paths consults the layer's own diff.

  defp selects?(_workflow, _entry, _top, "merge_group", _changed), do: true

  defp selects?(workflow, entry, top, "layer", changed) do

    case workflow.run_on do

      "every_layer" -> true

      "top_layer_only" -> entry.position == top

      "bottom_layer_only" -> entry.position == 1

      "changed_paths" -> Enum.any?(changed, &path_selected?(&1, workflow.paths))

      "merge_group_only" -> false

    end

  end

  # A declared path selects a changed file when it names the file exactly

  # or is a directory prefix.

  defp path_selected?(changed_path, declared_paths) do

    Enum.any?(declared_paths, fn declared ->

      prefix = String.trim_trailing(declared, "/")

      changed_path == prefix or String.starts_with?(changed_path, prefix <> "/")

    end)

  end

  defp changed_paths(repository, entry) do

    path = Repos.bare_path(repository.storage_key)

    args = [

      "diff",

      "--name-only",

      "--end-of-options",

      entry.boundary_oid,

      entry.observed_head_oid

    ]

    case Repos.git(path, args) do

      {output, 0} -> {:ok, String.split(output, "\n", trim: true)}

      _other -> {:error, {:diff_failed, entry.position}}

    end

  end

  ## Snapshots

  # Each snapshot is the cumulative repository state through a layer:

  # current trunk plus every layer up to and including this position. A

  # head that already sits on the chain (its boundary is the previous

  # snapshot) is its own snapshot; otherwise the layer replays onto the

  # chain and the result is published as a synthetic commit.

  defp snapshots(repository, entries, trunk_oid) do

    entries

    |> Enum.reduce_while({:ok, trunk_oid, %{}}, fn entry, {:ok, parent, acc} ->

      case snapshot_for(repository, entry, parent) do

        {:ok, snapshot} ->

          {:cont, {:ok, snapshot.tested_oid, Map.put(acc, entry.position, snapshot)}}

        {:error, reason} ->

          {:halt, {:error, reason}}

      end

    end)

    |> case do

      {:ok, _parent, snapshots} -> {:ok, snapshots}

      {:error, reason} -> {:error, reason}

    end

  end

  defp snapshot_for(_repository, %{boundary_oid: boundary} = entry, parent)

       when boundary == parent do

    {:ok, %{tested_oid: entry.observed_head_oid, synthetic: false}}

  end

  defp snapshot_for(repository, entry, parent) do

    case GitPlane.replay(

           repository.storage_key,

           entry.boundary_oid,

           entry.observed_head_oid,

           parent

         ) do

      {:ok, %{new_head: new_head}} ->

        {:ok, %{tested_oid: new_head, synthetic: true}}

      {:conflict, conflict} ->

        {:error, {:snapshot_conflict, entry.position, Map.get(conflict, :paths, [])}}

      {:error, reason} ->

        {:error, {:snapshot_failed, entry.position, reason}}

    end

  end

  ## Materialization

  defp materialize(repository, stack, entries, workflows, snapshots, plan) do

    top = entries |> Enum.map(& &1.position) |> Enum.max()

    entries

    |> Enum.reduce_while({:ok, [], []}, fn entry, {:ok, runs, skipped} ->

      case layer_runs(repository, stack, entry, workflows, snapshots, plan, top) do

        {:ok, layer_runs, layer_skipped} ->

          {:cont, {:ok, runs ++ layer_runs, skipped ++ layer_skipped}}

        {:error, reason} ->

          {:halt, {:error, reason}}

      end

    end)

    |> case do

      {:ok, runs, skipped} -> {:ok, runs, skipped}

      {:error, reason} -> {:error, reason}

    end

  end

  defp layer_runs(repository, stack, entry, workflows, snapshots, plan, top) do

    with {:ok, changed} <- layer_changed_paths(repository, entry, workflows, plan.context) do

      workflows

      |> Enum.reduce_while({:ok, [], []}, fn workflow, {:ok, runs, skipped} ->

        if selects?(workflow, entry, top, plan.context, changed) do

          case ensure_run(repository, stack, entry, workflow, snapshots, plan) do

            {:ok, run} -> {:cont, {:ok, [run | runs], skipped}}

            {:error, reason} -> {:halt, {:error, reason}}

          end

        else

          skip = %{position: entry.position, workflow: workflow.name, policy: workflow.run_on}

          {:cont, {:ok, runs, [skip | skipped]}}

        end

      end)

      |> case do

        {:ok, runs, skipped} -> {:ok, Enum.reverse(runs), Enum.reverse(skipped)}

        {:error, reason} -> {:error, reason}

      end

    end

  end

  defp layer_changed_paths(repository, entry, workflows, "layer") do

    if Enum.any?(workflows, &(&1.run_on == "changed_paths")) do

      changed_paths(repository, entry)

    else

      {:ok, []}

    end

  end

  defp layer_changed_paths(_repository, _entry, _workflows, _context), do: {:ok, []}

  defp ensure_run(repository, stack, entry, workflow, snapshots, plan) do

    snapshot = Map.fetch!(snapshots, entry.position)

    case existing_run(entry.pull_request_id, workflow.name, plan, entry.observed_head_oid) do

      %CheckRun{} = run -> {:ok, run}

      nil -> create_run(repository, stack, entry, workflow, snapshot, plan)

    end

  end

  defp existing_run(pull_request_id, workflow_name, plan, head_oid) do

    Repo.one(

      from run in CheckRun,

        where:

          run.pull_request_id == ^pull_request_id and

            run.workflow_name == ^workflow_name and

            run.context == ^plan.context and

            run.head_oid == ^head_oid and

            run.effective_base_oid == ^plan.trunk_oid and

            run.workflow_definition_oid == ^plan.definition_oid and

            run.state != "stale"

    )

  end

  defp create_run(repository, stack, entry, workflow, snapshot, plan) do

    run_id = Ecto.UUID.generate()

    synthetic_ref =

      if snapshot.synthetic do

        {:ok, ref} = GitPlane.internal_ref(["checks", run_id])

        ref

      end

    changeset =

      CheckRun.changeset(%CheckRun{}, %{

        id: run_id,

        repository_id: repository.id,

        stack_id: stack.id,

        pull_request_id: entry.pull_request_id,

        workflow_name: workflow.name,

        run_on: workflow.run_on,

        required: workflow.required,

        run_reason: "policy",

        context: plan.context,

        head_oid: entry.observed_head_oid,

        effective_base_oid: plan.trunk_oid,

        workflow_definition_oid: plan.definition_oid,

        tested_oid: snapshot.tested_oid,

        synthetic_ref: synthetic_ref

      })

    with {:ok, run} <- Repo.insert(changeset),

         :ok <- publish_synthetic_ref(repository, run) do

      {:ok, run}

    else

      {:error, %Ecto.Changeset{} = changeset} -> {:error, {:run_invalid, changeset}}

      {:error, reason} -> {:error, reason}

    end

  end

  # The synthetic ref publishes exactly once, from absent, and nothing ever

  # moves it: the snapshot a run tested is immutable for the run's lifetime.

  defp publish_synthetic_ref(_repository, %CheckRun{synthetic_ref: nil}), do: :ok

  defp publish_synthetic_ref(repository, %CheckRun{} = run) do

    updates = [%{ref: run.synthetic_ref, expected_old: :absent, new: run.tested_oid}]

    case GitPlane.batch_update_refs(repository.storage_key, updates, @principal) do

      {:ok, _result} -> :ok

      {:error, reason} -> {:error, {:synthetic_ref_failed, reason}}

    end

  end

  ## Invalidation

  defp mark_stale(stack, trunk_oid) do

    {count, _rows} =

      Repo.update_all(

        from(run in CheckRun,

          where:

            run.stack_id == ^stack.id and

              run.state in ["pending", "passed"] and

              run.effective_base_oid != ^trunk_oid

        ),

        set: [state: "stale", updated_at: DateTime.utc_now()]

      )

    count

  end

  defp resolve(repository, ref) do

    case Browse.resolve_commit(repository, ref) do

      {:ok, oid} -> {:ok, oid}

      _other -> {:error, {:unresolved_ref, ref}}

    end

  end

end

    20260823072000,

    20260823073000,

    20260823074000,

    20260823120247

    20260823120247,

    20260823122038

  ],

  "required_tables": [

    "users",

defmodule OpenAgents.Repo.Migrations.CreateStackCheckRuns do

  use Ecto.Migration

  def change do

    create table(:stack_check_runs, primary_key: false) do

      add :id, :binary_id, primary_key: true

      add :repository_id,

          references(:repositories, type: :binary_id, on_delete: :delete_all),

          null: false

      add :stack_id,

          references(:pull_request_stacks, type: :binary_id, on_delete: :delete_all),

          null: false

      add :pull_request_id,

          references(:pull_requests, type: :binary_id, on_delete: :delete_all),

          null: false

      add :workflow_name, :string, null: false

      add :run_on, :string, null: false

      add :required, :boolean, null: false, default: false

      add :run_reason, :string, null: false

      add :context, :string, null: false, default: "layer"

      add :head_oid, :binary, null: false

      add :effective_base_oid, :binary, null: false

      add :workflow_definition_oid, :binary, null: false

      add :tested_oid, :binary, null: false

      add :synthetic_ref, :string

      add :state, :string, null: false, default: "pending"

      add :concluded_at, :utc_datetime_usec

      timestamps(type: :utc_datetime_usec)

    end

    create unique_index(

             :stack_check_runs,

             [

               :pull_request_id,

               :workflow_name,

               :context,

               :head_oid,

               :effective_base_oid,

               :workflow_definition_oid

             ],

             name: :stack_check_runs_identity_index

           )

    create index(:stack_check_runs, [:stack_id, :state])

    create constraint(:stack_check_runs, :stack_check_runs_state_check,

             check: "state IN ('pending', 'passed', 'failed', 'stale')"

           )

    create constraint(:stack_check_runs, :stack_check_runs_context_check,

             check: "context IN ('layer', 'merge_group')"

           )

    create constraint(:stack_check_runs, :stack_check_runs_run_on_check,

             check:

               "run_on IN ('every_layer', 'top_layer_only', 'bottom_layer_only', 'changed_paths', 'merge_group_only')"

           )

  end

end

defmodule OpenAgents.Stacks.ChecksTest do

  @moduledoc """

  Stack-aware checks (#53): full-context run identity, trunk-advance

  invalidation, immutable synthetic snapshot refs whose trees layer the

  current trunk plus every layer through the checked pull request, and

  every declared `run_on` policy.

  """

  use OpenAgents.DataCase, async: false

  alias OpenAgents.Forge.Repos

  alias OpenAgents.PullRequests.PullRequest

  alias OpenAgents.Repo

  alias OpenAgents.Stacks

  alias OpenAgents.Stacks.CheckRun

  alias OpenAgents.Stacks.Checks

  import OpenAgents.AccountsFixtures

  import OpenAgents.IssuesFixtures

  setup do

    base = Path.join(System.tmp_dir!(), "stack-checks-#{System.unique_integer([:positive])}")

    previous_data = Application.get_env(:openagents, :forge_data_dir)

    previous_wal = Application.get_env(:openagents, :forge_wal_dir)

    Application.put_env(:openagents, :forge_data_dir, Path.join(base, "data"))

    Application.put_env(:openagents, :forge_wal_dir, Path.join(base, "wal"))

    on_exit(fn ->

      restore_env(:forge_data_dir, previous_data)

      restore_env(:forge_wal_dir, previous_wal)

      File.rm_rf(base)

    end)

    actor = repository_user_fixture("checks-actor")

    repository = repository_with_member_fixture(actor)

    %{actor: actor, repository: repository}

  end

  describe "workflow definitions" do

    test "an absent definitions file returns no_workflows", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack} = seed_stack(repository, actor, workflows: nil)

      assert {:error, :no_workflows} = Checks.plan(repository, stack)

    end

    test "a malformed definitions file is rejected", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack} = seed_stack(repository, actor, workflows: "not json")

      assert {:error, {:invalid_workflows, :malformed_json}} = Checks.plan(repository, stack)

    end

    test "an unknown run_on policy is rejected", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack} =

        seed_stack(repository, actor,

          workflows: workflows_json([%{"name" => "unit", "run_on" => "sometimes"}])

        )

      assert {:error, {:invalid_workflows, {:unknown_run_on, "unit", "sometimes"}}} =

               Checks.plan(repository, stack)

    end

  end

  describe "run identity" do

    test "runs key to head, effective base, and workflow definition", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack, oids: oids, pull_requests: [pr_1, pr_2]} =

        seed_stack(repository, actor, workflows: every_layer_workflows())

      {:ok, plan} = Checks.plan(repository, stack)

      assert plan.invalidated == 0

      assert plan.skipped == []

      assert length(plan.runs) == 2

      run_1 = Enum.find(plan.runs, &(&1.pull_request_id == pr_1.id))

      run_2 = Enum.find(plan.runs, &(&1.pull_request_id == pr_2.id))

      assert run_1.head_oid == oids["layer-1"]

      assert run_2.head_oid == oids["layer-2"]

      assert run_1.effective_base_oid == plan.trunk_oid

      assert run_2.effective_base_oid == plan.trunk_oid

      assert run_1.workflow_definition_oid == plan.definition_oid

      assert run_1.state == "pending"

      # A healthy stack's heads already contain every lower layer, so the

      # heads are their own snapshots and no synthetic ref publishes.

      assert run_1.tested_oid == oids["layer-1"]

      assert run_2.tested_oid == oids["layer-2"]

      assert is_nil(run_1.synthetic_ref)

      assert is_nil(run_2.synthetic_ref)

    end

    test "planning twice reuses runs instead of duplicating identity", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack} = seed_stack(repository, actor, workflows: every_layer_workflows())

      {:ok, first} = Checks.plan(repository, stack)

      {:ok, second} = Checks.plan(repository, stack)

      assert Enum.map(first.runs, & &1.id) |> Enum.sort() ==

               Enum.map(second.runs, & &1.id) |> Enum.sort()

      assert Repo.aggregate(CheckRun, :count) == 2

    end

    test "a trunk advance invalidates existing runs and re-keys new ones", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack, path: path, oids: oids} =

        seed_stack(repository, actor, workflows: every_layer_workflows())

      {:ok, before_advance} = Checks.plan(repository, stack)

      old_trunk = before_advance.trunk_oid

      advance = commit(path, oids["main"], "Advance trunk", %{"advance.md" => "advance\n"})

      {_, 0} = Repos.git(path, ["update-ref", "refs/heads/main", advance])

      {:ok, after_advance} = Checks.plan(repository, stack)

      assert after_advance.trunk_oid == advance

      assert after_advance.invalidated == 2

      for run <- before_advance.runs do

        assert Repo.get!(CheckRun, run.id).state == "stale"

      end

      for run <- after_advance.runs do

        assert run.effective_base_oid == advance

        assert run.effective_base_oid != old_trunk

        assert run.state == "pending"

      end

    end

    test "refresh alone marks affected runs stale after a trunk advance", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack, path: path, oids: oids} =

        seed_stack(repository, actor, workflows: every_layer_workflows())

      {:ok, plan} = Checks.plan(repository, stack)

      {:ok, run} = Checks.report(hd(plan.runs), "passed")

      assert run.state == "passed"

      advance = commit(path, oids["main"], "Advance trunk", %{"advance.md" => "advance\n"})

      {_, 0} = Repos.git(path, ["update-ref", "refs/heads/main", advance])

      {:ok, %{trunk_oid: ^advance, invalidated: 2}} = Checks.refresh(repository, stack)

      assert Repo.get!(CheckRun, run.id).state == "stale"

    end

  end

  describe "synthetic snapshot refs" do

    test "a stack behind trunk tests a synthetic snapshot of trunk plus its layers",

         context do

      %{repository: repository, actor: actor} = context

      %{stack: stack, path: path, oids: oids, pull_requests: [pr_1, pr_2]} =

        seed_stack(repository, actor, workflows: every_layer_workflows())

      advance = commit(path, oids["main"], "Advance trunk", %{"advance.md" => "advance\n"})

      {_, 0} = Repos.git(path, ["update-ref", "refs/heads/main", advance])

      {:ok, plan} = Checks.plan(repository, stack)

      run_1 = Enum.find(plan.runs, &(&1.pull_request_id == pr_1.id))

      run_2 = Enum.find(plan.runs, &(&1.pull_request_id == pr_2.id))

      for run <- [run_1, run_2] do

        assert run.synthetic_ref == "refs/internal/checks/" <> run.id

        assert run.tested_oid != run.head_oid

        assert show(path, ["rev-parse", run.synthetic_ref]) == run.tested_oid

      end

      # The snapshot tree is the current trunk plus every layer through the

      # checked pull request.

      names_1 = tree_names(path, run_1.tested_oid)

      assert "advance.md" in names_1

      assert "layer-1.md" in names_1

      refute "layer-2.md" in names_1

      names_2 = tree_names(path, run_2.tested_oid)

      assert "advance.md" in names_2

      assert "layer-1.md" in names_2

      assert "layer-2.md" in names_2

    end

    test "a synthetic ref publishes once and never moves", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack, path: path, oids: oids} =

        seed_stack(repository, actor, workflows: every_layer_workflows())

      advance = commit(path, oids["main"], "Advance trunk", %{"advance.md" => "advance\n"})

      {_, 0} = Repos.git(path, ["update-ref", "refs/heads/main", advance])

      {:ok, first} = Checks.plan(repository, stack)

      {:ok, second} = Checks.plan(repository, stack)

      assert Enum.map(first.runs, & &1.id) |> Enum.sort() ==

               Enum.map(second.runs, & &1.id) |> Enum.sort()

      for run <- second.runs do

        assert show(path, ["rev-parse", run.synthetic_ref]) == run.tested_oid

        assert Repo.get!(CheckRun, run.id).tested_oid == run.tested_oid

      end

    end

  end

  describe "run_on policies" do

    test "top_layer_only plans only the top layer, with explicit skips", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack, pull_requests: [_pr_1, pr_2]} =

        seed_stack(repository, actor,

          workflows: workflows_json([%{"name" => "e2e", "run_on" => "top_layer_only"}])

        )

      {:ok, plan} = Checks.plan(repository, stack)

      assert [run] = plan.runs

      assert run.pull_request_id == pr_2.id

      assert plan.skipped == [%{position: 1, workflow: "e2e", policy: "top_layer_only"}]

    end

    test "bottom_layer_only plans only the bottom layer", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack, pull_requests: [pr_1, _pr_2]} =

        seed_stack(repository, actor,

          workflows: workflows_json([%{"name" => "smoke", "run_on" => "bottom_layer_only"}])

        )

      {:ok, plan} = Checks.plan(repository, stack)

      assert [run] = plan.runs

      assert run.pull_request_id == pr_1.id

      assert plan.skipped == [%{position: 2, workflow: "smoke", policy: "bottom_layer_only"}]

    end

    test "changed_paths plans only layers that touch the declared paths", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack, pull_requests: [pr_1, _pr_2]} =

        seed_stack(repository, actor,

          workflows:

            workflows_json([

              %{

                "name" => "layer-1-only",

                "run_on" => "changed_paths",

                "paths" => ["layer-1.md"]

              }

            ])

        )

      {:ok, plan} = Checks.plan(repository, stack)

      assert [run] = plan.runs

      assert run.pull_request_id == pr_1.id

      assert plan.skipped == [

               %{position: 2, workflow: "layer-1-only", policy: "changed_paths"}

             ]

    end

    test "merge_group_only skips every layer outside a merge group", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack} =

        seed_stack(repository, actor,

          workflows:

            workflows_json([

              %{"name" => "queue-gate", "run_on" => "merge_group_only", "required" => true}

            ])

        )

      {:ok, plan} = Checks.plan(repository, stack)

      assert plan.runs == []

      assert plan.skipped == [

               %{position: 1, workflow: "queue-gate", policy: "merge_group_only"},

               %{position: 2, workflow: "queue-gate", policy: "merge_group_only"}

             ]

    end

    test "the merge_group context runs every workflow: required checks never skip",

         context do

      %{repository: repository, actor: actor} = context

      %{stack: stack} =

        seed_stack(repository, actor,

          workflows:

            workflows_json([

              %{"name" => "queue-gate", "run_on" => "merge_group_only", "required" => true},

              %{"name" => "e2e", "run_on" => "top_layer_only", "required" => true}

            ])

        )

      {:ok, plan} = Checks.plan(repository, stack, context: "merge_group")

      assert plan.skipped == []

      assert length(plan.runs) == 4

      assert Enum.all?(plan.runs, &(&1.context == "merge_group"))

      # Layer-context runs and merge-group runs have distinct identities.

      {:ok, layer_plan} = Checks.plan(repository, stack)

      assert [%{workflow_name: "e2e", context: "layer"}] = layer_plan.runs

    end

  end

  describe "reporting" do

    test "a pending run concludes once", context do

      %{repository: repository, actor: actor} = context

      %{stack: stack} = seed_stack(repository, actor, workflows: every_layer_workflows())

      {:ok, plan} = Checks.plan(repository, stack)

      [run | _rest] = plan.runs

      {:ok, passed} = Checks.report(run, "passed")

      assert passed.state == "passed"

      refute is_nil(passed.concluded_at)

      assert {:error, :not_pending} = Checks.report(passed, "failed")

    end

  end

  ## Seeding