diff --git a/app/domain/schemas/operational_task.py b/app/domain/schemas/operational_task.py index 4eae4c0..0bddabd 100644 --- a/app/domain/schemas/operational_task.py +++ b/app/domain/schemas/operational_task.py @@ -10,7 +10,7 @@ from app.domain.schemas.shared import NonEmptyStr, PositiveInt, StrictModel from app.domain.schemas.operational_data import TaskDataSnapshot from app.domain.schemas.workbench_runtime import ( WorkbenchDateTime, - WorkbenchRuntimeWorkOrderSubtask, + WorkOrderActivity, ) @@ -290,19 +290,17 @@ class ExternalWorkOrderSnapshot(StrictModel): work_order_type: Literal["general", "water_quality_complaint", "burst_report"] = "general" location: NonEmptyStr | None = None scenario_code: NonEmptyStr | None = None - subtasks: Annotated[ - list[WorkbenchRuntimeWorkOrderSubtask], Field(min_length=1, max_length=50) - ] | None = None + feedback_requirements: list[NonEmptyStr] = Field(default_factory=list) + activity: list[WorkOrderActivity] = Field(default_factory=list) origin: Literal["synthetic"] updated_at: WorkbenchDateTime - @model_validator(mode="after") - def unique_subtasks(self): - if self.subtasks is not None and len( - {item.id for item in self.subtasks} - ) != len(self.subtasks): - raise ValueError("work order subtask ids must be unique") - return self + @model_validator(mode="before") + @classmethod + def accept_legacy_snapshot(cls, value): + if isinstance(value, dict): + return {key: item for key, item in value.items() if key != "subtasks"} + return value class OperationalTaskRun(StrictModel): diff --git a/app/domain/schemas/workbench_runtime.py b/app/domain/schemas/workbench_runtime.py index 3b084e5..a25080b 100644 --- a/app/domain/schemas/workbench_runtime.py +++ b/app/domain/schemas/workbench_runtime.py @@ -2,7 +2,7 @@ from typing import Annotated, Literal -from pydantic import AwareDatetime, BeforeValidator, JsonValue, model_validator +from pydantic import AwareDatetime, BeforeValidator, Field, JsonValue, model_validator from app.domain.schemas.shared import ( NonEmptyStr, @@ -129,42 +129,13 @@ class WorkbenchAnalysisHistoryPage(StrictModel): next_offset: NonNegativeInt | None -class WorkbenchRuntimeWorkOrderStage(StrictModel): - id: NonEmptyStr +class WorkOrderActivity(StrictModel): sequence: PositiveInt + occurred_at: WorkbenchDateTime + actor: NonEmptyStr + kind: Literal["created", "updated", "feedback"] title: NonEmptyStr - status: Literal["done", "current", "pending", "blocked", "skipped"] - owner: NonEmptyStr - time_label: NonEmptyStr - description: NonEmptyStr - - -class WorkbenchRuntimeWorkOrderSubtask(StrictModel): - id: NonEmptyStr - title: NonEmptyStr - summary: NonEmptyStr - status: Literal["pending", "running", "blocked", "completed", "cancelled"] - scheduled_at: WorkbenchDateTime - location: NonEmptyStr - assignee: NonEmptyStr - duration_minutes: PositiveInt - reply_window_minutes: PositiveInt - stages: list[WorkbenchRuntimeWorkOrderStage] - reply_requirements: list[NonEmptyStr] - - @model_validator(mode="after") - def validate_stages(self): - if not self.stages: - raise ValueError("subtask stages must not be empty") - if [stage.sequence for stage in self.stages] != list( - range(1, len(self.stages) + 1) - ): - raise ValueError("subtask stages must use contiguous sequence numbers") - if len({stage.id for stage in self.stages}) != len(self.stages): - raise ValueError("subtask stage ids must be unique") - if sum(stage.status == "current" for stage in self.stages) > 1: - raise ValueError("subtask stages may contain at most one current stage") - return self + detail: NonEmptyStr class WorkbenchRuntimeWorkOrder(StrictModel): @@ -181,16 +152,28 @@ class WorkbenchRuntimeWorkOrder(StrictModel): duration_minutes: PositiveInt source: NonEmptyStr dispatcher: NonEmptyStr + location: NonEmptyStr | None = None + assignee: NonEmptyStr | None = None priority: Literal["normal", "urgent"] - subtasks: list[WorkbenchRuntimeWorkOrderSubtask] + feedback_requirements: list[NonEmptyStr] = Field(default_factory=list) + activity: list[WorkOrderActivity] = Field(default_factory=list) - @model_validator(mode="after") - def validate_subtasks(self): - if not self.subtasks: - raise ValueError("work order subtasks must not be empty") - if len({subtask.id for subtask in self.subtasks}) != len(self.subtasks): - raise ValueError("work order subtask ids must be unique") - return self + @model_validator(mode="before") + @classmethod + def accept_legacy_document(cls, value): + if not isinstance(value, dict) or "subtasks" not in value: + return value + document = {key: item for key, item in value.items() if key != "subtasks"} + subtasks = value.get("subtasks") or [] + first = subtasks[0] if subtasks else {} + document.setdefault("location", first.get("location")) + document.setdefault("assignee", first.get("assignee")) + document.setdefault("feedback_requirements", list(dict.fromkeys( + requirement for subtask in subtasks + for requirement in subtask.get("reply_requirements", []) + ))) + document.setdefault("activity", []) + return document class WorkbenchRuntimeSnapshot(StrictModel): diff --git a/contracts/manifest.json b/contracts/manifest.json index 7812a00..be183dc 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "server": { "file": "server-v1.openapi.json", - "sha256": "3669e1b9ba31dad83f9c53028fd974b8bacac0f9d3069f644e49254baf6bb82f" + "sha256": "c37f5626398e1762584a844bb908ca4f4fb68e203f4d37a489561669a434312b" } } } diff --git a/contracts/peripheral-sample.openapi.json b/contracts/peripheral-sample.openapi.json index 10984a6..026c576 100644 --- a/contracts/peripheral-sample.openapi.json +++ b/contracts/peripheral-sample.openapi.json @@ -251,6 +251,15 @@ ], "title": "Assignee" }, + "feedback_requirements": { + "items": { + "maxLength": 4000, + "minLength": 1, + "type": "string" + }, + "title": "Feedback Requirements", + "type": "array" + }, "location": { "anyOf": [ { @@ -318,22 +327,6 @@ "title": "Status", "type": "string" }, - "subtasks": { - "anyOf": [ - { - "items": { - "$ref": "#/components/schemas/WorkbenchRuntimeWorkOrderSubtask" - }, - "maxItems": 50, - "minItems": 1, - "type": "array" - }, - { - "type": "null" - } - ], - "title": "Subtasks" - }, "summary": { "maxLength": 4000, "minLength": 1, @@ -1654,16 +1647,27 @@ "title": "Clear Scheduled At", "type": "boolean" }, - "clear_subtasks": { - "default": false, - "title": "Clear Subtasks", - "type": "boolean" - }, "expected_revision": { "minimum": 1.0, "title": "Expected Revision", "type": "integer" }, + "feedback_requirements": { + "anyOf": [ + { + "items": { + "maxLength": 4000, + "minLength": 1, + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Feedback Requirements" + }, "location": { "anyOf": [ { @@ -1694,6 +1698,19 @@ ], "title": "Priority" }, + "progress_note": { + "anyOf": [ + { + "maxLength": 4000, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Progress Note" + }, "request_id": { "format": "uuid", "title": "Request Id", @@ -1743,22 +1760,6 @@ ], "title": "Status" }, - "subtasks": { - "anyOf": [ - { - "items": { - "$ref": "#/components/schemas/WorkbenchRuntimeWorkOrderSubtask" - }, - "maxItems": 50, - "minItems": 1, - "type": "array" - }, - { - "type": "null" - } - ], - "title": "Subtasks" - }, "summary": { "anyOf": [ { @@ -1796,6 +1797,13 @@ "WorkOrder": { "additionalProperties": false, "properties": { + "activity": { + "items": { + "$ref": "#/components/schemas/WorkOrderActivity" + }, + "title": "Activity", + "type": "array" + }, "assignee": { "anyOf": [ { @@ -1816,6 +1824,13 @@ "title": "Created By", "type": "string" }, + "feedback_requirements": { + "items": { + "type": "string" + }, + "title": "Feedback Requirements", + "type": "array" + }, "location": { "anyOf": [ { @@ -1887,22 +1902,6 @@ "title": "Status", "type": "string" }, - "subtasks": { - "anyOf": [ - { - "items": { - "$ref": "#/components/schemas/WorkbenchRuntimeWorkOrderSubtask" - }, - "maxItems": 50, - "minItems": 1, - "type": "array" - }, - { - "type": "null" - } - ], - "title": "Subtasks" - }, "summary": { "title": "Summary", "type": "string" @@ -1955,6 +1954,55 @@ "title": "WorkOrder", "type": "object" }, + "WorkOrderActivity": { + "additionalProperties": false, + "properties": { + "actor": { + "minLength": 1, + "title": "Actor", + "type": "string" + }, + "detail": { + "minLength": 1, + "title": "Detail", + "type": "string" + }, + "kind": { + "enum": [ + "created", + "updated", + "feedback" + ], + "title": "Kind", + "type": "string" + }, + "occurred_at": { + "format": "date-time", + "title": "Occurred At", + "type": "string" + }, + "sequence": { + "minimum": 1.0, + "title": "Sequence", + "type": "integer" + }, + "title": { + "minLength": 1, + "title": "Title", + "type": "string" + } + }, + "required": [ + "sequence", + "occurred_at", + "actor", + "kind", + "title", + "detail" + ], + "title": "WorkOrderActivity", + "type": "object" + }, "WorkOrderPage": { "additionalProperties": false, "properties": { @@ -1984,149 +2032,6 @@ ], "title": "WorkOrderPage", "type": "object" - }, - "WorkbenchRuntimeWorkOrderStage": { - "additionalProperties": false, - "properties": { - "description": { - "minLength": 1, - "title": "Description", - "type": "string" - }, - "id": { - "minLength": 1, - "title": "Id", - "type": "string" - }, - "owner": { - "minLength": 1, - "title": "Owner", - "type": "string" - }, - "sequence": { - "minimum": 1.0, - "title": "Sequence", - "type": "integer" - }, - "status": { - "enum": [ - "done", - "current", - "pending", - "blocked", - "skipped" - ], - "title": "Status", - "type": "string" - }, - "time_label": { - "minLength": 1, - "title": "Time Label", - "type": "string" - }, - "title": { - "minLength": 1, - "title": "Title", - "type": "string" - } - }, - "required": [ - "id", - "sequence", - "title", - "status", - "owner", - "time_label", - "description" - ], - "title": "WorkbenchRuntimeWorkOrderStage", - "type": "object" - }, - "WorkbenchRuntimeWorkOrderSubtask": { - "additionalProperties": false, - "properties": { - "assignee": { - "minLength": 1, - "title": "Assignee", - "type": "string" - }, - "duration_minutes": { - "minimum": 1.0, - "title": "Duration Minutes", - "type": "integer" - }, - "id": { - "minLength": 1, - "title": "Id", - "type": "string" - }, - "location": { - "minLength": 1, - "title": "Location", - "type": "string" - }, - "reply_requirements": { - "items": { - "minLength": 1, - "type": "string" - }, - "title": "Reply Requirements", - "type": "array" - }, - "reply_window_minutes": { - "minimum": 1.0, - "title": "Reply Window Minutes", - "type": "integer" - }, - "scheduled_at": { - "format": "date-time", - "title": "Scheduled At", - "type": "string" - }, - "stages": { - "items": { - "$ref": "#/components/schemas/WorkbenchRuntimeWorkOrderStage" - }, - "title": "Stages", - "type": "array" - }, - "status": { - "enum": [ - "pending", - "running", - "blocked", - "completed", - "cancelled" - ], - "title": "Status", - "type": "string" - }, - "summary": { - "minLength": 1, - "title": "Summary", - "type": "string" - }, - "title": { - "minLength": 1, - "title": "Title", - "type": "string" - } - }, - "required": [ - "id", - "title", - "summary", - "status", - "scheduled_at", - "location", - "assignee", - "duration_minutes", - "reply_window_minutes", - "stages", - "reply_requirements" - ], - "title": "WorkbenchRuntimeWorkOrderSubtask", - "type": "object" } } }, diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json index f6bbb46..b91e6e6 100644 --- a/contracts/server-v1.openapi.json +++ b/contracts/server-v1.openapi.json @@ -2131,7 +2131,7 @@ "work_order_snapshot": { "anyOf": [ { - "$ref": "#/components/schemas/ExternalWorkOrderSnapshot-Input" + "$ref": "#/components/schemas/ExternalWorkOrderSnapshot" }, { "type": "null" @@ -2235,9 +2235,16 @@ "title": "ExternalFeedbackResult", "type": "object" }, - "ExternalWorkOrderSnapshot-Input": { + "ExternalWorkOrderSnapshot": { "additionalProperties": false, "properties": { + "activity": { + "items": { + "$ref": "#/components/schemas/WorkOrderActivity" + }, + "title": "Activity", + "type": "array" + }, "assignee": { "anyOf": [ { @@ -2250,6 +2257,14 @@ ], "title": "Assignee" }, + "feedback_requirements": { + "items": { + "minLength": 1, + "type": "string" + }, + "title": "Feedback Requirements", + "type": "array" + }, "location": { "anyOf": [ { @@ -2318,165 +2333,6 @@ "title": "Status", "type": "string" }, - "subtasks": { - "anyOf": [ - { - "items": { - "$ref": "#/components/schemas/WorkbenchRuntimeWorkOrderSubtask" - }, - "maxItems": 50, - "minItems": 1, - "type": "array" - }, - { - "type": "null" - } - ], - "title": "Subtasks" - }, - "summary": { - "minLength": 1, - "title": "Summary", - "type": "string" - }, - "title": { - "minLength": 1, - "title": "Title", - "type": "string" - }, - "updated_at": { - "format": "date-time", - "title": "Updated At", - "type": "string" - }, - "work_order_id": { - "format": "uuid", - "title": "Work Order Id", - "type": "string" - }, - "work_order_type": { - "default": "general", - "enum": [ - "general", - "water_quality_complaint", - "burst_report" - ], - "title": "Work Order Type", - "type": "string" - } - }, - "required": [ - "work_order_id", - "revision", - "title", - "summary", - "status", - "priority", - "origin", - "updated_at" - ], - "title": "ExternalWorkOrderSnapshot", - "type": "object" - }, - "ExternalWorkOrderSnapshot-Output": { - "additionalProperties": false, - "properties": { - "assignee": { - "anyOf": [ - { - "minLength": 1, - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Assignee" - }, - "location": { - "anyOf": [ - { - "minLength": 1, - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Location" - }, - "origin": { - "const": "synthetic", - "title": "Origin", - "type": "string" - }, - "priority": { - "enum": [ - "low", - "normal", - "high", - "urgent" - ], - "title": "Priority", - "type": "string" - }, - "revision": { - "minimum": 1.0, - "title": "Revision", - "type": "integer" - }, - "scenario_code": { - "anyOf": [ - { - "minLength": 1, - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Scenario Code" - }, - "scheduled_at": { - "anyOf": [ - { - "format": "date-time", - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Scheduled At" - }, - "status": { - "enum": [ - "draft", - "open", - "in_progress", - "awaiting_reply", - "completed", - "cancelled" - ], - "title": "Status", - "type": "string" - }, - "subtasks": { - "anyOf": [ - { - "items": { - "$ref": "#/components/schemas/WorkbenchRuntimeWorkOrderSubtask" - }, - "maxItems": 50, - "minItems": 1, - "type": "array" - }, - { - "type": "null" - } - ], - "title": "Subtasks" - }, "summary": { "minLength": 1, "title": "Summary", @@ -3705,7 +3561,7 @@ "work_order_snapshot": { "anyOf": [ { - "$ref": "#/components/schemas/ExternalWorkOrderSnapshot-Output" + "$ref": "#/components/schemas/ExternalWorkOrderSnapshot" }, { "type": "null" @@ -6577,6 +6433,55 @@ "title": "WebSearchRequest", "type": "object" }, + "WorkOrderActivity": { + "additionalProperties": false, + "properties": { + "actor": { + "minLength": 1, + "title": "Actor", + "type": "string" + }, + "detail": { + "minLength": 1, + "title": "Detail", + "type": "string" + }, + "kind": { + "enum": [ + "created", + "updated", + "feedback" + ], + "title": "Kind", + "type": "string" + }, + "occurred_at": { + "format": "date-time", + "title": "Occurred At", + "type": "string" + }, + "sequence": { + "minimum": 1.0, + "title": "Sequence", + "type": "integer" + }, + "title": { + "minLength": 1, + "title": "Title", + "type": "string" + } + }, + "required": [ + "sequence", + "occurred_at", + "actor", + "kind", + "title", + "detail" + ], + "title": "WorkOrderActivity", + "type": "object" + }, "WorkbenchAlertSource": { "additionalProperties": false, "properties": { @@ -7217,6 +7122,25 @@ "WorkbenchRuntimeWorkOrder": { "additionalProperties": false, "properties": { + "activity": { + "items": { + "$ref": "#/components/schemas/WorkOrderActivity" + }, + "title": "Activity", + "type": "array" + }, + "assignee": { + "anyOf": [ + { + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Assignee" + }, "code": { "minLength": 1, "title": "Code", @@ -7237,11 +7161,31 @@ "title": "Duration Minutes", "type": "integer" }, + "feedback_requirements": { + "items": { + "minLength": 1, + "type": "string" + }, + "title": "Feedback Requirements", + "type": "array" + }, "id": { "minLength": 1, "title": "Id", "type": "string" }, + "location": { + "anyOf": [ + { + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Location" + }, "priority": { "enum": [ "normal", @@ -7291,13 +7235,6 @@ "title": "Status", "type": "string" }, - "subtasks": { - "items": { - "$ref": "#/components/schemas/WorkbenchRuntimeWorkOrderSubtask" - }, - "title": "Subtasks", - "type": "array" - }, "summary": { "minLength": 1, "title": "Summary", @@ -7328,155 +7265,11 @@ "duration_minutes", "source", "dispatcher", - "priority", - "subtasks" + "priority" ], "title": "WorkbenchRuntimeWorkOrder", "type": "object" }, - "WorkbenchRuntimeWorkOrderStage": { - "additionalProperties": false, - "properties": { - "description": { - "minLength": 1, - "title": "Description", - "type": "string" - }, - "id": { - "minLength": 1, - "title": "Id", - "type": "string" - }, - "owner": { - "minLength": 1, - "title": "Owner", - "type": "string" - }, - "sequence": { - "minimum": 1.0, - "title": "Sequence", - "type": "integer" - }, - "status": { - "enum": [ - "done", - "current", - "pending", - "blocked", - "skipped" - ], - "title": "Status", - "type": "string" - }, - "time_label": { - "minLength": 1, - "title": "Time Label", - "type": "string" - }, - "title": { - "minLength": 1, - "title": "Title", - "type": "string" - } - }, - "required": [ - "id", - "sequence", - "title", - "status", - "owner", - "time_label", - "description" - ], - "title": "WorkbenchRuntimeWorkOrderStage", - "type": "object" - }, - "WorkbenchRuntimeWorkOrderSubtask": { - "additionalProperties": false, - "properties": { - "assignee": { - "minLength": 1, - "title": "Assignee", - "type": "string" - }, - "duration_minutes": { - "minimum": 1.0, - "title": "Duration Minutes", - "type": "integer" - }, - "id": { - "minLength": 1, - "title": "Id", - "type": "string" - }, - "location": { - "minLength": 1, - "title": "Location", - "type": "string" - }, - "reply_requirements": { - "items": { - "minLength": 1, - "type": "string" - }, - "title": "Reply Requirements", - "type": "array" - }, - "reply_window_minutes": { - "minimum": 1.0, - "title": "Reply Window Minutes", - "type": "integer" - }, - "scheduled_at": { - "format": "date-time", - "title": "Scheduled At", - "type": "string" - }, - "stages": { - "items": { - "$ref": "#/components/schemas/WorkbenchRuntimeWorkOrderStage" - }, - "title": "Stages", - "type": "array" - }, - "status": { - "enum": [ - "pending", - "running", - "blocked", - "completed", - "cancelled" - ], - "title": "Status", - "type": "string" - }, - "summary": { - "minLength": 1, - "title": "Summary", - "type": "string" - }, - "title": { - "minLength": 1, - "title": "Title", - "type": "string" - } - }, - "required": [ - "id", - "title", - "summary", - "status", - "scheduled_at", - "location", - "assignee", - "duration_minutes", - "reply_window_minutes", - "stages", - "reply_requirements" - ], - "title": "WorkbenchRuntimeWorkOrderSubtask", - "type": "object" - }, "WorkbenchSyncSnapshot": { "additionalProperties": false, "properties": { diff --git a/docs/operational-task-workers.md b/docs/operational-task-workers.md index 08dd42a..96b2912 100644 --- a/docs/operational-task-workers.md +++ b/docs/operational-task-workers.md @@ -81,6 +81,8 @@ conda run -n server python scripts/sync_peripheral_work_orders.py 使用相同工作身份，并配置 `OPERATIONAL_ACTION_PERIPHERAL_URL`。同步器读取分页工单，将外围版本投影到 `external_work_order` 任务。合成来源在摘要和证据中明确标注；工单完成只表示该外部任务完成，不代表其他调查自动完成。 +通用合成检查工单不属于预约任务。早期种子脚本创建的通信、压力和零流量检查工单已经退役，种子脚本不再恢复它们；同步器按不可编辑的专用 seed actor 跳过旧种子工单。投诉、爆管事件以及调查动作实际创建的现场工单继续同步。`018_remove_synthetic_check_task_projections.sql` 只清理专用 seed actor 创建的旧检查投影，不按用户可编辑的标题或状态删除任务，也不影响 `workbench.work_orders` 中保留的三条演示工单。 + 水质投诉和爆管上报类型的外围工单同时创建关联调查，即使当时没有浏览器打开也会入队。两类事件从外围样板测试面板模拟上报；工况页负责展示调查。调查中创建的外围工单通过回执编号关联，后续反馈送回原调查，由 Agent 重新分析。 ## 旧调查迁移 diff --git a/docs/operational-tasks.md b/docs/operational-tasks.md index 7adf218..7f250ac 100644 --- a/docs/operational-tasks.md +++ b/docs/operational-tasks.md @@ -6,6 +6,8 @@ - 定时任务：当前产品继续使用前端 mock。`scheduled_flow` 及后端计划接口为已有保留能力，不接当前前端、不由当前 worker 调度或执行。 - `external_work_order`：外围预约、维修等工单的只读任务投影。`source.kind + source.external_id` 在项目内唯一，进度以外部版本反馈为准。 + +预约任务采用单层结构：一张外围工单对应一条任务，工单快照直接提供状态、位置、负责人、预约时间、回执要求和按发生时间记录的 `activity`。活动仅由工单创建、修改或显式现场反馈产生；不能按当前状态补造派发、执行、复令三个阶段。旧快照中的 `subtasks` 仅在读取时兼容并丢弃，不回写外围工单或历史版本；后续外围修订同步为新结构。 - `event_investigation`：投诉、爆管上报或人工发起的调查。Agent 查询受控数据，发布执行图、证据引用和报告；现场反馈会创建新的执行 generation，让 Agent 重新分析。 创建请求使用 `request_id` 幂等。任务执行使用独立的 `run_id + execution_request_id`：Agent 必须领取当前 run，并通过 run 级心跳维持 60 秒租约。正常结束必须显式 complete；缺少同代最终计划或报告会以 `ARTIFACT_INCOMPLETE` 失败。`POST /operational-tasks/leases/recover` 会处理过期执行，连续三代超时后停止自动重试。 @@ -60,6 +62,7 @@ psql "$PROJECT_BIZ_DSN" -v ON_ERROR_STOP=1 -f resources/sql/011_workbench_work_o psql "$PROJECT_BIZ_DSN" -v ON_ERROR_STOP=1 -f resources/sql/014_operational_tasks.sql psql "$PROJECT_BIZ_DSN" -v ON_ERROR_STOP=1 -f resources/sql/015_workbench_events_and_task_runs.sql psql "$PROJECT_BIZ_DSN" -v ON_ERROR_STOP=1 -f resources/sql/017_workbench_schema_comments.sql +psql "$PROJECT_BIZ_DSN" -v ON_ERROR_STOP=1 -f resources/sql/018_remove_synthetic_check_task_projections.sql ``` -`017_workbench_schema_comments.sql` 只增加数据库可见的表、列和 schema 注释，不改变数据、约束或索引。迁移不会创建或修改外围系统的业务表，也不授予 `PUBLIC` 权限。生产运行前需按部署身份显式授予新表权限。SQL 文件包含自己的事务边界，不能把外层 `ROLLBACK` 当作演练；具体要求见数据库变更门禁。 +`017_workbench_schema_comments.sql` 只增加数据库可见的表、列和 schema 注释，不改变数据、约束或索引。`018_remove_synthetic_check_task_projections.sql` 读取 `peripheral_sample.work_orders` 的不可编辑创建者标识，删除错误进入预约任务的旧种子检查投影，不修改外围表，也不删除演示工单、事件工单或调查动作创建的工单。迁移不授予 `PUBLIC` 权限。生产运行前需按部署身份显式授予新表权限。SQL 文件包含自己的事务边界，不能把外层 `ROLLBACK` 当作演练；具体要求见数据库变更门禁。 diff --git a/docs/peripheral-generated-data.md b/docs/peripheral-generated-data.md index 143a813..f6ac976 100644 --- a/docs/peripheral-generated-data.md +++ b/docs/peripheral-generated-data.md @@ -44,7 +44,9 @@ CREATE TABLE peripheral_sample.demo_datasets ( conda run -n server python scripts/seed_peripheral_sample.py --project-id ``` -脚本使用固定 UUIDv5 请求号和专用 synthetic actor，重复执行会读取原幂等回执，不重复创建。它核验或补充 `water-demo-v1`，并创建 3 张测试工单和 12 条带 `DEMO-` 前缀的可编辑测试数据；不创建缺少合法 Action 来源的外围核查任务，不写正式工单、资产或监测表。 +脚本使用固定 UUIDv5 请求号和专用 synthetic actor，重复执行会读取原幂等回执，不重复创建。它核验或补充 `water-demo-v1`，并创建 12 条带 `DEMO-` 前缀的可编辑测试数据；不再创建早期用于检查通信、压力和零流量展示的三张测试工单，也不创建缺少合法 Action 来源的外围核查任务，不写正式工单、资产或监测表。 + +通过工单接口新建的预约工单不再包含子任务和固定阶段。创建请求可直接填写 `feedback_requirements`；修改请求可填写 `progress_note`，它只追加一条有时间、操作者和正文的活动记录，不作为工单顶层字段保存。旧工单读取时去掉 `subtasks`，历史修订仍保留原样。 ## 演示历史窗口配置 diff --git a/docs/workbench-database-rollout.md b/docs/workbench-database-rollout.md index d6682a7..67252bf 100644 --- a/docs/workbench-database-rollout.md +++ b/docs/workbench-database-rollout.md @@ -8,6 +8,10 @@ `017_workbench_schema_comments.sql` 只为 `workbench` schema 及其 17 张表和关键列增加数据库可见注释。它先在 `tjwater_v2` 应用；其他项目数据库保持原状，直到 `tjwater_v2` 验证通过并获得明确扩展确认。 +`018_remove_synthetic_check_task_projections.sql` 清理曾被错误同步为预约任务的通用合成检查投影。它按完整 `peripheral_sample:` 标识关联外围源，只匹配 `general` 类型、没有调查动作回执关联且由专用 seed actor 创建的旧检查任务；不使用用户可编辑的标题或工单状态作为删除依据。`workbench.work_orders` 中三条演示工单、投诉与爆管事件以及调查产生的现场采样工单均不在清理范围。同步器使用同一专用 seed actor 阻止旧种子检查再次生成。 + +预约工单结构调整无需 DDL。`seed_workbench_work_orders.py --project-id <已核对的项目 UUID>` 只重建该项目的三条演示预约投影，保留工单 ID 和标题，改为单层字段与显式活动记录。执行前必须核对脚本删除谓词实际命中的 ID；它不修改 `peripheral_sample.work_orders`、历史修订或调查任务。其他项目不得随 `tjwater_v2` 自动重建。 + 后续数据库变更默认只写入 `tjwater_v2`。其他项目保持现状，必须等 `tjwater_v2` 验证通过并由维护者明确确认后，再逐库扩展。禁止把“所有活跃项目”作为写操作的默认范围。 ## 变更顺序 diff --git a/docs/workbench-operations-storage.md b/docs/workbench-operations-storage.md index e56c8f1..588f2f1 100644 --- a/docs/workbench-operations-storage.md +++ b/docs/workbench-operations-storage.md @@ -12,11 +12,12 @@ DDL 位于 `resources/sql/008_workbench_alert_events.sql`。任务通过 `POST / `resources/sql/017_workbench_schema_comments.sql` 在 `008`、`011`、`014` 和 `015` 之后执行，为工作台 schema、17 张表和关键列增加数据库可见的用途说明。 -- 一张主工单包含一个或多个 `subtasks`。 -- 每个子任务保存自己的位置、班组、时间窗和复令要求。 -- 每个子任务包含一个或多个有序 `stages`，`sequence` 从 1 连续递增，阶段编号在子任务内唯一，最多只有一个 `current` 阶段。 -- 阶段名称不写死为固定枚举。演示数据使用派发、执行、复令，正式工单适配器可以按业务定义其他阶段。 +- 一张预约工单对应一条单层投影，直接保存位置、负责人、预约时间和回执要求。 +- `activity` 只记录已经发生的创建、修改和现场反馈，按递增 `sequence` 排列；展示层不得根据工单总状态补造派发、执行或复令阶段。 +- 旧文档中的 `subtasks` 在兼容读取时被丢弃，不再进入新接口契约；历史修订保留原始审计内容。 -预约任务演示数据通过 `scripts/seed_workbench_work_orders.py` 直接写入。工作台会显示三张原有 mock 工单，分别处于派发、执行和复令阶段；每张工单都保留自己的子任务与有序阶段。脚本可重复执行，并会先清理旧的外围合成工单投影。外围种子脚本仍会恢复被软删除的固定演示资源。 +预约任务演示数据通过 `scripts/seed_workbench_work_orders.py` 直接写入。工作台继续显示三张原有演示工单，保留既有 ID、标题和状态，并改用回执要求与真实活动记录表达过程。脚本可重复执行，并会先清理旧的外围合成工单投影。`scripts/seed_peripheral_sample.py` 不再创建通信、压力和零流量检查工单。 + +截至 2026-09-24，本次数据清理和三张演示工单结构重建仅应用于物理数据库 `tjwater_v2`，项目为 `324ff7ee-c2d4-5184-a84b-48fb14a2933e`。核验结果为三张演示工单均使用单层结构，旧种子检查投影为 0；其他项目数据库未执行本次变更。扩大范围必须遵循 [`workbench-database-rollout.md`](workbench-database-rollout.md) 的逐库门禁。 工作台运行账号对工单表只需 `SELECT`，同步账号使用独立受控写权限；对告警事件表需要 `SELECT, INSERT`，对告警状态表需要 `SELECT, INSERT, UPDATE`。 diff --git a/resources/sql/018_remove_synthetic_check_task_projections.sql b/resources/sql/018_remove_synthetic_check_task_projections.sql new file mode 100644 index 0000000..c09215a --- /dev/null +++ b/resources/sql/018_remove_synthetic_check_task_projections.sql @@ -0,0 +1,74 @@ +-- Remove synthetic check fixtures that were incorrectly projected as appointments. + +BEGIN; + +CREATE TEMP TABLE synthetic_check_task_cleanup ON COMMIT DROP AS +SELECT + task.project_id, + task.task_id, + task.revision, + task.document->>'title' AS title +FROM workbench.operational_tasks AS task +WHERE task.task_type = 'external_work_order' + AND task.document#>>'{source,kind}' = 'external_system' + AND COALESCE(task.document#>>'{source,event_kind}', 'general') = 'general' + AND EXISTS ( + SELECT 1 + FROM peripheral_sample.work_orders AS source_order + WHERE source_order.project_id = task.project_id + AND task.document#>>'{source,external_id}' = + 'peripheral_sample:' || source_order.work_order_id::text + AND source_order.document->>'created_by' = + 'f1a20d4c-b7b7-5b67-b134-92e5ac541162' + ) + AND NOT EXISTS ( + SELECT 1 + FROM workbench.operational_task_actions AS action + WHERE action.project_id = task.project_id + AND action.document->>'kind' = 'create_work_order' + AND action.document#>>'{receipt,adapter}' = 'peripheral_sample' + AND action.document#>>'{receipt,receipt_id}' = + split_part(task.document#>>'{source,external_id}', ':', 2) + ); + +DELETE FROM workbench.operational_task_requests AS request +USING synthetic_check_task_cleanup AS cleanup +WHERE request.project_id = cleanup.project_id + AND ( + request.scope LIKE 'task:' || cleanup.task_id::text || ':%' + OR request.response->>'task_id' = cleanup.task_id::text + OR request.response#>>'{task,task_id}' = cleanup.task_id::text + OR request.response#>>'{notification,task_id}' = cleanup.task_id::text + ); + +DELETE FROM workbench.operational_tasks AS task +USING synthetic_check_task_cleanup AS cleanup +WHERE (task.project_id, task.task_id) = (cleanup.project_id, cleanup.task_id); + +INSERT INTO workbench.domain_event_log ( + project_id, + event_id, + topic, + aggregate_type, + aggregate_id, + aggregate_revision, + metadata +) +SELECT + project_id, + ( + substr(md5(project_id::text || ':' || task_id::text || ':synthetic-check-cleanup'), 1, 8) + || '-' || substr(md5(project_id::text || ':' || task_id::text || ':synthetic-check-cleanup'), 9, 4) + || '-5' || substr(md5(project_id::text || ':' || task_id::text || ':synthetic-check-cleanup'), 14, 3) + || '-a' || substr(md5(project_id::text || ':' || task_id::text || ':synthetic-check-cleanup'), 18, 3) + || '-' || substr(md5(project_id::text || ':' || task_id::text || ':synthetic-check-cleanup'), 21, 12) + )::uuid, + 'operational.task.changed', + 'operational_task', + task_id::text, + revision + 1, + jsonb_build_object('change', 'deleted', 'reason', 'synthetic_check_projection_cleanup') +FROM synthetic_check_task_cleanup +ON CONFLICT (project_id, event_id) DO NOTHING; + +COMMIT; diff --git a/samples/peripheral/schemas.py b/samples/peripheral/schemas.py index 17df8d1..3773f89 100644 --- a/samples/peripheral/schemas.py +++ b/samples/peripheral/schemas.py @@ -5,7 +5,7 @@ from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictFloat, StrictInt, model_validator from app.domain.schemas.shared import ObjectRef, TimeRange, WireDateTime -from app.domain.schemas.workbench_runtime import WorkbenchRuntimeWorkOrderSubtask +from app.domain.schemas.workbench_runtime import WorkOrderActivity Text = Annotated[str, Field(min_length=1, max_length=4000)] PositiveInt = Annotated[StrictInt, Field(ge=1)] @@ -122,18 +122,6 @@ WorkOrderStatus = Literal[ ] Priority = Literal["low", "normal", "high", "urgent"] WorkOrderType = Literal["general", "water_quality_complaint", "burst_report"] -WorkOrderSubtasks = Annotated[ - list[WorkbenchRuntimeWorkOrderSubtask], Field(min_length=1, max_length=50) -] - - -def _validate_unique_subtasks( - subtasks: list[WorkbenchRuntimeWorkOrderSubtask] | None, -) -> None: - if subtasks is not None and len({item.id for item in subtasks}) != len(subtasks): - raise ValueError("work order subtask ids must be unique") - - class CreateWorkOrder(Model): request_id: UUID title: Annotated[str, Field(min_length=1, max_length=200)] @@ -145,13 +133,12 @@ class CreateWorkOrder(Model): work_order_type: WorkOrderType = "general" location: Annotated[str, Field(min_length=1, max_length=200)] | None = None scenario_code: Annotated[str, Field(min_length=1, max_length=128)] | None = None - subtasks: WorkOrderSubtasks | None = None + feedback_requirements: list[Text] = Field(default_factory=list) @model_validator(mode="after") def aware_schedule(self): if self.scheduled_at is not None and self.scheduled_at.tzinfo is None: raise ValueError("scheduled_at must include timezone") - _validate_unique_subtasks(self.subtasks) return self @@ -170,8 +157,8 @@ class UpdateWorkOrder(Model): clear_location: StrictBool = False scenario_code: Annotated[str, Field(min_length=1, max_length=128)] | None = None clear_scenario_code: StrictBool = False - subtasks: WorkOrderSubtasks | None = None - clear_subtasks: StrictBool = False + feedback_requirements: list[Text] | None = None + progress_note: Text | None = None @model_validator(mode="after") def valid_update(self): @@ -186,11 +173,8 @@ class UpdateWorkOrder(Model): raise ValueError("clear_location conflicts with location") if self.clear_scenario_code and self.scenario_code is not None: raise ValueError("clear_scenario_code conflicts with scenario_code") - if self.clear_subtasks and self.subtasks is not None: - raise ValueError("clear_subtasks conflicts with subtasks") if self.scheduled_at is not None and self.scheduled_at.tzinfo is None: raise ValueError("scheduled_at must include timezone") - _validate_unique_subtasks(self.subtasks) return self @@ -217,12 +201,15 @@ class WorkOrder(Model): work_order_type: WorkOrderType = "general" location: str | None = None scenario_code: str | None = None - subtasks: WorkOrderSubtasks | None = None + feedback_requirements: list[str] = Field(default_factory=list) + activity: list[WorkOrderActivity] = Field(default_factory=list) - @model_validator(mode="after") - def unique_subtasks(self): - _validate_unique_subtasks(self.subtasks) - return self + @model_validator(mode="before") + @classmethod + def accept_legacy_document(cls, value): + if isinstance(value, dict): + return {key: item for key, item in value.items() if key != "subtasks"} + return value class WorkOrderPage(Model): diff --git a/samples/peripheral/seed_identity.py b/samples/peripheral/seed_identity.py new file mode 100644 index 0000000..06e4cbf --- /dev/null +++ b/samples/peripheral/seed_identity.py @@ -0,0 +1,7 @@ +"""Stable identity used only by the explicit peripheral demo-data seed.""" + +from uuid import NAMESPACE_URL, uuid5 + + +SEED_NAMESPACE = uuid5(NAMESPACE_URL, "tjwater/peripheral-sample/seed/v1") +SEED_ACTOR_ID = uuid5(SEED_NAMESPACE, "actor") diff --git a/samples/peripheral/store.py b/samples/peripheral/store.py index d54c7e5..6e02397 100644 --- a/samples/peripheral/store.py +++ b/samples/peripheral/store.py @@ -24,6 +24,11 @@ from .schemas import ( TRANSITIONS = {"pending": ["processing"], "processing": ["pending_review"], "pending_review": ["processing", "completed"], "completed": []} +WORK_ORDER_FIELD_LABELS = { + "title": "标题", "summary": "说明", "status": "状态", "priority": "优先级", + "assignee": "负责人", "scheduled_at": "预约时间", "location": "位置", + "scenario_code": "场景编号", "feedback_requirements": "回执要求", +} class PostgresStore: @@ -268,8 +273,13 @@ class PostgresStore: @staticmethod def _active_managed_document(document: dict[str, Any]) -> dict[str, Any]: """Project a current active row without legacy soft-delete metadata.""" - return {key: value for key, value in document.items() - if key not in {"deleted_at", "deleted_by"}} + active = {key: value for key, value in document.items() + if key not in {"deleted_at", "deleted_by"}} + if "work_order_id" in active: + active.pop("subtasks", None) + active.setdefault("feedback_requirements", []) + active.setdefault("activity", []) + return active def _list_managed(self, who: Identity, table: str, id_column: str, limit: int, cursor: UUID | None) -> dict[str, Any]: @@ -314,6 +324,8 @@ class PostgresStore: "work_order_id": str(uuid4()), "project_id": who.project_id, "revision": 1, "origin": "synthetic", "created_at": now, "updated_at": now, "created_by": who.user_id, "updated_by": who.user_id, + "activity": [{"sequence": 1, "occurred_at": now, "actor": who.user_id, + "kind": "created", "title": "工单创建", "detail": "预约工单已创建。"}], } return self._save_work_order(conn, who, document) return self._mutate(who, "create_work_order", command, write) @@ -326,7 +338,7 @@ class PostgresStore: ensure(document["revision"] == command.expected_revision, "REVISION_CONFLICT", "工单版本已变化，请重新读取", 409) changes = command.model_dump(mode="json", exclude_unset=True, exclude={"request_id", "expected_revision", "clear_assignee", "clear_scheduled_at", - "clear_location", "clear_scenario_code", "clear_subtasks"}) + "clear_location", "clear_scenario_code", "progress_note"}) if command.clear_assignee: changes["assignee"] = None if command.clear_scheduled_at: @@ -335,10 +347,22 @@ class PostgresStore: changes["location"] = None if command.clear_scenario_code: changes["scenario_code"] = None - if command.clear_subtasks: - changes["subtasks"] = None + changed_fields = [key for key, value in changes.items() if document.get(key) != value] document.update(changes) - document.update(revision=document["revision"] + 1, updated_at=datetime.now(timezone.utc).isoformat(), + now = datetime.now(timezone.utc).isoformat() + revision = document["revision"] + 1 + if command.progress_note: + title, detail, kind = "现场反馈", command.progress_note, "feedback" + elif changed_fields: + title, detail, kind = "工单信息更新", "更新了" + "、".join( + WORK_ORDER_FIELD_LABELS.get(field, field) for field in changed_fields + ) + "。", "updated" + else: + title, detail, kind = "工单信息确认", "工单信息已确认，未记录新的现场进展。", "updated" + document["activity"].append({"sequence": revision, "occurred_at": now, + "actor": who.user_id, "kind": kind, + "title": title, "detail": detail}) + document.update(revision=revision, updated_at=now, updated_by=who.user_id) return self._save_work_order(conn, who, document) return self._mutate(who, f"update_work_order:{work_order_id}", command, write) diff --git a/scripts/seed_peripheral_sample.py b/scripts/seed_peripheral_sample.py index 6fa71df..40b3dcb 100644 --- a/scripts/seed_peripheral_sample.py +++ b/scripts/seed_peripheral_sample.py @@ -5,7 +5,7 @@ import argparse import json import sys from pathlib import Path -from uuid import NAMESPACE_URL, UUID, uuid5 +from uuid import UUID, uuid5 from psycopg.types.json import Jsonb @@ -16,49 +16,18 @@ from app.services.content_digest import digest from samples.peripheral.auth import Identity from samples.peripheral.config import SampleSettings from samples.peripheral.demo_data import build_demo_content -from samples.peripheral.schemas import CreateTestDataRecord, CreateWorkOrder +from samples.peripheral.schemas import CreateTestDataRecord +from samples.peripheral.seed_identity import SEED_ACTOR_ID, SEED_NAMESPACE from samples.peripheral.store import PostgresStore -SEED_NAMESPACE = uuid5(NAMESPACE_URL, "tjwater/peripheral-sample/seed/v1") -ACTOR_ID = uuid5(SEED_NAMESPACE, "actor") +ACTOR_ID = SEED_ACTOR_ID def request_id(kind: str) -> UUID: return uuid5(SEED_NAMESPACE, kind) -WORK_ORDERS = ( - { - "key": "work-order/demo-communication", - "title": "合成通信核查", - "summary": "检查 DEMO-P001 的预置通信中断情节，仅用于外围样板联调。", - "status": "open", - "priority": "high", - "assignee": "外围样板班组", - "scheduled_at": "2026-09-17T03:00:00Z", - }, - { - "key": "work-order/demo-pressure", - "title": "合成压力偏差复核", - "summary": "复核 DEMO-P001、DEMO-P002 相邻测试点的固定合成读数，不代表现场异常。", - "status": "in_progress", - "priority": "urgent", - "assignee": "演示核查班组", - "scheduled_at": "2026-09-17T04:00:00Z", - }, - { - "key": "work-order/flow-zero", - "title": "合成零流量样本确认", - "summary": "确认 DEMO-F001 中零值与缺测值可被界面明确区分。", - "status": "completed", - "priority": "normal", - "assignee": "外围样板班组", - "scheduled_at": "2026-09-17T06:00:00Z", - }, -) - - TEST_DATA = ( ("DEMO-P001", "合成压力 1 · 00:00", "2026-09-17T00:00:00Z", 24.37, "m", "valid"), ("DEMO-P001", "合成压力 1 · 00:15", "2026-09-17T00:15:00Z", 24.18, "m", "valid"), @@ -97,31 +66,22 @@ def previous_seed_result( return row["result"] -def restore_seeded_resources( +def restore_seeded_test_data( store: PostgresStore, project_id: str, - *, - table: str, - id_column: str, - resource_ids: list[str], + record_ids: list[str], ) -> None: """Restore deterministic seed rows that were soft-deleted after an earlier run.""" - allowed_resources = { - ("work_orders", "work_order_id"), - ("test_data_records", "record_id"), - } - if (table, id_column) not in allowed_resources: - raise ValueError("unsupported seeded resource") - if not resource_ids: + if not record_ids: return with store.connection() as connection: connection.execute( - f"""UPDATE peripheral_sample.{table} + """UPDATE peripheral_sample.test_data_records SET document=document - 'deleted_at' - 'deleted_by', deleted_at=NULL, deleted_by=NULL - WHERE project_id=%s AND {id_column}=ANY(%s::uuid[])""", - (project_id, resource_ids), + WHERE project_id=%s AND record_id=ANY(%s::uuid[])""", + (project_id, record_ids), ) @@ -160,34 +120,6 @@ def main() -> int: who = Identity(project_id, str(ACTOR_ID), True) dataset_digest = ensure_demo_dataset(store, project_id) - order_ids = [] - for item in WORK_ORDERS: - seed_request_id = request_id(item["key"]) - command = CreateWorkOrder( - request_id=seed_request_id, - title=item["title"], - summary=item["summary"], - status=item["status"], - priority=item["priority"], - assignee=item["assignee"], - scheduled_at=item["scheduled_at"], - ) - previous = previous_seed_result( - store, - project_id, - seed_request_id=seed_request_id, - command_key="create_work_order", - ) - result = previous or store.create_work_order(who, command) - order_ids.append(result["work_order_id"]) - restore_seeded_resources( - store, - project_id, - table="work_orders", - id_column="work_order_id", - resource_ids=order_ids, - ) - record_ids = [] for point_id, label, observed_at, value, unit, quality in TEST_DATA: seed_request_id = request_id(f"test-data/{point_id}/{observed_at}") @@ -209,20 +141,13 @@ def main() -> int: ) result = previous or store.create_test_data(who, command) record_ids.append(result["record_id"]) - restore_seeded_resources( - store, - project_id, - table="test_data_records", - id_column="record_id", - resource_ids=record_ids, - ) + restore_seeded_test_data(store, project_id, record_ids) result = { "project_id": project_id, "actor_id": str(ACTOR_ID), "origin": "synthetic", "demo_dataset_digest": dataset_digest, - "work_orders": {"count": len(order_ids), "ids": order_ids}, "test_data_records": {"count": len(record_ids), "ids": record_ids}, } print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True)) diff --git a/scripts/seed_workbench_work_orders.py b/scripts/seed_workbench_work_orders.py index 8e7fce0..c674512 100644 --- a/scripts/seed_workbench_work_orders.py +++ b/scripts/seed_workbench_work_orders.py @@ -108,119 +108,16 @@ WORK_ORDER_TEMPLATES = ( ) -def _clock(value: datetime) -> str: - return value.astimezone(SHANGHAI).strftime("%H:%M") - - -def _clock_range(start: datetime, duration_minutes: int) -> str: - return f"{_clock(start)}-{_clock(start + timedelta(minutes=duration_minutes))}" - - -def _stage_statuses(phase: str, position: str) -> tuple[str, str, str]: - if position == "before": - return "done", "done", "done" - if position == "after": - return "pending", "pending", "pending" - if phase == "dispatch": - return "current", "pending", "pending" - if phase == "execution": - return "done", "current", "pending" - return "done", "done", "current" - - -def _subtask_status(phase: str, position: str) -> str: - if position == "before": - return "completed" - if position == "after" or phase == "dispatch": - return "pending" - return "running" - - -def _build_subtasks(template: dict, scheduled_at: datetime) -> list[dict]: - requirements = template["reply_requirements"] - phase = template["phase"] - active_index = { - "dispatch": 0, - "execution": min(1, len(requirements) - 1), - "reply": len(requirements) - 1, - }[phase] - subtask_duration = max( - 5, - (template["duration_minutes"] + len(requirements) - 1) // len(requirements), - ) - subtasks = [] - - for index, requirement in enumerate(requirements): - position = ( - "before" - if index < active_index - else "after" - if index > active_index - else "active" - ) - subtask_start = scheduled_at + timedelta(minutes=index * subtask_duration) - execution_end = subtask_start + timedelta(minutes=subtask_duration) - reply_end = execution_end + timedelta(minutes=template["reply_window_minutes"]) - dispatch_status, execution_status, reply_status = _stage_statuses( - phase, position - ) - subtasks.append( - { - "id": f"{template['id']}-subtask-{index + 1}", - "title": requirement, - "summary": f"完成“{requirement}”并保留可复核的阶段记录。", - "status": _subtask_status(phase, position), - "scheduled_at": subtask_start, - "location": template["location"], - "assignee": template["assignee"], - "duration_minutes": subtask_duration, - "reply_window_minutes": template["reply_window_minutes"], - "stages": [ - { - "id": "dispatch", - "sequence": 1, - "title": "派发", - "status": dispatch_status, - "owner": template["dispatcher"], - "time_label": _clock( - subtask_start - - timedelta(minutes=template["dispatch_lead_minutes"]) - ), - "description": ( - f"确认{template['location']}任务范围、优先级和到场窗口，" - f"派发给{template['assignee']}。" - ), - }, - { - "id": "execution", - "sequence": 2, - "title": "执行中", - "status": execution_status, - "owner": template["assignee"], - "time_label": _clock_range(subtask_start, subtask_duration), - "description": template["summary"], - }, - { - "id": "reply", - "sequence": 3, - "title": "待复令", - "status": reply_status, - "owner": f"{template['assignee']} → 调度中心", - "time_label": _clock_range( - execution_end, - template["reply_window_minutes"], - ), - "description": ( - "提交处置结果、现场照片和影响复核，" - f"最迟 {_clock(reply_end)} 前完成复令。" - ), - }, - ], - "reply_requirements": [f"回传“{requirement}”的执行结果"], - } - ) - - return subtasks +def _build_activity(template: dict, scheduled_at: datetime, updated_at: datetime) -> list[dict]: + created_at = scheduled_at - timedelta(minutes=template["dispatch_lead_minutes"]) + activity = [{"sequence": 1, "occurred_at": created_at, + "actor": template["dispatcher"], "kind": "created", + "title": "演示工单创建", "detail": template["summary"]}] + if template["phase"] != "dispatch": + activity.append({"sequence": 2, "occurred_at": updated_at, + "actor": template["assignee"], "kind": "updated", + "title": "演示状态更新", "detail": template["detail"]}) + return activity def build_demo_work_orders() -> list[dict]: @@ -256,7 +153,10 @@ def build_demo_work_orders() -> list[dict]: "source": template["source"], "dispatcher": template["dispatcher"], "priority": template["priority"], - "subtasks": _build_subtasks(template, scheduled_at), + "location": template["location"], + "assignee": template["assignee"], + "feedback_requirements": template["reply_requirements"], + "activity": _build_activity(template, scheduled_at, updated_at), } documents.append( WorkbenchRuntimeWorkOrder.model_validate(projected).model_dump(mode="json") @@ -352,7 +252,9 @@ def main() -> int: "code": document["code"], "title": document["title"], "status": document["status"], - "subtask_count": len(document["subtasks"]), + "feedback_requirement_count": len( + document["feedback_requirements"] + ), } for document in documents ], diff --git a/scripts/sync_peripheral_work_orders.py b/scripts/sync_peripheral_work_orders.py index 76533db..95bfe45 100644 --- a/scripts/sync_peripheral_work_orders.py +++ b/scripts/sync_peripheral_work_orders.py @@ -10,6 +10,8 @@ from uuid import NAMESPACE_URL, uuid5 import httpx +from samples.peripheral.seed_identity import SEED_ACTOR_ID + try: from scripts.run_operational_worker import CredentialProvider, WorkerSettings except ModuleNotFoundError: @@ -19,18 +21,34 @@ except ModuleNotFoundError: logger = logging.getLogger("peripheral-task-sync") STATUS = {"draft": "scheduled", "open": "assigned", "in_progress": "in_progress", "awaiting_reply": "awaiting_feedback", "completed": "completed", "cancelled": "cancelled"} +EVENT_WORK_ORDER_TYPES = frozenset({"water_quality_complaint", "burst_report"}) + + +def should_project_work_order(order: dict, linked_order_ids: set[str]) -> bool: + """Exclude retired seed checks while preserving user and workflow orders.""" + work_order_id = order["work_order_id"] + work_order_type = order.get("work_order_type", "general") + if work_order_type in EVENT_WORK_ORDER_TYPES or work_order_id in linked_order_ids: + return True + return not ( + work_order_type == "general" + and order.get("origin") == "synthetic" + and order.get("created_by") == str(SEED_ACTOR_ID) + ) def work_order_snapshot(order: dict) -> dict: - """Preserve external process facts without deriving stages from total status.""" + """Preserve the flat order and its recorded activity without inventing stages.""" required = ( "work_order_id", "revision", "title", "summary", "status", "priority", "origin", "updated_at", ) snapshot = {key: order[key] for key in required} snapshot.update({key: order.get(key) for key in ( - "assignee", "scheduled_at", "location", "scenario_code", "subtasks", + "assignee", "scheduled_at", "location", "scenario_code", )}) + snapshot["feedback_requirements"] = order.get("feedback_requirements") or [] + snapshot["activity"] = order.get("activity") or [] snapshot["work_order_type"] = order.get("work_order_type", "general") return snapshot @@ -39,7 +57,10 @@ async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, periphe credentials: CredentialProvider) -> int: headers = await credentials.headers(client) tasks_url = f"{settings.server_url.rstrip('/')}/api/v1/operational-tasks" - existing, event_tasks, investigations, offset = {}, {}, [], 0 + existing: dict[str, dict] = {} + event_tasks: dict[str | None, dict] = {} + investigations: list[dict] = [] + offset = 0 while True: response = await client.get(tasks_url, headers=headers, params={"limit": 100, "offset": offset}) response.raise_for_status() @@ -63,6 +84,7 @@ async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, periphe receipt = action.get("receipt") if action["kind"] == "create_work_order" and receipt and receipt["adapter"] == "peripheral_sample": linked.setdefault(receipt["receipt_id"], []).append(task["task_id"]) + linked_order_ids = set(linked) cursor, synchronized = None, 0 while True: @@ -76,14 +98,16 @@ async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, periphe for order in page["items"]: if order["project_id"] != settings.project_id or order["origin"] != "synthetic": raise ValueError("peripheral work order scope or origin mismatch") + if not should_project_work_order(order, linked_order_ids): + continue external_id = f"peripheral_sample:{order['work_order_id']}" # Complaint intake is an event even if no browser was open when the # external system created it. Use the same source key as the UI. event_kind = order.get("work_order_type") if ( - event_kind in {"water_quality_complaint", "burst_report"} + event_kind in EVENT_WORK_ORDER_TYPES and order["work_order_id"] not in event_tasks - and order["work_order_id"] not in linked + and order["work_order_id"] not in linked_order_ids ): event_request_id = str(uuid5(NAMESPACE_URL, f"{settings.project_id}:{external_id}:investigation")) event = await client.post(tasks_url, headers=headers, json={ diff --git a/tests/integration/test_synthetic_check_cleanup.py b/tests/integration/test_synthetic_check_cleanup.py new file mode 100644 index 0000000..732dd03 --- /dev/null +++ b/tests/integration/test_synthetic_check_cleanup.py @@ -0,0 +1,326 @@ +"""Opt-in boundary checks for the synthetic-check cleanup migration.""" + +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from uuid import UUID, uuid4 + +import psycopg +import pytest +from psycopg.rows import dict_row + +from app.core.config import get_pgconn_string +from samples.peripheral.seed_identity import SEED_ACTOR_ID + + +pytestmark = pytest.mark.skipif( + os.environ.get("OPERATIONAL_TASKS_STORAGE_TEST") != "1", + reason="explicit PostgreSQL test opt-in required", +) + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +def _migration_sql(*names: str) -> str: + root = Path(__file__).parents[2] + sql = "\n".join((root / f"resources/sql/{name}").read_text() for name in names) + return sql.replace("\nBEGIN;", "").replace("\nCOMMIT;", "") + + +async def _insert_task( + conn: psycopg.AsyncConnection, + *, + project_id: UUID, + external_id: str, + event_kind: str, + title: str, + task_type: str = "external_work_order", +) -> UUID: + task_id = uuid4() + request_id = uuid4() + execution_request_id = uuid4() + now = datetime.now(timezone.utc) + document = { + "project_id": str(project_id), + "task_id": str(task_id), + "task_type": task_type, + "workflow": "external-work-order" if task_type == "external_work_order" else event_kind, + "status": "completed", + "lifecycle_status": "completed", + "priority": "normal", + "revision": 1, + "source": { + "kind": "external_system", + "external_id": external_id, + "event_kind": event_kind, + }, + "title": title, + } + await conn.execute( + """INSERT INTO workbench.operational_tasks ( + project_id,task_id,create_request_id,task_type,workflow,status,priority, + revision,execution_generation,execution_request_id,created_at,updated_at, + document,lifecycle_status + ) VALUES (%s,%s,%s,%s,%s,'completed','normal',1,1,%s,%s,%s,%s::jsonb,'completed')""", + ( + project_id, + task_id, + request_id, + task_type, + document["workflow"], + execution_request_id, + now, + now, + json.dumps(document), + ), + ) + return task_id + + +async def _insert_source_order( + conn: psycopg.AsyncConnection, + *, + project_id: UUID, + work_order_id: UUID, + title: str, + created_by: UUID = SEED_ACTOR_ID, + deleted: bool = True, +) -> None: + now = datetime.now(timezone.utc) + document = { + "project_id": str(project_id), + "work_order_id": str(work_order_id), + "revision": 1, + "title": title, + "created_by": str(created_by), + "origin": "synthetic", + } + await conn.execute( + """INSERT INTO peripheral_sample.work_orders ( + project_id,work_order_id,revision,document,deleted_at + ) VALUES (%s,%s,1,%s::jsonb,%s)""", + (project_id, work_order_id, json.dumps(document), now if deleted else None), + ) + + +@pytest.mark.anyio +async def test_cleanup_deletes_only_unlinked_exact_namespace_and_is_idempotent(): + base_migration = _migration_sql( + "014_operational_tasks.sql", + "015_workbench_events_and_task_runs.sql", + ) + cleanup_migration = _migration_sql("018_remove_synthetic_check_task_projections.sql") + project_id = uuid4() + conn = await psycopg.AsyncConnection.connect( + get_pgconn_string(), row_factory=dict_row, connect_timeout=3 + ) + try: + async with conn.transaction(force_rollback=True): + await conn.execute(base_migration) + await conn.execute("CREATE SCHEMA IF NOT EXISTS peripheral_sample") + await conn.execute( + """CREATE TABLE IF NOT EXISTS peripheral_sample.work_orders ( + project_id uuid NOT NULL, + work_order_id uuid NOT NULL, + revision bigint NOT NULL, + document jsonb NOT NULL, + deleted_at timestamptz, + PRIMARY KEY (project_id,work_order_id) + )""" + ) + removed_order_id = uuid4() + await _insert_source_order( + conn, + project_id=project_id, + work_order_id=removed_order_id, + title="合成通信核查", + deleted=False, + ) + removed_id = await _insert_task( + conn, + project_id=project_id, + external_id=f"peripheral_sample:{removed_order_id}", + event_kind="general", + title=" 合成通信核查", + ) + deleted_artifact_order_id = uuid4() + await _insert_source_order( + conn, + project_id=project_id, + work_order_id=deleted_artifact_order_id, + title="合成联调：预约工单编辑已对齐", + created_by=uuid4(), + ) + deleted_user_order_id = await _insert_task( + conn, + project_id=project_id, + external_id=f"peripheral_sample:{deleted_artifact_order_id}", + event_kind="general", + title="合成联调：预约工单编辑已对齐", + ) + suffix_alias_id = await _insert_task( + conn, + project_id=project_id, + external_id=f"peripheral_sample:{removed_order_id}:suffix", + event_kind="general", + title="合成但不是规范来源标识", + ) + similar_namespace_id = await _insert_task( + conn, + project_id=project_id, + external_id=f"peripheralXsample:{uuid4()}", + event_kind="general", + title="合成但属于其他命名空间", + ) + event_order_id = uuid4() + await _insert_source_order( + conn, + project_id=project_id, + work_order_id=event_order_id, + title="合成水质投诉", + ) + event_id = await _insert_task( + conn, + project_id=project_id, + external_id=f"peripheral_sample:{event_order_id}", + event_kind="water_quality_complaint", + title="合成水质投诉", + ) + linked_order_id = uuid4() + await _insert_source_order( + conn, + project_id=project_id, + work_order_id=linked_order_id, + title="合成现场采样", + ) + linked_id = await _insert_task( + conn, + project_id=project_id, + external_id=f"peripheral_sample:{linked_order_id}", + event_kind="general", + title="合成现场采样", + ) + normal_order_id = uuid4() + await _insert_source_order( + conn, + project_id=project_id, + work_order_id=normal_order_id, + title="常规预约", + created_by=uuid4(), + ) + normal_id = await _insert_task( + conn, + project_id=project_id, + external_id=f"peripheral_sample:{normal_order_id}", + event_kind="general", + title="常规预约", + ) + action_id = uuid4() + now = datetime.now(timezone.utc) + action_document = { + "action_id": str(action_id), + "version": 1, + "kind": "create_work_order", + "approval_status": "approved", + "execution_status": "succeeded", + "receipt": { + "adapter": "peripheral_sample", + "receipt_id": str(linked_order_id), + }, + } + await conn.execute( + """INSERT INTO workbench.operational_task_actions ( + project_id,task_id,action_id,version,approval_status,execution_status, + document,created_at,updated_at + ) VALUES (%s,%s,%s,1,'approved','succeeded',%s::jsonb,%s,%s)""", + (project_id, linked_id, action_id, json.dumps(action_document), now, now), + ) + await conn.execute( + """INSERT INTO workbench.operational_task_requests ( + project_id,scope,request_id,request_digest,response + ) VALUES (%s,%s,%s,'test',%s::jsonb)""", + ( + project_id, + f"task:{removed_id}:create", + uuid4(), + json.dumps({"task_id": str(removed_id)}), + ), + ) + await conn.execute( + """INSERT INTO workbench.operational_task_requests ( + project_id,scope,request_id,request_digest,response + ) VALUES (%s,%s,%s,'test',%s::jsonb)""", + ( + project_id, + f"notification:{uuid4()}:ack", + uuid4(), + json.dumps({"notification": {"task_id": str(removed_id)}}), + ), + ) + + await conn.execute(cleanup_migration) + + rows = await ( + await conn.execute( + """SELECT task_id FROM workbench.operational_tasks + WHERE project_id=%s AND task_id=ANY(%s)""", + ( + project_id, + [ + removed_id, + deleted_user_order_id, + suffix_alias_id, + similar_namespace_id, + event_id, + linked_id, + normal_id, + ], + ), + ) + ).fetchall() + assert {row["task_id"] for row in rows} == { + similar_namespace_id, + deleted_user_order_id, + suffix_alias_id, + event_id, + linked_id, + normal_id, + } + request_count = await ( + await conn.execute( + """SELECT count(*) AS count FROM workbench.operational_task_requests + WHERE project_id=%s AND ( + response->>'task_id'=%s + OR response#>>'{notification,task_id}'=%s + )""", + (project_id, str(removed_id), str(removed_id)), + ) + ).fetchone() + assert request_count["count"] == 0 + cleanup_event_count = await ( + await conn.execute( + """SELECT count(*) AS count FROM workbench.domain_event_log + WHERE project_id=%s + AND metadata->>'reason'='synthetic_check_projection_cleanup'""", + (project_id,), + ) + ).fetchone() + assert cleanup_event_count["count"] == 1 + + await conn.execute("DROP TABLE synthetic_check_task_cleanup") + await conn.execute(cleanup_migration) + repeated_event_count = await ( + await conn.execute( + """SELECT count(*) AS count FROM workbench.domain_event_log + WHERE project_id=%s + AND metadata->>'reason'='synthetic_check_projection_cleanup'""", + (project_id,), + ) + ).fetchone() + assert repeated_event_count["count"] == 1 + finally: + await conn.close() diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py index a792bb8..47f0aed 100644 --- a/tests/unit/test_operational_tasks.py +++ b/tests/unit/test_operational_tasks.py @@ -647,27 +647,9 @@ def _work_order_snapshot(work_order_id, now): "work_order_type": "water_quality_complaint", "location": "联调小区", "scenario_code": "WQ-SECONDARY-SUPPLY-001", - "subtasks": [{ - "id": "sampling", - "title": "现场采样", - "summary": "采集并复测", - "status": "running", - "scheduled_at": (now + timedelta(hours=1)).isoformat(), - "location": "联调小区", - "assignee": "水质班组", - "duration_minutes": 60, - "reply_window_minutes": 120, - "stages": [{ - "id": "reply", - "sequence": 1, - "title": "等待回令", - "status": "current", - "owner": "水质班组", - "time_label": "待反馈", - "description": "上传复测结果", - }], - "reply_requirements": ["上传检测结果"], - }], + "feedback_requirements": ["上传检测结果"], + "activity": [{"sequence": 1, "occurred_at": now.isoformat(), "actor": "水质班组", + "kind": "feedback", "title": "现场反馈", "detail": "上传复测结果"}], "origin": "synthetic", "updated_at": now.isoformat(), } @@ -707,7 +689,8 @@ async def test_external_feedback_exposes_validated_work_order_snapshot_and_refre assert result.task.scheduled_for == now + timedelta(hours=1) assert result.task.priority == "urgent" assert result.task.work_order_snapshot is not None - assert result.task.work_order_snapshot.subtasks[0].stages[0].status == "current" + assert result.task.work_order_snapshot.activity[0].title == "现场反馈" + assert result.task.work_order_snapshot.feedback_requirements == ["上传检测结果"] @pytest.mark.anyio diff --git a/tests/unit/test_peripheral_managed_resources.py b/tests/unit/test_peripheral_managed_resources.py index 277d662..23aa2cd 100644 --- a/tests/unit/test_peripheral_managed_resources.py +++ b/tests/unit/test_peripheral_managed_resources.py @@ -25,7 +25,7 @@ class ManagedStore: return command.model_dump(mode="json", exclude={"request_id"}) | { "work_order_id": str(uuid4()), "project_id": str(PROJECT_ID), "revision": 1, "origin": "synthetic", "created_at": now, "updated_at": now, - "created_by": str(USER_ID), "updated_by": str(USER_ID), + "created_by": str(USER_ID), "updated_by": str(USER_ID), "activity": [], } def update_work_order(self, who, work_order_id, command): @@ -35,7 +35,7 @@ class ManagedStore: "status": command.status or "open", "priority": command.priority or "normal", "assignee": None, "scheduled_at": None, "work_order_type": "general", "location": command.location, "scenario_code": command.scenario_code, - "subtasks": command.subtasks}})()) | {"revision": command.expected_revision + 1, + "feedback_requirements": command.feedback_requirements or []}})()) | {"revision": command.expected_revision + 1, "work_order_id": str(work_order_id)} def delete_work_order(self, who, work_order_id, command): @@ -86,32 +86,25 @@ def test_managed_work_order_and_data_routes_are_typed_crud(): "status": "open", "priority": "high", "assignee": None, "scheduled_at": None, "work_order_type": "water_quality_complaint", "location": "演示小区", "scenario_code": "WQ-SECONDARY-SUPPLY-001", - "subtasks": [{ - "id": "sampling", "title": "现场采样", "summary": "采集二次供水样本", - "status": "running", "scheduled_at": "2026-09-18T01:00:00Z", - "location": "演示小区", "assignee": "水质班组", "duration_minutes": 60, - "reply_window_minutes": 120, "reply_requirements": ["上传检测结果"], - "stages": [{"id": "dispatch", "sequence": 1, "title": "派发", - "status": "current", "owner": "调度员", "time_label": "待执行", - "description": "确认采样人员"}], - }], + "feedback_requirements": ["上传检测结果"], }) assert created.status_code == 201 and created.json()["origin"] == "synthetic" assert created.json()["work_order_type"] == "water_quality_complaint" assert created.json()["location"] == "演示小区" assert created.json()["scenario_code"] == "WQ-SECONDARY-SUPPLY-001" - assert created.json()["subtasks"][0]["stages"][0]["status"] == "current" + assert created.json()["feedback_requirements"] == ["上传检测结果"] + assert "subtasks" not in created.json() work_order_id = created.json()["work_order_id"] changed = client.patch(f"/api/sample/v1/work-orders/{work_order_id}", json={ "request_id": str(uuid4()), "expected_revision": 1, "status": "awaiting_reply", - "location": "复测点", "clear_scenario_code": True, "clear_subtasks": True, + "location": "复测点", "clear_scenario_code": True, "progress_note": "等待现场检测结果", "clear_assignee": False, "clear_scheduled_at": False, }) assert changed.status_code == 200 and changed.json()["revision"] == 2 assert changed.json()["status"] == "awaiting_reply" update_command = next(call[3] for call in fake.calls if call[0] == "update_work_order") assert update_command.location == "复测点" - assert update_command.clear_scenario_code and update_command.clear_subtasks + assert update_command.clear_scenario_code and update_command.progress_note == "等待现场检测结果" deleted = client.request("DELETE", f"/api/sample/v1/work-orders/{work_order_id}", json={ "request_id": str(uuid4()), "expected_revision": 2, }) diff --git a/tests/unit/test_peripheral_task_sync.py b/tests/unit/test_peripheral_task_sync.py index 2b7e41a..789ecd1 100644 --- a/tests/unit/test_peripheral_task_sync.py +++ b/tests/unit/test_peripheral_task_sync.py @@ -5,7 +5,12 @@ import httpx import pytest from scripts.run_operational_worker import CredentialProvider, WorkerSettings -from scripts.sync_peripheral_work_orders import sync_once, work_order_snapshot +from scripts.sync_peripheral_work_orders import ( + should_project_work_order, + sync_once, + work_order_snapshot, +) +from samples.peripheral.seed_identity import SEED_ACTOR_ID from app.domain.schemas.operational_task import ExternalFeedbackCommand @@ -14,6 +19,29 @@ def anyio_backend(): return "asyncio" +def test_unlinked_synthetic_checks_are_not_operational_tasks(): + order_id = str(uuid4()) + order = { + "work_order_id": order_id, + "origin": "synthetic", + "created_by": str(SEED_ACTOR_ID), + "title": "合成压力偏差复核", + "work_order_type": "general", + } + + assert not should_project_work_order(order, set()) + assert should_project_work_order(order, {order_id}) + assert should_project_work_order( + {**order, "work_order_type": "water_quality_complaint"}, set() + ) + assert should_project_work_order( + {**order, "title": "现场采样：二次供水出水及末端用户水样"}, set() + ) is False + assert should_project_work_order( + {**order, "created_by": str(uuid4()), "title": "合成管材接驳预约"}, set() + ) + + @pytest.mark.anyio async def test_sync_creates_one_task_and_only_applies_new_external_revisions(): project_id, order_id, task_id = map(str, (uuid4(), uuid4(), uuid4())) @@ -55,12 +83,52 @@ async def test_sync_creates_one_task_and_only_applies_new_external_revisions(): assert snapshot["assignee"] == "测试班组" assert snapshot["location"] == "测试位置" assert snapshot["scheduled_at"] == "2026-09-24T08:00:00Z" - assert snapshot["subtasks"] is None + assert snapshot["feedback_requirements"] == [] + assert snapshot["activity"] == [] # Exercise the real API schema, including the prefixed external identity. accepted = ExternalFeedbackCommand.model_validate(writes[-1][1]) assert str(accepted.work_order_snapshot.work_order_id) == order_id +@pytest.mark.anyio +async def test_sync_skips_unlinked_synthetic_check_before_writing(): + project_id, order_id = str(uuid4()), str(uuid4()) + order = { + "project_id": project_id, + "work_order_id": order_id, + "origin": "synthetic", + "created_by": str(SEED_ACTOR_ID), + "revision": 1, + "title": "合成通信核查", + "summary": "仅用于外围样板联调", + "priority": "normal", + "status": "open", + "created_at": "2026-09-23T08:00:00Z", + "updated_at": "2026-09-23T08:00:00Z", + "scheduled_at": None, + "work_order_type": "general", + } + + def handler(request): + assert request.method == "GET" + if request.url.host == "sample.test": + return httpx.Response(200, json={"items": [order], "next_cursor": None}) + return httpx.Response(200, json={"items": [], "next_offset": None}) + + settings = WorkerSettings( + "http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY" + ) + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + synchronized = await sync_once( + client, + settings, + "http://sample.test", + CredentialProvider(settings), + ) + + assert synchronized == 0 + + @pytest.mark.anyio async def test_sample_sync_rejects_cross_project_data_before_writing(): def handler(request): @@ -74,17 +142,29 @@ async def test_sample_sync_rejects_cross_project_data_before_writing(): @pytest.mark.anyio -@pytest.mark.parametrize("investigation_status", ["waiting_human", "queued"]) -async def test_created_work_order_receipt_routes_feedback_back_to_investigation(investigation_status): +@pytest.mark.parametrize( + ("investigation_status", "expected_feedback_count", "routes_to_investigation"), + [ + ("waiting_human", 2, True), + ("queued", 2, True), + ("completed", 1, False), + ], +) +async def test_created_work_order_receipt_preserves_projection_and_routes_active_feedback( + investigation_status, + expected_feedback_count, + routes_to_investigation, +): project_id, order_id, investigation_id, projection_id = map(str, (uuid4(), uuid4(), uuid4(), uuid4())) received = [] def handler(request): if request.url.host == "sample.test": return httpx.Response(200, json={"items": [{ "project_id": project_id, "work_order_id": order_id, "origin": "synthetic", + "created_by": str(uuid4()), "revision": 2, "summary": "采样已完成", "status": "completed", "updated_at": "2026-09-23T09:00:00Z", "title": "合成采样工单", "priority": "normal", - "work_order_type": "water_quality_complaint", "created_at": "2026-09-23T08:00:00Z", + "work_order_type": "general", "created_at": "2026-09-23T08:00:00Z", "scheduled_at": None, }], "next_cursor": None}) if request.method == "POST": @@ -106,8 +186,10 @@ async def test_created_work_order_receipt_routes_feedback_back_to_investigation( ], "next_offset": None}) settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY") async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: - assert await sync_once(client, settings, "http://sample.test", CredentialProvider(settings)) == 2 - assert any(investigation_id in path for path in received) + assert await sync_once( + client, settings, "http://sample.test", CredentialProvider(settings) + ) == expected_feedback_count + assert any(investigation_id in path for path in received) is routes_to_investigation assert any(projection_id in path for path in received) @@ -142,11 +224,20 @@ async def test_external_events_start_investigations_without_browser(event_kind): assert investigation["trigger"]["occurred_at"] == "2026-09-23T07:30:00Z" -def test_snapshot_preserves_nested_process_without_inventing_completed_stages(): +def test_snapshot_preserves_recorded_activity_without_legacy_subtasks(): order = { "work_order_id": str(uuid4()), "revision": 4, "title": "合成预约", "summary": "仅测试过程同步", "status": "in_progress", "priority": "normal", "origin": "synthetic", "updated_at": "2026-09-23T09:00:00Z", "work_order_type": "general", + "feedback_requirements": ["回填现场检查结果"], + "activity": [{ + "sequence": 1, + "occurred_at": "2026-09-23T08:00:00Z", + "actor": "测试调度", + "kind": "created", + "title": "工单创建", + "detail": "已创建现场检查工单。", + }], "subtasks": [{ "id": "visit", "title": "现场检查", "summary": "合成测试，未现场执行", "status": "pending", "scheduled_at": "2026-09-24T08:00:00Z", @@ -157,6 +248,6 @@ def test_snapshot_preserves_nested_process_without_inventing_completed_stages(): }], } snapshot = work_order_snapshot(order) - assert snapshot["subtasks"] == order["subtasks"] - assert snapshot["subtasks"][0]["stages"][0]["status"] == "pending" - assert work_order_snapshot({key: value for key, value in order.items() if key != "subtasks"})["subtasks"] is None + assert snapshot["feedback_requirements"] == ["回填现场检查结果"] + assert snapshot["activity"] == order["activity"] + assert "subtasks" not in snapshot diff --git a/tests/unit/test_peripheral_work_order_activity.py b/tests/unit/test_peripheral_work_order_activity.py new file mode 100644 index 0000000..bdb5470 --- /dev/null +++ b/tests/unit/test_peripheral_work_order_activity.py @@ -0,0 +1,49 @@ +from uuid import uuid4 + +from samples.peripheral.auth import Identity +from samples.peripheral.config import SampleSettings +from samples.peripheral.schemas import CreateWorkOrder, UpdateWorkOrder, WorkOrder +from samples.peripheral.store import PostgresStore + + +def test_create_and_update_record_flat_order_activity(monkeypatch): + store = PostgresStore(SampleSettings()) + actor = Identity(str(uuid4()), str(uuid4()), True) + saved = {} + + monkeypatch.setattr(store, "_mutate", lambda who, key, command, operation: operation(None, "normal")) + + def save(_conn, _who, document): + saved.clear() + saved.update(WorkOrder.model_validate(document).model_dump(mode="json")) + return dict(saved) + + monkeypatch.setattr(store, "_save_work_order", save) + monkeypatch.setattr(store, "_managed_document", lambda *_args: dict(saved)) + + created = store.create_work_order(actor, CreateWorkOrder( + request_id=uuid4(), title="阀门预约复核", summary="现场复核开度", + location="北区阀门", assignee="北区班组", + feedback_requirements=["回传开度照片"], + )) + assert created["feedback_requirements"] == ["回传开度照片"] + assert created["activity"][0]["kind"] == "created" + assert "subtasks" not in created + + updated = store.update_work_order(actor, created["work_order_id"], UpdateWorkOrder( + request_id=uuid4(), expected_revision=1, status="awaiting_reply", + progress_note="已复核开度，等待回执确认", + )) + assert updated["status"] == "awaiting_reply" + assert [entry["sequence"] for entry in updated["activity"]] == [1, 2] + assert updated["activity"][1]["kind"] == "feedback" + assert updated["activity"][1]["detail"] == "已复核开度，等待回执确认" + assert "progress_note" not in updated + + +def test_legacy_order_reads_flat_without_claiming_unrecorded_progress(): + legacy = {"work_order_id": str(uuid4()), "subtasks": [{"stages": [{"status": "done"}]}]} + projected = PostgresStore._active_managed_document(legacy) + assert "subtasks" not in projected + assert projected["feedback_requirements"] == [] + assert projected["activity"] == [] diff --git a/tests/unit/test_seed_peripheral_sample.py b/tests/unit/test_seed_peripheral_sample.py index 04d3e23..a92a9e9 100644 --- a/tests/unit/test_seed_peripheral_sample.py +++ b/tests/unit/test_seed_peripheral_sample.py @@ -1,12 +1,10 @@ from contextlib import contextmanager from uuid import uuid4 -import pytest - from scripts.seed_peripheral_sample import ( ACTOR_ID, previous_seed_result, - restore_seeded_resources, + restore_seeded_test_data, ) @@ -42,8 +40,8 @@ class FakeStore: def test_previous_seed_result_reuses_legacy_receipt(): resource_id = str(uuid4()) store = FakeStore({ - "command_key": "create_work_order", - "result": {"work_order_id": resource_id}, + "command_key": "create_test_data", + "result": {"record_id": resource_id}, }) seed_request_id = uuid4() @@ -51,10 +49,10 @@ def test_previous_seed_result_reuses_legacy_receipt(): store, "project-1", seed_request_id=seed_request_id, - command_key="create_work_order", + command_key="create_test_data", ) - assert result == {"work_order_id": resource_id} + assert result == {"record_id": resource_id} assert store.readonly is True assert store.connection_value.calls[0][1] == ( "project-1", @@ -63,29 +61,14 @@ def test_previous_seed_result_reuses_legacy_receipt(): ) -def test_restore_seeded_resources_only_restores_allowed_tables(): +def test_restore_seeded_test_data_clears_soft_delete_fields(): store = FakeStore() resource_ids = [str(uuid4()), str(uuid4())] - restore_seeded_resources( - store, - "project-1", - table="work_orders", - id_column="work_order_id", - resource_ids=resource_ids, - ) + restore_seeded_test_data(store, "project-1", resource_ids) statement, parameters = store.connection_value.calls[0] - assert "UPDATE peripheral_sample.work_orders" in statement + assert "UPDATE peripheral_sample.test_data_records" in statement assert "document=document - 'deleted_at' - 'deleted_by'" in statement assert "deleted_at=NULL, deleted_by=NULL" in statement assert parameters == ("project-1", resource_ids) - - with pytest.raises(ValueError, match="unsupported seeded resource"): - restore_seeded_resources( - store, - "project-1", - table="requests", - id_column="request_id", - resource_ids=resource_ids, - ) diff --git a/tests/unit/test_seed_workbench_work_orders.py b/tests/unit/test_seed_workbench_work_orders.py index 14e2567..834ba2f 100644 --- a/tests/unit/test_seed_workbench_work_orders.py +++ b/tests/unit/test_seed_workbench_work_orders.py @@ -28,25 +28,18 @@ def test_build_demo_work_orders_restores_original_mock_tasks(): "running", "awaiting_reply", ] - assert all(len(document["subtasks"]) == 3 for document in documents) + assert all(len(document["feedback_requirements"]) == 3 for document in documents) + assert all("subtasks" not in document for document in documents) assert all("合成" not in str(document) for document in documents) -def test_demo_work_orders_keep_one_current_stage_per_active_task(): +def test_demo_work_orders_show_recorded_activity_without_synthetic_stages(): documents = build_demo_work_orders() for document in documents: - current_stages = [ - stage - for subtask in document["subtasks"] - for stage in subtask["stages"] - if stage["status"] == "current" - ] - assert len(current_stages) == 1 - - assert documents[0]["subtasks"][0]["stages"][0]["status"] == "current" - assert documents[1]["subtasks"][1]["stages"][1]["status"] == "current" - assert documents[2]["subtasks"][2]["stages"][2]["status"] == "current" + assert document["activity"] + assert all(entry["kind"] in {"created", "updated"} for entry in document["activity"]) + assert "stages" not in document def test_replace_removes_legacy_projection_before_inserting_demo_tasks(): diff --git a/tests/unit/test_workbench_runtime.py b/tests/unit/test_workbench_runtime.py index 6b7a29f..a37315d 100644 --- a/tests/unit/test_workbench_runtime.py +++ b/tests/unit/test_workbench_runtime.py @@ -201,47 +201,11 @@ async def test_snapshot_projects_persisted_work_orders_and_alerts(): "source": "生产工单系统", "dispatcher": "调度中心", "priority": "urgent", - "subtasks": [{ - "id": "inspect-valve", - "title": "阀门开度复核", - "summary": "现场复核阀门开度。", - "status": "running", - "scheduled_at": NOW.replace(hour=3), - "location": "北区阀门 V-1", - "assignee": "北区班组", - "duration_minutes": 45, - "reply_window_minutes": 20, - "stages": [ - { - "id": "dispatch", - "sequence": 1, - "title": "派发", - "status": "done", - "owner": "调度中心", - "time_label": "10:20", - "description": "已派发。", - }, - { - "id": "execution", - "sequence": 2, - "title": "执行中", - "status": "current", - "owner": "北区班组", - "time_label": "10:30—11:15", - "description": "现场执行中。", - }, - { - "id": "reply", - "sequence": 3, - "title": "待复令", - "status": "pending", - "owner": "北区班组 → 调度中心", - "time_label": "11:15—11:35", - "description": "等待执行结果。", - }, - ], - "reply_requirements": ["回传阀门开度", "复核压力"], - }], + "location": "北区阀门 V-1", + "assignee": "北区班组", + "feedback_requirements": ["回传阀门开度", "复核压力"], + "activity": [{"sequence": 1, "occurred_at": NOW, "actor": "调度中心", + "kind": "created", "title": "工单创建", "detail": "等待现场复核。"}], } persisted_alert = { "alert_id": "alarm-1", @@ -282,3 +246,5 @@ async def test_snapshot_projects_persisted_work_orders_and_alerts(): "severity": "warning", }] assert snapshot["work_orders"][0]["code"] == "WO-20260917-001" + assert snapshot["work_orders"][0]["activity"][0]["title"] == "工单创建" + assert "subtasks" not in snapshot["work_orders"][0]