diff --git a/SPEC.md b/SPEC.md index cfc13a02d5..3435dab07b 100644 --- a/SPEC.md +++ b/SPEC.md @@ -501,7 +501,7 @@ Fields: Validation: -- Repo names MUST be unique. +- Repo names MUST be unique, including after workspace path sanitization. - `repositories:` MUST contain at least one entry. - At most one repo may set `default: true`. - With multiple repos, unscoped non-default repos are rejected. diff --git a/docs/configuration.md b/docs/configuration.md index f6973d981a..ccae3e1c58 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -111,7 +111,8 @@ repositories: fetch_before_dispatch: true ``` -- `key`: unique repo key used in dashboards, run records, and prompt context. +- `key`: unique repo key used in dashboards, run records, and prompt context. Keys must also remain + unique after workspace path sanitization. - `workflow`: path to that repo's `WORKFLOW.md`; defaults to `WORKFLOW.md`. - `default`: at most one repo can be the fallback route. - `base_branch`: optional branch used for review-agent diff context and as the base a @@ -120,8 +121,9 @@ repositories: - `route`: Linear team, project, label, or assignee selectors. - `workspace`: per-repo override for workspace population. -Routing validation rejects duplicate keys, identical routes, ambiguous team catch-alls, multiple -defaults, and multi-repo global worktree settings that do not provide per-repo workspace overrides. +Routing validation rejects duplicate keys, workspace-sanitized key collisions, identical routes, +ambiguous team catch-alls, multiple defaults, and multi-repo global worktree settings that do not +provide per-repo workspace overrides. ### `workspaces` diff --git a/lib/symphony_elixir/ci_poller.ex b/lib/symphony_elixir/ci_poller.ex index a9a7c95cf2..8f39868bd8 100644 --- a/lib/symphony_elixir/ci_poller.ex +++ b/lib/symphony_elixir/ci_poller.ex @@ -15,6 +15,11 @@ defmodule SymphonyElixir.CiPoller do @closed_pr_states ["CLOSED", "MERGED"] @github_error_backoff_threshold 3 @max_github_error_backoff_ms 300_000 + # Grace window after a CI-failure dispatch during which escalation is held off, + # giving the orchestrator time to pick up the In Progress issue and mark the + # rework run "running". Without it, the poll immediately following the final + # retry's dispatch could escalate before the agent starts and abandon it. + @dispatch_start_grace_ms 120_000 @status_table :ci_poller_status defmodule State do @@ -328,13 +333,7 @@ defmodule SymphonyElixir.CiPoller do flaky_retry?(settings) and not rerun_attempted_for_sha?(record, commit_sha) -> rerun_failed_ci(record, ci_status, failed_checks, settings, opts, now) - dispatched_for_sha?(record, commit_sha) -> - attrs = - ci_status_attrs(record, ci_status, %{status: "failure_already_handled", failed_checks: failed_checks}, now) - - complete_ci_update(opts, record, attrs, {:already_handled, Map.get(record, :issue_id), commit_sha}) - - ci_retry_count(record) >= settings.ci.max_retries and Map.get(record, :status) != "escalated" -> + escalate_ci_failure?(record, settings, commit_sha, opts, now) -> escalate_ci_failure(record, ci_status, failed_checks, settings, opts, now) Map.get(record, :status) == "escalated" -> @@ -343,6 +342,12 @@ defmodule SymphonyElixir.CiPoller do complete_ci_update(opts, record, attrs, {:already_handled, Map.get(record, :issue_id), commit_sha}) + dispatched_for_sha?(record, commit_sha) -> + attrs = + ci_status_attrs(record, ci_status, %{status: "failure_already_handled", failed_checks: failed_checks}, now) + + complete_ci_update(opts, record, attrs, {:already_handled, Map.get(record, :issue_id), commit_sha}) + true -> dispatch_ci_failure(record, ci_status, failed_checks, settings, opts, now) end @@ -686,6 +691,8 @@ defmodule SymphonyElixir.CiPoller do ci_status, %{ status: "state_transition_error", + ci_retry_count: ci_retry_count(record), + dispatched_shas: string_list(Map.get(record, :dispatched_shas, [])), failed_checks: failed_checks, last_action: action, last_action_at: nil @@ -989,6 +996,31 @@ defmodule SymphonyElixir.CiPoller do defp rerun_attempted_for_sha?(record, sha), do: sha in string_list(Map.get(record, :rerun_attempted_shas, [])) defp dispatched_for_sha?(record, sha), do: sha in string_list(Map.get(record, :dispatched_shas, [])) + # Escalate once retries are exhausted, but only when no rework is in flight and + # the latest dispatch has had time to start (see recently_dispatched?/3). + defp escalate_ci_failure?(record, settings, commit_sha, opts, now) do + ci_retry_count(record) >= settings.ci.max_retries and + Map.get(record, :status) != "escalated" and + not rework_in_progress?(record, opts) and + not recently_dispatched?(record, commit_sha, now) + end + + # A dispatch for this SHA landed within the start-grace window, so the rework + # agent may not have reached "running" yet. Hold off escalation until either it + # does (covered by rework_in_progress?/2) or the grace window lapses, so the + # final retry's just-dispatched agent is not escalated out from under itself. + defp recently_dispatched?(record, commit_sha, now) do + dispatched_for_sha?(record, commit_sha) and + Map.get(record, :last_action) == "dispatch" and + within_dispatch_grace?(Map.get(record, :last_action_at), now) + end + + defp within_dispatch_grace?(%DateTime{} = last_action_at, %DateTime{} = now) do + DateTime.diff(now, last_action_at, :millisecond) < @dispatch_start_grace_ms + end + + defp within_dispatch_grace?(_last_action_at, _now), do: false + defp ci_owned_record?(record) do ci_retry_count(record) > 0 or Map.get(record, :status) in ["dispatch_requested", "escalated", "state_transition_error"] end @@ -1000,8 +1032,9 @@ defmodule SymphonyElixir.CiPoller do active_agent_run?(issue_id, repo_key, opts) or pending_rework_review?(issue_id, repo_key, opts) end - # Runs and PR reviews are prefetched once per poll cycle so the green path - # does not rescan storage for every CI check (see rework_in_progress?/2). + # Runs and PR reviews are prefetched once per poll cycle so the green-deferral + # and escalation paths do not rescan storage for every CI check (see + # rework_in_progress?/2). defp put_prefetched_rework_sources(opts, _run_store, _repo_key, []), do: opts defp put_prefetched_rework_sources(opts, run_store, repo_key, _checks) do @@ -1056,7 +1089,6 @@ defmodule SymphonyElixir.CiPoller do {:ok, reviews} -> Enum.any?(reviews, fn review -> Map.get(review, :issue_id) == issue_id and - Map.get(review, :status) == "rework_requested" and pending_reviewer_comments?(Map.get(review, :pending_reviewer_comments)) end) diff --git a/lib/symphony_elixir/config/system_schema.ex b/lib/symphony_elixir/config/system_schema.ex index 50518d953b..b97cbaf5a6 100644 --- a/lib/symphony_elixir/config/system_schema.ex +++ b/lib/symphony_elixir/config/system_schema.ex @@ -7,6 +7,7 @@ defmodule SymphonyElixir.Config.SystemSchema do alias SymphonyElixir.Config.Schema alias SymphonyElixir.Workflow + alias SymphonyElixir.Workspace @primary_key false @allowed_keys ~w( @@ -403,6 +404,7 @@ defmodule SymphonyElixir.Config.SystemSchema do |> cast_embed(:repos, with: &Repo.changeset/2, required: true) |> validate_length(:repos, min: 1) |> validate_unique_repo_names() + |> validate_unique_repo_workspace_keys() |> validate_single_default_repo() end @@ -773,13 +775,7 @@ defmodule SymphonyElixir.Config.SystemSchema do defp validate_unique_repo_names(changeset) do duplicate_names = changeset - |> get_change(:repos, []) - |> Enum.flat_map(fn repo_changeset -> - case get_field(repo_changeset, :name) do - name when is_binary(name) and name != "" -> [name] - _name -> [] - end - end) + |> repo_names() |> duplicate_values() case duplicate_names do @@ -788,6 +784,19 @@ defmodule SymphonyElixir.Config.SystemSchema do end end + defp validate_unique_repo_workspace_keys(changeset) do + duplicate_workspace_keys = + changeset + |> repo_names() + |> Enum.map(&repo_workspace_key/1) + |> duplicate_values() + + case duplicate_workspace_keys do + [] -> changeset + _duplicates -> add_error(changeset, :repos, "keys must not collide after workspace normalization") + end + end + defp validate_single_default_repo(changeset) do default_count = changeset @@ -803,6 +812,21 @@ defmodule SymphonyElixir.Config.SystemSchema do defp truthy_change?(changeset, field), do: get_field(changeset, field) == true + # Mirror the real workspace path normalization so validation predicts the same + # collisions that Workspace.safe_identifier/1 would produce on disk. + defp repo_workspace_key(name), do: Workspace.safe_identifier(name) + + defp repo_names(changeset) do + changeset + |> get_change(:repos, []) + |> Enum.flat_map(fn repo_changeset -> + case get_field(repo_changeset, :name) do + name when is_binary(name) and name != "" -> [name] + _name -> [] + end + end) + end + defp duplicate_values(values) do {_seen, duplicates} = Enum.reduce(values, {MapSet.new(), MapSet.new()}, fn value, {seen, duplicates} -> diff --git a/lib/symphony_elixir/orchestrator.ex b/lib/symphony_elixir/orchestrator.ex index f2597ad308..6214bd75ec 100644 --- a/lib/symphony_elixir/orchestrator.ex +++ b/lib/symphony_elixir/orchestrator.ex @@ -155,7 +155,7 @@ defmodule SymphonyElixir.Orchestrator do state = seed_watching_from_completed_run_metadata(state) - mark_interrupted_runs(repo_key) + mark_interrupted_runs_for_configured_repos(repo_key) tick_token = make_ref() send(self(), {:tick, tick_token}) schedule_snapshot_publish(config.observability.snapshot_publish_ms) @@ -4258,17 +4258,42 @@ defmodule SymphonyElixir.Orchestrator do defp retry_attempt(attempt) when is_integer(attempt) and attempt > 0, do: attempt defp retry_attempt(_attempt), do: 1 + defp mark_interrupted_runs_for_configured_repos(default_repo_key) do + default_repo_key + |> configured_repo_keys() + |> Enum.each(&mark_interrupted_runs/1) + end + + defp configured_repo_keys(default_repo_key) do + case Config.repos() do + {:ok, repos} -> + repos + |> Enum.map(&Map.get(&1, :name)) + |> Enum.filter(&(is_binary(&1) and String.trim(&1) != "")) + |> Enum.uniq() + |> case do + [] -> [default_repo_key] + repo_keys -> repo_keys + end + + {:error, reason} -> + Logger.warning("Failed to read configured repos for startup run interruption; using primary repo #{default_repo_key}: #{inspect(reason)}") + + [default_repo_key] + end + end + defp mark_interrupted_runs(repo_key) do case RunStore.interrupt_running_runs(repo_key, "orchestrator restarted before worker exit") do {:ok, 0} -> :ok {:ok, count} -> - Logger.warning("Marked #{count} previously running agent run(s) as failed after orchestrator startup") + Logger.warning("Marked #{count} previously running agent run(s) as failed after orchestrator startup repo_key=#{repo_key}") :ok {:error, reason} -> - Logger.warning("Failed to mark interrupted runs in run store: #{inspect(reason)}") + Logger.warning("Failed to mark interrupted runs in run store repo_key=#{repo_key}: #{inspect(reason)}") :ok end end diff --git a/lib/symphony_elixir/pr_review_poller.ex b/lib/symphony_elixir/pr_review_poller.ex index d3290c7085..401e6875b2 100644 --- a/lib/symphony_elixir/pr_review_poller.ex +++ b/lib/symphony_elixir/pr_review_poller.ex @@ -1013,20 +1013,35 @@ defmodule SymphonyElixir.PrReviewPoller do end defp transition_issue_for_action(record, attrs, opts, now, action) do - issue_id = Map.get(record, :issue_id) - if dispatch_paused?(opts) do defer_transition_action(record, attrs, opts, now, action) else - tracker = Keyword.get(opts, :tracker, Tracker) + persist_and_transition_action(record, attrs, opts, now, action) + end + end - case tracker.update_issue_state(issue_id, @active_state) do - :ok -> - complete_transition_action(record, attrs, opts, now, action) + defp persist_and_transition_action(record, attrs, opts, now, action) do + pending_attrs = pending_transition_action_attrs(attrs, action, now) - {:error, reason} -> - record_transition_error(record, attrs, opts, now, action, reason) - end + case update_review(opts, record, pending_attrs) do + :ok -> + transition_persisted_action(record, pending_attrs, opts, now, action) + + {:error, reason} -> + {:state_transition_update_error, Map.get(record, :issue_id), action_atom(action), reason} + end + end + + defp transition_persisted_action(record, pending_attrs, opts, now, action) do + tracker = Keyword.get(opts, :tracker, Tracker) + issue_id = Map.get(record, :issue_id) + + case tracker.update_issue_state(issue_id, @active_state) do + :ok -> + complete_transition_action(record, pending_attrs, opts, now, action) + + {:error, reason} -> + record_transition_error(record, pending_attrs, opts, now, action, reason) end end @@ -1047,11 +1062,7 @@ defmodule SymphonyElixir.PrReviewPoller do end defp complete_transition_action(record, attrs, opts, now, action) do - case update_review( - opts, - record, - transition_action_attrs(record, attrs, action, now) - ) do + case update_review(opts, record, transition_success_attrs(record, attrs, action, now)) do :ok -> maybe_emit_reviewer_commented(record, attrs, action, now) {:state_transitioned, Map.get(record, :issue_id), action_atom(action), @active_state} @@ -1081,20 +1092,31 @@ defmodule SymphonyElixir.PrReviewPoller do end end - defp transition_action_attrs(record, attrs, action, now) do - attrs - |> Map.merge(%{ + defp pending_transition_action_attrs(attrs, action, now) do + Map.merge(attrs, %{ + status: "#{action}_transition_pending", + target_issue_state: @active_state, + last_action: nil, + last_action_at: nil, + error: nil, + updated_at: now + }) + end + + defp transition_success_attrs(record, attrs, action, now) do + %{ status: "#{action}_requested", target_issue_state: @active_state, + error: nil, last_action: action, last_action_at: now, updated_at: now - }) - |> maybe_mark_conflict_dispatched(record, action) + } + |> maybe_mark_conflict_dispatched(record, attrs, action) end - defp maybe_mark_conflict_dispatched(attrs, record, "conflict") do - conflict_key = Map.get(attrs, :last_conflict_key) + defp maybe_mark_conflict_dispatched(attrs, record, pending_attrs, "conflict") do + conflict_key = Map.get(pending_attrs, :last_conflict_key) attrs |> Map.put(:conflict_retry_count, conflict_retry_count(record) + 1) @@ -1102,7 +1124,7 @@ defmodule SymphonyElixir.PrReviewPoller do |> Map.put(:error, nil) end - defp maybe_mark_conflict_dispatched(attrs, _record, _action), do: attrs + defp maybe_mark_conflict_dispatched(attrs, _record, _pending_attrs, _action), do: attrs defp cleanup_review(record, opts, now, reason) do workspace = Keyword.get(opts, :workspace, Workspace) diff --git a/test/symphony_elixir/ci_poller_test.exs b/test/symphony_elixir/ci_poller_test.exs index 4a185e782c..26ce338115 100644 --- a/test/symphony_elixir/ci_poller_test.exs +++ b/test/symphony_elixir/ci_poller_test.exs @@ -667,7 +667,7 @@ defmodule SymphonyElixir.CiPollerTest do issue_identifier: issue.identifier, pr_url: List.first(issue.pr_urls), workspace_path: "/tmp/workspaces/ACME-2401", - status: "rework_requested", + status: "rework_transition_pending", pending_reviewer_comments: [%{id: "comment-1", body: "Please adjust."}], updated_at: now }) @@ -817,7 +817,7 @@ defmodule SymphonyElixir.CiPollerTest do assert CiPoller.log_excerpt_for_test(log, 5) == "ERROR: broken\n??\nlast" end - test "Linear transition failure leaves dispatch state recorded so the SHA is not redispatched" do + test "Linear transition failure clears dispatch marker so the SHA can be redispatched" do now = ~U[2026-05-06 09:00:00Z] issue = in_review_issue() Application.put_env(:symphony_elixir, :ci_test_issues, []) @@ -853,8 +853,8 @@ defmodule SymphonyElixir.CiPollerTest do assert [ %{ status: "state_transition_error", - ci_retry_count: 1, - dispatched_shas: ["abc123"], + ci_retry_count: 0, + dispatched_shas: [], ci_failure: %{commit_sha: "abc123"}, log_excerpt: log_excerpt } @@ -862,10 +862,11 @@ defmodule SymphonyElixir.CiPollerTest do assert is_binary(log_excerpt) and log_excerpt != "" - # A subsequent poll past the backoff window must not re-dispatch the same SHA. + # A subsequent poll past the backoff window should re-dispatch the same SHA + # because the previous Linear transition never landed. later = DateTime.add(poll_time, 2, :minute) - assert {:ok, %{actions: [{:already_handled, "issue-2401", "abc123"}]}} = + assert {:ok, %{actions: [{:state_transitioned, "issue-2401", :ci_failure, "In Progress"}]}} = CiPoller.poll_once( tracker: FakeTracker, github: FakeGitHub, @@ -873,7 +874,11 @@ defmodule SymphonyElixir.CiPollerTest do now: later ) - refute_receive {:issue_state_update, _, _} + assert_receive {:ci_failure_at_transition, "issue-2401", %{commit_sha: "abc123"}} + assert_receive {:issue_state_update, "issue-2401", "In Progress"} + + assert [%{status: "dispatch_requested", ci_retry_count: 1, dispatched_shas: ["abc123"]}] = + RunStore.list_ci_checks() end test "transient GitHub errors are recorded without dispatching" do @@ -928,9 +933,7 @@ defmodule SymphonyElixir.CiPollerTest do ] = RunStore.list_ci_checks() end - test "dispatched SHA is protected from escalation on the next poll" do - # Race-protection: after a dispatch lands, the next poll for the same SHA - # must not escalate (and kill the in-flight agent run). + test "dispatched SHA escalates after max retries once no agent is running" do now = ~U[2026-05-06 09:00:00Z] issue = in_review_issue() Application.put_env(:symphony_elixir, :ci_test_issues, [issue]) @@ -960,11 +963,97 @@ defmodule SymphonyElixir.CiPollerTest do updated_at: now }) + assert {:ok, %{actions: [{:escalated, "issue-2401", "In Review"}]}} = + CiPoller.poll_once(tracker: FakeTracker, github: FakeGitHub, now: DateTime.add(now, 1, :minute)) + + assert_receive {:issue_state_update, "issue-2401", "In Review"} + assert [%{status: "escalated", ci_retry_count: 1}] = RunStore.list_ci_checks() + end + + test "dispatched SHA is protected from escalation while an agent is running" do + now = ~U[2026-05-06 09:00:00Z] + issue = in_review_issue() + Application.put_env(:symphony_elixir, :ci_test_issues, [issue]) + Application.put_env(:symphony_elixir, :ci_test_status, failed_status("abc123")) + + write_workflow_file!(Workflow.workflow_file_path(), + tracker_kind: "memory", + pr_review_mode: "polling", + ci: %{enabled: true, log_excerpt_lines: 3, max_retries: 1, escalation_state: "In Review"} + ) + + put_run(issue, now, "running") + + assert :ok = + RunStore.put_ci_check(%{ + repo_key: @repo_key, + issue_id: issue.id, + issue_identifier: issue.identifier, + issue_url: issue.url, + pr_url: List.first(issue.pr_urls), + workspace_path: "/tmp/workspaces/ACME-2401", + status: "dispatch_requested", + ci_retry_count: 1, + rerun_attempted_shas: ["abc123"], + dispatched_shas: ["abc123"], + last_observed_sha: "abc123", + updated_at: now + }) + + assert {:ok, %{actions: [{:already_handled, "issue-2401", "abc123"}]}} = + CiPoller.poll_once(tracker: FakeTracker, github: FakeGitHub, now: DateTime.add(now, 1, :minute)) + + refute_receive {:issue_state_update, _, _} + assert [%{status: "failure_already_handled", ci_retry_count: 1}] = RunStore.list_ci_checks() + end + + test "freshly dispatched SHA is protected from escalation until the start-grace window lapses" do + now = ~U[2026-05-06 09:00:00Z] + issue = in_review_issue() + Application.put_env(:symphony_elixir, :ci_test_issues, [issue]) + Application.put_env(:symphony_elixir, :ci_test_status, failed_status("abc123")) + + write_workflow_file!(Workflow.workflow_file_path(), + tracker_kind: "memory", + pr_review_mode: "polling", + ci: %{enabled: true, log_excerpt_lines: 3, max_retries: 1, escalation_state: "In Review"} + ) + + # No running agent yet: the orchestrator has not picked up the In Progress + # issue, so the rework run is not "running" in the dispatch->running gap. + put_run(issue, now) + + assert :ok = + RunStore.put_ci_check(%{ + repo_key: @repo_key, + issue_id: issue.id, + issue_identifier: issue.identifier, + issue_url: issue.url, + pr_url: List.first(issue.pr_urls), + workspace_path: "/tmp/workspaces/ACME-2401", + status: "dispatch_requested", + ci_retry_count: 1, + rerun_attempted_shas: ["abc123"], + dispatched_shas: ["abc123"], + last_observed_sha: "abc123", + last_action: "dispatch", + last_action_at: now, + updated_at: now + }) + + # Within the grace window the just-dispatched agent must not be escalated. assert {:ok, %{actions: [{:already_handled, "issue-2401", "abc123"}]}} = CiPoller.poll_once(tracker: FakeTracker, github: FakeGitHub, now: DateTime.add(now, 1, :minute)) refute_receive {:issue_state_update, _, _} assert [%{status: "failure_already_handled", ci_retry_count: 1}] = RunStore.list_ci_checks() + + # Past the grace window with still no running agent, escalation proceeds. + assert {:ok, %{actions: [{:escalated, "issue-2401", "In Review"}]}} = + CiPoller.poll_once(tracker: FakeTracker, github: FakeGitHub, now: DateTime.add(now, 3, :minute)) + + assert_receive {:issue_state_update, "issue-2401", "In Review"} + assert [%{status: "escalated", ci_retry_count: 1}] = RunStore.list_ci_checks() end test "escalated status survives a transient poll error" do diff --git a/test/symphony_elixir/config_split_test.exs b/test/symphony_elixir/config_split_test.exs index 0bd9252698..ffe914de08 100644 --- a/test/symphony_elixir/config_split_test.exs +++ b/test/symphony_elixir/config_split_test.exs @@ -352,6 +352,18 @@ defmodule SymphonyElixir.ConfigSplitTest do assert duplicate_message =~ "repositories keys must be unique" + assert {:error, {:invalid_symphony_config, normalized_duplicate_message}} = + SystemSchema.parse( + system_config(%{ + "repositories" => [ + repo_config("api/main"), + repo_config("api_main") + ] + }) + ) + + assert normalized_duplicate_message =~ "repositories keys must not collide after workspace normalization" + assert {:error, {:invalid_symphony_config, default_message}} = SystemSchema.parse( system_config(%{ diff --git a/test/symphony_elixir/orchestrator_status_test.exs b/test/symphony_elixir/orchestrator_status_test.exs index 1f276b8b04..e22400a339 100644 --- a/test/symphony_elixir/orchestrator_status_test.exs +++ b/test/symphony_elixir/orchestrator_status_test.exs @@ -3802,10 +3802,13 @@ defmodule SymphonyElixir.OrchestratorStatusTest do send(pid, :run_poll_cycle) running_record = - wait_for_run_record(fn - %{issue_id: "issue-interrupted-run", status: "running"} -> true - _record -> false - end) + wait_for_run_record( + fn + %{issue_id: "issue-interrupted-run", status: "running"} -> true + _record -> false + end, + 2_000 + ) GenServer.stop(pid) terminate_task_supervisor_children() @@ -3832,6 +3835,88 @@ defmodule SymphonyElixir.OrchestratorStatusTest do end end + test "orchestrator startup marks interrupted runs for every configured repo" do + test_root = + Path.join( + System.tmp_dir!(), + "symphony-elixir-multi-repo-interrupted-run-recovery-#{System.unique_integer([:positive])}" + ) + + write_workflow_file!(Workflow.workflow_file_path(), + tracker_kind: "memory", + workspace_root: test_root, + poll_interval_ms: 60_000, + quality_gate: %{enabled: false}, + repos: [ + [key: "default", workflow: Workflow.workflow_file_path(), team: "Test"], + [key: "api", workflow: Workflow.workflow_file_path(), team: "API"] + ] + ) + + Application.put_env(:symphony_elixir, :memory_tracker_issues, []) + :ok = RunStore.clear() + + now = DateTime.utc_now() + + assert :ok = + RunStore.put_run(%{ + repo_key: "default", + run_id: "run-default-interrupted", + issue_id: "issue-default-interrupted-run", + issue_identifier: "MT-502", + title: "Default interrupted run", + status: "running", + started_at: now, + updated_at: now + }) + + assert :ok = + RunStore.put_run(%{ + repo_key: "api", + run_id: "run-api-interrupted", + issue_id: "issue-api-interrupted-run", + issue_identifier: "API-502", + title: "API interrupted run", + status: "running", + started_at: now, + updated_at: now + }) + + orchestrator_name = Module.concat(__MODULE__, :MultiRepoInterruptedRunRecoveryOrchestrator) + {:ok, pid} = Orchestrator.start_link(name: orchestrator_name) + + try do + default_recovered_record = + wait_for_run_record(fn + %{run_id: "run-default-interrupted", status: "failure", error: "orchestrator restarted before worker exit"} -> + true + + _record -> + false + end) + + assert default_recovered_record.issue_identifier == "MT-502" + assert %DateTime{} = default_recovered_record.ended_at + + api_recovered_record = + wait_for_run_record("api", fn + %{run_id: "run-api-interrupted", status: "failure", error: "orchestrator restarted before worker exit"} -> + true + + _record -> + false + end) + + assert api_recovered_record.issue_identifier == "API-502" + assert %DateTime{} = api_recovered_record.ended_at + + GenServer.stop(pid) + after + terminate_task_supervisor_children() + File.rm_rf(test_root) + end + end + test "orchestrator snapshot includes poll countdown and checking status" do orchestrator_name = Module.concat(__MODULE__, :PollingSnapshotOrchestrator) {:ok, pid} = Orchestrator.start_link(name: orchestrator_name) @@ -5752,9 +5837,21 @@ defmodule SymphonyElixir.OrchestratorStatusTest do end end - defp wait_for_run_record(predicate, timeout_ms \\ 500) when is_function(predicate, 1) do + defp wait_for_run_record(predicate) when is_function(predicate, 1) do + wait_for_run_record(predicate, 500) + end + + defp wait_for_run_record(predicate, timeout_ms) when is_function(predicate, 1) and is_integer(timeout_ms) do + wait_for_run_record(Config.repo_key!(), predicate, timeout_ms) + end + + defp wait_for_run_record(repo_key, predicate) when is_binary(repo_key) and is_function(predicate, 1) do + wait_for_run_record(repo_key, predicate, 500) + end + + defp wait_for_run_record(repo_key, predicate, timeout_ms) when is_binary(repo_key) and is_function(predicate, 1) do deadline_ms = System.monotonic_time(:millisecond) + timeout_ms - do_wait_for_run_record(predicate, deadline_ms) + do_wait_for_run_record(repo_key, predicate, deadline_ms) end defp running_entry_for_token_test(%Issue{} = issue, %DateTime{} = started_at) do @@ -5771,9 +5868,9 @@ defmodule SymphonyElixir.OrchestratorStatusTest do } end - defp do_wait_for_run_record(predicate, deadline_ms) do + defp do_wait_for_run_record(repo_key, predicate, deadline_ms) do record = - RunStore.list_runs(:all) + RunStore.list_runs(repo_key, :all) |> Enum.find(predicate) cond do @@ -5781,11 +5878,11 @@ defmodule SymphonyElixir.OrchestratorStatusTest do record System.monotonic_time(:millisecond) >= deadline_ms -> - flunk("timed out waiting for run store record: #{inspect(RunStore.list_runs(:all))}") + flunk("timed out waiting for run store record: #{inspect(RunStore.list_runs(repo_key, :all))}") true -> Process.sleep(5) - do_wait_for_run_record(predicate, deadline_ms) + do_wait_for_run_record(repo_key, predicate, deadline_ms) end end diff --git a/test/symphony_elixir/pr_review_poller_test.exs b/test/symphony_elixir/pr_review_poller_test.exs index 08fdd5ff4f..7261965c36 100644 --- a/test/symphony_elixir/pr_review_poller_test.exs +++ b/test/symphony_elixir/pr_review_poller_test.exs @@ -33,11 +33,22 @@ defmodule SymphonyElixir.PrReviewPollerTest do {:error, :linear_unavailable} else recipient = Application.fetch_env!(:symphony_elixir, :pr_review_test_recipient) + maybe_send_pending_context_at_transition(recipient, issue_id) send(recipient, {:issue_state_update, issue_id, state_name}) :ok end end + defp maybe_send_pending_context_at_transition(recipient, issue_id) do + if Application.get_env(:symphony_elixir, :pr_review_test_capture_pending_at_transition, false) do + comments = SymphonyElixir.PrReviewPoller.pending_reviewer_comments(issue_id) + conflict = SymphonyElixir.PrReviewPoller.pending_pr_conflict(issue_id) + + send(recipient, {:pending_reviewer_comments_at_transition, issue_id, comments}) + send(recipient, {:pending_pr_conflict_at_transition, issue_id, conflict}) + end + end + defp take_failure(key, value) do failures = Application.get_env(:symphony_elixir, key, []) @@ -324,6 +335,7 @@ defmodule SymphonyElixir.PrReviewPollerTest do Application.delete_env(:symphony_elixir, :pr_review_test_update_status_failures) Application.delete_env(:symphony_elixir, :pr_review_test_update_attr_failures) Application.delete_env(:symphony_elixir, :pr_review_test_state_update_failures) + Application.delete_env(:symphony_elixir, :pr_review_test_capture_pending_at_transition) Application.delete_env(:symphony_elixir, :pr_review_test_delete_failures) Application.delete_env(:symphony_elixir, :pr_review_test_github_error) Application.delete_env(:symphony_elixir, :pr_review_test_pause) @@ -588,6 +600,7 @@ defmodule SymphonyElixir.PrReviewPollerTest do latest_comment_at = DateTime.add(now, -31, :minute) Application.put_env(:symphony_elixir, :pr_review_test_issues, [in_review_issue(updated_at: now)]) + Application.put_env(:symphony_elixir, :pr_review_test_capture_pending_at_transition, true) :ok = put_review(now) Application.put_env( @@ -613,6 +626,9 @@ defmodule SymphonyElixir.PrReviewPollerTest do assert {:ok, %{actions: [{:state_transitioned, "issue-1780", :rework, "In Progress"}]}} = PrReviewPoller.poll_once(tracker: FakeTracker, github: FakeGitHub, now: now) + assert_receive {:pending_reviewer_comments_at_transition, "issue-1780", [%{id: "comment-1", author: "human-reviewer", body: "Please refactor this before merge."}]} + + assert_receive {:pending_pr_conflict_at_transition, "issue-1780", nil} assert_receive {:issue_state_update, "issue-1780", "In Progress"} assert [ @@ -1147,6 +1163,7 @@ defmodule SymphonyElixir.PrReviewPollerTest do now = ~U[2026-05-01 09:00:00Z] issue = in_review_issue(updated_at: now) Application.put_env(:symphony_elixir, :pr_review_test_issues, [issue]) + Application.put_env(:symphony_elixir, :pr_review_test_capture_pending_at_transition, true) :ok = put_review(now) Application.put_env( @@ -1165,6 +1182,17 @@ defmodule SymphonyElixir.PrReviewPollerTest do assert {:ok, %{actions: [{:state_transitioned, "issue-1780", :conflict, "In Progress"}]}} = PrReviewPoller.poll_once(tracker: FakeTracker, github: FakeGitHub, now: now) + assert_receive {:pending_reviewer_comments_at_transition, "issue-1780", []} + + assert_receive {:pending_pr_conflict_at_transition, "issue-1780", + %{ + head_ref: "auto/ACME-1780", + head_sha: "head-sha", + base_ref: "main", + base_sha: "base-sha", + conflict_key: "head-sha|base-sha" + }} + assert_receive {:issue_state_update, "issue-1780", "In Progress"} assert [ @@ -3281,7 +3309,7 @@ defmodule SymphonyElixir.PrReviewPollerTest do now = ~U[2026-05-01 09:00:00Z] Application.put_env(:symphony_elixir, :pr_review_test_issues, [in_review_issue(updated_at: now)]) Application.put_env(:symphony_elixir, :pr_review_test_activity, open_activity(now, review_decision: "APPROVED")) - Application.put_env(:symphony_elixir, :pr_review_test_update_status_failures, ["merge_requested"]) + Application.put_env(:symphony_elixir, :pr_review_test_update_status_failures, ["merge_transition_pending"]) Application.put_env(:symphony_elixir, :pr_review_test_review_records, %{ "issue-1780" => review_record(now) @@ -3304,12 +3332,12 @@ defmodule SymphonyElixir.PrReviewPollerTest do Process.cancel_timer(next_state.timer_ref) end) - assert_receive {:issue_state_update, "issue-1780", "In Progress"} + refute_receive {:issue_state_update, _, _}, 50 assert log =~ "PR review transition update error issue_id=issue-1780 action=merge" assert log =~ "update_pr_review_failed" end - test "does not report state transition success when final review update fails" do + test "failed pending action persistence prevents Linear transition" do now = ~U[2026-05-01 09:00:00Z] issue = in_review_issue(updated_at: now) Application.put_env(:symphony_elixir, :pr_review_test_issues, [issue]) @@ -3319,7 +3347,7 @@ defmodule SymphonyElixir.PrReviewPollerTest do issue.id => review_record(now) }) - Application.put_env(:symphony_elixir, :pr_review_test_update_status_failures, ["merge_requested"]) + Application.put_env(:symphony_elixir, :pr_review_test_update_status_failures, ["merge_transition_pending"]) assert {:ok, %{ @@ -3334,7 +3362,7 @@ defmodule SymphonyElixir.PrReviewPollerTest do now: now ) - assert_receive {:issue_state_update, "issue-1780", "In Progress"} + refute_receive {:issue_state_update, _, _}, 50 assert [ %{ @@ -3353,6 +3381,148 @@ defmodule SymphonyElixir.PrReviewPollerTest do assert_receive {:issue_state_update, "issue-1780", "In Progress"} end + test "failed transition success marker keeps pending reviewer comments durable and retryable" do + now = ~U[2026-05-01 09:00:00Z] + latest_comment_at = DateTime.add(now, -31, :minute) + issue = in_review_issue(updated_at: now) + Application.put_env(:symphony_elixir, :pr_review_test_issues, [issue]) + + Application.put_env( + :symphony_elixir, + :pr_review_test_activity, + open_activity(latest_comment_at, + comments: [ + %{ + id: "comment-1", + kind: "comment", + author: "human-reviewer", + body: "Please refactor this before merge.", + url: "https://github.com/example/repo/pull/1780#issuecomment-1", + created_at: latest_comment_at, + updated_at: latest_comment_at + } + ] + ) + ) + + Application.put_env(:symphony_elixir, :pr_review_test_review_records, %{ + issue.id => review_record(now) + }) + + Application.put_env(:symphony_elixir, :pr_review_test_update_status_failures, ["rework_requested"]) + + assert {:ok, + %{ + actions: [ + {:state_transition_update_error, "issue-1780", :rework, {:update_pr_review_failed, :disk_full}} + ] + }} = + PrReviewPoller.poll_once( + tracker: FakeTracker, + run_store: StatefulRunStore, + github: FakeGitHub, + now: now + ) + + assert_receive {:issue_state_update, "issue-1780", "In Progress"} + + assert [ + %{ + status: "rework_transition_pending", + last_action: nil, + pending_reviewer_comments: [%{id: "comment-1"}] + } = record + ] = StatefulRunStore.list_pr_reviews() + + assert Map.get(record, :last_action_at) == nil + + assert [%{id: "comment-1"}] = + PrReviewPoller.pending_reviewer_comments("issue-1780", run_store: StatefulRunStore) + + assert {:ok, %{actions: [{:state_transitioned, "issue-1780", :rework, "In Progress"}]}} = + PrReviewPoller.poll_once( + tracker: FakeTracker, + run_store: StatefulRunStore, + github: FakeGitHub, + now: DateTime.add(now, 5, :second) + ) + + assert_receive {:issue_state_update, "issue-1780", "In Progress"} + + assert [%{last_action: "rework", last_action_at: %DateTime{}, pending_reviewer_comments: [%{id: "comment-1"}]}] = + StatefulRunStore.list_pr_reviews() + end + + test "failed Linear transition keeps pending reviewer comments durable and retryable" do + now = ~U[2026-05-01 09:00:00Z] + latest_comment_at = DateTime.add(now, -31, :minute) + issue = in_review_issue(updated_at: now) + Application.put_env(:symphony_elixir, :pr_review_test_issues, [issue]) + + Application.put_env( + :symphony_elixir, + :pr_review_test_activity, + open_activity(latest_comment_at, + comments: [ + %{ + id: "comment-1", + kind: "comment", + author: "human-reviewer", + body: "Please refactor this before merge.", + url: "https://github.com/example/repo/pull/1780#issuecomment-1", + created_at: latest_comment_at, + updated_at: latest_comment_at + } + ] + ) + ) + + Application.put_env(:symphony_elixir, :pr_review_test_review_records, %{ + issue.id => review_record(now) + }) + + # Fail the Linear transition itself (after the pending context is persisted). + Application.put_env(:symphony_elixir, :pr_review_test_state_update_failures, ["issue-1780"]) + + assert {:ok, %{actions: [{:state_transition_error, "issue-1780", :rework, :linear_unavailable}]}} = + PrReviewPoller.poll_once( + tracker: FakeTracker, + run_store: StatefulRunStore, + github: FakeGitHub, + now: now + ) + + # The Linear transition never landed, so no issue state update is emitted. + refute_receive {:issue_state_update, _, _}, 50 + + assert [ + %{ + status: "state_transition_error", + error: ":linear_unavailable", + last_action: "rework", + last_action_at: nil, + pending_reviewer_comments: [%{id: "comment-1"}] + } + ] = StatefulRunStore.list_pr_reviews() + + assert [%{id: "comment-1"}] = + PrReviewPoller.pending_reviewer_comments("issue-1780", run_store: StatefulRunStore) + + # A subsequent poll retries the transition; with Linear recovered it lands. + assert {:ok, %{actions: [{:state_transitioned, "issue-1780", :rework, "In Progress"}]}} = + PrReviewPoller.poll_once( + tracker: FakeTracker, + run_store: StatefulRunStore, + github: FakeGitHub, + now: DateTime.add(now, 5, :second) + ) + + assert_receive {:issue_state_update, "issue-1780", "In Progress"} + + assert [%{last_action: "rework", last_action_at: %DateTime{}, pending_reviewer_comments: [%{id: "comment-1"}]}] = + StatefulRunStore.list_pr_reviews() + end + defp put_review(now, attrs \\ %{}) do now |> review_record(attrs)