fix(tasks): persist work-order waits without restarting runs

Dispatch receipts previously requeued active investigations and the final-only completion gate then failed waiting generations. Preserve the run, accept current interim wait reports, and wake only on actionable feedback or recovery evidence. Regression tests cover dispatch, same-generation reports, receipt revisions, cancellation and replacement ordering.

app/services/operational_tasks.py

@@ -560,6 +560,8 @@ class OperationalTaskService:
                 return OperationalTaskRun.model_validate(replay["run"])
             await self.repo.lock(f"run:{run_id}")
             run = await self._run(task_id, run_id)
+            require(run.get("failure_code") != "SUPERSEDED_BY_REQUEUE", "RUN_SUPERSEDED",
+                    "New task input superseded this run; the replacement run will continue", status=409)
             require(run["revision"] == command.expected_revision, "VERSION_CONFLICT",
                     "Run revision changed", status=409)
             require(run["status"] in {"claimed", "running"}, "INVALID_TRANSITION",
@@ -1439,6 +1441,7 @@ class OperationalTaskService:
 
         async with self.repo.transaction():
             await self.repo.lock(f"request:{scope}:{command.request_id}")
+            await self.repo.lock(f"task:{task_id}")
             await self.repo.lock(f"action:{action_id}")
             current = await self._action(task_id, action_id)
             require(current.version == action.version and current.execution_status == "unknown",
@@ -1474,8 +1477,12 @@ class OperationalTaskService:
                 )
             if outcome.status in {"succeeded", "failed", "unsupported"}:
                 task_document = await self._task(task_id)
-                track_work_order(task_document, _json(current))
-                await self._requeue_task(task_document, outcome_recorded_at)
+                tracked = track_work_order(task_document, _json(current))
+                if (tracked or task_document["status"] == "cancelled"
+                        or (task_document.get("source") or {}).get("deleted_at")):
+                    await self._keep_work_order_wait(task_document, outcome_recorded_at)
+                else:
+                    await self._requeue_task(task_document, outcome_recorded_at)
                 await self.repo.update_task(
                     _task_json(OperationalTaskSummary.model_validate(task_document))
                 )
@@ -1511,6 +1518,7 @@ class OperationalTaskService:
             replay = await self._request_replay(scope, command.request_id, body)
             if replay:
                 return HumanApprovalAction.model_validate(replay["action"])
+            await self.repo.lock(f"task:{task_id}")
             await self.repo.lock(f"action:{action_id}")
             current = await self._action(task_id, action_id)
             require(current.version == action.version, "VERSION_CONFLICT",
@@ -1548,8 +1556,12 @@ class OperationalTaskService:
                 )
             if outcome.status in {"succeeded", "failed", "unsupported"}:
                 task_document = await self._task(task_id)
-                track_work_order(task_document, _json(current))
-                await self._requeue_task(task_document, outcome_recorded_at)
+                tracked = track_work_order(task_document, _json(current))
+                if (tracked or task_document["status"] == "cancelled"
+                        or (task_document.get("source") or {}).get("deleted_at")):
+                    await self._keep_work_order_wait(task_document, outcome_recorded_at)
+                else:
+                    await self._requeue_task(task_document, outcome_recorded_at)
                 await self.repo.update_task(
                     _task_json(OperationalTaskSummary.model_validate(task_document))
                 )
@@ -1653,8 +1665,8 @@ class OperationalTaskService:
                             "EVIDENCE_CONFLICT", "Evidence ids are immutable", status=409)
                 else:
                     await self.repo.insert_evidence(str(task_id), _json(evidence))
-            self._apply_feedback_status(task, command.status)
             if task["task_type"] == "external_work_order":
+                self._apply_feedback_status(task, command.status)
                 task["source"] = {
                     **task["source"],
                     "external_revision": command.source_revision,
@@ -1674,10 +1686,27 @@ class OperationalTaskService:
                 # Recover older tasks whose successful dispatch predates tracking.
                 for item in await self.repo.list_actions(str(task_id)):
                     track_work_order(task, item)
-                accept_completion(task, body)
+                linked_work_order = accept_completion(task, body, recorded_at=now)
                 if matching_source:
                     task["source"] = {**task["source"], "external_revision": command.source_revision}
-                await self._requeue_task(task, now)
+                feedback_order = (task.get("handling_context") or {}).get("work_orders", {}).get(
+                    command.source_event_id.removeprefix("peripheral_sample:"), {})
+                if linked_work_order and (feedback_order.get("superseded_by")
+                                         or task["handling_context"].get("phase") == "completed"):
+                    # A receipt edit with the same completion time does not
+                    # reopen an incident that already passed recovery review.
+                    task.update(revision=task["revision"] + 1, updated_at=now)
+                elif (linked_work_order and command.status not in {"failed", "cancelled"}
+                        and task["handling_context"].get("phase") in {"waiting_work_order", "waiting_scada"}):
+                    # Progress is persisted, not a reason to restart analysis.
+                    # Completed field work waits for causal SCADA evidence;
+                    # reconcile_recoveries wakes the same task when it is ready.
+                    await self._keep_work_order_wait(task, now)
+                elif (linked_work_order and task["handling_context"].get("phase") == "verification_ready"
+                      and await self.repo.get_active_run(str(task_id)) is not None):
+                    await self._keep_work_order_wait(task, now)
+                else:
+                    await self._requeue_task(task, now)
             else:
                 task.update(revision=task["revision"] + 1, updated_at=now)
             await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
@@ -2099,8 +2128,22 @@ class OperationalTaskService:
                         and plan.source_session_id == command.source_session_id
                     ))
                 )
+                actions = [HumanApprovalAction.model_validate(item)
+                           for item in await self.repo.list_actions(str(task_id))]
+                work_order_wait = task["task_type"] == "event_investigation" and (
+                    (task.get("handling_context") or {}).get("phase") in {"waiting_work_order", "waiting_scada"}
+                    or any(action.kind == "create_work_order" and (
+                        action.approval_status == "pending" or (
+                            action.approval_status == "approved"
+                            and action.execution_status in {"not_submitted", "submitted", "unknown"}
+                        )) for action in actions)
+                )
                 if pending_question:
                     lifecycle = "waiting_human"
+                elif same_generation and work_order_wait and report.report.stage == "interim":
+                    # Ending a run is distinct from closing an incident. Keep
+                    # the current stage report and release the execution lease.
+                    lifecycle = "waiting_human"
                 elif not same_generation or report.report.stage != "final":
                     final_outcome = "failed"
                     lifecycle = "failed"
@@ -2110,10 +2153,6 @@ class OperationalTaskService:
                     )
                 else:
                     await self._verify_scada_report_presentation(task, run, _json(report.report), now)
-                    actions = [
-                        HumanApprovalAction.model_validate(item)
-                        for item in await self.repo.list_actions(str(task_id))
-                    ]
                     access_requests = [
                         TaskAccessRequest.model_validate(item)
                         for item in await self._list_access_requests(task_id)
@@ -2177,7 +2216,7 @@ class OperationalTaskService:
                     now,
                 )
             elif (lifecycle == "waiting_human" and not pending_question
-                  and (task.get("handling_context") or {}).get("phase") != "waiting_scada"):
+                  and (task.get("handling_context") or {}).get("phase") not in {"waiting_work_order", "waiting_scada"}):
                 pending_actions = any(
                     action.approval_status == "pending" for action in actions
                 )
@@ -2411,6 +2450,21 @@ class OperationalTaskService:
         }[status]
         task["agent_lease_expires_at"] = None
 
+    async def _keep_work_order_wait(self, task: dict, now: datetime) -> None:
+        """Persist a field wait without revoking a run or allocating a generation."""
+        if task["status"] != "cancelled" and not (task.get("source") or {}).get("deleted_at"):
+            active = await self.repo.get_active_run(str(task["task_id"]))
+            if active is not None:
+                task.update(
+                    lifecycle_status="open", status="queued" if active["status"] == "queued" else "running",
+                    agent_session_id=active.get("agent_session_id"),
+                    agent_lease_expires_at=active.get("lease_expires_at"),
+                )
+            else:
+                task.update(lifecycle_status="waiting_human", status="waiting_human",
+                            agent_session_id=None, agent_lease_expires_at=None)
+        task.update(revision=task["revision"] + 1, updated_at=now)
+
     async def _requeue_task(self, task: dict, now: datetime) -> None:
         await task_questions.cancel_pending(self, UUID(str(task["task_id"])), now)
         active_run = await self.repo.get_active_run(str(task["task_id"]))