@timestamps_opts [type: :utc_datetime_usec]

  @results ~w(live reverted needs_rolling_replace failed)

  @deployment_types ~w(direct_load relup rolling_replacement)

  schema "forge_deploys" do

    field :repo, :string

    field :expected_nodes, {:array, :string}, default: []

    field :node_results, :map, default: %{}

    field :result, :string

    field :deployment_type, :string

    field :canary, :string

    field :error_code, :string

    field :rollback_verified, :boolean

  @doc "All deploy results."

  def results, do: @results

  @doc "All deployment types."

  def deployment_types, do: @deployment_types

  def changeset(receipt, attrs) do

    now = DateTime.utc_now()

      :expected_nodes,

      :node_results,

      :result,

      :deployment_type,

      :canary,

      :error_code,

      :rollback_verified,

      :completed_at

    ])

    |> validate_inclusion(:result, @results)

    |> validate_inclusion(:deployment_type, @deployment_types)

    |> validate_format(:sha, ~r/^[0-9a-f]{40}$/)

    |> validate_format(:artifact_digest, ~r/^[0-9a-f]{64}$/)

    |> validate_format(:manifest_digest, ~r/^[0-9a-f]{64}$/)

    |> validate_node_results()

    |> unique_constraint(:deployment_id)

    |> check_constraint(:result, name: :forge_deploys_result)

    |> check_constraint(:deployment_type, name: :forge_deploys_deployment_type)

    |> check_constraint(:artifact_digest, name: :forge_deploys_artifact_digest)

    |> check_constraint(:manifest_digest, name: :forge_deploys_manifest_digest)

    |> check_constraint(:node_results, name: :forge_deploys_node_bounds)

      message = "hot_load_failed code=" <> OpenAgents.OperationalLog.code(error)

      Logger.error("forge_hot_load_failed code=#{OpenAgents.OperationalLog.code(error)}")

      advance(target_id, "failed", %{"error" => message})

      insert_receipt(repo, sha, target_id, modules, [], "failed", nil, nil)

      insert_receipt(repo, sha, target_id, modules, [], "failed", "direct_load", nil, nil)

      broadcast_deploy(repo, sha, "failed")

  catch

    :refused -> :ok

           "reasons" => reasons

         }) do

      :ok ->

        insert_receipt(repo, sha, target_id, modules, [], "needs_rolling_replace", nil, nil)

        # Classification only — no deployment ran, so no deployment_type.

        insert_receipt(repo, sha, target_id, modules, [], "needs_rolling_replace", nil, nil, nil)

        broadcast_deploy(repo, sha, "needs_rolling_replace")

      :error ->

    message = "artifact_verification_failed code=" <> OpenAgents.OperationalLog.code(reason)

    Logger.error(message)

    advance(target_id, "failed", %{"error" => message})

    insert_receipt(repo, sha, target_id, modules, [], "failed", nil, nil)

    insert_receipt(repo, sha, target_id, modules, [], "failed", "direct_load", nil, nil)

    broadcast_deploy(repo, sha, "failed")

  end

    end

  end

  defp insert_receipt(repo, sha, target_id, modules, nodes, result, canary, push_ms) do

  defp insert_receipt(

         repo,

         sha,

         target_id,

         modules,

         nodes,

         result,

         deployment_type,

         canary,

         push_ms

       ) do

    %DeployReceipt{}

    |> DeployReceipt.changeset(%{

      repo: repo,

      modules: modules,

      nodes: nodes,

      result: result,

      deployment_type: deployment_type,

      canary: canary,

      push_to_live_ms: push_ms

    })

      nodes: outcome.nodes,

      expected_nodes: outcome.expected_nodes,

      node_results: outcome.node_results,

      deployment_type: "direct_load",

      canary: canary || outcome.canary,

      error_code: outcome.error_code,

      rollback_verified: outcome.rollback_verified,

  @doc "Roll an immutable image across the exact expected node set."

  def run(request, opts \\ []) do

    started = System.monotonic_time(:millisecond)

    with :ok <- validate_request(request),

         {:ok, _receipt} <- gate_verify(request.sha, opts),

         {:ok, provider} <- provider(opts),

         :ok <- initial_membership(request, provider, opts) do

      replace_nodes(request.expected_nodes, request, provider, opts, %{})

      request.expected_nodes

      |> replace_nodes(request, provider, opts, %{})

      |> put_duration(started)

    end

  end

  defp put_duration({verdict, result}, started) when is_map(result) do

    {verdict, Map.put(result, :duration_ms, System.monotonic_time(:millisecond) - started)}

  end

  defp replace_nodes([], request, _provider, _opts, results) do

    {:ok, public_result(request, "live", results, nil, nil)}

  end

            artifact_digest: relup.artifact_digest,

            completed_at: now,

            deployment_id: deployment_id,

            deployment_type: "relup",

            error_code: relup.error_code,

            expected_nodes: relup.expected_nodes,

            manifest_digest: relup.package_manifest_digest,

            artifact_digest: build.artifact_digest,

            completed_at: now,

            deployment_id: deployment_id,

            deployment_type: "rolling_replacement",

            error_code: rolling.error_code,

            expected_nodes: rolling.expected_nodes,

            manifest_digest: manifest_digest,

            rollback_verified:

              rolling.status == "failed" and rolling.recovery == "last_known_good_restored",

            sha: target.sha,

            started_at: now,

            started_at:

              if(is_integer(rolling.duration_ms),

                do: DateTime.add(now, -rolling.duration_ms, :millisecond),

                else: now

              ),

            target_id: target.id

          }

    node_results = result_value(result, :node_results)

    error_code = result_value(result, :error_code)

    recovery = result_value(result, :recovery)

    duration_ms = result_value(result, :duration_ms)

    cond do

      schema != "openagents.rolling-replacement.v1" ->

      not valid_node_results?(node_results) ->

        {:error, :invalid_rolling_result}

      not (is_nil(duration_ms) or valid_duration?(duration_ms)) ->

        {:error, :invalid_rolling_result}

      status == "live" and

          (Enum.any?(node_results, fn {_node, node_status} -> node_status != "ready" end) or

             not is_nil(error_code) or not is_nil(recovery)) ->

      true ->

        {:ok,

         %{

           duration_ms: duration_ms,

           error_code: error_code,

           expected_nodes: node_results |> Map.keys() |> Enum.sort(),

           image_digest: image_digest,

    }

  end

  # Deployment history stays content-free: short sha, result, lane, module

  # count, and timings only. `duration_ms` measures deployment start through

  # completion; classification-only receipts (needs_rolling_replace) ran no

  # deployment, so they carry no type and no duration.

  defp public_deploy(deploy) do

    %{

      "sha" => short_sha(deploy.sha),

      "result" => deploy.result,

      "type" => deploy.deployment_type,

      "modules" => length(deploy.modules),

      "push_to_live_ms" => deploy.push_to_live_ms,

      "duration_ms" => deploy_duration_ms(deploy),

      "completed_at" => iso8601_or_nil(deploy.completed_at),

      "at" => DateTime.to_iso8601(deploy.inserted_at)

    }

  end

  defp deploy_duration_ms(%{result: "needs_rolling_replace"}), do: nil

  defp deploy_duration_ms(%{started_at: %DateTime{} = started, completed_at: %DateTime{} = done}),

    do: DateTime.diff(done, started, :millisecond)

  defp deploy_duration_ms(_deploy), do: nil

  defp iso8601_or_nil(%DateTime{} = at), do: DateTime.to_iso8601(at)

  defp iso8601_or_nil(_at), do: nil

  defp short_sha(sha) when is_binary(sha), do: String.slice(sha, 0, 12)

  defp short_sha(_sha), do: nil

  defp deploy_result_variant("needs_rolling_replace"), do: :warning

  defp deploy_result_variant(_result), do: :danger

  # Classification-only receipts ran no deployment; a receipt without a

  # recorded lane (rows older than the column) renders an honest "unknown".

  defp deploy_type_text(%{"result" => "needs_rolling_replace"}), do: "classification only"

  defp deploy_type_text(%{"type" => "direct_load"}), do: "direct load"

  defp deploy_type_text(%{"type" => "relup"}), do: "relup"

  defp deploy_type_text(%{"type" => "rolling_replacement"}), do: "rolling replacement"

  defp deploy_type_text(_deploy), do: "unknown lane"

  defp completed_text(iso) when is_binary(iso) do

    case DateTime.from_iso8601(iso) do

      {:ok, at, _offset} -> Calendar.strftime(at, "%Y-%m-%d %H:%M:%S UTC")

      _error -> "—"

    end

  end

  defp completed_text(_iso), do: "—"

  defp overall_badge_variant("ok"), do: :success

  defp overall_badge_variant("rolling"), do: :info

  defp overall_badge_variant("recovering"), do: :warning

              deploy policy: direct hot load → relup → rolling replacement

            </p>

            <div :if={@projection["forge"]["recent_deploys"] != []} class="status-forge__history">

              <h3>Recent deploys</h3>

              <ul class="status-events">

                <li :for={deploy <- @projection["forge"]["recent_deploys"]}>

            <div id="status-forge-history" class="status-forge__history">

              <h3>Recent deployments</h3>

              <.empty

                :if={@projection["forge"]["recent_deploys"] == []}

                id="status-forge-no-deploys"

                title="No deployment receipts yet"

              >

                Deployment receipts appear here newest first as Forge records them.

              </.empty>

              <ul :if={@projection["forge"]["recent_deploys"] != []} class="status-events">

                <li

                  :for={deploy <- @projection["forge"]["recent_deploys"]}

                  data-deploy-type={deploy["type"] || "none"}

                >

                  <span class="status-events__stamp"><code>{deploy["sha"]}</code></span>

                  <.badge variant={deploy_result_variant(deploy["result"])}>

                    {deploy["result"]}

                  </.badge>

                  {deploy["modules"]} module{if deploy["modules"] == 1, do: "", else: "s"} · push→live {ms_text(

                    deploy["push_to_live_ms"]

                  {deploy_type_text(deploy)} · {deploy["modules"]} module{if deploy["modules"] == 1,

                    do: "",

                    else: "s"} · completed {completed_text(deploy["completed_at"])} · took {ms_text(

                    deploy["duration_ms"]

                  )}

                </li>

              </ul>

    20260823021021,

    20260823034851,

    20260823040635,

    20260823042207,

    20260823043000

  ],

  "required_tables": [

defmodule OpenAgents.Repo.Migrations.AddForgeDeploysDeploymentType do

  use Ecto.Migration

  # Nullable on purpose: receipts are immutable (no backfill), so rows written

  # before this column render an honest "unknown" type rather than a guess.

  def change do

    alter table(:forge_deploys) do

      add :deployment_type, :string

    end

    create constraint(:forge_deploys, :forge_deploys_deployment_type,

             check:

               "deployment_type IS NULL OR deployment_type IN ('direct_load', 'relup', 'rolling_replacement')"

           )

  end

end

      assert NetworkStatus.projection(refresh: true)["forge"]["state"] == "off"

    end

    test "recent deployments expose lane, completion time, and duration, newest first" do

      sha = String.duplicate("c", 40)

      target =

        %Target{}

        |> Target.changeset(%{

          repo: "openagents.com",

          sha: sha,

          promoted_by: "operator:1",

          status: "live"

        })

        |> Repo.insert!()

      now = DateTime.utc_now()

      rows = [

        # Oldest: a pre-column legacy row (no lane recorded).

        %{result: "live", started_at: DateTime.add(now, -100, :millisecond), completed_at: now},

        # A classification-only receipt — no deployment ran.

        %{result: "needs_rolling_replace"},

        %{

          result: "live",

          deployment_type: "direct_load",

          started_at: DateTime.add(now, -1_500, :millisecond),

          completed_at: now

        },

        %{

          result: "failed",

          deployment_type: "relup",

          started_at: DateTime.add(now, -2_000, :millisecond),

          completed_at: now

        },

        # Newest: a rolling replacement.

        %{

          result: "live",

          deployment_type: "rolling_replacement",

          started_at: DateTime.add(now, -60_000, :millisecond),

          completed_at: now

        }

      ]

      for row <- rows do

        %DeployReceipt{}

        |> DeployReceipt.changeset(

          Map.merge(%{repo: "openagents.com", sha: sha, target_id: target.id}, row)

        )

        |> Repo.insert!()

        # Distinct inserted_at ordering under usec timestamps.

        Process.sleep(2)

      end

      deploys = NetworkStatus.projection(refresh: true)["forge"]["recent_deploys"]

      assert [rolling, relup, direct, classification, legacy] = deploys

      assert %{"type" => "rolling_replacement", "result" => "live", "duration_ms" => 60_000} =

               rolling

      assert %{"type" => "relup", "result" => "failed", "duration_ms" => 2_000} = relup

      assert %{"type" => "direct_load", "result" => "live", "duration_ms" => 1_500} = direct

      # Classification-only receipts stay distinguishable: no lane, no

      # duration, and their own result.

      assert %{"type" => nil, "result" => "needs_rolling_replace", "duration_ms" => nil} =

               classification

      # Rows older than the lane column degrade honestly to a nil type while

      # keeping their measured duration.

      assert %{"type" => nil, "result" => "live", "duration_ms" => 100} = legacy

      for deploy <- deploys do

        assert {:ok, _at, _offset} = DateTime.from_iso8601(deploy["completed_at"])

      end

    end

    test "loop metrics: last is newest, median over live deploys only" do

      sha = String.duplicate("b", 40)

      assert html =~ "Rapid deploys"

      assert has_element?(view, "#status-forge-state")

      assert html =~ "No deploys yet"

      assert has_element?(view, "#status-forge-no-deploys")

      assert html =~ "No deployment receipts yet"

    end

    test "renders the pipeline position for an advancing target", %{conn: conn} do

      assert render(view) =~ "hot deploy #{String.duplicate("c", 12)}: live"

    end

    test "renders deployment history with lane, completion time, and duration", %{conn: conn} do

      target = seed_target("live")

      now = DateTime.utc_now()

      receipts = [

        # Oldest: classification only.

        %{result: "needs_rolling_replace"},

        %{

          result: "live",

          deployment_type: "direct_load",

          started_at: DateTime.add(now, -13_242, :millisecond),

          completed_at: now

        },

        # Newest: a settled relup.

        %{

          result: "live",

          deployment_type: "relup",

          started_at: DateTime.add(now, -2_500, :millisecond),

          completed_at: now

        }

      ]

      for receipt <- receipts do

        %DeployReceipt{}

        |> DeployReceipt.changeset(

          Map.merge(%{repo: "openagents.com", sha: target.sha, target_id: target.id}, receipt)

        )

        |> Repo.insert!()

        # Distinct inserted_at ordering under usec timestamps.

        Process.sleep(2)

      end

      {:ok, view, html} = live(conn, ~p"/status")

      assert html =~ "Recent deployments"

      assert has_element?(view, "#status-forge-history li[data-deploy-type='direct_load']")

      assert has_element?(view, "#status-forge-history li[data-deploy-type='relup']")

      assert has_element?(view, "#status-forge-history li[data-deploy-type='none']")

      assert html =~ "direct load"

      assert html =~ "classification only"

      assert html =~ "took 13.2s"

      assert html =~ "took 2.5s"

      assert html =~ "completed #{Calendar.strftime(now, "%Y-%m-%d")}"

      # Newest first: the relup receipt renders before the direct load.

      {relup_at, _} = :binary.match(html, ~s(data-deploy-type="relup"))

      {direct_at, _} = :binary.match(html, ~s(data-deploy-type="direct_load"))

      assert relup_at < direct_at

    end

    test "the deployment list updates when Forge records a new receipt", %{conn: conn} do

      target = seed_target("live", sha: String.duplicate("d", 40))

      {:ok, view, html} = live(conn, ~p"/status")

      refute html =~ ~s(data-deploy-type="direct_load")

      %DeployReceipt{}

      |> DeployReceipt.changeset(%{

        repo: "openagents.com",

        sha: target.sha,

        target_id: target.id,

        result: "live",

        deployment_type: "direct_load"

      })

      |> Repo.insert!()

      Phoenix.PubSub.broadcast(

        OpenAgents.PubSub,

        "forge:deploys",

        {:forge_deploy, %{repo: "openagents.com", sha: target.sha, result: "live"}}

      )

      html = render(view)

      assert html =~ ~s(data-deploy-type="direct_load")

      assert html =~ "direct load"

    end

    test "the page is content-free: no operator identity, no module names", %{conn: conn} do

      target = seed_target("live", promoted_by: "operator:55554444")