OpenAgents

Commit

Restack stacks with cascading server-side rebases

db91a8581718 · Devin AI · · parent c0a0451384dd 

Restack stacks with cascading server-side rebases

One operation rebases a stack from the trunk upward: durable
stack_operations rows carry the request, snapshot, planned result,
and conflict workspace; the waterfall verifies each live branch head,
replays only commits past the stored boundary onto the new parent,
builds replacement commits under hidden internal refs, and moves every
public branch through one batch-CAS ref transaction. Conflicts pause
the operation with a durable workspace that continue resumes from
(re-verifying branch heads) and abort rolls back. Completed operations
emit pull_request_stack.rebased and pull_request.synchronize events.

Server restacks create unsigned commits that preserve the original
author identity; signature-mandatory flows keep rebasing locally
(docs/stacked-prs.md section 12.5).

Closes #50

Co-Authored-By: Christopher David <chris@openagents.com>
Co-Authored-By
Christopher David <chris@openagents.com>
Closes
#50

Deploy story

What this commit did to the running system — joined from the forge receipt chain, the part a commit page elsewhere cannot show.

Not deployed through the forge lane

No push, promotion, build, or deploy receipt references this commit (receipts are scanned over a bounded recent window). Changes shipped by full node replacement carry their proof in the release gate receipt instead.

Changed files

  • modified lib/openagents/forge/git_plane.ex
  • modified lib/openagents/forge/supervisor.ex
  • added lib/openagents/stacks/operation.ex
  • added lib/openagents/stacks/operation_worker.ex
  • added lib/openagents/stacks/restack.ex
  • modified lib/openagents/stacks/stack_event.ex
  • modified lib/openagents_web/api_route_authority.ex
  • modified lib/openagents_web/controllers/stack_controller.ex
  • modified lib/openagents_web/controllers/stack_json.ex
  • modified lib/openagents_web/router.ex
  • added priv/repo/migrations/20260823072000_create_stack_operations.exs
  • added test/openagents/stacks/restack_test.exs
  • modified test/openagents_web/controllers/stack_controller_test.exs

Diff

13 files changed, +1974 -6

lib/openagents/forge/git_plane.ex modified +11

@@ -56,6 +56,17 @@ defmodule OpenAgents.Forge.GitPlane do

56 56 
    end
57 57 
  end
58 58
59 
  @doc "The parent commit OIDs of a commit, in order (empty for a root commit)."
60 
  def parents(repo, rev) do
61 
    with :ok <- check_rev(rev),
62 
         :ok <- Sync.ensure_fresh(repo) do
63 
      case git(repo, ["show", "-s", "--format=%P", "--end-of-options", rev]) do
64 
        {output, 0} -> {:ok, output |> String.split() |> Enum.filter(&(&1 != ""))}
65 
        _other -> {:error, :not_found}
66 
      end
67 
    end
68 
  end
69
59 70 
  @doc "Whether `ancestor` is an ancestor of (or equal to) `descendant`."
60 71 
  def ancestor?(repo, ancestor, descendant) do
61 72 
    with :ok <- check_rev(ancestor),
lib/openagents/forge/supervisor.ex modified +2 -1

@@ -22,7 +22,8 @@ defmodule OpenAgents.Forge.Supervisor do

22 22 
    if Application.get_env(:openagents, :repository_provisioner_enabled, true) do
23 23 
      [
24 24 
        {OpenAgents.Repositories.Provisioner, []},
25 
        {OpenAgents.Repositories.ImportWorkspaceJanitor, []}
25 
        {OpenAgents.Repositories.ImportWorkspaceJanitor, []},
26 
        {OpenAgents.Stacks.OperationWorker, []}
26 27 
      ]
27 28 
    else
28 29 
      []
lib/openagents/stacks/operation.ex added +94

@@ -0,0 +1,94 @@

1 
defmodule OpenAgents.Stacks.Operation do
2 
  @moduledoc """
3 
  One durable server-side stack operation.
4
5 
  The row is the recovery record: it carries the request, the snapshot the
6 
  worker took before touching anything, the planned result, and — for a
7 
  paused rebase — the conflict workspace. A worker crash leaves the row
8 
  claimable again after its lease expires, so the operation resumes instead
9 
  of vanishing.
10 
  """
11 
  use Ecto.Schema
12 
  import Ecto.Changeset
13
14 
  @primary_key {:id, :binary_id, autogenerate: true}
15 
  @foreign_key_type :binary_id
16
17 
  @kinds ~w(create append restructure rebase merge queue unstack dissolve repair)
18 
  @states ~w(pending running waiting_for_conflict_resolution waiting_for_checks succeeded partially_succeeded failed cancelled)
19 
  @active_states ~w(pending running waiting_for_conflict_resolution)
20
21 
  schema "stack_operations" do
22 
    belongs_to :stack, OpenAgents.Stacks.Stack
23 
    field :kind, :string
24 
    field :state, :string, default: "pending"
25 
    field :target_position, :integer
26 
    field :expected_stack_version, :integer
27 
    field :idempotency_key, :string
28
29 
    field :request, :map, default: %{}
30 
    field :snapshot, :map
31 
    field :planned_result, :map
32 
    field :conflict, :map
33 
    field :error, :map
34
35 
    belongs_to :created_by_user, OpenAgents.Accounts.User
36
37 
    field :attempt_count, :integer, default: 0
38 
    field :retry_at, :utc_datetime_usec
39 
    field :claimed_at, :utc_datetime_usec
40 
    field :started_at, :utc_datetime_usec
41 
    field :completed_at, :utc_datetime_usec
42
43 
    timestamps(type: :utc_datetime_usec)
44 
  end
45
46 
  def kinds, do: @kinds
47 
  def states, do: @states
48 
  def active_states, do: @active_states
49
50 
  def changeset(operation, attrs) do
51 
    operation
52 
    |> cast(attrs, [
53 
      :kind,
54 
      :state,
55 
      :target_position,
56 
      :expected_stack_version,
57 
      :idempotency_key,
58 
      :request,
59 
      :retry_at
60 
    ])
61 
    |> put_change(:stack_id, Map.fetch!(attrs, :stack_id))
62 
    |> put_change(:created_by_user_id, Map.get(attrs, :created_by_user_id))
63 
    |> validate_required([:stack_id, :kind, :state, :expected_stack_version, :idempotency_key])
64 
    |> validate_inclusion(:kind, @kinds)
65 
    |> validate_inclusion(:state, @states)
66 
    |> validate_number(:expected_stack_version, greater_than_or_equal_to: 1)
67 
    |> validate_length(:idempotency_key, min: 1, max: 255)
68 
    |> unique_constraint([:stack_id, :idempotency_key])
69 
    |> unique_constraint(:stack_id, name: :stack_operations_active_stack_index)
70 
    |> check_constraint(:kind, name: :stack_operations_kind_check)
71 
    |> check_constraint(:state, name: :stack_operations_state_check)
72 
    |> foreign_key_constraint(:stack_id)
73 
    |> foreign_key_constraint(:created_by_user_id)
74 
  end
75
76 
  def transition_changeset(operation, attrs) do
77 
    operation
78 
    |> cast(attrs, [
79 
      :state,
80 
      :snapshot,
81 
      :planned_result,
82 
      :conflict,
83 
      :error,
84 
      :attempt_count,
85 
      :retry_at,
86 
      :claimed_at,
87 
      :started_at,
88 
      :completed_at
89 
    ])
90 
    |> validate_required([:state])
91 
    |> validate_inclusion(:state, @states)
92 
    |> check_constraint(:state, name: :stack_operations_state_check)
93 
  end
94 
end
lib/openagents/stacks/operation_worker.ex added +144

@@ -0,0 +1,144 @@

1 
defmodule OpenAgents.Stacks.OperationWorker do
2 
  @moduledoc """
3 
  Claims and executes durable stack operations.
4
5 
  The worker polls `stack_operations` for pending rows (and running rows
6 
  whose lease expired, which recovers a crashed worker) and executes them
7 
  through `OpenAgents.Stacks.Restack`. Claiming uses `FOR UPDATE SKIP
8 
  LOCKED`, so multiple nodes never execute the same operation twice inside
9 
  one lease window.
10 
  """
11 
  use GenServer
12
13 
  import Ecto.Query
14
15 
  require Logger
16
17 
  alias OpenAgents.OperationalLog
18 
  alias OpenAgents.Repo
19 
  alias OpenAgents.Stacks.Operation
20 
  alias OpenAgents.Stacks.Restack
21
22 
  @lease_seconds 120
23 
  @maximum_drain 100
24
25 
  def start_link(options) do
26 
    name = Keyword.get(options, :name, __MODULE__)
27 
    gen_server_options = if name, do: [name: name], else: []
28 
    GenServer.start_link(__MODULE__, options, gen_server_options)
29 
  end
30
31 
  def drain(server \\ __MODULE__), do: GenServer.call(server, :drain, 30_000)
32
33 
  def run_once(executor \\ &Restack.execute/1) when is_function(executor, 1) do
34 
    case claim_next() do
35 
      nil ->
36 
        :idle
37
38 
      %Operation{} = operation ->
39 
        _result = safe_execute(executor, operation)
40 
        :processed
41 
    end
42 
  end
43
44 
  @impl true
45 
  def init(options) do
46 
    state = %{
47 
      executor: Keyword.get(options, :executor, &Restack.execute/1),
48 
      poll_interval_ms: Keyword.get(options, :poll_interval_ms, poll_interval_ms())
49 
    }
50
51 
    schedule(state.poll_interval_ms)
52 
    {:ok, state}
53 
  end
54
55 
  @impl true
56 
  def handle_call(:drain, _from, state) do
57 
    {:reply, {:ok, drain_now(state.executor, 0)}, state}
58 
  end
59
60 
  @impl true
61 
  def handle_info(:poll, state) do
62 
    _result = run_once(state.executor)
63 
    schedule(state.poll_interval_ms)
64 
    {:noreply, state}
65 
  end
66
67 
  defp drain_now(_executor, count) when count >= @maximum_drain, do: count
68
69 
  defp drain_now(executor, count) do
70 
    case run_once(executor) do
71 
      :processed -> drain_now(executor, count + 1)
72 
      :idle -> count
73 
    end
74 
  end
75
76 
  defp claim_next do
77 
    now = DateTime.utc_now()
78 
    stale_before = DateTime.add(now, -@lease_seconds, :second)
79
80 
    {:ok, operation} =
81 
      Repo.transaction(fn ->
82 
        operation =
83 
          Repo.one(
84 
            from operation in Operation,
85 
              where:
86 
                (operation.state == "pending" and operation.retry_at <= ^now) or
87 
                  (operation.state == "running" and operation.claimed_at < ^stale_before),
88 
              order_by: [asc: operation.retry_at, asc: operation.inserted_at, asc: operation.id],
89 
              limit: 1,
90 
              lock: "FOR UPDATE SKIP LOCKED"
91 
          )
92
93 
        case operation do
94 
          nil ->
95 
            nil
96
97 
          %Operation{} = claimed ->
98 
            claimed
99 
            |> Operation.transition_changeset(%{
100 
              state: "running",
101 
              attempt_count: claimed.attempt_count + 1,
102 
              claimed_at: now,
103 
              started_at: claimed.started_at || now
104 
            })
105 
            |> Repo.update!()
106 
        end
107 
      end)
108
109 
    operation
110 
  end
111
112 
  defp safe_execute(executor, operation) do
113 
    executor.(operation)
114 
  rescue
115 
    error ->
116 
      Logger.warning(
117 
        "stack_operation_crashed operation=#{operation.id} code=#{OperationalLog.code(error)}"
118 
      )
119
120 
      mark_crashed(operation)
121 
      {:error, :operation_exception}
122 
  catch
123 
    kind, _reason ->
124 
      Logger.warning("stack_operation_crashed operation=#{operation.id} code=#{kind}")
125 
      mark_crashed(operation)
126 
      {:error, :operation_exception}
127 
  end
128
129 
  defp mark_crashed(operation) do
130 
    operation
131 
    |> Operation.transition_changeset(%{
132 
      state: "failed",
133 
      error: %{"code" => "operation_exception"},
134 
      completed_at: DateTime.utc_now()
135 
    })
136 
    |> Repo.update!()
137 
  end
138
139 
  defp schedule(interval), do: Process.send_after(self(), :poll, interval)
140
141 
  defp poll_interval_ms do
142 
    Application.get_env(:openagents, :stack_operation_worker_poll_interval_ms, 1_000)
143 
  end
144 
end
lib/openagents/stacks/restack.ex added +828

@@ -0,0 +1,828 @@

1 
defmodule OpenAgents.Stacks.Restack do
2 
  @moduledoc """
3 
  Cascading server-side stack rebase.
4
5 
  One durable operation walks the stack bottom to top: it verifies each live
6 
  branch still equals the stored observed head, replays only the commits
7 
  after the stored boundary onto the new parent, and plans one ref update
8 
  per branch. New commits build under hidden internal refs first; every
9 
  public branch then moves through one atomic compare-and-swap batch, so a
10 
  concurrent push rejects the whole batch and the user's branch survives.
11
12 
  A conflict pauses the operation in `waiting_for_conflict_resolution` with
13 
  a persisted workspace — old boundary, old head, proposed parent, in-flight
14 
  commit, conflict paths, and the steps that already succeeded — so continue
15 
  resumes exactly where the replay stopped and abort rolls back without any
16 
  public ref having moved.
17
18 
  Replayed commits are unsigned and carry the forge committer identity while
19 
  preserving each original author and message (`docs/stacked-prs.md` section
20 
  12.5); repositories that require author signatures rebase locally instead.
21 
  """
22 
  import Ecto.Query, warn: false
23
24 
  alias OpenAgents.Accounts.User
25 
  alias OpenAgents.Forge.GitPlane
26 
  alias OpenAgents.PullRequests.PullRequest
27 
  alias OpenAgents.Repo
28 
  alias OpenAgents.Repositories
29 
  alias OpenAgents.Repositories.Repository
30 
  alias OpenAgents.Stacks.Operation
31 
  alias OpenAgents.Stacks.Stack
32 
  alias OpenAgents.Stacks.StackEntry
33 
  alias OpenAgents.Stacks.StackEvent
34
35 
  @doc """
36 
  Requests a rebase of an open stack onto its current trunk tip.
37
38 
  The request inserts one durable `Operation` row in state `pending`; a
39 
  worker claims and executes it. A retried idempotency key replays the
40 
  original operation; the same key with a different request fails with
41 
  `:idempotency_conflict`. Only one operation may be active per stack.
42 
  """
43 
  def request_from_api(%Repository{} = repository, number, params, %User{} = actor, key)
44 
      when is_integer(number) and is_binary(key) do
45 
    with :ok <- authorize(repository, actor),
46 
         {:ok, request} <- parse_rebase_request(params) do
47 
      Repo.transaction(fn ->
48 
        lock_repository_stacks(repository.id)
49
50 
        with {:ok, stack} <- get_stack_for_update(repository, number),
51 
             :ok <- validate_open(stack),
52 
             {:ok, replay} <- check_idempotency(stack, key, request),
53 
             :ok <- ensure_no_active_operation(stack, replay),
54 
             :ok <- validate_expected_version(request["expected_stack_version"], stack) do
55 
          case replay do
56 
            %Operation{} = operation ->
57 
              {operation, :replayed}
58
59 
            nil ->
60 
              operation = insert_operation!(stack, actor, key, request)
61 
              set_health!(stack, "operation_in_progress")
62 
              {operation, :created}
63 
          end
64 
        else
65 
          {:error, reason} -> Repo.rollback(reason)
66 
        end
67 
      end)
68 
    end
69 
  end
70
71 
  @doc "Fetches one operation of a stack, scoped by repository and stack number."
72 
  def get_operation(%Repository{} = repository, number, operation_id) do
73 
    case Ecto.UUID.cast(operation_id) do
74 
      {:ok, operation_id} -> get_valid_operation(repository, number, operation_id)
75 
      :error -> {:error, :operation_not_found}
76 
    end
77 
  end
78
79 
  defp get_valid_operation(repository, number, operation_id) do
80 
    operation =
81 
      Repo.one(
82 
        from operation in Operation,
83 
          join: stack in assoc(operation, :stack),
84 
          where:
85 
            operation.id == ^operation_id and stack.number == ^number and
86 
              stack.repository_id == ^repository.id
87 
      )
88
89 
    case operation do
90 
      nil -> {:error, :operation_not_found}
91 
      %Operation{} -> {:ok, operation}
92 
    end
93 
  end
94
95 
  @doc """
96 
  Resumes a paused rebase with a caller-supplied resolution commit.
97
98 
  The resolution commit must already exist in the repository and its parent
99 
  must be the persisted `onto` — the tip the in-flight commit failed to
100 
  replay onto. The operation returns to `pending` and the worker resumes
101 
  from the persisted workspace; it re-verifies every branch head before
102 
  applying anything.
103 
  """
104 
  def continue_from_api(%Repository{} = repository, number, operation_id, params, %User{} = actor) do
105 
    with :ok <- authorize(repository, actor),
106 
         {:ok, resolution} <- required_oid(params, "resolution_oid") do
107 
      Repo.transaction(fn ->
108 
        with {:ok, operation} <- get_operation_for_update(repository, number, operation_id),
109 
             :ok <- validate_waiting(operation),
110 
             :ok <- validate_resolution(repository, operation, resolution) do
111 
          conflict = Map.put(operation.conflict, "resolution_oid", resolution)
112
113 
          operation
114 
          |> Operation.transition_changeset(%{
115 
            state: "pending",
116 
            conflict: conflict,
117 
            retry_at: DateTime.utc_now()
118 
          })
119 
          |> Repo.update!()
120 
        else
121 
          {:error, reason} -> Repo.rollback(reason)
122 
        end
123 
      end)
124 
    end
125 
  end
126
127 
  @doc """
128 
  Aborts a pending or paused operation.
129
130 
  No public ref has moved before an operation succeeds, so abort only marks
131 
  the row `cancelled` and restores the stack health observed at request
132 
  time (or `conflicted` when the pause proved a conflict exists).
133 
  """
134 
  def abort_from_api(%Repository{} = repository, number, operation_id, %User{} = actor) do
135 
    with :ok <- authorize(repository, actor) do
136 
      Repo.transaction(fn ->
137 
        with {:ok, operation} <- get_operation_for_update(repository, number, operation_id),
138 
             :ok <- validate_abortable(operation) do
139 
          stack = Repo.one!(from stack in Stack, where: stack.id == ^operation.stack_id)
140 
          set_health!(stack, abort_health(operation))
141
142 
          operation
143 
          |> Operation.transition_changeset(%{
144 
            state: "cancelled",
145 
            completed_at: DateTime.utc_now()
146 
          })
147 
          |> Repo.update!()
148 
        else
149 
          {:error, reason} -> Repo.rollback(reason)
150 
        end
151 
      end)
152 
    end
153 
  end
154
155 
  @doc """
156 
  Executes one claimed rebase operation to a terminal or paused state.
157
158 
  The caller (the operation worker) has already marked the row `running`.
159 
  Every state this function persists is recoverable: a crash before the
160 
  final metadata transaction leaves either nothing moved or the paused
161 
  conflict row, and a stale-leased `running` row re-executes from the
162 
  persisted request and workspace.
163 
  """
164 
  def execute(%Operation{kind: "rebase"} = operation) do
165 
    stack = load_stack(operation.stack_id)
166 
    repository = Repo.one!(from r in Repository, where: r.id == ^stack.repository_id)
167
168 
    with :ok <- validate_executable(operation, stack),
169 
         {:ok, trunk_tip} <- resolve_trunk(repository, stack),
170 
         operation = record_snapshot!(operation, stack, trunk_tip),
171 
         {:ok, steps} <- plan(operation, repository, stack, trunk_tip),
172 
         {:ok, applied} <- apply_refs(operation, repository, steps) do
173 
      finish(operation, stack, trunk_tip, steps, applied)
174 
    else
175 
      {:pause, conflict} -> pause(operation, conflict)
176 
      {:error, reason} -> fail(operation, stack, reason)
177 
    end
178 
  end
179
180 
  # The snapshot pins what the operation saw before touching anything, so a
181 
  # recovered worker can compare the live repository against the recorded
182 
  # starting point. The first execution records it; a resumed one keeps it.
183 
  defp record_snapshot!(%Operation{snapshot: nil} = operation, stack, trunk_tip) do
184 
    snapshot = %{
185 
      "trunk_oid" => trunk_tip,
186 
      "stack_version" => stack.version,
187 
      "entries" =>
188 
        Enum.map(stack.entries, fn entry ->
189 
          %{
190 
            "position" => entry.position,
191 
            "ref" => "refs/heads/" <> entry.pull_request.head_ref,
192 
            "boundary_oid" => entry.boundary_oid,
193 
            "observed_head_oid" => entry.observed_head_oid
194 
          }
195 
        end)
196 
    }
197
198 
    operation
199 
    |> Operation.transition_changeset(%{state: operation.state, snapshot: snapshot})
200 
    |> Repo.update!()
201 
  end
202
203 
  defp record_snapshot!(%Operation{} = operation, _stack, _trunk_tip), do: operation
204
205 
  ## Planning
206
207 
  defp plan(operation, repository, stack, trunk_tip) do
208 
    case operation.conflict do
209 
      %{"resolution_oid" => resolution} = conflict ->
210 
        resume_plan(repository, stack, conflict, resolution)
211
212 
      _no_resolution ->
213 
        waterfall(repository, stack.entries, trunk_tip, [])
214 
    end
215 
  end
216
217 
  defp waterfall(_repository, [], _new_parent, steps), do: {:ok, Enum.reverse(steps)}
218
219 
  defp waterfall(repository, [entry | rest], new_parent, steps) do
220 
    with :ok <- verify_live_head(repository, entry),
221 
         {:ok, step} <- plan_entry(repository, entry, new_parent, steps) do
222 
      waterfall(repository, rest, step.new_head, [step | steps])
223 
    end
224 
  end
225
226 
  defp plan_entry(_repository, entry, new_parent, _steps)
227 
       when entry.boundary_oid == new_parent do
228 
    {:ok, step(entry, new_parent, entry.observed_head_oid)}
229 
  end
230
231 
  defp plan_entry(repository, entry, new_parent, steps) do
232 
    case GitPlane.replay(
233 
           repository.storage_key,
234 
           entry.boundary_oid,
235 
           entry.observed_head_oid,
236 
           new_parent
237 
         ) do
238 
      {:ok, %{new_head: new_head}} ->
239 
        {:ok, step(entry, new_parent, new_head)}
240
241 
      {:conflict, conflict} ->
242 
        {:pause, conflict_workspace(entry, new_parent, conflict, Enum.reverse(steps))}
243
244 
      {:error, reason} ->
245 
        {:error, {:replay_failed, entry.position, reason}}
246 
    end
247 
  end
248
249 
  defp step(entry, new_boundary, new_head) do
250 
    %{
251 
      position: entry.position,
252 
      entry_id: entry.id,
253 
      pull_request_id: entry.pull_request_id,
254 
      pull_request_number: entry.pull_request.issue.number,
255 
      ref: "refs/heads/" <> entry.pull_request.head_ref,
256 
      old_head: entry.observed_head_oid,
257 
      old_boundary: entry.boundary_oid,
258 
      new_boundary: new_boundary,
259 
      new_head: new_head
260 
    }
261 
  end
262
263 
  defp verify_live_head(repository, entry) do
264 
    ref = "refs/heads/" <> entry.pull_request.head_ref
265
266 
    case GitPlane.resolve_commit(repository.storage_key, ref) do
267 
      {:ok, oid} when oid == entry.observed_head_oid -> :ok
268 
      {:ok, actual} -> {:error, {:head_changed, ref, actual}}
269 
      {:error, _reason} -> {:error, {:missing_ref, ref}}
270 
    end
271 
  end
272
273 
  defp conflict_workspace(entry, new_parent, conflict, prior_steps) do
274 
    %{
275 
      "position" => entry.position,
276 
      "entry_id" => entry.id,
277 
      "pull_request_number" => entry.pull_request.issue.number,
278 
      "old_boundary" => entry.boundary_oid,
279 
      "old_head" => entry.observed_head_oid,
280 
      "proposed_parent" => new_parent,
281 
      "onto" => conflict.onto,
282 
      "commit" => conflict.commit,
283 
      "paths" => conflict.paths,
284 
      "messages" => conflict.messages,
285 
      "replayed" => Enum.map(conflict.replayed, &%{"old" => &1.old, "new" => &1.new}),
286 
      "steps" => Enum.map(prior_steps, &stringify_step/1)
287 
    }
288 
  end
289
290 
  defp stringify_step(step) do
291 
    %{
292 
      "position" => step.position,
293 
      "entry_id" => step.entry_id,
294 
      "pull_request_id" => step.pull_request_id,
295 
      "pull_request_number" => step.pull_request_number,
296 
      "ref" => step.ref,
297 
      "old_head" => step.old_head,
298 
      "old_boundary" => step.old_boundary,
299 
      "new_boundary" => step.new_boundary,
300 
      "new_head" => step.new_head
301 
    }
302 
  end
303
304 
  defp resume_plan(repository, stack, conflict, resolution) do
305 
    position = Map.fetch!(conflict, "position")
306 
    prior_steps = conflict |> Map.fetch!("steps") |> Enum.map(&atomize_step/1)
307 
    entry = Enum.find(stack.entries, &(&1.position == position))
308 
    rest = Enum.filter(stack.entries, &(&1.position > position))
309
310 
    with :ok <- verify_prior_steps(repository, prior_steps),
311 
         {:ok, entry} <- require_entry(entry, conflict),
312 
         :ok <- verify_live_head(repository, entry),
313 
         {:ok, step} <- resume_entry(repository, entry, conflict, resolution) do
314 
      waterfall(repository, rest, step.new_head, [step | Enum.reverse(prior_steps)])
315 
    end
316 
  end
317
318 
  defp atomize_step(step) do
319 
    %{
320 
      position: Map.fetch!(step, "position"),
321 
      entry_id: Map.fetch!(step, "entry_id"),
322 
      pull_request_id: Map.fetch!(step, "pull_request_id"),
323 
      pull_request_number: Map.fetch!(step, "pull_request_number"),
324 
      ref: Map.fetch!(step, "ref"),
325 
      old_head: Map.fetch!(step, "old_head"),
326 
      old_boundary: Map.fetch!(step, "old_boundary"),
327 
      new_boundary: Map.fetch!(step, "new_boundary"),
328 
      new_head: Map.fetch!(step, "new_head")
329 
    }
330 
  end
331
332 
  defp require_entry(nil, conflict),
333 
    do: {:error, {:entry_removed, Map.fetch!(conflict, "position")}}
334
335 
  defp require_entry(%StackEntry{} = entry, _conflict), do: {:ok, entry}
336
337 
  defp verify_prior_steps(repository, steps) do
338 
    Enum.reduce_while(steps, :ok, fn step, :ok ->
339 
      case GitPlane.resolve_commit(repository.storage_key, step.ref) do
340 
        {:ok, oid} when oid == step.old_head -> {:cont, :ok}
341 
        {:ok, actual} -> {:halt, {:error, {:head_changed, step.ref, actual}}}
342 
        {:error, _reason} -> {:halt, {:error, {:missing_ref, step.ref}}}
343 
      end
344 
    end)
345 
  end
346
347 
  defp resume_entry(repository, entry, conflict, resolution) do
348 
    in_flight = Map.fetch!(conflict, "commit")
349 
    proposed_parent = Map.fetch!(conflict, "proposed_parent")
350
351 
    case GitPlane.replay(
352 
           repository.storage_key,
353 
           in_flight,
354 
           entry.observed_head_oid,
355 
           resolution
356 
         ) do
357 
      {:ok, %{new_head: new_head}} ->
358 
        {:ok, step(entry, proposed_parent, new_head)}
359
360 
      {:conflict, next_conflict} ->
361 
        prior_steps = Map.fetch!(conflict, "steps")
362
363 
        workspace =
364 
          entry
365 
          |> conflict_workspace(proposed_parent, next_conflict, [])
366 
          |> Map.put("steps", prior_steps)
367
368 
        {:pause, workspace}
369
370 
      {:error, reason} ->
371 
        {:error, {:replay_failed, entry.position, reason}}
372 
    end
373 
  end
374
375 
  ## Ref application
376
377 
  defp apply_refs(operation, repository, steps) do
378 
    changed = Enum.filter(steps, &(&1.new_head != &1.old_head))
379
380 
    if changed == [] do
381 
      {:ok, %{moved: false}}
382 
    else
383 
      with {:ok, temp_refs} <- retain_new_commits(operation, repository, changed),
384 
           :ok <- move_public_refs(repository, changed, temp_refs) do
385 
        {:ok, %{moved: true}}
386 
      end
387 
    end
388 
  end
389
390 
  # New commits become reachable (and WAL-persisted) under hidden internal
391 
  # refs before any public branch moves, so a crash between the two batches
392 
  # never strands the planned commits.
393 
  defp retain_new_commits(operation, repository, changed) do
394 
    temp_refs =
395 
      Enum.map(changed, fn step ->
396 
        {:ok, ref} =
397 
          GitPlane.internal_ref([
398 
            "operations",
399 
            operation.id,
400 
            "a#{operation.attempt_count}",
401 
            "p#{step.position}"
402 
          ])
403
404 
        %{ref: ref, expected_old: :absent, new: step.new_head}
405 
      end)
406
407 
    case GitPlane.batch_update_refs(repository.storage_key, temp_refs, principal(operation)) do
408 
      {:ok, _result} -> {:ok, temp_refs}
409 
      {:error, reason} -> {:error, {:retention_failed, reason}}
410 
    end
411 
  end
412
413 
  # One atomic batch: every public branch moves from its verified old head
414 
  # to its replayed head, and the retention refs delete in the same
415 
  # transaction. Any concurrent push fails the expected-old check and
416 
  # rejects the whole batch, preserving the user's branch.
417 
  defp move_public_refs(repository, changed, temp_refs) do
418 
    updates =
419 
      Enum.map(changed, fn step ->
420 
        %{ref: step.ref, expected_old: step.old_head, new: step.new_head}
421 
      end) ++
422 
        Enum.map(temp_refs, fn temp ->
423 
          %{ref: temp.ref, expected_old: temp.new, new: :delete}
424 
        end)
425
426 
    case GitPlane.batch_update_refs(repository.storage_key, updates, "stack-restack") do
427 
      {:ok, _result} ->
428 
        :ok
429
430 
      {:error, {:expected_mismatch, ref, actual}} ->
431 
        cleanup_temp_refs(repository, temp_refs)
432 
        {:error, {:head_changed, ref, actual}}
433
434 
      {:error, reason} ->
435 
        cleanup_temp_refs(repository, temp_refs)
436 
        {:error, {:ref_update_failed, reason}}
437 
    end
438 
  end
439
440 
  defp cleanup_temp_refs(repository, temp_refs) do
441 
    deletes = Enum.map(temp_refs, &%{ref: &1.ref, expected_old: &1.new, new: :delete})
442 
    _result = GitPlane.batch_update_refs(repository.storage_key, deletes, "stack-restack")
443 
    :ok
444 
  end
445
446 
  defp principal(operation), do: "stack-operation-" <> operation.id
447
448 
  ## Terminal transitions
449
450 
  defp finish(operation, stack, trunk_tip, steps, applied) do
451 
    result =
452 
      Repo.transaction(fn ->
453 
        lock_repository_stacks(stack.repository_id)
454 
        current = Repo.one!(from s in Stack, where: s.id == ^stack.id, lock: "FOR UPDATE")
455
456 
        if current.version != operation.expected_stack_version do
457 
          Repo.rollback({:version_moved, current.version})
458 
        end
459
460 
        stack_after = bump_version!(current)
461
462 
        Enum.each(steps, &apply_step_metadata!/1)
463 
        record_events!(stack_after, operation, trunk_tip, steps)
464
465 
        operation
466 
        |> Operation.transition_changeset(%{
467 
          state: "succeeded",
468 
          conflict: nil,
469 
          planned_result: %{
470 
            "trunk_oid" => trunk_tip,
471 
            "moved" => applied.moved,
472 
            "steps" => Enum.map(steps, &stringify_step/1)
473 
          },
474 
          completed_at: DateTime.utc_now()
475 
        })
476 
        |> Repo.update!()
477 
      end)
478
479 
    case result do
480 
      {:ok, operation} ->
481 
        {:ok, operation}
482
483 
      {:error, {:version_moved, _version} = reason} when applied.moved ->
484 
        partially_succeed(operation, trunk_tip, steps, reason)
485
486 
      {:error, reason} ->
487 
        fail(operation, stack, {:metadata_failed, reason})
488 
    end
489 
  end
490
491 
  # The refs already moved but the stack metadata advanced underneath the
492 
  # operation, so the branch updates stand while the metadata reconciliation
493 
  # is left to the caller.
494 
  defp partially_succeed(operation, trunk_tip, steps, reason) do
495 
    operation =
496 
      operation
497 
      |> Operation.transition_changeset(%{
498 
        state: "partially_succeeded",
499 
        error: error_map(reason),
500 
        planned_result: %{
501 
          "trunk_oid" => trunk_tip,
502 
          "moved" => true,
503 
          "steps" => Enum.map(steps, &stringify_step/1)
504 
        },
505 
        completed_at: DateTime.utc_now()
506 
      })
507 
      |> Repo.update!()
508
509 
    {:error, operation}
510 
  end
511
512 
  defp apply_step_metadata!(step) do
513 
    Repo.one!(from entry in StackEntry, where: entry.id == ^step.entry_id, lock: "FOR UPDATE")
514 
    |> StackEntry.changeset(%{
515 
      boundary_oid: step.new_boundary,
516 
      observed_head_oid: step.new_head
517 
    })
518 
    |> Repo.update!()
519
520 
    {1, _rows} =
521 
      Repo.update_all(
522 
        from(pr in PullRequest, where: pr.id == ^step.pull_request_id),
523 
        set: [head_sha: step.new_head, base_sha: step.new_boundary]
524 
      )
525
526 
    :ok
527 
  end
528
529 
  defp record_events!(stack, operation, trunk_tip, steps) do
530 
    changed = Enum.filter(steps, &(&1.new_head != &1.old_head))
531
532 
    record_event!(stack, operation, "pull_request_stack.rebased", %{
533 
      "operation_id" => operation.id,
534 
      "trunk_oid" => trunk_tip,
535 
      "steps" => Enum.map(steps, &stringify_step/1)
536 
    })
537
538 
    Enum.each(changed, fn step ->
539 
      record_event!(stack, operation, "pull_request.synchronize", %{
540 
        "operation_id" => operation.id,
541 
        "pull_request" => step.pull_request_number,
542 
        "ref" => step.ref,
543 
        "before" => step.old_head,
544 
        "after" => step.new_head
545 
      })
546 
    end)
547 
  end
548
549 
  defp record_event!(stack, operation, event_type, payload) do
550 
    %StackEvent{}
551 
    |> StackEvent.changeset(%{
552 
      stack_id: stack.id,
553 
      actor_user_id: operation.created_by_user_id,
554 
      event_type: event_type,
555 
      stack_version: stack.version,
556 
      payload: payload
557 
    })
558 
    |> Repo.insert!()
559 
  end
560
561 
  defp pause(operation, conflict) do
562 
    {:ok, operation} =
563 
      Repo.transaction(fn ->
564 
        stack = Repo.one!(from s in Stack, where: s.id == ^operation.stack_id, lock: "FOR UPDATE")
565 
        set_health!(stack, "conflicted")
566
567 
        operation
568 
        |> Operation.transition_changeset(%{
569 
          state: "waiting_for_conflict_resolution",
570 
          conflict: conflict
571 
        })
572 
        |> Repo.update!()
573 
      end)
574
575 
    {:waiting, operation}
576 
  end
577
578 
  defp fail(operation, stack, reason) do
579 
    {:ok, operation} =
580 
      Repo.transaction(fn ->
581 
        current = Repo.one!(from s in Stack, where: s.id == ^stack.id, lock: "FOR UPDATE")
582 
        set_health!(current, failure_health(reason))
583
584 
        operation
585 
        |> Operation.transition_changeset(%{
586 
          state: "failed",
587 
          error: error_map(reason),
588 
          completed_at: DateTime.utc_now()
589 
        })
590 
        |> Repo.update!()
591 
      end)
592
593 
    {:error, operation}
594 
  end
595
596 
  defp failure_health({:head_changed, _ref, _actual}), do: "head_changed"
597 
  defp failure_health({:missing_ref, _ref}), do: "missing_ref"
598 
  defp failure_health(_reason), do: "needs_rebase"
599
600 
  defp error_map({:head_changed, ref, actual}),
601 
    do: %{"code" => "head_changed", "ref" => ref, "actual" => stringify_actual(actual)}
602
603 
  defp error_map({:missing_ref, ref}), do: %{"code" => "missing_ref", "ref" => ref}
604
605 
  defp error_map({:replay_failed, position, reason}),
606 
    do: %{"code" => "replay_failed", "position" => position, "reason" => inspect(reason)}
607
608 
  defp error_map({:retention_failed, reason}),
609 
    do: %{"code" => "retention_failed", "reason" => inspect(reason)}
610
611 
  defp error_map({:ref_update_failed, reason}),
612 
    do: %{"code" => "ref_update_failed", "reason" => inspect(reason)}
613
614 
  defp error_map({:metadata_failed, reason}),
615 
    do: %{"code" => "metadata_failed", "reason" => inspect(reason)}
616
617 
  defp error_map({:version_moved, version}),
618 
    do: %{"code" => "version_moved", "stack_version" => version}
619
620 
  defp error_map(reason) when is_atom(reason), do: %{"code" => Atom.to_string(reason)}
621 
  defp error_map(reason), do: %{"code" => "operation_failed", "reason" => inspect(reason)}
622
623 
  defp stringify_actual(:absent), do: "absent"
624 
  defp stringify_actual(oid), do: oid
625
626 
  ## Validation
627
628 
  defp validate_executable(operation, stack) do
629 
    cond do
630 
      stack.state != "open" -> {:error, :stack_not_open}
631 
      stack.version != operation.expected_stack_version -> {:error, :stale_stack_version}
632 
      stack.entries == [] -> {:error, :empty_stack}
633 
      true -> :ok
634 
    end
635 
  end
636
637 
  defp resolve_trunk(repository, stack) do
638 
    case GitPlane.resolve_commit(repository.storage_key, "refs/heads/" <> stack.trunk_ref) do
639 
      {:ok, oid} -> {:ok, oid}
640 
      {:error, _reason} -> {:error, {:missing_ref, "refs/heads/" <> stack.trunk_ref}}
641 
    end
642 
  end
643
644 
  defp authorize(repository, actor) do
645 
    if Repositories.writable?(repository, actor), do: :ok, else: {:error, :forbidden}
646 
  end
647
648 
  defp parse_rebase_request(params) do
649 
    case Map.get(params, "expected_stack_version") do
650 
      nil ->
651 
        {:ok, %{"expected_stack_version" => nil}}
652
653 
      version when is_integer(version) and version >= 1 ->
654 
        {:ok, %{"expected_stack_version" => version}}
655
656 
      _other ->
657 
        {:error, :invalid_request}
658 
    end
659 
  end
660
661 
  defp required_oid(params, key) do
662 
    case Map.get(params, key) do
663 
      value when is_binary(value) and byte_size(value) in [40, 64] ->
664 
        case Base.decode16(value, case: :lower) do
665 
          {:ok, _raw} -> {:ok, value}
666 
          :error -> {:error, :invalid_request}
667 
        end
668
669 
      _other ->
670 
        {:error, :invalid_request}
671 
    end
672 
  end
673
674 
  defp check_idempotency(stack, key, request) do
675 
    operation =
676 
      Repo.one(
677 
        from operation in Operation,
678 
          where: operation.stack_id == ^stack.id and operation.idempotency_key == ^key,
679 
          lock: "FOR UPDATE"
680 
      )
681
682 
    case operation do
683 
      nil ->
684 
        {:ok, nil}
685
686 
      %Operation{} = operation ->
687 
        if Map.drop(operation.request, ["previous_health"]) == request,
688 
          do: {:ok, operation},
689 
          else: {:error, :idempotency_conflict}
690 
    end
691 
  end
692
693 
  defp ensure_no_active_operation(_stack, %Operation{}), do: :ok
694
695 
  defp ensure_no_active_operation(stack, nil) do
696 
    active =
697 
      Repo.exists?(
698 
        from operation in Operation,
699 
          where:
700 
            operation.stack_id == ^stack.id and
701 
              operation.state in ^Operation.active_states()
702 
      )
703
704 
    if active, do: {:error, :operation_in_progress}, else: :ok
705 
  end
706
707 
  defp validate_expected_version(nil, _stack), do: :ok
708 
  defp validate_expected_version(version, %Stack{version: version}), do: :ok
709 
  defp validate_expected_version(_version, %Stack{}), do: {:error, :stale_stack_version}
710
711 
  defp insert_operation!(stack, actor, key, request) do
712 
    %Operation{}
713 
    |> Operation.changeset(%{
714 
      stack_id: stack.id,
715 
      created_by_user_id: actor.id,
716 
      kind: "rebase",
717 
      state: "pending",
718 
      expected_stack_version: request["expected_stack_version"] || stack.version,
719 
      idempotency_key: key,
720 
      request: Map.put(request, "previous_health", stack.health),
721 
      retry_at: DateTime.utc_now()
722 
    })
723 
    |> Repo.insert!()
724 
  end
725
726 
  defp get_operation_for_update(repository, number, operation_id) do
727 
    with {:ok, operation} <- get_operation(repository, number, operation_id) do
728 
      {:ok,
729 
       Repo.one!(
730 
         from candidate in Operation,
731 
           where: candidate.id == ^operation.id,
732 
           lock: "FOR UPDATE"
733 
       )}
734 
    end
735 
  end
736
737 
  defp validate_waiting(%Operation{state: "waiting_for_conflict_resolution"}), do: :ok
738 
  defp validate_waiting(%Operation{}), do: {:error, :operation_not_waiting}
739
740 
  defp validate_abortable(%Operation{state: state})
741 
       when state in ["pending", "waiting_for_conflict_resolution"],
742 
       do: :ok
743
744 
  defp validate_abortable(%Operation{}), do: {:error, :operation_not_abortable}
745
746 
  defp validate_resolution(repository, operation, resolution) do
747 
    onto = Map.fetch!(operation.conflict, "onto")
748
749 
    with {:ok, _oid} <- resolve_or_error(repository, resolution),
750 
         {:ok, parents} <- parents_of(repository, resolution) do
751 
      if parents == [onto], do: :ok, else: {:error, :resolution_parent_mismatch}
752 
    end
753 
  end
754
755 
  defp resolve_or_error(repository, oid) do
756 
    case GitPlane.resolve_commit(repository.storage_key, oid) do
757 
      {:ok, full} when full == oid -> {:ok, full}
758 
      {:ok, _other} -> {:error, :resolution_not_found}
759 
      {:error, _reason} -> {:error, :resolution_not_found}
760 
    end
761 
  end
762
763 
  defp parents_of(repository, oid) do
764 
    case GitPlane.parents(repository.storage_key, oid) do
765 
      {:ok, parents} -> {:ok, parents}
766 
      {:error, _reason} -> {:error, :resolution_not_found}
767 
    end
768 
  end
769
770 
  defp abort_health(%Operation{state: "waiting_for_conflict_resolution"}), do: "conflicted"
771
772 
  defp abort_health(%Operation{request: request}),
773 
    do: Map.get(request, "previous_health", "needs_rebase")
774
775 
  ## Shared helpers
776
777 
  defp load_stack(stack_id) do
778 
    Stack
779 
    |> Repo.get!(stack_id)
780 
    |> Repo.preload(
781 
      entries:
782 
        from(entry in StackEntry,
783 
          where: is_nil(entry.removed_at),
784 
          order_by: [asc: entry.position],
785 
          preload: [pull_request: :issue]
786 
        )
787 
    )
788 
  end
789
790 
  defp get_stack_for_update(%Repository{id: repository_id}, number) do
791 
    case Repo.one(
792 
           from stack in Stack,
793 
             where: stack.repository_id == ^repository_id and stack.number == ^number,
794 
             lock: "FOR UPDATE"
795 
         ) do
796 
      nil -> {:error, :stack_not_found}
797 
      stack -> {:ok, stack}
798 
    end
799 
  end
800
801 
  defp validate_open(%Stack{state: "open"}), do: :ok
802 
  defp validate_open(%Stack{}), do: {:error, :stack_not_open}
803
804 
  defp lock_repository_stacks(repository_id) do
805 
    key = "pull_request_stacks:#{repository_id}"
806 
    Repo.query!("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", [key])
807 
    :ok
808 
  end
809
810 
  defp set_health!(%Stack{} = stack, health) do
811 
    stack
812 
    |> Stack.changeset(%{health: health})
813 
    |> Repo.update!()
814 
  end
815
816 
  defp bump_version!(%Stack{id: id, version: version, state: "open"}) do
817 
    {1, [stack]} =
818 
      Repo.update_all(
819 
        from(stack in Stack,
820 
          where: stack.id == ^id and stack.version == ^version and stack.state == "open",
821 
          select: stack
822 
        ),
823 
        set: [version: version + 1, health: "healthy", updated_at: DateTime.utc_now()]
824 
      )
825
826 
    stack
827 
  end
828 
end
lib/openagents/stacks/stack_event.ex modified +1 -1

@@ -13,7 +13,7 @@ defmodule OpenAgents.Stacks.StackEvent do

13 13 
  @foreign_key_type :binary_id
14 14 
  @timestamps_opts [type: :utc_datetime_usec]
15 15
16 
  @event_types ~w(pull_request_stack.created pull_request_stack.appended)
16 
  @event_types ~w(pull_request_stack.created pull_request_stack.appended pull_request_stack.rebased pull_request.synchronize)
17 17
18 18 
  schema "pull_request_stack_events" do
19 19 
    belongs_to :stack, OpenAgents.Stacks.Stack
lib/openagents_web/api_route_authority.ex modified +7

@@ -66,6 +66,8 @@ defmodule OpenAgentsWeb.ApiRouteAuthority do

66 66 
      "get /api/v3/repos/:owner/:repo/pulls/:pull_number" => :optional_bearer,
67 67 
      "get /api/v3/repos/:owner/:repo/stacks" => :optional_bearer,
68 68 
      "get /api/v3/repos/:owner/:repo/stacks/:stack_number" => :optional_bearer,
69 
      "get /api/v3/repos/:owner/:repo/stacks/:stack_number/operations/:operation_id" =>
70 
        :optional_bearer,
69 71 
      "get /api/v3/repos/:owner/:repo/projectsV2" => :optional_bearer,
70 72 
      "get /api/v3/repos/:owner/:repo/projectsV2/:project_number" => :optional_bearer,
71 73 
      "get /api/v3/repos/:owner/:repo/projectsV2/:project_number/items" => :optional_bearer,

@@ -107,6 +109,11 @@ defmodule OpenAgentsWeb.ApiRouteAuthority do

107 109 
      "post /api/v3/repos/:owner/:repo/pulls" => :required_bearer,
108 110 
      "post /api/v3/repos/:owner/:repo/stacks" => :required_bearer,
109 111 
      "post /api/v3/repos/:owner/:repo/stacks/:stack_number/append" => :required_bearer,
112 
      "post /api/v3/repos/:owner/:repo/stacks/:stack_number/rebase" => :required_bearer,
113 
      "post /api/v3/repos/:owner/:repo/stacks/:stack_number/operations/:operation_id/continue" =>
114 
        :required_bearer,
115 
      "post /api/v3/repos/:owner/:repo/stacks/:stack_number/operations/:operation_id/abort" =>
116 
        :required_bearer,
110 117 
      # pipe_through :deployments_api — tenant deployment authority only. No
111 118 
      # route here is anonymous, and none of them reaches the operator fleet
112 119 
      # promotion surface.
lib/openagents_web/controllers/stack_controller.ex modified +88 -4

@@ -3,6 +3,7 @@ defmodule OpenAgentsWeb.StackController do

3 3
4 4 
  alias OpenAgents.Repositories
5 5 
  alias OpenAgents.Stacks
6 
  alias OpenAgents.Stacks.Restack
6 7 
  alias OpenAgentsWeb.ControllerHelpers
7 8
8 9 
  def index(conn, %{"owner" => owner, "repo" => repo}) do

@@ -56,6 +57,76 @@ defmodule OpenAgentsWeb.StackController do

56 57 
    Ecto.NoResultsError -> not_found(conn)
57 58 
  end
58 59
60 
  def rebase(conn, %{"owner" => owner, "repo" => repo, "stack_number" => number} = params) do
61 
    repository = Repositories.get_visible_by_path!(owner, repo, conn.assigns.current_user)
62
63 
    with {:ok, idempotency_key} <- idempotency_key(conn),
64 
         {:ok, {operation, replay_state}} <-
65 
           Restack.request_from_api(
66 
             repository,
67 
             ControllerHelpers.integer_param!(number),
68 
             params,
69 
             conn.assigns.current_user,
70 
             idempotency_key
71 
           ) do
72 
      conn
73 
      |> put_status(:accepted)
74 
      |> render(:operation, operation: operation, replay_state: replay_state)
75 
    else
76 
      {:error, reason} -> render_error(conn, reason)
77 
    end
78 
  rescue
79 
    Ecto.NoResultsError -> not_found(conn)
80 
  end
81
82 
  def show_operation(conn, %{"owner" => owner, "repo" => repo} = params) do
83 
    repository = Repositories.get_visible_by_path!(owner, repo, conn.assigns[:current_user])
84
85 
    case Restack.get_operation(
86 
           repository,
87 
           ControllerHelpers.integer_param!(params["stack_number"]),
88 
           params["operation_id"]
89 
         ) do
90 
      {:ok, operation} -> render(conn, :operation, operation: operation, replay_state: nil)
91 
      {:error, reason} -> render_error(conn, reason)
92 
    end
93 
  rescue
94 
    Ecto.NoResultsError -> not_found(conn)
95 
  end
96
97 
  def continue_operation(conn, %{"owner" => owner, "repo" => repo} = params) do
98 
    repository = Repositories.get_visible_by_path!(owner, repo, conn.assigns.current_user)
99
100 
    case Restack.continue_from_api(
101 
           repository,
102 
           ControllerHelpers.integer_param!(params["stack_number"]),
103 
           params["operation_id"],
104 
           params,
105 
           conn.assigns.current_user
106 
         ) do
107 
      {:ok, operation} -> render(conn, :operation, operation: operation, replay_state: nil)
108 
      {:error, reason} -> render_error(conn, reason)
109 
    end
110 
  rescue
111 
    Ecto.NoResultsError -> not_found(conn)
112 
  end
113
114 
  def abort_operation(conn, %{"owner" => owner, "repo" => repo} = params) do
115 
    repository = Repositories.get_visible_by_path!(owner, repo, conn.assigns.current_user)
116
117 
    case Restack.abort_from_api(
118 
           repository,
119 
           ControllerHelpers.integer_param!(params["stack_number"]),
120 
           params["operation_id"],
121 
           conn.assigns.current_user
122 
         ) do
123 
      {:ok, operation} -> render(conn, :operation, operation: operation, replay_state: nil)
124 
      {:error, reason} -> render_error(conn, reason)
125 
    end
126 
  rescue
127 
    Ecto.NoResultsError -> not_found(conn)
128 
  end
129
59 130 
  defp idempotency_key(conn) do
60 131 
    case get_req_header(conn, "idempotency-key") do
61 132 
      [key] when byte_size(key) in 1..200 ->

@@ -74,15 +145,19 @@ defmodule OpenAgentsWeb.StackController do

74 145 
  defp render_error(conn, :forbidden),
75 146 
    do: error(conn, :forbidden, "You cannot modify stacks in this repository.")
76 147
77 
  defp render_error(conn, reason) when reason in [:stack_not_found, :pull_request_not_found],
78 
    do: not_found(conn)
148 
  defp render_error(conn, reason)
149 
       when reason in [:stack_not_found, :pull_request_not_found, :operation_not_found],
150 
       do: not_found(conn)
79 151
80 152 
  defp render_error(conn, reason)
81 153 
       when reason in [
82 154 
              :idempotency_conflict,
83 155 
              :stale_stack_version,
84 156 
              :expected_head_mismatch,
85 
              :stack_not_open
157 
              :stack_not_open,
158 
              :operation_in_progress,
159 
              :operation_not_waiting,
160 
              :operation_not_abortable
86 161 
            ],
87 162 
       do: conflict(conn, reason)
88 163

@@ -99,7 +174,9 @@ defmodule OpenAgentsWeb.StackController do

99 174 
              :duplicate_branch,
100 175 
              :broken_base_chain,
101 176 
              :already_stacked,
102 
              :not_stack_top
177 
              :not_stack_top,
178 
              :resolution_not_found,
179 
              :resolution_parent_mismatch
103 180 
            ] do
104 181 
    conn
105 182 
    |> put_status(:unprocessable_entity)

@@ -128,6 +205,13 @@ defmodule OpenAgentsWeb.StackController do

128 205 
  defp message(:broken_base_chain), do: "The direct-base chain is broken."
129 206 
  defp message(:already_stacked), do: "A pull request already belongs to an active stack."
130 207 
  defp message(:not_stack_top), do: "The pull request does not target the current top head."
208 
  defp message(:operation_in_progress), do: "Another operation is active on this stack."
209 
  defp message(:operation_not_waiting), do: "The operation is not waiting for a resolution."
210 
  defp message(:operation_not_abortable), do: "The operation can no longer be aborted."
211 
  defp message(:resolution_not_found), do: "The resolution commit does not exist."
212
213 
  defp message(:resolution_parent_mismatch),
214 
    do: "The resolution commit does not build on the persisted parent."
131 215
132 216 
  defp not_found(conn), do: error(conn, :not_found, "Not Found")
133 217 
  defp error(conn, status, message), do: conn |> put_status(status) |> json(%{message: message})
lib/openagents_web/controllers/stack_json.ex modified +21

@@ -13,6 +13,27 @@ defmodule OpenAgentsWeb.StackJSON do

13 13 
    end
14 14 
  end
15 15
16 
  def render("operation.json", %{operation: operation} = assigns) do
17 
    json = %{
18 
      id: operation.id,
19 
      kind: operation.kind,
20 
      state: operation.state,
21 
      expected_stack_version: operation.expected_stack_version,
22 
      target_position: operation.target_position,
23 
      conflict: operation.conflict,
24 
      planned_result: operation.planned_result,
25 
      error: operation.error,
26 
      created_at: operation.inserted_at,
27 
      started_at: operation.started_at,
28 
      completed_at: operation.completed_at
29 
    }
30
31 
    case Map.get(assigns, :replay_state) do
32 
      nil -> json
33 
      replay_state -> Map.put(json, :replayed, replay_state == :replayed)
34 
    end
35 
  end
36
16 37 
  defp stack(stack, assigns) do
17 38 
    base_url = String.trim_trailing(OpenAgentsWeb.Endpoint.url(), "/")
18 39 
    owner = assigns.owner
lib/openagents_web/router.ex modified +15

@@ -385,6 +385,11 @@ defmodule OpenAgentsWeb.Router do

385 385 
    get "/repos/:owner/:repo/pulls/:pull_number", PullRequestController, :show
386 386 
    get "/repos/:owner/:repo/stacks", StackController, :index
387 387 
    get "/repos/:owner/:repo/stacks/:stack_number", StackController, :show
388
389 
    get "/repos/:owner/:repo/stacks/:stack_number/operations/:operation_id",
390 
        StackController,
391 
        :show_operation
392
388 393 
    get "/repos/:owner/:repo/projectsV2", ProjectController, :index
389 394 
    get "/repos/:owner/:repo/projectsV2/:project_number", ProjectController, :show
390 395 
    get "/repos/:owner/:repo/projectsV2/:project_number/items", ProjectController, :items

@@ -439,6 +444,16 @@ defmodule OpenAgentsWeb.Router do

439 444 
    patch "/repos/:owner/:repo/pulls/:pull_number", PullRequestController, :update
440 445 
    post "/repos/:owner/:repo/stacks", StackController, :create
441 446 
    post "/repos/:owner/:repo/stacks/:stack_number/append", StackController, :append
447 
    post "/repos/:owner/:repo/stacks/:stack_number/rebase", StackController, :rebase
448
449 
    post "/repos/:owner/:repo/stacks/:stack_number/operations/:operation_id/continue",
450 
         StackController,
451 
         :continue_operation
452
453 
    post "/repos/:owner/:repo/stacks/:stack_number/operations/:operation_id/abort",
454 
         StackController,
455 
         :abort_operation
456
442 457 
    post "/repos/:owner/:repo/issues/:issue_number/comments", CommentController, :create
443 458 
    put "/repos/:owner/:repo/issues/comments/:id", CommentController, :update
444 459 
    patch "/repos/:owner/:repo/issues/comments/:id", CommentController, :update
priv/repo/migrations/20260823072000_create_stack_operations.exs added +55

@@ -0,0 +1,55 @@

1 
defmodule OpenAgents.Repo.Migrations.CreateStackOperations do
2 
  use Ecto.Migration
3
4 
  def change do
5 
    create table(:stack_operations, primary_key: false) do
6 
      add :id, :binary_id, primary_key: true
7
8 
      add :stack_id,
9 
          references(:pull_request_stacks, type: :binary_id, on_delete: :delete_all),
10 
          null: false
11
12 
      add :kind, :string, null: false
13 
      add :state, :string, null: false, default: "pending"
14 
      add :target_position, :integer
15 
      add :expected_stack_version, :bigint, null: false
16 
      add :idempotency_key, :string, null: false
17
18 
      add :request, :map, null: false, default: %{}
19 
      add :snapshot, :map
20 
      add :planned_result, :map
21 
      add :conflict, :map
22 
      add :error, :map
23
24 
      add :created_by_user_id,
25 
          references(:users, type: :binary_id, on_delete: :nilify_all)
26
27 
      add :attempt_count, :integer, null: false, default: 0
28 
      add :retry_at, :utc_datetime_usec
29 
      add :claimed_at, :utc_datetime_usec
30 
      add :started_at, :utc_datetime_usec
31 
      add :completed_at, :utc_datetime_usec
32
33 
      timestamps(type: :utc_datetime_usec)
34 
    end
35
36 
    create unique_index(:stack_operations, [:stack_id, :idempotency_key])
37
38 
    create unique_index(:stack_operations, [:stack_id],
39 
             where: "state IN ('pending', 'running', 'waiting_for_conflict_resolution')",
40 
             name: :stack_operations_active_stack_index
41 
           )
42
43 
    create index(:stack_operations, [:state, :retry_at])
44
45 
    create constraint(:stack_operations, :stack_operations_kind_check,
46 
             check:
47 
               "kind IN ('create', 'append', 'restructure', 'rebase', 'merge', 'queue', 'unstack', 'dissolve', 'repair')"
48 
           )
49
50 
    create constraint(:stack_operations, :stack_operations_state_check,
51 
             check:
52 
               "state IN ('pending', 'running', 'waiting_for_conflict_resolution', 'waiting_for_checks', 'succeeded', 'partially_succeeded', 'failed', 'cancelled')"
53 
           )
54 
  end
55 
end
test/openagents/stacks/restack_test.exs added +608

@@ -0,0 +1,608 @@

1 
defmodule OpenAgents.Stacks.RestackTest do
2 
  @moduledoc """
3 
  Cascading server-side rebase (#50): the bottom-to-top waterfall, conflict
4 
  pause with a durable workspace, continue and abort, head re-verification
5 
  on resume, atomic public ref movement with retention refs, operation
6 
  idempotency, and crash recovery through the lease.
7 
  """
8
9 
  use OpenAgents.DataCase, async: false
10
11 
  alias OpenAgents.Forge.Repos
12 
  alias OpenAgents.PullRequests.PullRequest
13 
  alias OpenAgents.Repo
14 
  alias OpenAgents.Stacks
15 
  alias OpenAgents.Stacks.Operation
16 
  alias OpenAgents.Stacks.OperationWorker
17 
  alias OpenAgents.Stacks.Restack
18 
  alias OpenAgents.Stacks.Stack
19 
  alias OpenAgents.Stacks.StackEntry
20 
  alias OpenAgents.Stacks.StackEvent
21
22 
  import Ecto.Query
23 
  import OpenAgents.AccountsFixtures
24 
  import OpenAgents.IssuesFixtures
25
26 
  setup do
27 
    base = Path.join(System.tmp_dir!(), "stack-restack-#{System.unique_integer([:positive])}")
28
29 
    previous_data = Application.get_env(:openagents, :forge_data_dir)
30 
    previous_wal = Application.get_env(:openagents, :forge_wal_dir)
31 
    Application.put_env(:openagents, :forge_data_dir, Path.join(base, "data"))
32 
    Application.put_env(:openagents, :forge_wal_dir, Path.join(base, "wal"))
33
34 
    on_exit(fn ->
35 
      restore_env(:forge_data_dir, previous_data)
36 
      restore_env(:forge_wal_dir, previous_wal)
37 
      File.rm_rf(base)
38 
    end)
39
40 
    actor = repository_user_fixture("restack-actor")
41 
    repository = repository_with_member_fixture(actor)
42
43 
    %{actor: actor, repository: repository}
44 
  end
45
46 
  describe "a clean restack" do
47 
    test "moves every branch, updates metadata, and emits events", context do
48 
      %{repository: repository, actor: actor} = context
49 
      %{path: path, oids: oids, stack: stack} = seed_stack(repository, actor)
50
51 
      trunk_tip = advance_trunk(path, oids["main"], "trunk.md")
52
53 
      {:ok, {operation, :created}} =
54 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-1")
55
56 
      assert operation.state == "pending"
57 
      assert reload(Stack, stack.id).health == "operation_in_progress"
58
59 
      assert :processed = OperationWorker.run_once()
60
61 
      operation = reload(Operation, operation.id)
62 
      assert operation.state == "succeeded"
63 
      assert operation.snapshot["trunk_oid"] == trunk_tip
64 
      assert operation.planned_result["moved"] == true
65
66 
      stack = reload(Stack, stack.id)
67 
      assert stack.version == 2
68 
      assert stack.health == "healthy"
69
70 
      new_layer_1 = show(path, ["rev-parse", "refs/heads/layer-1"])
71 
      new_layer_2 = show(path, ["rev-parse", "refs/heads/layer-2"])
72 
      refute new_layer_1 == oids["layer-1"]
73 
      refute new_layer_2 == oids["layer-2"]
74
75 
      assert show(path, ["rev-parse", "refs/heads/layer-1^"]) == trunk_tip
76 
      assert show(path, ["rev-parse", "refs/heads/layer-2^"]) == new_layer_1
77
78 
      [entry_1, entry_2] = entries(stack.id)
79 
      assert entry_1.boundary_oid == trunk_tip
80 
      assert entry_1.observed_head_oid == new_layer_1
81 
      assert entry_2.boundary_oid == new_layer_1
82 
      assert entry_2.observed_head_oid == new_layer_2
83
84 
      pull_request_1 = Repo.one!(from pr in PullRequest, where: pr.id == ^entry_1.pull_request_id)
85 
      assert pull_request_1.head_sha == new_layer_1
86 
      assert pull_request_1.base_sha == trunk_tip
87
88 
      assert Repo.exists?(
89 
               from event in StackEvent,
90 
                 where:
91 
                   event.event_type == "pull_request_stack.rebased" and
92 
                     event.stack_id == ^stack.id and event.stack_version == 2
93 
             )
94
95 
      synchronize_count =
96 
        Repo.aggregate(
97 
          from(event in StackEvent,
98 
            where: event.event_type == "pull_request.synchronize" and event.stack_id == ^stack.id
99 
          ),
100 
          :count
101 
        )
102
103 
      assert synchronize_count == 2
104
105 
      # The replacement commits keep the original author and message.
106 
      assert show(path, ["show", "-s", "--format=%an <%ae>", new_layer_1]) ==
107 
               "Test Author <author@example.test>"
108
109 
      assert show(path, ["show", "-s", "--format=%s", new_layer_2]) == "Layer layer-2"
110 
    end
111
112 
    test "leaves no retention ref behind", context do
113 
      %{repository: repository, actor: actor} = context
114 
      %{path: path, oids: oids, stack: stack} = seed_stack(repository, actor)
115
116 
      advance_trunk(path, oids["main"], "trunk.md")
117
118 
      {:ok, {_operation, :created}} =
119 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-refs-1")
120
121 
      assert :processed = OperationWorker.run_once()
122
123 
      {internal, 0} = Repos.git(path, ["for-each-ref", "refs/internal/"])
124 
      assert internal == ""
125 
    end
126
127 
    test "an already-current stack succeeds without moving refs", context do
128 
      %{repository: repository, actor: actor} = context
129 
      %{path: path, oids: oids, stack: stack} = seed_stack(repository, actor)
130
131 
      {:ok, {operation, :created}} =
132 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-noop-1")
133
134 
      assert :processed = OperationWorker.run_once()
135
136 
      operation = reload(Operation, operation.id)
137 
      assert operation.state == "succeeded"
138 
      assert operation.planned_result["moved"] == false
139
140 
      assert show(path, ["rev-parse", "refs/heads/layer-1"]) == oids["layer-1"]
141 
      assert show(path, ["rev-parse", "refs/heads/layer-2"]) == oids["layer-2"]
142 
      assert reload(Stack, stack.id).health == "healthy"
143 
    end
144 
  end
145
146 
  describe "a mid-stack conflict" do
147 
    test "pauses with a durable workspace and no public ref moved", context do
148 
      %{repository: repository, actor: actor} = context
149 
      %{path: path, oids: oids, stack: stack} = seed_conflicting_stack(repository, actor)
150
151 
      {:ok, {operation, :created}} =
152 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-conflict-1")
153
154 
      assert :processed = OperationWorker.run_once()
155
156 
      operation = reload(Operation, operation.id)
157 
      assert operation.state == "waiting_for_conflict_resolution"
158
159 
      conflict = operation.conflict
160 
      assert conflict["position"] == 2
161 
      assert conflict["old_boundary"] == oids["layer-1"]
162 
      assert conflict["old_head"] == oids["layer-2"]
163 
      assert conflict["commit"] == oids["layer-2"]
164 
      assert conflict["paths"] == ["shared.txt"]
165 
      assert [%{"ref" => "refs/heads/layer-1"}] = conflict["steps"]
166
167 
      assert reload(Stack, stack.id).health == "conflicted"
168
169 
      # No public branch moved: the pause happened before any batch.
170 
      assert show(path, ["rev-parse", "refs/heads/layer-1"]) == oids["layer-1"]
171 
      assert show(path, ["rev-parse", "refs/heads/layer-2"]) == oids["layer-2"]
172 
    end
173
174 
    test "continue resumes from the resolution and completes the stack", context do
175 
      %{repository: repository, actor: actor} = context
176 
      %{path: path, oids: oids, stack: stack} = seed_conflicting_stack(repository, actor)
177
178 
      {:ok, {operation, :created}} =
179 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-continue-1")
180
181 
      assert :processed = OperationWorker.run_once()
182 
      operation = reload(Operation, operation.id)
183 
      assert operation.state == "waiting_for_conflict_resolution"
184
185 
      onto = operation.conflict["onto"]
186 
      resolution = commit(path, onto, "Resolve shared.txt", %{"shared.txt" => "resolved\n"})
187
188 
      assert {:ok, %Operation{state: "pending"}} =
189 
               Restack.continue_from_api(
190 
                 repository,
191 
                 stack.number,
192 
                 operation.id,
193 
                 %{"resolution_oid" => resolution},
194 
                 actor
195 
               )
196
197 
      assert :processed = OperationWorker.run_once()
198
199 
      operation = reload(Operation, operation.id)
200 
      assert operation.state == "succeeded"
201
202 
      assert show(path, ["rev-parse", "refs/heads/layer-2"]) == resolution
203 
      refute show(path, ["rev-parse", "refs/heads/layer-1"]) == oids["layer-1"]
204
205 
      stack = reload(Stack, stack.id)
206 
      assert stack.health == "healthy"
207 
      assert stack.version == 2
208 
    end
209
210 
    test "continue rejects a resolution that does not build on the persisted parent",
211 
         context do
212 
      %{repository: repository, actor: actor} = context
213 
      %{path: path, oids: oids, stack: stack} = seed_conflicting_stack(repository, actor)
214
215 
      {:ok, {operation, :created}} =
216 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-badres-1")
217
218 
      assert :processed = OperationWorker.run_once()
219 
      operation = reload(Operation, operation.id)
220
221 
      stray = commit(path, oids["main"], "Wrong parent", %{"stray.txt" => "stray\n"})
222
223 
      assert {:error, :resolution_parent_mismatch} =
224 
               Restack.continue_from_api(
225 
                 repository,
226 
                 stack.number,
227 
                 operation.id,
228 
                 %{"resolution_oid" => stray},
229 
                 actor
230 
               )
231
232 
      assert {:error, :resolution_not_found} =
233 
               Restack.continue_from_api(
234 
                 repository,
235 
                 stack.number,
236 
                 operation.id,
237 
                 %{"resolution_oid" => String.duplicate("0", 40)},
238 
                 actor
239 
               )
240 
    end
241
242 
    test "a resumed operation re-verifies every branch head first", context do
243 
      %{repository: repository, actor: actor} = context
244 
      %{path: path, oids: oids, stack: stack} = seed_conflicting_stack(repository, actor)
245
246 
      {:ok, {operation, :created}} =
247 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-reverify-1")
248
249 
      assert :processed = OperationWorker.run_once()
250 
      operation = reload(Operation, operation.id)
251 
      onto = operation.conflict["onto"]
252 
      resolution = commit(path, onto, "Resolve shared.txt", %{"shared.txt" => "resolved\n"})
253
254 
      assert {:ok, _operation} =
255 
               Restack.continue_from_api(
256 
                 repository,
257 
                 stack.number,
258 
                 operation.id,
259 
                 %{"resolution_oid" => resolution},
260 
                 actor
261 
               )
262
263 
      # A concurrent push lands on layer-1 before the worker resumes.
264 
      concurrent = commit(path, oids["layer-1"], "Concurrent push", %{"race.txt" => "race\n"})
265 
      {_, 0} = Repos.git(path, ["update-ref", "refs/heads/layer-1", concurrent])
266
267 
      assert :processed = OperationWorker.run_once()
268
269 
      operation = reload(Operation, operation.id)
270 
      assert operation.state == "failed"
271 
      assert operation.error["code"] == "head_changed"
272
273 
      assert reload(Stack, stack.id).health == "head_changed"
274 
      assert show(path, ["rev-parse", "refs/heads/layer-1"]) == concurrent
275 
      assert show(path, ["rev-parse", "refs/heads/layer-2"]) == oids["layer-2"]
276 
    end
277
278 
    test "abort cancels the paused operation and records the conflict", context do
279 
      %{repository: repository, actor: actor} = context
280 
      %{stack: stack} = seed_conflicting_stack(repository, actor)
281
282 
      {:ok, {operation, :created}} =
283 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-abort-1")
284
285 
      assert :processed = OperationWorker.run_once()
286
287 
      assert {:ok, %Operation{state: "cancelled"}} =
288 
               Restack.abort_from_api(repository, stack.number, operation.id, actor)
289
290 
      assert reload(Stack, stack.id).health == "conflicted"
291
292 
      assert {:error, :operation_not_abortable} =
293 
               Restack.abort_from_api(repository, stack.number, operation.id, actor)
294 
    end
295 
  end
296
297 
  describe "concurrent pushes" do
298 
    test "a moved branch fails the operation and the user branch survives", context do
299 
      %{repository: repository, actor: actor} = context
300 
      %{path: path, oids: oids, stack: stack} = seed_stack(repository, actor)
301
302 
      advance_trunk(path, oids["main"], "trunk.md")
303
304 
      {:ok, {operation, :created}} =
305 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-race-1")
306
307 
      concurrent = commit(path, oids["layer-2"], "User push", %{"user.txt" => "mine\n"})
308 
      {_, 0} = Repos.git(path, ["update-ref", "refs/heads/layer-2", concurrent])
309
310 
      assert :processed = OperationWorker.run_once()
311
312 
      operation = reload(Operation, operation.id)
313 
      assert operation.state == "failed"
314 
      assert operation.error["code"] == "head_changed"
315 
      assert operation.error["ref"] == "refs/heads/layer-2"
316
317 
      assert reload(Stack, stack.id).health == "head_changed"
318 
      assert show(path, ["rev-parse", "refs/heads/layer-1"]) == oids["layer-1"]
319 
      assert show(path, ["rev-parse", "refs/heads/layer-2"]) == concurrent
320 
    end
321
322 
    test "a deleted branch fails the operation as missing_ref", context do
323 
      %{repository: repository, actor: actor} = context
324 
      %{path: path, oids: oids, stack: stack} = seed_stack(repository, actor)
325
326 
      advance_trunk(path, oids["main"], "trunk.md")
327
328 
      {:ok, {operation, :created}} =
329 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-missing-1")
330
331 
      {_, 0} = Repos.git(path, ["update-ref", "-d", "refs/heads/layer-2"])
332
333 
      assert :processed = OperationWorker.run_once()
334
335 
      operation = reload(Operation, operation.id)
336 
      assert operation.state == "failed"
337 
      assert operation.error["code"] == "missing_ref"
338 
      assert reload(Stack, stack.id).health == "missing_ref"
339 
    end
340 
  end
341
342 
  describe "request validation" do
343 
    test "replays the same idempotency key and rejects a changed request", context do
344 
      %{repository: repository, actor: actor} = context
345 
      %{stack: stack} = seed_stack(repository, actor)
346
347 
      {:ok, {operation, :created}} =
348 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-idem-1")
349
350 
      {:ok, {replayed, :replayed}} =
351 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-idem-1")
352
353 
      assert replayed.id == operation.id
354 
      assert Repo.aggregate(Operation, :count) == 1
355
356 
      assert {:error, :idempotency_conflict} =
357 
               Restack.request_from_api(
358 
                 repository,
359 
                 stack.number,
360 
                 %{"expected_stack_version" => 1},
361 
                 actor,
362 
                 "restack-idem-1"
363 
               )
364 
    end
365
366 
    test "rejects a second active operation and a stale expected version", context do
367 
      %{repository: repository, actor: actor} = context
368 
      %{stack: stack} = seed_stack(repository, actor)
369
370 
      assert {:error, :stale_stack_version} =
371 
               Restack.request_from_api(
372 
                 repository,
373 
                 stack.number,
374 
                 %{"expected_stack_version" => 9},
375 
                 actor,
376 
                 "restack-active-3"
377 
               )
378
379 
      {:ok, {_operation, :created}} =
380 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-active-1")
381
382 
      assert {:error, :operation_in_progress} =
383 
               Restack.request_from_api(repository, stack.number, %{}, actor, "restack-active-2")
384 
    end
385
386 
    test "rejects a caller without write access", context do
387 
      %{repository: repository, actor: actor} = context
388 
      %{stack: stack} = seed_stack(repository, actor)
389 
      outsider = repository_user_fixture("restack-outsider")
390
391 
      assert {:error, :forbidden} =
392 
               Restack.request_from_api(repository, stack.number, %{}, outsider, "restack-out-1")
393 
    end
394
395 
    test "a version that moves before execution fails the operation", context do
396 
      %{repository: repository, actor: actor} = context
397 
      %{stack: stack} = seed_stack(repository, actor)
398
399 
      {:ok, {operation, :created}} =
400 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-stale-1")
401
402 
      {1, _rows} =
403 
        Repo.update_all(from(s in Stack, where: s.id == ^stack.id), set: [version: 5])
404
405 
      assert :processed = OperationWorker.run_once()
406
407 
      operation = reload(Operation, operation.id)
408 
      assert operation.state == "failed"
409 
      assert operation.error["code"] == "stale_stack_version"
410 
    end
411 
  end
412
413 
  describe "crash recovery" do
414 
    test "a stale running lease is reclaimed and executed to completion", context do
415 
      %{repository: repository, actor: actor} = context
416 
      %{path: path, oids: oids, stack: stack} = seed_stack(repository, actor)
417
418 
      trunk_tip = advance_trunk(path, oids["main"], "trunk.md")
419
420 
      {:ok, {operation, :created}} =
421 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-crash-1")
422
423 
      # A worker claimed the row and crashed: the lease is stale.
424 
      stale = DateTime.add(DateTime.utc_now(), -600, :second)
425
426 
      {1, _rows} =
427 
        Repo.update_all(
428 
          from(o in Operation, where: o.id == ^operation.id),
429 
          set: [state: "running", claimed_at: stale, attempt_count: 1]
430 
        )
431
432 
      assert :processed = OperationWorker.run_once()
433
434 
      operation = reload(Operation, operation.id)
435 
      assert operation.state == "succeeded"
436 
      assert operation.attempt_count == 2
437 
      assert show(path, ["rev-parse", "refs/heads/layer-1^"]) == trunk_tip
438 
    end
439
440 
    test "an executor crash marks the operation failed", context do
441 
      %{repository: repository, actor: actor} = context
442 
      %{stack: stack} = seed_stack(repository, actor)
443
444 
      {:ok, {operation, :created}} =
445 
        Restack.request_from_api(repository, stack.number, %{}, actor, "restack-boom-1")
446
447 
      assert :processed = OperationWorker.run_once(fn _operation -> raise "boom" end)
448
449 
      operation = reload(Operation, operation.id)
450 
      assert operation.state == "failed"
451 
      assert operation.error["code"] == "operation_exception"
452 
    end
453 
  end
454
455 
  ## Fixtures
456
457 
  # main ── layer-1 ── layer-2, each layer adding one file.
458 
  defp seed_stack(repository, actor) do
459 
    path = Repos.ensure_repo!(repository.storage_key, repository.default_branch)
460
461 
    main = commit(path, nil, "Seed repository", %{"README.md" => "readme\n"})
462 
    {_, 0} = Repos.git(path, ["update-ref", "refs/heads/main", main])
463
464 
    layer_1 = commit(path, main, "Layer layer-1", %{"layer-1.md" => "one\n"})
465 
    {_, 0} = Repos.git(path, ["update-ref", "refs/heads/layer-1", layer_1])
466
467 
    layer_2 = commit(path, layer_1, "Layer layer-2", %{"layer-2.md" => "two\n"})
468 
    {_, 0} = Repos.git(path, ["update-ref", "refs/heads/layer-2", layer_2])
469
470 
    oids = %{"main" => main, "layer-1" => layer_1, "layer-2" => layer_2}
471 
    build_stack(repository, actor, path, oids)
472 
  end
473
474 
  # layer-2 rewrites shared.txt, and the trunk advance rewrites it too, so
475 
  # layer-1 replays clean and layer-2 conflicts.
476 
  defp seed_conflicting_stack(repository, actor) do
477 
    path = Repos.ensure_repo!(repository.storage_key, repository.default_branch)
478
479 
    main = commit(path, nil, "Seed repository", %{"shared.txt" => "base\n"})
480 
    {_, 0} = Repos.git(path, ["update-ref", "refs/heads/main", main])
481
482 
    layer_1 = commit(path, main, "Layer layer-1", %{"layer-1.md" => "one\n"})
483 
    {_, 0} = Repos.git(path, ["update-ref", "refs/heads/layer-1", layer_1])
484
485 
    layer_2 = commit(path, layer_1, "Layer layer-2", %{"shared.txt" => "layer two\n"})
486 
    {_, 0} = Repos.git(path, ["update-ref", "refs/heads/layer-2", layer_2])
487
488 
    trunk = commit(path, main, "Trunk rewrite", %{"shared.txt" => "trunk\n"})
489 
    {_, 0} = Repos.git(path, ["update-ref", "refs/heads/main", trunk])
490
491 
    oids = %{"main" => main, "trunk" => trunk, "layer-1" => layer_1, "layer-2" => layer_2}
492 
    build_stack(repository, actor, path, oids)
493 
  end
494
495 
  defp build_stack(repository, actor, path, oids) do
496 
    bottom = pull_request(repository, "layer-1", "main", oids["main"], oids["layer-1"])
497 
    top = pull_request(repository, "layer-2", "layer-1", oids["layer-1"], oids["layer-2"])
498
499 
    {:ok, stack} = Stacks.create(repository, [bottom, top], actor)
500
501 
    %{path: path, oids: oids, stack: stack, pull_requests: [bottom, top]}
502 
  end
503
504 
  defp advance_trunk(path, parent, file) do
505 
    trunk_tip = commit(path, parent, "Trunk advance", %{file => "trunk\n"})
506 
    {_, 0} = Repos.git(path, ["update-ref", "refs/heads/main", trunk_tip])
507 
    trunk_tip
508 
  end
509
510 
  defp pull_request(repository, head_ref, base_ref, base_sha, head_sha) do
511 
    issue = issue_fixture(repository, %{title: "PR #{head_ref}"})
512
513 
    {:ok, pull_request} =
514 
      %PullRequest{}
515 
      |> PullRequest.changeset(%{
516 
        repository_id: repository.id,
517 
        issue_id: issue.id,
518 
        head_repository_id: repository.id,
519 
        head_ref: head_ref,
520 
        head_sha: head_sha,
521 
        base_ref: base_ref,
522 
        base_sha: base_sha,
523 
        state: "open"
524 
      })
525 
      |> Repo.insert()
526
527 
    Repo.preload(pull_request, :issue)
528 
  end
529
530 
  # Commits a tree that layers the given files over the parent's tree.
531 
  defp commit(path, parent, message, files) do
532 
    parent_entries =
533 
      if parent do
534 
        {listing, 0} = Repos.git(path, ["ls-tree", parent])
535
536 
        listing
537 
        |> String.split("\n", trim: true)
538 
        |> Map.new(fn line ->
539 
          [meta, name] = String.split(line, "\t", parts: 2)
540 
          {name, meta <> "\t" <> name}
541 
        end)
542 
      else
543 
        %{}
544 
      end
545
546 
    new_entries =
547 
      Map.new(files, fn {name, content} ->
548 
        blob = git!(path, ["hash-object", "-w", "--stdin"], content)
549 
        {name, "100644 blob #{blob}\t#{name}"}
550 
      end)
551
552 
    listing =
553 
      parent_entries
554 
      |> Map.merge(new_entries)
555 
      |> Map.values()
556 
      |> Enum.map_join("", &(&1 <> "\n"))
557
558 
    tree = git!(path, ["mktree"], listing)
559 
    parent_args = if parent, do: ["-p", parent], else: []
560
561 
    git!(path, ["commit-tree", tree] ++ parent_args ++ ["-m", message], "",
562 
      env: [
563 
        {"GIT_AUTHOR_NAME", "Test Author"},
564 
        {"GIT_AUTHOR_EMAIL", "author@example.test"},
565 
        {"GIT_COMMITTER_NAME", "Test Author"},
566 
        {"GIT_COMMITTER_EMAIL", "author@example.test"}
567 
      ]
568 
    )
569 
  end
570
571 
  defp entries(stack_id) do
572 
    Repo.all(
573 
      from entry in StackEntry,
574 
        where: entry.stack_id == ^stack_id and is_nil(entry.removed_at),
575 
        order_by: [asc: entry.position]
576 
    )
577 
  end
578
579 
  defp reload(schema, id), do: Repo.get!(schema, id)
580
581 
  defp show(path, args) do
582 
    {output, 0} = Repos.git(path, args)
583 
    String.trim(output)
584 
  end
585
586 
  defp git!(git_dir, args, input, options \\ []) do
587 
    input_path =
588 
      Path.join(System.tmp_dir!(), "restack-input-#{System.unique_integer([:positive])}")
589
590 
    File.write!(input_path, input)
591
592 
    try do
593 
      {output, 0} =
594 
        System.cmd(
595 
          "sh",
596 
          ["-c", ~s(exec git --git-dir "$GIT_DIR" "$@" < "$INPUT"), "sh"] ++ args,
597 
          env: [{"GIT_DIR", git_dir}, {"INPUT", input_path}] ++ Keyword.get(options, :env, [])
598 
        )
599
600 
      String.trim(output)
601 
    after
602 
      File.rm(input_path)
603 
    end
604 
  end
605
606 
  defp restore_env(key, nil), do: Application.delete_env(:openagents, key)
607 
  defp restore_env(key, value), do: Application.put_env(:openagents, key, value)
608 
end
test/openagents_web/controllers/stack_controller_test.exs modified +100

@@ -314,6 +314,106 @@ defmodule OpenAgentsWeb.StackControllerTest do

314 314 
    end
315 315 
  end
316 316
317 
  describe "POST /api/v3/repos/:owner/:repo/stacks/:stack_number/rebase" do
318 
    test "accepts a rebase, exposes the operation, and replays retries", %{conn: conn} do
319 
      repository = repository_fixture()
320 
      oids = seed_chain(repository, ["layer-1"])
321 
      [pr_1] = pull_request_chain(repository, oids, ["layer-1"])
322 
      conn = put_forge_api_token(conn, "stack-rebase", repository)
323
324 
      assert %{"number" => 1} =
325 
               conn
326 
               |> put_req_header("idempotency-key", "rebase-create-1")
327 
               |> post(path(repository), %{trunk_ref: "main", pull_requests: [pr_1]})
328 
               |> json_response(201)
329
330 
      rebase_conn =
331 
        conn
332 
        |> put_req_header("idempotency-key", "rebase-1")
333 
        |> post("#{path(repository)}/1/rebase", %{})
334
335 
      assert %{
336 
               "id" => operation_id,
337 
               "kind" => "rebase",
338 
               "state" => "pending",
339 
               "replayed" => false
340 
             } = json_response(rebase_conn, 202)
341
342 
      assert %{"health" => "operation_in_progress"} =
343 
               json_response(get(conn, "#{path(repository)}/1"), 200)
344
345 
      replay_conn =
346 
        conn
347 
        |> put_req_header("idempotency-key", "rebase-1")
348 
        |> post("#{path(repository)}/1/rebase", %{})
349
350 
      assert %{"id" => ^operation_id, "replayed" => true} = json_response(replay_conn, 202)
351
352 
      second_conn =
353 
        conn
354 
        |> put_req_header("idempotency-key", "rebase-2")
355 
        |> post("#{path(repository)}/1/rebase", %{})
356
357 
      assert %{"code" => "operation_in_progress"} = json_response(second_conn, 409)
358
359 
      show_conn = get(conn, "#{path(repository)}/1/operations/#{operation_id}")
360 
      assert %{"id" => ^operation_id, "state" => "pending"} = json_response(show_conn, 200)
361
362 
      missing_conn =
363 
        get(conn, "#{path(repository)}/1/operations/00000000-0000-0000-0000-000000000000")
364
365 
      assert json_response(missing_conn, 404)
366 
    end
367
368 
    test "continue requires a paused operation, and abort cancels", %{conn: conn} do
369 
      repository = repository_fixture()
370 
      oids = seed_chain(repository, ["layer-1"])
371 
      [pr_1] = pull_request_chain(repository, oids, ["layer-1"])
372 
      conn = put_forge_api_token(conn, "stack-rebase-ops", repository)
373
374 
      assert %{"number" => 1} =
375 
               conn
376 
               |> put_req_header("idempotency-key", "rebase-ops-create")
377 
               |> post(path(repository), %{trunk_ref: "main", pull_requests: [pr_1]})
378 
               |> json_response(201)
379
380 
      assert %{"id" => operation_id} =
381 
               conn
382 
               |> put_req_header("idempotency-key", "rebase-ops-1")
383 
               |> post("#{path(repository)}/1/rebase", %{})
384 
               |> json_response(202)
385
386 
      continue_conn =
387 
        post(conn, "#{path(repository)}/1/operations/#{operation_id}/continue", %{
388 
          resolution_oid: oids["layer-1"]
389 
        })
390
391 
      assert %{"code" => "operation_not_waiting"} = json_response(continue_conn, 409)
392
393 
      abort_conn = post(conn, "#{path(repository)}/1/operations/#{operation_id}/abort", %{})
394 
      assert %{"id" => ^operation_id, "state" => "cancelled"} = json_response(abort_conn, 200)
395
396 
      assert %{"health" => "healthy"} = json_response(get(conn, "#{path(repository)}/1"), 200)
397
398 
      again_conn = post(conn, "#{path(repository)}/1/operations/#{operation_id}/abort", %{})
399 
      assert %{"code" => "operation_not_abortable"} = json_response(again_conn, 409)
400 
    end
401
402 
    test "refuses a caller without write access", %{conn: conn} do
403 
      repository = repository_fixture()
404 
      oids = seed_chain(repository, ["layer-1"])
405 
      [_pr_1] = pull_request_chain(repository, oids, ["layer-1"])
406 
      conn = put_forge_api_token(conn, "stack-rebase-outsider")
407
408 
      conn =
409 
        conn
410 
        |> put_req_header("idempotency-key", "rebase-forbidden-1")
411 
        |> post("#{path(repository)}/1/rebase", %{})
412
413 
      assert json_response(conn, 403)
414 
    end
415 
  end
416
317 417 
  describe "GET /api/v3/repos/:owner/:repo/stacks" do
318 418 
    test "reads are public for a public repository", %{conn: conn} do
319 419 
      repository = repository_fixture()

This page updates live while a promote is in flight · changelog