fix: harden orchestration state recovery

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.

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`
 

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)
 

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} ->

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

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,28 +1092,39 @@ 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)
     |> Map.put(:dispatched_conflict_keys, append_string(Map.get(record, :dispatched_conflict_keys, []), conflict_key))
     |> 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)