From db092b0f5e0ab033f553221a2e78bd5e24db054a Mon Sep 17 00:00:00 2001 From: Huarch Date: Thu, 24 Sep 2026 15:18:18 +0800 Subject: [PATCH] feat(workbench): keep only actionable notifications --- app/domain/schemas/operational_task.py | 4 +- app/services/operational_tasks.py | 102 +++++++++++++----- contracts/manifest.json | 2 +- contracts/server-v1.openapi.json | 6 +- docs/operational-notification-routing.md | 30 ++++++ ...020_prune_non_actionable_notifications.sql | 51 +++++++++ .../test_operational_tasks_storage.py | 25 +++-- tests/unit/test_operational_tasks.py | 10 ++ 8 files changed, 192 insertions(+), 38 deletions(-) create mode 100644 docs/operational-notification-routing.md create mode 100644 resources/sql/020_prune_non_actionable_notifications.sql diff --git a/app/domain/schemas/operational_task.py b/app/domain/schemas/operational_task.py index 0bddabd..859b68e 100644 --- a/app/domain/schemas/operational_task.py +++ b/app/domain/schemas/operational_task.py @@ -498,7 +498,9 @@ class OperationalTaskNotification(StrictModel): task_id: UUID kind: Literal[ "task_created", "action_pending", "report_updated", "task_completed", - "task_failed", "run_delivery_degraded" + "task_failed", "run_delivery_degraded", "access_approval_pending", + "action_approval_pending", "human_input_required", + "external_coordination_failed" ] title: NonEmptyStr description: NonEmptyStr diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index d0a76ea..c781780 100644 --- a/app/services/operational_tasks.py +++ b/app/services/operational_tasks.py @@ -179,9 +179,6 @@ class OperationalTaskService: run = self._new_run(task_id, 1, now) await self.repo.insert_run(_json(run)) await self._run_event(run, now, queued=True) - await self._notify( - task_id, "task_created", command.title, "工况任务已进入待执行队列", now - ) await self.repo.save_request( scope, request_id, digest(body), {"task_id": str(task_id)} ) @@ -602,10 +599,6 @@ class OperationalTaskService: occurred_at=now, metadata={"task_id": str(task_id), "run_id": str(active_run["run_id"])}, ) - await self._notify( - task_id, "report_updated", command.report.title, - command.report.current_conclusion, now, - ) if current_plan is None: stored_plan = await self.repo.get_latest_plan(str(task_id)) if stored_plan is not None: @@ -704,7 +697,7 @@ class OperationalTaskService: ) await self._task_event(task, now) await self._notify( - task_id, "action_pending", task["title"], command.reason, now, + task_id, "action_approval_pending", task["title"], command.reason, now, required_permission=approval_permission, ) await self._save_request( @@ -783,9 +776,27 @@ class OperationalTaskService: ) task = await self.detail(task_id) - outcome = await dispatcher.dispatch( - context=self.context, task=task, action=action, credential=credential - ) + try: + outcome = await dispatcher.dispatch( + context=self.context, task=task, action=action, credential=credential + ) + except Exception: + async with self.repo.transaction(): + await self.repo.lock(f"action:{action_id}") + current = await self._action(task_id, action_id) + if ( + current.version == action.version + and current.execution_status == "unknown" + ): + await self._notify( + task_id, + "external_coordination_failed", + task.title, + "外部系统提交中断，执行结果未知，请核对回执", + self._now(), + required_permission=action.required_execution_permission, + ) + raise async with self.repo.transaction(): await self.repo.lock(f"request:{scope}:{command.request_id}") @@ -801,14 +812,24 @@ class OperationalTaskService: "execution_detail": outcome.detail, }) await self.repo.update_action(str(task_id), _json(current)) - await self._action_event(current, self._now()) + outcome_recorded_at = self._now() + await self._action_event(current, outcome_recorded_at) + if outcome.status in {"unknown", "failed", "unsupported"}: + await self._notify( + task_id, + "external_coordination_failed", + task.title, + outcome.detail, + outcome_recorded_at, + required_permission=action.required_execution_permission, + ) if outcome.status in {"succeeded", "failed", "unsupported"}: task_document = await self._task(task_id) - await self._requeue_task(task_document, self._now()) + await self._requeue_task(task_document, outcome_recorded_at) await self.repo.update_task( _task_json(OperationalTaskSummary.model_validate(task_document)) ) - await self._task_event(task_document, self._now()) + await self._task_event(task_document, outcome_recorded_at) await self._save_request(scope, command.request_id, body, {"action": _json(current)}) return current @@ -842,19 +863,34 @@ class OperationalTaskService: current = await self._action(task_id, action_id) require(current.version == action.version, "VERSION_CONFLICT", "Action changed while reconciling", status=409) + previous_execution_status = current.execution_status current = current.model_copy(update={ "version": current.version + 1, "execution_status": outcome.status, "receipt": outcome.receipt, "execution_detail": outcome.detail, }) await self.repo.update_action(str(task_id), _json(current)) - await self._action_event(current, self._now()) + outcome_recorded_at = self._now() + await self._action_event(current, outcome_recorded_at) + if ( + outcome.status in {"unknown", "failed", "unsupported"} + and outcome.status != previous_execution_status + ): + task = await self._task(task_id) + await self._notify( + task_id, + "external_coordination_failed", + task["title"], + outcome.detail, + outcome_recorded_at, + required_permission=action.required_execution_permission, + ) if outcome.status in {"succeeded", "failed", "unsupported"}: task_document = await self._task(task_id) - await self._requeue_task(task_document, self._now()) + await self._requeue_task(task_document, outcome_recorded_at) await self.repo.update_task( _task_json(OperationalTaskSummary.model_validate(task_document)) ) - await self._task_event(task_document, self._now()) + await self._task_event(task_document, outcome_recorded_at) await self._save_request( scope, command.request_id, body, {"action": _json(current)} ) @@ -961,9 +997,9 @@ class OperationalTaskService: occurred_at=now, metadata={"task_id": str(task_id), "namespace": "operational-task"}, ) - if task["status"] in {"completed", "failed"}: + if task["status"] == "failed": await self._notify( - task_id, "task_completed" if task["status"] == "completed" else "task_failed", + task_id, "task_failed", task["title"], command.detail, now, ) response = { @@ -1195,9 +1231,6 @@ class OperationalTaskService: "version": schedule.version + 1, "updated_at": now, }) await self.repo.advance_schedule(_json(next_schedule)) - await self._notify( - task_id, "task_created", task.title, "定时流程已进入待执行队列", now - ) await self._run_event(run, now, queued=True) await self._task_event(_task_json(task), now) task_ids.append(task_id) @@ -1283,14 +1316,33 @@ class OperationalTaskService: await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task))) await self._run_event(run_model, now) await self._task_event(task, now) - if lifecycle in {"completed", "failed"}: + if lifecycle == "failed": await self._notify( task_id, - "task_completed" if lifecycle == "completed" else "task_failed", + "task_failed", task["title"], - "工况任务流程状态已更新", + failure_detail or "工况任务执行失败", now, ) + elif lifecycle == "waiting_human": + pending_actions = any( + action.approval_status == "pending" for action in actions + ) + pending_access = any( + request.get("status") == "pending" + for request in ( + await self.repo.list_access_requests(str(task_id)) + if hasattr(self.repo, "list_access_requests") else [] + ) + ) + if not pending_actions and not pending_access: + await self._notify( + task_id, + "human_input_required", + task["title"], + "任务等待人工补充信息或业务确认", + now, + ) await self._save_request(scope, command.request_id, body, {"task_id": str(task_id)}) return await self.detail(task_id) diff --git a/contracts/manifest.json b/contracts/manifest.json index be183dc..392fe6a 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "server": { "file": "server-v1.openapi.json", - "sha256": "c37f5626398e1762584a844bb908ca4f4fb68e203f4d37a489561669a434312b" + "sha256": "4a7ef16dc9ebf9731359c314ab600730dc9b06dd4b0396c0e4e847d57063ee12" } } } diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json index b91e6e6..c5dc38c 100644 --- a/contracts/server-v1.openapi.json +++ b/contracts/server-v1.openapi.json @@ -3624,7 +3624,11 @@ "report_updated", "task_completed", "task_failed", - "run_delivery_degraded" + "run_delivery_degraded", + "access_approval_pending", + "action_approval_pending", + "human_input_required", + "external_coordination_failed" ], "title": "Kind", "type": "string" diff --git a/docs/operational-notification-routing.md b/docs/operational-notification-routing.md new file mode 100644 index 0000000..4f92b5d --- /dev/null +++ b/docs/operational-notification-routing.md @@ -0,0 +1,30 @@ +# 异常处置通知路由 + +“异常处置”只展示需要当前用户判断、授权或处置的活动事项。通知表仍是用户收件箱，不替代 `domain_event_log`；列表按当前用户权限和已读状态过滤。 + +## 进入异常处置的类型 + +| 分类 | 通知类型 | 严重度 | 说明 | +| --- | --- | --- | --- | +| 任务异常 | `task_failed`、`run_delivery_degraded` | 严重 | 任务失败、连续超时、产物不完整或 Agent 投递持续失败 | +| 人工审核 | `access_approval_pending`、`action_approval_pending` | 警告 | 数据访问授权、工单派发或设备动作审批 | +| 业务告警 | 活动的 `workbench.alerts` | 告警定义 | 压力、流量、水质、设备或通信等仍未解除的业务异常 | +| 人工补充 | `human_input_required` | 警告 | 计划存在等待人工处理的节点，且没有更具体的待审批动作或访问申请 | +| 外部协同异常 | `external_coordination_failed` | 严重 | 外部动作回执未知、提交失败或适配器不支持，需要人工核对 | + +兼容读取旧 `action_pending`，并按“人工审核”展示。它不再由新代码写入。 + +## 不进入异常处置的类型 + +- `task_created`：任务创建后直接进入工况列表。 +- `report_updated`：报告修订只更新任务详情。 +- `task_completed`：正常完成不产生异常事项。 +- queued、running 等普通状态变化和短暂 SSE 重连不创建用户通知。 + +新写入不再产生上述三类普通生命周期通知。`020_prune_non_actionable_notifications.sql` 只删除历史 `task_created`、`report_updated` 和 `task_completed` 通知及其级联已读记录，并为受影响项目写入一次批量失效事件。任务、报告、动作、业务告警和事件历史均不删除。 + +## 数据库应用范围 + +2026-09-24 本轮只允许在 `tjwater_v2` 执行 `020`。执行前必须确认 `current_database()`、完成备份，并在事务后核对：普通生命周期通知为零，审批、任务异常和外部协同异常仍保留。其他项目数据库待 `tjwater_v2` 验证通过后再单独评估，不批量应用。 + +本次实际执行结果：删除 16 条普通生命周期通知，保留 2 条旧 `action_pending` 审批通知，活动业务告警仍为 0 条，并写入 `event_seq=39` 的批量失效事件。通知表和用户已读表的迁移前逻辑备份位于 `/home/zhifu/TJWaterBackups/notification-routing-20260924-gZvjsn/tjwater_v2-notifications-before-020.json`，SHA-256 为 `046d6c3a0888b981791f3f39ee1972f304aab18a7e6af035da742c069b97a369`。系统缺少 `pg_dump` 客户端，因此本轮没有生成新的整库 custom dump；上一轮整库备份未被覆盖。 diff --git a/resources/sql/020_prune_non_actionable_notifications.sql b/resources/sql/020_prune_non_actionable_notifications.sql new file mode 100644 index 0000000..c1cebad --- /dev/null +++ b/resources/sql/020_prune_non_actionable_notifications.sql @@ -0,0 +1,51 @@ +-- Remove historical task lifecycle notices that do not require user action. + +BEGIN; + +CREATE TEMP TABLE notification_cleanup_projects ON COMMIT DROP AS +SELECT + project_id, + count(*)::bigint AS removed_count +FROM workbench.operational_task_notifications +WHERE document->>'kind' IN ('task_created', 'report_updated', 'task_completed') +GROUP BY project_id; + +DELETE FROM workbench.operational_task_notifications +WHERE document->>'kind' IN ('task_created', 'report_updated', 'task_completed'); + +WITH inserted_events AS ( + 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 || ':notification-routing-v1'), 1, 8) + || '-' || substr(md5(project_id::text || ':notification-routing-v1'), 9, 4) + || '-5' || substr(md5(project_id::text || ':notification-routing-v1'), 14, 3) + || '-a' || substr(md5(project_id::text || ':notification-routing-v1'), 18, 3) + || '-' || substr(md5(project_id::text || ':notification-routing-v1'), 21, 12) + )::uuid, + 'operational.notification.changed', + 'operational_notification_batch', + project_id::text, + removed_count, + jsonb_build_object( + 'change', 'batch_pruned', + 'reason', 'non_actionable_notification_cleanup', + 'removed_count', removed_count + ) + FROM notification_cleanup_projects + ON CONFLICT (project_id, event_id) DO NOTHING + RETURNING project_id, event_seq +) +SELECT pg_notify('workbench_events', project_id::text || ':' || event_seq::text) +FROM inserted_events; + +COMMIT; diff --git a/tests/integration/test_operational_tasks_storage.py b/tests/integration/test_operational_tasks_storage.py index f5b34e7..9781c27 100644 --- a/tests/integration/test_operational_tasks_storage.py +++ b/tests/integration/test_operational_tasks_storage.py @@ -132,11 +132,16 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( assert member_notifications assert all(item.task_id == task.task_id for item in member_notifications) action_pending = next( - item for item in member_notifications if item.kind == "action_pending" + item for item in member_notifications + if item.kind == "action_approval_pending" ) - task_created = next( - item for item in member_notifications if item.kind == "task_created" + human_input = next( + item for item in member_notifications + if item.kind == "human_input_required" ) + assert not any(item.kind in { + "task_created", "report_updated", "task_completed" + } for item in member_notifications) assert action_pending.required_permission == action.required_approval_permission viewer_context = ProjectContext( @@ -153,7 +158,7 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( viewer_notifications, _ = await viewer_service.list_notifications( unread_only=True, limit=100, offset=0 ) - assert task_created.notification_id in { + assert human_input.notification_id in { item.notification_id for item in viewer_notifications } assert action_pending.notification_id not in { @@ -161,9 +166,9 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( } acknowledged = await service.acknowledge_notification( - task_created.notification_id, + human_input.notification_id, NotificationAcknowledgeCommand( - request_id=uuid4(), expected_version=task_created.version + request_id=uuid4(), expected_version=human_input.version ), ) assert acknowledged.status == "read" @@ -179,21 +184,21 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( ) persisted_created = next( item for item in persisted_notifications - if item.notification_id == task_created.notification_id + if item.notification_id == human_input.notification_id ) assert persisted_created.status == "read" - assert persisted_created.version == task_created.version + 1 + assert persisted_created.version == human_input.version + 1 assert persisted_created.read_by == f"principal:{context.user_id}" unread_notifications, _ = await persisted_service.list_notifications( unread_only=True, limit=100, offset=0 ) - assert task_created.notification_id not in { + assert human_input.notification_id not in { item.notification_id for item in unread_notifications } viewer_after_member_ack, _ = await viewer_service.list_notifications( unread_only=True, limit=100, offset=0 ) - assert task_created.notification_id in { + assert human_input.notification_id in { item.notification_id for item in viewer_after_member_ack } action = await service.decide_action(task.task_id, action.action_id, ActionDecisionCommand( diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py index 47f0aed..0736b8d 100644 --- a/tests/unit/test_operational_tasks.py +++ b/tests/unit/test_operational_tasks.py @@ -357,6 +357,7 @@ async def test_complete_without_same_generation_artifacts_fails_run_explicitly() assert repository.run["failure_code"] == "ARTIFACT_INCOMPLETE" assert completed.lifecycle_status == "failed" assert completed.active_run is None + assert [item["kind"] for item in repository.notifications] == ["task_failed"] @pytest.mark.anyio @@ -451,6 +452,9 @@ async def test_plan_only_keeps_lease_until_same_generation_report_closes_run(): ) assert completed.status == "waiting_human" assert completed.agent_lease_expires_at is None + assert [item["kind"] for item in repository.notifications] == [ + "human_input_required" + ] @pytest.mark.anyio @@ -521,6 +525,9 @@ async def test_action_proposal_keeps_running_lease_until_plan_and_report_close_r ) assert closed.status == "waiting_human" assert closed.actions[0].action_id == action.action_id + assert [item["kind"] for item in repository.notifications] == [ + "action_approval_pending" + ] @pytest.mark.anyio @@ -786,6 +793,9 @@ async def test_interrupted_submit_replays_unknown_without_dispatching_twice(): assert dispatcher.calls == 1 assert replay.execution_status == "unknown" assert replay.execution_request_id is not None + assert [item["kind"] for item in repository.notifications] == [ + "external_coordination_failed" + ] @pytest.mark.anyio