From 471ad92a47e620fb3ae93b9bbf874528e4b3adbb Mon Sep 17 00:00:00 2001 From: Huarch Date: Thu, 12 Nov 2026 13:45:27 +0800 Subject: [PATCH] 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 | 29 ++- app/services/operational_tasks.py | 78 +++++- .../burst-event-handling/SKILL.md | 8 +- .../work-order-skills/scada-anomaly/SKILL.md | 8 +- scripts/sync_peripheral_work_orders.py | 2 +- .../unit/test_operational_work_order_waits.py | 243 ++++++++++++++++++ tests/unit/test_peripheral_task_sync.py | 1 + 7 files changed, 345 insertions(+), 24 deletions(-) create mode 100644 tests/unit/test_operational_work_order_waits.py diff --git a/app/services/operational_recovery.py b/app/services/operational_recovery.py index 1a07c5b..6d4295d 100644 --- a/app/services/operational_recovery.py +++ b/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) diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index 3a23a5d..cb5685e 100644 --- a/app/services/operational_tasks.py +++ b/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"])) diff --git a/resources/work-order-skills/burst-event-handling/SKILL.md b/resources/work-order-skills/burst-event-handling/SKILL.md index 5ef2054..ecc8c7e 100644 --- a/resources/work-order-skills/burst-event-handling/SKILL.md +++ b/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 表示已闭环。照片、阀位或复测记录未提供时仅记为资料缺项，不增设关闭前置条件。来源系统若另有明确行政结案流程，按其实际状态另记，不推断为当前任务尚未闭环。 diff --git a/resources/work-order-skills/scada-anomaly/SKILL.md b/resources/work-order-skills/scada-anomaly/SKILL.md index 0946dc4..45d644e 100644 --- a/resources/work-order-skills/scada-anomaly/SKILL.md +++ b/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表示已闭环。阀位、漏点照片、现场复测数值等未提供时仅记为资料缺项，不新增现场核验、人工确认或更多采样的闭环条件。 diff --git a/scripts/sync_peripheral_work_orders.py b/scripts/sync_peripheral_work_orders.py index 568b5d1..440cf7f 100644 --- a/scripts/sync_peripheral_work_orders.py +++ b/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 diff --git a/tests/unit/test_operational_work_order_waits.py b/tests/unit/test_operational_work_order_waits.py new file mode 100644 index 0000000..58914ca --- /dev/null +++ b/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" diff --git a/tests/unit/test_peripheral_task_sync.py b/tests/unit/test_peripheral_task_sync.py index f41eedd..5fb23cf 100644 --- a/tests/unit/test_peripheral_task_sync.py +++ b/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), ],