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_recovery.py

@@ -19,36 +19,53 @@ if TYPE_CHECKING:
 RECOVERY_WORKFLOWS = {"burst_report", "scada_anomaly"}
 
 
-def track_work_order(task: dict, action: dict) -> None:
+def track_work_order(task: dict, action: dict) -> bool:
     if task.get("workflow") not in RECOVERY_WORKFLOWS or action["kind"] != "create_work_order":
-        return
+        return False
     receipt = action.get("receipt") or {}
     if action.get("execution_status") != "succeeded" or receipt.get("adapter") != "peripheral_sample":
-        return
+        return False
     context = task.setdefault("handling_context", {})
     orders = context.setdefault("work_orders", {})
     order_id = receipt["receipt_id"]
     if order_id not in orders:
+        proposed_at = action.get("proposed_at")
+        for prior in orders.values():
+            terminal_at = prior.get("terminal_recorded_at")
+            # Only a new decision after the failed field attempt replaces it.
+            # A delayed receipt for an older independent dispatch does not.
+            if (not prior.get("superseded_by") and prior["status"] in {"failed", "cancelled"}
+                    and proposed_at and terminal_at
+                    and datetime.fromisoformat(proposed_at) > datetime.fromisoformat(terminal_at)):
+                prior["superseded_by"] = order_id
         orders[order_id] = {"status": "awaiting_completion", "action_id": str(action["action_id"])}
         context.update(phase="waiting_work_order", title="等待现场工单返回")
         context.pop("verification_evidence_id", None)
+    return True
 
 
-def accept_completion(task: dict, feedback: dict) -> bool:
+def accept_completion(task: dict, feedback: dict, *, recorded_at: datetime | None = None) -> bool:
     context = task.get("handling_context") or {}
     order_id = feedback["source_event_id"].removeprefix("peripheral_sample:")
     order = context.get("work_orders", {}).get(order_id)
     if order is None:
         return False
     order.update(status=feedback["status"], source_revision=feedback["source_revision"])
+    if feedback["status"] in {"failed", "cancelled"}:
+        if recorded_at is not None and not order.get("terminal_recorded_at"):
+            order["terminal_recorded_at"] = recorded_at.isoformat()
+    else:
+        order.pop("terminal_recorded_at", None)
     # The synchronizer preserves the field activity's business time on ordinary
     # edits. A revised field receipt can explicitly correct the verification start.
     if feedback["status"] == "completed":
         order["returned_at"] = feedback["occurred_at"]
     else:
         order.pop("returned_at", None)
-    orders = list(context["work_orders"].values())
-    if all(item["status"] == "completed" for item in orders):
+    if order.get("superseded_by"):
+        return True
+    orders = [item for item in context["work_orders"].values() if not item.get("superseded_by")]
+    if orders and all(item["status"] == "completed" for item in orders):
         returned_at = max((item["returned_at"] for item in orders), key=datetime.fromisoformat)
         previous = context.get("returned_at")
         if (previous is None or datetime.fromisoformat(previous) != datetime.fromisoformat(returned_at)

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,7 +1477,11 @@ class OperationalTaskService:
                 )
             if outcome.status in {"succeeded", "failed", "unsupported"}:
                 task_document = await self._task(task_id)
-                track_work_order(task_document, _json(current))
+                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,7 +1556,11 @@ class OperationalTaskService:
                 )
             if outcome.status in {"succeeded", "failed", "unsupported"}:
                 task_document = await self._task(task_id)
-                track_work_order(task_document, _json(current))
+                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,9 +1686,26 @@ 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}
+                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)
@@ -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"]))

resources/work-order-skills/burst-event-handling/SKILL.md

@@ -2,13 +2,15 @@
 name: burst-event-handling
 description: 根据爆管上报工单处理事件，完成资产关联、隔离方案、影响说明、处置建议与反馈复核。
 metadata:
-  version: "1.1.2"
+  version: "1.1.3"
 ---
 
 # 爆管事件处理
 
 适用于工单触发的 `workflow=burst_report`。以工单上报的事件位置和资产线索开展处置，依次完成下列步骤；每一步给出实际结果或具体待补项，已有证据可直接复用。SCADA 异常发现使用独立的 scada-anomaly 流程。
 
+先按handling_context接续：等待阶段按第6节保存本轮阶段报告；verification_ready复用已有异常分析与资产、隔离证据，仅补反馈和恢复复核，不重跑初次调查步骤。真正新增的独立事实另行核查。
+
 正文、表格、图表及引用时间统一按Asia/Shanghai（UTC+8）阅读。工具返回的+08:00时间保持原值；UTC来源转换后同步更新日期与窗口，例如2026-07-16T00:00:00Z对应7月16日08:00。结构化查询与sample_times保留明确时区，跨日按转换后的日期展示。
 
 ## 1. 接收与核对
@@ -42,12 +44,14 @@ metadata:
 
 先保存 interim/concluded 阶段报告，再按需要提出核验、通知、供水保障、隔离与抢修、恢复复测等动作草稿。关联现场工作采用 work_order_type=general，保留原爆管上报工单作为事件来源。草稿标题写“地点＋动作”，正文以三条概括任务、执行条件和回执要求；审批与实际执行按 Server 流程完成。
 
-报告固定给出事件概况、资产关联、隔离方案、影响说明、处置建议和待办。摘要以150字以内为目标，指标和资产明细放表格，证据展开保留来源。待办应指向具体事实、参数或动作。已完成本轮可执行分析且依赖结果已读取时发布 final，现场处理进展由后续反馈接续。
+报告固定给出事件概况、资产关联、隔离方案、影响说明、处置建议和待办。摘要以150字以内为目标，指标和资产明细放表格，证据展开保留来源。待办应指向具体事实、参数或动作。本轮可执行分析完成且依赖结果已读取后，若仍待审批、现场工单或恢复监测，按第6节以interim结束本轮，恢复闭环后发布final；无此依赖时按后台调查停止条件发布final。
 
 ## 6. 反馈与闭环复核
 
 读取摘要中的 handling_context。waiting_work_order 表示已派单、等待有效完工反馈，创建回执的 completed_at 只证明创建动作完成。现场处理和恢复供水以关联工单的 completed 反馈为准；returned_at 是现场反馈的业务时间，演示时不得换成电脑当前时间。
 
+草稿待审批、waiting_work_order或waiting_scada且当前无独立可执行工作时，保存本轮interim/concluded阶段报告并关联真实action_id，然后结束本轮执行，任务保持等待并释放租约。不能只更新progress或沿用旧轮报告，不循环查询工单或重复异常分析。创建成功及普通进度更新不启动新一轮；工单失败或取消时重新决策，完工后由Server等待恢复证据就绪，再唤醒同一任务。
+
 waiting_scada 表示恢复供水和监测复核阶段，以 returned_at 为反馈时刻 T，读取严格晚于 T 的最近一个 SCADA 采样时刻作为复核来源，说明测点、实际采样时间、读数、单位和正常范围；不等待额外周期，不使用 T 之前、T 当刻或更晚的正常数据替代。后台按事件前15天同期正常包络核验该时刻。最近时刻缺测或异常时保留实际问题，不宣称复核成功。保存本轮阶段成果，等待后续数据接续，不反复派发同一工单。
 
 verification_ready 时读取 verification_evidence_id 对应的证据与快照，核对工单反馈、T、最近采样时刻和测点，展示该时刻读数与正常范围；具备真实多时刻数据时再展示曲线，工单 completed 且恢复证据 status=passed 时发布 final/concluded 恢复闭环报告并引用该证据。Server 完成运行后关闭任务并解除关联告警，phase=completed 表示已闭环。照片、阀位或复测记录未提供时仅记为资料缺项，不增设关闭前置条件。来源系统若另有明确行政结案流程，按其实际状态另记，不推断为当前任务尚未闭环。

resources/work-order-skills/scada-anomaly/SKILL.md

@@ -2,13 +2,15 @@
 name: scada-anomaly
 description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。
 metadata:
-  version: "1.0.15"
+  version: "1.0.16"
 ---
 
 # SCADA 数据异常调查
 
 读取任务与 replay_context，核对设备、单位、异常时刻、knowledge_cutoff 和 batch_sequence。使用本轮已提交的 SCADA 观测开展调查，同一持续事件沿用已有调查。证据来源字段用于记录数据取得方式，不构成停止分析的条件。
 
+先按handling_context接续：待审批/工单/监测且无独立工作时，保存本轮interim/concluded及action_id后退出，不轮询。verification_ready复用原异常证据，只补恢复复核；初次分析按下文。
+
 报告与图表的阅读时区统一为Asia/Shanghai（UTC+8）。工具返回的+08:00时间按原值使用；Z时间换算后同步更新日期、历史系列名和横轴标签，例如2026-07-16T00:00:00Z对应7月16日08:00。sample_times记录相同真实时间点并使用+08:00偏移，查询与冻结边界按带时区的时间点比较。图表标题、横轴和系列名只写日期、时刻与单位，不写“上海时间”“Asia/Shanghai”或“UTC+8”；横轴命名为“时间”。
 
 ## 数据核验与诊断
@@ -17,7 +19,7 @@ metadata:
 
 通过 query_scada_readings 省略device_ids读取冻结异常窗内全部SCADA设备的完整属性和授权观测，覆盖压力、流量、水质等全部类型；正常对照也采用全量读取。Server自动读完所有设备页和观测页，完整返回devices、readings、coverage，结果仍受8MiB整份上限约束，超限明确报错后按时间分段，不能丢弃设备。query_complete表示查询已读完，缺测与无有效值另行判断。查询结束时间止于knowledge_cutoff，数据批次止于batch_sequence；冻结范围外设备单列，不回退实时数据。按需核查开始前15天基线，复用已有完整监测回执。结合monitored_value、cleaned_value和质量标记判断变化，可靠采集的低压与水质变化保留为业务异常。
 
-全量响应中的devices包含完整目录属性，readings包含实际读数，coverage列出全部查询设备、各设备读数数量、缺测和冻结范围外设备。不能用代表设备推断全网。单独读取list_scada_devices时，默认每页100个，持续翻页到page.has_more=false；属性缩略不等于已读取完整设备信息。全量读取后可显式指定device_ids做局部复核，局部读数仍按page.has_more接续，不能替代全量输入。最终报告引用覆盖本轮knowledge_cutoff和batch_sequence的全量证据，Server会拒绝仅有局部或旧批次证据的final。
+全量devices含完整目录，readings含观测，coverage含各设备读数数量、缺测及冻结范围外清单。不得由代表测点推断全网。单查list_scada_devices默认每页100，读至page.has_more=false，属性缩略不代表完整信息。全量后可按device_ids局部复核并持续翻页，不能替代全量输入。final引用本轮knowledge_cutoff及batch_sequence的全量证据；Server拒绝局部或旧批次证据。
 
 分别说明压力异常模型得分、压力绝对下降及变化率、已配置出厂边界流量变化，再说明联合证据支持程度。使用实际权重、嫌疑门槛、动态阈值、参与测点与各自基线。综合分用于爆管嫌疑判断；得分越低越异常，目标告警与参考日分别解释。水质等其他异常按实际证据选择原因分支。
 
@@ -49,7 +51,7 @@ metadata:
 
 ## 完工后的恢复复核
 
-收到关联工单的completed反馈后，先读取get_task_context摘要中的handling_context，核对实际完工业务时间returned_at，不以工单创建、预约或电脑当前时间代替。Server通过现有恢复监测读取项目时序库中严格晚于returned_at的最近一帧及事件前15天同相位对照，独立保存恢复证据；例如13:00完工、15分钟采样时核对13:15。这条恢复链路适用于回放来源工单，不要求推进回放，原异常knowledge_cutoff与batch_sequence保持不变。
+关联工单completed后，以handling_context.returned_at为实际完工业务时间，不用创建、预约或电脑当前时间代替。Server读取严格晚于该时刻的最近一帧及事件前15天同相位对照，独立保存恢复证据。回放来源工单同样适用，不推进回放，原knowledge_cutoff与batch_sequence不变。
 
 waiting_work_order等待完工；waiting_scada按last_check说明尚无后续采样、数据缺项或仍异常，等待监测复核。verification_ready时用get_task_evidence读取verification_evidence_id完整快照；工单completed且恢复证据status=passed时，发布final/concluded恢复闭环报告，同时引用原异常全量证据与当前恢复证据。Server完成运行后关闭任务并解除关联告警，phase=completed表示已闭环。阀位、漏点照片、现场复测数值等未提供时仅记为资料缺项，不新增现场核验、人工确认或更多采样的闭环条件。
 

scripts/sync_peripheral_work_orders.py

@@ -114,7 +114,7 @@ async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, periphe
                     existing[task["source"].get("external_id")] = task
                 elif task["task_type"] == "event_investigation":
                     event_tasks[task.get("source", {}).get("external_id")] = task
-                    if task["status"] in {"queued", "waiting_human", "running"} or (
+                    if task["status"] in {"queued", "waiting_human", "running", "failed"} or (
                         work_order_ids is not None and task["status"] == "completed"
                     ):
                         investigations[task["task_id"]] = task

tests/unit/test_operational_work_order_waits.py

@@ -0,0 +1,243 @@
+"""A field-work wait ends the run without closing or restarting the incident."""
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+
+from app.auth.permissions import OPERATIONAL_ACTION_WORK_ORDER_EXECUTE
+from app.domain.domain_errors import DomainError
+from app.domain.schemas.operational_task import (
+    ActionExecuteCommand, ActionReconcileCommand, ActionReceipt, AgentRunCommand,
+    ExternalFeedbackCommand, HumanApprovalAction,
+)
+from app.infra.integrations.operational_actions import ActionDispatchOutcome
+from app.services.operational_recovery import track_work_order
+from app.services.operational_tasks import OperationalTaskService
+from tests.support.operational_task_repository import TaskRepository, _active_run, _context, _task_document
+from tests.unit.test_operational_tasks import _seed_concluded_report
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+def setup_wait(*, active=True, phase="waiting_work_order"):
+    now = datetime(2026, 11, 12, 5, 14, tzinfo=timezone.utc)
+    context = _context()
+    task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    task.update(workflow="scada_anomaly", lifecycle_status="open")
+    run = _active_run(task, now)
+    if not active:
+        run.update(status="succeeded", lease_expires_at=None, finished_at=now.isoformat())
+        task.update(status="waiting_human", lifecycle_status="waiting_human",
+                    agent_session_id=None, agent_lease_expires_at=None)
+    order_id = str(uuid4())
+    action = HumanApprovalAction(
+        action_id=uuid4(), task_id=task["task_id"], version=3, kind="create_work_order",
+        reason="现场检查", payload={"title": "检查", "description": "核验候选点", "work_order_type": "general"},
+        evidence_refs=[], required_approval_permission="work-order.approve",
+        required_execution_permission=OPERATIONAL_ACTION_WORK_ORDER_EXECUTE,
+        approval_status="approved", execution_status="succeeded",
+        proposed_by="principal:test", proposed_at=now,
+        receipt=ActionReceipt(adapter="peripheral_sample", receipt_id=order_id,
+            submitted_at=now, completed_at=now, external_status="open", detail="创建成功"),
+    )
+    repository = TaskRepository(task, run, action=action.model_dump(mode="json"))
+    track_work_order(task, repository.action)
+    task["handling_context"]["phase"] = phase
+    if phase == "waiting_scada":
+        task["handling_context"]["returned_at"] = now.isoformat()
+    service = OperationalTaskService(repository, context, {OPERATIONAL_ACTION_WORK_ORDER_EXECUTE}, clock=lambda: now)
+    return service, repository, now, action
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("method", ["submit", "reconcile"])
+@pytest.mark.parametrize("active", [True, False])
+async def test_successful_dispatch_preserves_run_and_parks_only_idle_task(method, active):
+    service, repository, now, action = setup_wait(active=active)
+    repository.task["handling_context"] = {}
+    _seed_concluded_report(repository, now)
+    repository.action.update(execution_status="not_submitted" if method == "submit" else "unknown", receipt=None)
+    dispatcher = SimpleNamespace(**{method if method == "reconcile" else "dispatch": AsyncMock(
+        return_value=ActionDispatchOutcome("succeeded", "创建成功", action.receipt))})
+    run_before = repository.run.copy()
+    command_type = ActionExecuteCommand if method == "submit" else ActionReconcileCommand
+    command = command_type(request_id=uuid4(), expected_version=action.version)
+    await getattr(service, f"{method}_action")(action.task_id, action.action_id, command, dispatcher, None)
+    assert repository.task["handling_context"]["phase"] == "waiting_work_order"
+    assert repository.task["execution_generation"] == 1
+    assert repository.run == run_before
+    assert repository.task["status"] == ("running" if active else "waiting_human")
+    if active:
+        await service.heartbeat_run(action.task_id, repository.run["run_id"], AgentRunCommand(
+            request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1"))
+    else:
+        assert await repository.get_active_run(str(action.task_id)) is None
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("method", ["submit", "reconcile"])
+@pytest.mark.parametrize("outcome", ["succeeded", "failed", "unsupported"])
+async def test_late_dispatch_receipt_is_saved_without_reviving_cancelled_parent(method, outcome):
+    service, repository, now, action = setup_wait()
+    repository.action.update(execution_status="not_submitted" if method == "submit" else "unknown", receipt=None)
+    async def cancelled_while_dispatching(**kwargs):
+        repository.task.update(status="cancelled", lifecycle_status="cancelled")
+        repository.run.update(status="cancelled", lease_expires_at=None)
+        return ActionDispatchOutcome(outcome, "外部回执", action.receipt if outcome == "succeeded" else None)
+    dispatcher = SimpleNamespace(**{method if method == "reconcile" else "dispatch": cancelled_while_dispatching})
+    command_type = ActionExecuteCommand if method == "submit" else ActionReconcileCommand
+    result = await getattr(service, f"{method}_action")(action.task_id, action.action_id,
+        command_type(request_id=uuid4(), expected_version=action.version), dispatcher, None)
+    assert result.execution_status == outcome
+    assert repository.task["status"] == "cancelled"
+    assert repository.task["execution_generation"] == 1
+    assert await repository.get_active_run(str(action.task_id)) is None
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("phase", ["waiting_work_order", "waiting_scada", "pending_approval"])
+async def test_current_interim_report_finishes_run_as_waiting_without_closing_incident(phase):
+    service, repository, now, action = setup_wait()
+    if phase == "pending_approval":
+        repository.task["handling_context"] = {}
+        repository.action.update(approval_status="pending", execution_status="not_submitted", receipt=None)
+    else:
+        repository.task["handling_context"]["phase"] = phase
+    _seed_concluded_report(repository, now)
+    repository.reports[-1]["report"]["stage"] = "interim"
+    result = await service.complete_run(action.task_id, repository.run["run_id"], AgentRunCommand(
+        request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1"))
+    assert result.status == result.lifecycle_status == "waiting_human"
+    assert repository.run["status"] == "succeeded"
+    assert repository.run["lease_expires_at"] is None
+    assert result.active_run is None
+    assert result.execution_generation == 1
+    assert not any(n["kind"] == "task_failed" for n in repository.notifications)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("defect", ["missing", "old_generation", "other_session", "verification_ready"])
+async def test_wait_does_not_bypass_current_report_or_final_recovery_requirements(defect):
+    service, repository, now, action = setup_wait()
+    if defect != "missing":
+        _seed_concluded_report(repository, now)
+        repository.reports[-1]["report"]["stage"] = "interim"
+        if defect == "old_generation":
+            repository.task["execution_generation"] = repository.run["generation"] = 2
+            repository.reports[-1]["execution_generation"] = 1
+        elif defect == "other_session":
+            repository.reports[-1]["source_session_id"] = "another-session"
+        else:
+            repository.task["handling_context"]["phase"] = "verification_ready"
+    result = await service.complete_run(action.task_id, repository.run["run_id"], AgentRunCommand(
+        request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1"))
+    assert result.status == "failed"
+    assert repository.run["failure_code"] == "ARTIFACT_INCOMPLETE"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("status", ["assigned", "in_progress", "awaiting_feedback", "completed"])
+@pytest.mark.parametrize("active", [True, False])
+async def test_ordinary_child_feedback_waits_without_starting_or_cancelling_run(status, active):
+    service, repository, now, action = setup_wait(active=active)
+    if not active:
+        # Recover the persisted shape left by the reported ARTIFACT_INCOMPLETE failure.
+        repository.task.update(status="failed", lifecycle_status="failed")
+        repository.run.update(status="failed", failure_code="ARTIFACT_INCOMPLETE")
+    run_before = repository.run.copy()
+    result = await service.external_feedback(action.task_id, ExternalFeedbackCommand(
+        request_id=uuid4(), source_event_id=f"peripheral_sample:{action.receipt.receipt_id}",
+        source_revision=2, status=status, occurred_at=now, detail="工单进度更新"))
+    assert result.task.execution_generation == 1
+    assert repository.run == run_before
+    assert result.task.status == ("running" if active else "waiting_human")
+    assert result.task.handling_context["phase"] == ("waiting_scada" if status == "completed" else "waiting_work_order")
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("status", ["failed", "cancelled"])
+async def test_failed_or_cancelled_child_wakes_same_task_for_replanning(status):
+    service, repository, now, action = setup_wait(active=False)
+    result = await service.external_feedback(action.task_id, ExternalFeedbackCommand(
+        request_id=uuid4(), source_event_id=f"peripheral_sample:{action.receipt.receipt_id}",
+        source_revision=2, status=status, occurred_at=now, detail="现场处置无法继续"))
+    assert str(result.task.task_id) == str(action.task_id)
+    assert result.task.status == "queued" and result.task.execution_generation == 2
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("phase", ["verification_ready", "completed"])
+async def test_revised_receipt_with_ready_recovery_evidence_does_not_strand_task(phase):
+    service, repository, now, action = setup_wait(active=False, phase="waiting_scada")
+    repository.task["handling_context"].update(phase=phase, verification_evidence_id="recovery:saved")
+    if phase == "completed":
+        repository.task.update(status="completed", lifecycle_status="completed")
+    result = await service.external_feedback(action.task_id, ExternalFeedbackCommand(
+        request_id=uuid4(), source_event_id=f"peripheral_sample:{action.receipt.receipt_id}",
+        source_revision=3, status="completed", occurred_at=now, detail="补充完工说明，实际完工时间不变"))
+    assert result.task.status == ("completed" if phase == "completed" else "queued")
+    assert result.task.execution_generation == (1 if phase == "completed" else 2)
+    assert result.task.handling_context["verification_evidence_id"] == "recovery:saved"
+
+
+@pytest.mark.anyio
+async def test_same_completion_time_revision_does_not_interrupt_active_recovery_review():
+    service, repository, now, action = setup_wait(phase="waiting_scada")
+    repository.task["handling_context"].update(phase="verification_ready", verification_evidence_id="recovery:saved")
+    run_before = repository.run.copy()
+    result = await service.external_feedback(action.task_id, ExternalFeedbackCommand(
+        request_id=uuid4(), source_event_id=f"peripheral_sample:{action.receipt.receipt_id}",
+        source_revision=3, status="completed", occurred_at=now, detail="补充备注，完工时间不变"))
+    assert result.task.status == "running" and result.task.execution_generation == 1
+    assert repository.run == run_before
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("new_decision", [True, False])
+@pytest.mark.parametrize("terminal_status", ["failed", "cancelled"])
+async def test_only_new_dispatch_decisions_replace_terminated_work_orders(new_decision, terminal_status):
+    service, repository, now, action = setup_wait(active=False)
+    old_id = action.receipt.receipt_id
+    await service.external_feedback(action.task_id, ExternalFeedbackCommand(
+        request_id=uuid4(), source_event_id=f"peripheral_sample:{old_id}", source_revision=2,
+        status=terminal_status, occurred_at=now - timedelta(hours=1), detail="需重新安排"))
+    replacement_id = str(uuid4())
+    new_action = action.model_dump(mode="json")
+    new_action.update(action_id=str(uuid4()), proposed_at=(now + timedelta(seconds=1 if new_decision else -1)).isoformat())
+    new_action["receipt"]["receipt_id"] = replacement_id
+    track_work_order(repository.task, new_action)
+    repository.action = new_action
+    # Finish the replanning run before the replacement's completion arrives.
+    repository.run.update(status="succeeded", lease_expires_at=None)
+    await service.external_feedback(action.task_id, ExternalFeedbackCommand(
+        request_id=uuid4(), source_event_id=f"peripheral_sample:{replacement_id}", source_revision=2,
+        status="completed", occurred_at=now + timedelta(hours=1), detail="替代处置完成"))
+    phase = repository.task["handling_context"]["phase"]
+    assert phase == ("waiting_scada" if new_decision else "waiting_work_order")
+    if new_decision:
+        orders = repository.task["handling_context"]["work_orders"]
+        assert orders[old_id]["superseded_by"] == replacement_id
+        # A late historical edit cannot interrupt the replacement or change T.
+        context_before = {k: v for k, v in repository.task["handling_context"].items() if k != "work_orders"}
+        generation = repository.task["execution_generation"]
+        await service.external_feedback(action.task_id, ExternalFeedbackCommand(
+            request_id=uuid4(), source_event_id=f"peripheral_sample:{old_id}", source_revision=3,
+            status=terminal_status, occurred_at=now, detail="旧工单备注更新"))
+        assert repository.task["execution_generation"] == generation
+        assert {k: v for k, v in repository.task["handling_context"].items() if k != "work_orders"} == context_before
+
+
+@pytest.mark.anyio
+async def test_superseded_run_heartbeat_has_distinct_problem_code():
+    service, repository, now, action = setup_wait()
+    old_run = repository.run.copy()
+    await service._requeue_task(repository.task, now)
+    with pytest.raises(DomainError) as rejected:
+        await service.heartbeat_run(action.task_id, old_run["run_id"], AgentRunCommand(
+            request_id=uuid4(), expected_revision=old_run["revision"], source_session_id="session-1"))
+    assert rejected.value.code == "RUN_SUPERSEDED"

tests/unit/test_peripheral_task_sync.py

@@ -232,6 +232,7 @@ async def test_sample_sync_rejects_cross_project_data_before_writing():
     [
         ("waiting_human", 2, True),
         ("queued", 2, True),
+        ("failed", 2, True),
         ("completed", 2, True),
         ("cancelled", 1, False),
     ],