fix(orchestrator): preserve slice/task runtime state across contract re-populate (#2908)

orchestrator/routes/pipelines.py

@@ -285,6 +285,7 @@ def get_repo_checks(repo: str) -> list[dict[str, str]]:  # type: ignore[misc]
 if TYPE_CHECKING:
     from egg_container import MountSpec
     from egg_contracts.agent_roles import AgentRole as ContractAgentRole
+    from egg_contracts.models import Slice as ContractSlice
 
     try:
         from ..container_spawner import ContainerSpawner
@@ -19620,6 +19621,103 @@ def _populate_contract_from_plan_safe(
         return PopulateResult(PopulateOutcome.UNEXPECTED_EXCEPTION)
 
 
+def _merge_preserved_slice_runtime(
+    new_slices: "list[ContractSlice]",  # noqa: UP037
+    old_slices: "list[ContractSlice]",  # noqa: UP037
+) -> None:
+    """Carry runtime slice/task state from ``old_slices`` onto ``new_slices`` in place.
+
+    ``_populate_contract_from_plan`` re-parses the plan markdown into a
+    fresh set of slices on every call — and its safety-net caller fires
+    on *every* ``start_phase=implement`` restart (deliberately outside
+    the ``contract_synced`` guard). The plan is the source of truth for
+    slice/task STRUCTURE (names, descriptions, dependencies, acceptance
+    criteria); it always parses back as ``PENDING`` with the runtime
+    bookkeeping fields unset. Blindly assigning ``contract.slices =
+    <freshly parsed>`` therefore wipes every slice the slice loop had
+    already advanced — resetting COMPLETE slices to PENDING and dropping
+    the ``parent_branch_at_creation`` / ``integration_base_sha`` a real
+    run stamped — so a restarted pipeline re-runs slice-1 forever and can
+    never reach slice-2 (#2908).
+
+    Mirroring the PR-metadata preservation a few lines down in the
+    caller, this merges by slice id (and by task id within a slice): the
+    plan supplies STRUCTURE while RUNTIME state survives a re-populate.
+    Unmatched ids (a re-plan that adds or removes slices/tasks) simply
+    keep the plan's fresh ``PENDING`` defaults.
+
+    Task-level runtime fields covered (each is durably written by a
+    runtime path that the plan parser cannot reconstruct):
+
+    - ``status``, ``commit``, ``checkpoint_id``, ``review_cycles``,
+      ``escalated``, ``gaps`` — slice-loop / reviewer / tester
+      bookkeeping.
+    - ``role`` + ``delegation_attempts`` — paired SYSTEM-owned
+      impasse-delegation state. ``impasse_routing.py`` flips
+      ``task.role`` to the suggested alternative and bumps
+      ``delegation_attempts`` in the same ``apply_mutation`` cycle
+      under ``Role.SYSTEM`` (only SYSTEM owns these two fields); the
+      slice-loop dispatcher then routes the task to the new role.
+      Preserving the counter without the role would re-spawn the
+      original producer on restart and trip ``DELEGATION_LIMIT`` on
+      the next impasse, escalating to HITL even though no delegation
+      visibly happened — so both fields must survive together.
+    - ``notes`` — APPLIER writes Won't-Do drain failure reasons here
+      (``pipelines.py`` Won't-Do path) and agents write implementation
+      narrative via ``mcp__task__update_notes`` / ``egg-contract
+      update-notes``; the plan parser always emits ``""``.
+    - ``jira_action_status`` — APPLIER advances ``pending`` →
+      ``in_flight`` → ``applied``/``failed`` (#1557 risk_analyst R7);
+      idempotency depends on ``applied`` surviving re-populate so the
+      next apply skips it instead of re-creating the Jira issue.
+    - ``jira_key`` — APPLIER writes the freshly-allocated key back after
+      a ``create`` action so re-runs skip the create; plan parser emits
+      ``None`` on ``create`` actions, so re-populate would otherwise
+      strand the applier into creating duplicate tickets.
+    """
+    old_by_id = {s.id: s for s in (old_slices or [])}
+    for new_slice in new_slices:
+        old_slice = old_by_id.get(new_slice.id)
+        if old_slice is None:
+            continue
+        # Slice-level runtime state stamped by ``_run_one_slice_inner``
+        # and the bootstrap reconciler — never re-derivable from the plan.
+        new_slice.status = old_slice.status
+        new_slice.parent_branch_at_creation = old_slice.parent_branch_at_creation
+        new_slice.integration_base_sha = old_slice.integration_base_sha
+        new_slice.commit = old_slice.commit
+        new_slice.review_cycles = old_slice.review_cycles
+        # Defensive copy so post-merge mutations of the discarded ``old``
+        # contract don't alias-leak into the live ``new`` contract.
+        new_slice.review_feedback = list(old_slice.review_feedback)
+        new_slice.escalated = old_slice.escalated
+        new_slice.escalation_reason = old_slice.escalation_reason
+        # Task-level runtime state: match by task id so a re-plan that
+        # adds/removes tasks still preserves completion of the survivors.
+        old_tasks_by_id = {t.id: t for t in old_slice.tasks}
+        for new_task in new_slice.tasks:
+            old_task = old_tasks_by_id.get(new_task.id)
+            if old_task is None:
+                continue
+            new_task.status = old_task.status
+            new_task.commit = old_task.commit
+            new_task.checkpoint_id = old_task.checkpoint_id
+            new_task.review_cycles = old_task.review_cycles
+            new_task.escalated = old_task.escalated
+            # Paired SYSTEM-owned impasse-delegation state — preserving
+            # the counter without the role would silently undo the
+            # delegation on restart (see docstring).
+            new_task.role = old_task.role
+            new_task.delegation_attempts = old_task.delegation_attempts
+            new_task.gaps = list(old_task.gaps)
+            # Runtime narrative + applier idempotency anchors. The
+            # plan parser cannot reconstruct any of these — see the
+            # docstring for the per-field invariants.
+            new_task.notes = old_task.notes
+            new_task.jira_action_status = old_task.jira_action_status
+            new_task.jira_key = old_task.jira_key
+
+
 def _populate_contract_from_plan(
     repo_path: Path,
     pipeline_id: str,
@@ -19780,6 +19878,11 @@ def _populate_contract_from_plan(
                 # ``plan_review_feedback`` stash above is the durable
                 # signal the reviewer prompt picks up either way.
                 raise ForestValidationError("slice DAG is not a forest", errors=forest_errors)
+            # Preserve runtime slice/task progress across re-populates so
+            # the safety-net populator (which fires on every
+            # ``start_phase=implement`` restart) cannot reset COMPLETE
+            # slices to PENDING and strand the pipeline on slice-1 (#2908).
+            _merge_preserved_slice_runtime(contract_slices, contract.slices)
             contract.slices = contract_slices
             changed = True
 

orchestrator/tests/test_short_flow_contract_population.py

@@ -228,6 +228,193 @@ def test_populate_contract_from_plan_preserves_deferred_actions(self, tmp_path:
         # And the planner-emitted fields are still refreshed from the plan.
         assert contract_after.pr.title == "Add retry logic to API client"
 
+    def test_populate_contract_from_plan_preserves_slice_and_task_runtime(self, tmp_path: Path):
+        """A re-populate must preserve runtime slice/task progress (#2908).
+
+        The safety-net populator fires on every ``start_phase=implement``
+        restart and rebuilds ``contract.slices`` wholesale from the plan
+        (every slice/task parses back as PENDING). A prior version blindly
+        assigned the freshly parsed slices, resetting COMPLETE slices to
+        PENDING and dropping the ``parent_branch_at_creation`` /
+        ``integration_base_sha`` a real slice run stamped — so a restarted
+        pipeline re-ran slice-1 forever and could never reach slice-2. The
+        populator now merges by slice id / task id: STRUCTURE comes from
+        the plan, RUNTIME state survives the re-build.
+
+        This test stamps every preserved field on the slice and the task
+        to a distinctive value, so a future field-list typo in the merge
+        helper regresses one of the asserts here instead of silently
+        wiping runtime state.
+        """
+        from datetime import UTC, datetime
+
+        from egg_contracts.loader import create_contract, load_contract, save_contract
+        from egg_contracts.models import ReviewFeedback, SliceStatus, TaskGap, TaskStatus
+        from routes.pipelines import _populate_contract_from_plan
+
+        pipeline_id = "pipeline-slice-runtime-preserve"
+        create_contract(pipeline_id=pipeline_id, title="Test", repo_root=tmp_path)
+
+        drafts_dir = tmp_path / ".egg-state" / "drafts"
+        drafts_dir.mkdir(parents=True, exist_ok=True)
+        (drafts_dir / f"{pipeline_id}-plan.md").write_text(SAMPLE_PLAN)
+
+        # First populate — establishes slice-1 with task-1-1/task-1-2.
+        _populate_contract_from_plan(tmp_path, pipeline_id, "local")
+
+        # Stamp every preserved runtime field with a distinctive value
+        # so the assert block below pins each one independently.
+        contract = load_contract(pipeline_id, tmp_path)
+        sl = contract.slices[0]
+        assert sl.id == "slice-1"
+        # Slice-level fields the merge helper claims to preserve.
+        sl.status = SliceStatus.COMPLETE
+        sl.parent_branch_at_creation = "egg/issue-2908/work"
+        sl.integration_base_sha = "0" * 40
+        sl.commit = "a" * 40
+        sl.review_cycles = 2
+        sl.review_feedback = [
+            ReviewFeedback(
+                timestamp=datetime(2026, 1, 15, 12, 0, tzinfo=UTC),
+                task_id="task-1-1",
+                feedback="needs-changes-on-first-pass",
+                status=TaskStatus.INCOMPLETE,
+            ),
+        ]
+        sl.escalated = True
+        sl.escalation_reason = "max-cycles-exceeded"
+        # Task-level fields the merge helper claims to preserve.
+        t = sl.tasks[0]
+        t.status = TaskStatus.COMPLETE
+        t.commit = "1" * 40
+        t.checkpoint_id = "ckpt-deadbeef"
+        t.review_cycles = 3
+        t.escalated = True
+        # Paired SYSTEM-owned impasse-delegation state. ``SAMPLE_PLAN``
+        # doesn't specify a ``role:`` for task-1-1, so the plan parser
+        # emits ``role=None`` (the slice-loop dispatcher at
+        # ``pipelines.py:6045-6052`` consumes ``None`` as coder for
+        # filtering); a successful delegation would flip the durable
+        # value to ``"tester"`` and bump the counter — both must
+        # survive a re-populate together or the slice-loop dispatcher
+        # will re-spawn the original producer on restart.
+        t.role = "tester"
+        t.delegation_attempts = 1
+        t.gaps = [
+            TaskGap(
+                id="gap-1",
+                from_role="tester",
+                to_role="coder",
+                description="missing edge case for empty input",
+            ),
+        ]
+        # Runtime narrative + applier idempotency anchors (blocking
+        # items 1-3 from the #2923 review).
+        t.notes = "wontdo drain failed: gateway 502 on first apply"
+        t.jira_action_status = "applied"
+        t.jira_key = "ENG-7421"
+        save_contract(contract, tmp_path)
+
+        # Re-run the populator (start_phase=implement restart re-entry).
+        _populate_contract_from_plan(tmp_path, pipeline_id, "local")
+
+        after = load_contract(pipeline_id, tmp_path)
+        sl_after = after.slices[0]
+        # Slice-level preservation — one assert per preserved field.
+        assert sl_after.status == SliceStatus.COMPLETE
+        assert sl_after.parent_branch_at_creation == "egg/issue-2908/work"
+        assert sl_after.integration_base_sha == "0" * 40
+        assert sl_after.commit == "a" * 40
+        assert sl_after.review_cycles == 2
+        assert len(sl_after.review_feedback) == 1
+        assert sl_after.review_feedback[0].feedback == "needs-changes-on-first-pass"
+        assert sl_after.escalated is True
+        assert sl_after.escalation_reason == "max-cycles-exceeded"
+        # Task-level preservation — one assert per preserved field.
+        t_after = sl_after.tasks[0]
+        assert t_after.status == TaskStatus.COMPLETE
+        assert t_after.commit == "1" * 40
+        assert t_after.checkpoint_id == "ckpt-deadbeef"
+        assert t_after.review_cycles == 3
+        assert t_after.escalated is True
+        # Paired SYSTEM-owned impasse-delegation state survives together.
+        assert t_after.role == "tester"
+        assert t_after.delegation_attempts == 1
+        assert len(t_after.gaps) == 1
+        assert t_after.gaps[0].id == "gap-1"
+        # Notes + applier anchors — the #2923 blocking-item additions.
+        assert t_after.notes == "wontdo drain failed: gateway 502 on first apply"
+        assert t_after.jira_action_status == "applied"
+        assert t_after.jira_key == "ENG-7421"
+        # The untouched task keeps the plan's fresh PENDING default.
+        assert sl_after.tasks[1].status == TaskStatus.PENDING
+        # And STRUCTURE is still refreshed from the plan.
+        assert sl_after.tasks[0].id == "task-1-1"
+        assert "retry_with_backoff" in sl_after.tasks[0].description
+
+    def test_populate_contract_from_plan_handles_unmatched_slices(self, tmp_path: Path):
+        """Unmatched slice ids (re-plan adds/removes a slice) keep plan defaults.
+
+        Documents the merge helper's behaviour on the unmatched-slice
+        case alongside the matched-slice case covered above:
+
+        - A slice present in the OLD contract but absent from the new
+          plan (re-plan dropped it) simply doesn't appear post-merge.
+        - A slice present in the new plan but absent from the OLD
+          contract (re-plan added it) keeps the plan's fresh PENDING
+          defaults — the merge helper does not invent runtime state.
+
+        The current ``SAMPLE_PLAN`` emits a single slice, so we simulate
+        the dropped-slice case by stamping a fake old slice that the new
+        parse will not produce, and the added-slice case by relying on
+        the empty initial contract → populated post-plan slice path.
+        """
+        from egg_contracts.loader import create_contract, load_contract, save_contract
+        from egg_contracts.models import Slice, SliceStatus
+        from routes.pipelines import _populate_contract_from_plan
+
+        pipeline_id = "pipeline-slice-runtime-unmatched"
+        create_contract(pipeline_id=pipeline_id, title="Test", repo_root=tmp_path)
+
+        drafts_dir = tmp_path / ".egg-state" / "drafts"
+        drafts_dir.mkdir(parents=True, exist_ok=True)
+        (drafts_dir / f"{pipeline_id}-plan.md").write_text(SAMPLE_PLAN)
+
+        # Inject an "orphan" slice into the OLD contract — the new
+        # plan parse won't reproduce it, so the merge helper must
+        # tolerate the unmatched-old-id case and drop it cleanly.
+        contract = load_contract(pipeline_id, tmp_path)
+        contract.slices = [
+            Slice(
+                id="slice-99",
+                name="Orphan slice",
+                tasks=[],
+                status=SliceStatus.COMPLETE,
+            ),
+        ]
+        save_contract(contract, tmp_path)
+
+        # Populate now sees a fresh plan with slice-1 and an OLD
+        # contract that contains only slice-99 — the unmatched-new-id
+        # case (slice-1 has no OLD twin) and the unmatched-old-id case
+        # (slice-99 has no NEW twin) both fire.
+        _populate_contract_from_plan(tmp_path, pipeline_id, "local")
+
+        after = load_contract(pipeline_id, tmp_path)
+        slice_ids = [s.id for s in after.slices]
+        # The orphan OLD slice does not survive — the plan is the
+        # structural source of truth.
+        assert "slice-99" not in slice_ids
+        # The new slice exists with the plan's fresh PENDING default —
+        # the helper does not invent runtime state for a slice it
+        # never saw in the old contract.
+        assert "slice-1" in slice_ids
+        sl_new = after.get_slice("slice-1")
+        assert sl_new is not None
+        assert sl_new.status == SliceStatus.PENDING
+        assert sl_new.parent_branch_at_creation is None
+        assert sl_new.integration_base_sha is None
+
 
 class TestEnsureStatefilesRestoresPRMetadata:
     """_ensure_statefiles_on_branch re-populates PR metadata from plan draft.