190 lines
18 KiB
Markdown
190 lines
18 KiB
Markdown
# 工况工作进程与外部动作适配
|
||
|
||
工况执行不依赖浏览器保活。API 保存任务及审批事实，工作进程扫描持久化队列，再向 Agent 提交后端分配的执行请求编号。相同请求编号重试复用同一执行，不能根据轮询时间生成新的编号。
|
||
|
||
## 调查任务队列
|
||
|
||
第二轮及后续调查的触发条件、待执行轮次合并、旧运行替代，以及协作与消息的边界，见 [调查轮次与协作边界](operational-tasks.md#调查轮次与协作边界)。
|
||
|
||
worker 始终只处理 `event_investigation`，外部预约工单不占用模型并发。
|
||
执行列表支持可选 `task_type`、`task_created_from` 查询参数；租约恢复请求支持同名字段，
|
||
均在数据库分页之前过滤。省略字段保持原接口行为。时间下界使用任务创建时间，
|
||
旧任务产生新 run 不会越过边界，不删除历史或修改旧任务状态来缩小执行范围。
|
||
|
||
本机常驻模板位于 `infra/systemd/tjwater-operational-worker.service`，使用已有 `.env`
|
||
凭据与配置。安装到用户 systemd 后启用，避免同时启动第二个 worker。
|
||
用户服务未启用 linger 时，不保证退出登录后持续运行。
|
||
|
||
当前产品分为定时任务、调查任务、预约任务。定时任务保留前端 mock 数据，不调用后端计划创建或到期调度；调查任务读取数据库并进入本工作队列；预约任务读取数据库中的外围工单投影，以外部同步进度为准。已有后端 schedule API 暂不在当前产品主路径使用。
|
||
|
||
从后端仓库运行：
|
||
|
||
```bash
|
||
conda run -n server python scripts/run_operational_worker.py --once
|
||
conda run -n server python scripts/run_operational_worker.py
|
||
```
|
||
|
||
部署环境提供以下变量，不将实际值写入仓库。工作进程也读取后端仓库已忽略的 `.env`，已存在的进程环境变量优先：
|
||
|
||
| 变量 | 用途 |
|
||
| --- | --- |
|
||
| `OPERATIONAL_WORKER_PROJECT_ID` | 单项目模式的授权项目 UUID |
|
||
| `OPERATIONAL_WORKER_PROJECT_IDS` | 可选，显式授权项目 UUID 列表，逗号分隔；非空时优先于单项目配置，同步器和调查 worker 均按列表运行 |
|
||
| `OPERATIONAL_WORKER_SERVER_URL` | 主后端地址，默认本机 8000 |
|
||
| `OPERATIONAL_WORKER_AGENT_URL` | Agent 地址，默认本机 8787 |
|
||
| `OPERATIONAL_WORKER_RECONCILE_SECONDS` | SSE 中断或漏通知时的对账间隔，默认 30 秒 |
|
||
| `OPERATIONAL_WORKER_MAX_CONCURRENCY` | 同时投递的 run 上限，默认 4，可设置 1 至 32 |
|
||
| `OPERATIONAL_WORKER_TASK_CREATED_FROM` | 可选、带时区的任务创建时间下界（包含边界），限制排队、并发计数、凭据刷新和租约恢复；不是 run 创建时间 |
|
||
| `OPERATIONAL_WORKER_BEARER` | 已具有该项目权限的产品身份，仅用于受控开发或短期运行 |
|
||
| `OPERATIONAL_WORKER_TOKEN_URL` | 可选 OAuth 客户端凭据取令牌地址 |
|
||
| `OPERATIONAL_WORKER_DEV_TOKEN_URL` | 仅 `ENVIRONMENT=development` 且非空时覆盖取令牌地址；其他环境忽略，保留正式地址 |
|
||
| `OPERATIONAL_WORKER_CLIENT_ID`、`OPERATIONAL_WORKER_CLIENT_SECRET` | 长期运行身份；不配置固定 Bearer 时使用，并按有效期刷新令牌 |
|
||
| `OPERATIONAL_WORKER_TOKEN_REFRESH_SECONDS` | OAuth 缓存上限，默认 300 秒，可设置 30 至 600 秒；同时受令牌 expires_in 限制，应短于后端允许的最大令牌年龄 |
|
||
|
||
工作身份必须能通过现有产品身份与项目成员权限校验。配置令牌获取不等于自动创建产品用户、项目成员或授权。权限错误会保留任务用于后续重试，不会绕过认证。
|
||
|
||
多项目模式为每个项目维护独立事件游标、会话、消息游标及 OAuth 缓存，单项目暂时失败不阻止其他项目对账。`OPERATIONAL_WORKER_MAX_CONCURRENCY` 是每个项目的上限，总上限随显式配置的项目数增加；需要控制整体模型用量时相应调低。进程不自动枚举所有项目或扩大服务身份权限。修改项目列表后需同时重启 `tjwater-operational-worker.service` 和 `tjwater-peripheral-sync.service`。
|
||
|
||
调查数据读取沿用账号已有的项目权限，不再因 `service` 身份额外移除 `scada.view` 和
|
||
`simulation.view`。上下文的 `approved_data_scopes` 与数据查询采用同一权限集合；已有读取
|
||
权限无需逐任务数据审批或委托凭据。此规则不新增角色权限、不放开跨项目读取，也不改变
|
||
工单审批和设备控制权限。旧的显式批准委托路径保留兼容。
|
||
|
||
工作进程每轮从 Server 的 claimed/running 运行快照读取 agent_session_id，并使用当前 OAuth 凭据向 Agent 的 `POST /api/v1/session/{session_id}/credentials` 更新正在执行的会话凭据。令牌只在 header 中传递，body 为 `{}`，不会进入任务、报告或会话持久化文件。Agent 再次校验会话所属用户和项目；工作进程重启后从 Server 快照恢复关联，不依赖浏览器续期。固定 Bearer 仍受自身有效期限制。
|
||
|
||
Agent API 统一使用 `/api/v1` 前缀：任务投递为 `POST /api/v1/task-executions`，消息投递为 `POST /api/v1/task-messages/deliver`，凭据续期使用上述会话接口。旧 `/api/v1/agent/chat` 前缀已移除。更新脚本后须重新启动常驻 worker，现有 Python 进程不会自动加载文件变化。本机用户服务使用：
|
||
|
||
```bash
|
||
systemctl --user restart tjwater-operational-worker.service
|
||
systemctl --user status tjwater-operational-worker.service --no-pager
|
||
```
|
||
|
||
重启后检查日志中的实际请求路径；HTTP 404 属于不可自动重试的投递失败，已失败的 run 不会因重启自动恢复，需通过原有重新调查流程显式创建新一轮。
|
||
|
||
长期 OAuth 身份由身份管理员创建支持 `client_credentials` 的专用机密客户端；服务账号令牌应具有后端配置要求的 audience、UUID `sub` 和 `preferred_username`。再由产品管理员把该 `sub` 绑定到启用的产品用户及目标项目成员关系。只赋予本次调查所需项目权限，不授予系统管理员或设备命令权限。当前项目成员策略同时包含工单审批/执行权限，不能将其误称为已经细分的专用最小权限服务角色。
|
||
|
||
本机联调配置示意（密钥由管理员通过安全配置填入，禁止粘贴到聊天）：
|
||
|
||
```dotenv
|
||
OPERATIONAL_WORKER_PROJECT_ID=324ff7ee-c2d4-5184-a84b-48fb14a2933e
|
||
OPERATIONAL_WORKER_SERVER_URL=http://127.0.0.1:8000
|
||
OPERATIONAL_WORKER_AGENT_URL=http://127.0.0.1:8787
|
||
OPERATIONAL_WORKER_TOKEN_URL=https://<identity-host>/realms/<realm>/protocol/openid-connect/token
|
||
OPERATIONAL_WORKER_CLIENT_ID=<dedicated-service-client>
|
||
OPERATIONAL_WORKER_CLIENT_SECRET=<inject-securely>
|
||
OPERATIONAL_ACTION_PERIPHERAL_URL=http://127.0.0.1:8000/peripheral
|
||
OPERATIONAL_DATA_PERIPHERAL_URL=http://127.0.0.1:8000/peripheral
|
||
```
|
||
|
||
选择 OAuth 时不要同时配置 `OPERATIONAL_WORKER_BEARER`，否则固定 Bearer 优先。不要配置设备命令网关进行本次合成联调。
|
||
|
||
工作进程先恢复过期执行租约，再使用 keyset 游标扫描持久化 run 队列；`operational.run.queued` 事件会立即唤醒扫描，连接中断时每 30 秒对账。Agent 正在执行时每 20 秒续租，租约为 60 秒；进程中断后由后端分配新的执行代次，旧 run 和旧会话不能继续覆盖新一轮结果。网络错误及 5xx 从 5 秒指数退避到 5 分钟，确定性 4xx 终止该 run；连续 10 次投递失败产生运维通知。
|
||
|
||
Agent 正常结束必须调用 run `complete`。同一执行代次缺少最终计划或报告时，Server 将 run 以 `ARTIFACT_INCOMPLETE` 明确失败，不依赖租约超时猜测完成状态。心跳只增加 run revision，不改变任务业务 revision。
|
||
|
||
最终报告未提供 Mermaid 图（包括 `diagram=null`）时，Server 将同轮、同会话且不晚于报告发布时刻的最新有效进度图保存到报告中；图的证据引用必须已在报告范围内，节点状态保留记录原值，不把“记录时执行中”改成“已完成”。进度快照显式清空图后，不再从更旧的进度恢复。既有无图最终报告在详情、指定版本和历史列表读取时使用相同兼容规则，不改写历史数据库记录，也不借用新轮次的图。
|
||
|
||
前端默认跟随最新轮次的进展，新报告发布后自动显示新报告；没有本轮报告时保留并标注上一轮报告。本轮最终报告图优先于过程图。主动选择历史报告或展开阅读报告详情时保留阅读版本，通过“返回实时内容”或“查看最新内容”切换。
|
||
|
||
## 事件日志与清理
|
||
|
||
### 普通对话与调查运行的私有消息
|
||
|
||
启用消息接口前，需要按现有项目数据库发布规则应用
|
||
`resources/sql/024_agent_messages.sql`；本次代码修改不自动执行迁移或重启服务。
|
||
该表也是调查启动前收信、运行完成/失败和租约恢复的必需依赖，即使没有协作消息，
|
||
也必须先迁移再启用调查执行。Server 在领取运行、修改状态前检查消息表；缺失时返回
|
||
`503 DEPENDENCY_UNAVAILABLE`，提示具体迁移文件，不将排队运行改成“调查中”。
|
||
若旧运行已因缺表卡住，补齐迁移后现有 worker 会按租约恢复规则创建下一代运行；
|
||
不要手工改状态或重复提交投诉。应用迁移前须确认允许自动恢复及后续模型调用。
|
||
迁移保留每条消息的项目、会话归属、明确的 `task_id`/`run_id`、幂等请求 ID
|
||
及固定原生消息 ID。普通对话只能向排队或有效租约内的调查运行发送消息，
|
||
调查回复只能沿已存在的消息关系返回原对话。终态运行拒绝新消息，不创建新的运行代次。
|
||
|
||
`POST /api/v1/agent-messages` 与 `/{message_id}/reply` 要求项目任务写权限；
|
||
`GET /api/v1/agent-messages?session_id=...` 按认证用户的对话归属过滤，支持
|
||
`cursor`、`limit` 与 `pending_only`。Agent 仍须验证实际会话归属，Server
|
||
不把客户端提交的会话 ID 当成会话所有权证明。服务身份读取任务会话消息时，
|
||
Server 校验该会话属于当前活动运行。跨用户投递查询与任务回执要求项目工作服务身份。
|
||
原对话用户可以确认收到的回复，但不能改写目标会话或原生消息 ID。
|
||
|
||
既有 worker 每轮最多对账四页、每页 50 条私有待投递消息，即使调查并发已满仍继续对账，
|
||
不增加进程或配置项。临时网络错误保留消息等待重试；权限拒绝、不存在等确定性
|
||
拒绝保存失败结果。消息正文、发送人和会话归属不写入项目事件日志。
|
||
待投递接口按创建时间与消息 ID 返回 keyset 游标，worker 跨轮保留扫描位置；
|
||
扫到末页后下一轮从头重试。长期空闲对话的待收消息不会一直占住首屏、阻止后续调查投递。
|
||
|
||
`pending` 表示等待投递，`enqueued` 表示运行时已接受，`delivered` 表示消息已进入
|
||
模型上下文；这三者都不能作为业务事项已经处理完成的证明。Agent 应在确认进入
|
||
上下文后保存回执。同一运行仍有 `pending`/`enqueued` 入站消息时，`complete`
|
||
返回结构化 `PENDING_AGENT_MESSAGES` 409，Agent 需接收后重试完成。取消、租约
|
||
超时、失败或被新代次替代会将旧运行未投递的入站消息明确标记为失败。
|
||
|
||
`resources/sql/015_workbench_events_and_task_runs.sql` 在项目 BizDB 中增加 `domain_event_log` 和 `operational_task_runs`。后续迁移默认只应用到 `tjwater_v2`，扩展规则见 [`workbench-database-rollout.md`](workbench-database-rollout.md)。事件与业务投影在同一事务提交，`LISTEN/NOTIFY` 只负责唤醒；浏览器与 worker 始终按持久游标补读。每个 API 进程对活跃项目最多保留一条 LISTEN 连接，无订阅者 60 秒后关闭。
|
||
|
||
事件默认保留 7 天。清理脚本会从 Metadata DB 解析项目路由，并在发现多个项目共享同一 BizDB 时拒绝运行。先确认 `OPERATIONAL_WORKER_PROJECT_ID` 对应 `tjwater_v2`，然后运行定向 dry-run：
|
||
|
||
```bash
|
||
conda run -n server python scripts/cleanup_workbench_events.py \
|
||
--project-id "$OPERATIONAL_WORKER_PROJECT_ID" --dry-run
|
||
```
|
||
|
||
核对输出后，写入模式仍必须显式保留项目参数：
|
||
|
||
```bash
|
||
conda run -n server python scripts/cleanup_workbench_events.py \
|
||
--project-id "$OPERATIONAL_WORKER_PROJECT_ID"
|
||
```
|
||
|
||
禁止省略 `--project-id` 执行清理。数据库维护脚本同样遵循先在 `tjwater_v2` 验证、再逐库扩展的规则。
|
||
|
||
浏览器使用 `GET /api/v1/workbench/sync-snapshot` 建立原子快照，并以该游标连接 `GET /api/v1/workbench/events`。游标早于保留窗口、超前或对应事件已全部清理时返回 `reset_required`，客户端必须重建快照。SSE 需要 Bearer 与 `X-Project-Id`，且 `CORS_ALLOWED_ORIGINS` 必须配置为明确来源。
|
||
|
||
## 外围样板工单同步
|
||
|
||
```bash
|
||
conda run -n server python scripts/sync_peripheral_work_orders.py --once
|
||
conda run -n server python scripts/sync_peripheral_work_orders.py
|
||
```
|
||
|
||
使用相同工作身份，并配置 `OPERATIONAL_ACTION_PERIPHERAL_URL`。同步器订阅源工单变更事件，收到消息即读取分页工单，将外围版本投影到 `external_work_order` 任务；默认每 300 秒兜底对账。合成来源在摘要和证据中明确标注；工单完成只表示该外部任务完成，不代表其他调查自动完成。事件协议、`tjwater_v2` 范围与启用步骤见 [外围工单主动同步](peripheral-work-order-events.md)。
|
||
|
||
工况任务面板的刷新按钮只重新读取当前项目的工作台快照，不直接触发外围系统全量对账。外围工单由独立同步器使用服务身份响应后端事件，避免多标签页或普通用户刷新放大全量扫描，也避免浏览器项目与同步器固定项目不一致。
|
||
|
||
通用合成检查工单不属于预约任务。早期种子脚本创建的通信、压力和零流量检查工单已经退役，种子脚本不再恢复它们；同步器按不可编辑的专用 seed actor 跳过旧种子工单。投诉、爆管事件以及调查动作实际创建的现场工单继续同步。`018_remove_synthetic_check_task_projections.sql` 只清理专用 seed actor 创建的旧检查投影，不按用户可编辑的标题或状态删除任务。三张独立演示工单已迁入外围样板，由同一同步器维护。
|
||
|
||
水质投诉和爆管上报类型的外围工单同时创建关联调查，即使当时没有浏览器打开也会入队。两类事件从外围样板测试面板模拟上报；工况页负责展示调查。调查中创建的外围工单通过回执编号关联，后续反馈送回原调查，由 Agent 重新分析。
|
||
|
||
## 旧调查迁移
|
||
|
||
`scripts/import_agent_investigation.py <session_id> --dry-run` 只检查旧合成水质调查能否转换。去掉 `--dry-run` 后使用上述产品工作身份导入后端任务、流程、证据及阶段报告；不会修改原 Agent 文件，也不会覆盖已存在的任务计划。应在启用队列工作进程前导入，避免旧调查被当作新的待执行任务领取。
|
||
|
||
## 调查数据读取
|
||
|
||
`OPERATIONAL_DATA_SERVER_URL` 默认本机 8000；外围样板数据读取需配置 `OPERATIONAL_DATA_PERIPHERAL_URL`。人工批准任务级读取范围后，Server 使用仅保存在自身运行环境中的 `OPERATIONAL_DATA_DELEGATION_BEARER` 查询来源；Agent 服务不接收此凭证。该凭证必须拥有目标项目和来源的读取权限，缺失时已批准的查询返回 503。查询只允许预定义的数据源、最多 20 个设备和 24 小时时间范围，响应有大小限制。每次查询保存内容摘要对应的数据快照；缺失、拒绝、超限或未配置都返回明确错误，不生成正常数据结论。截断结果必须保留不完整标记。
|
||
|
||
## 人工确认后的动作
|
||
|
||
- `OPERATIONAL_ACTION_PERIPHERAL_URL`：使用既有 `/api/sample/v1/work-orders` 创建合成外围工单。成功回执说明工单已创建，不表示现场工作已经完成。
|
||
- `OPERATIONAL_ACTION_COMMAND_GATEWAY_URL`：可选设备命令网关。未配置时明确返回不支持执行，不能生成成功回执。
|
||
- `OPERATIONAL_ACTION_TIMEOUT_SECONDS`：外部请求超时，默认 10 秒。
|
||
|
||
命令网关契约：`POST /api/v1/commands` 接收 `request_id`、`project_id`、`task_id`、`action_id`、`action_version`、`approved_by`、`approved_at` 与类型化 `command`。网关必须按 `request_id` 幂等处理，并在执行前验证目标设备当前状态、前置条件和命令有效期。
|
||
|
||
回执包含 `request_id`、`project_id`、`receipt_id`、`status`（`accepted` / `running` / `completed` / `failed`）、`submitted_at`、可空 `completed_at` 和 `detail`。`completed` 必须有完成时间。网关回执查询路径为 `GET /api/v1/commands/requests/{request_id}`。
|
||
|
||
请求和回执时间必须包含时区。网关地址只来自服务端配置，Agent 或用户动作内容不能指定任意目的地址。产品 Bearer 仅在当前请求内传递，不进入任务、会话、提示词和日志。没有回执的网络故障保留为 `unknown`，不能当作失败后直接重新下发。
|
||
|
||
## 验证边界
|
||
|
||
适配器和工作进程测试使用隔离 HTTP transport 验证权限前提、幂等身份、丢响应恢复、项目隔离、接受与完成的区别。它们不等于生产设备网关或现场执行验收；接入具体网关后还需以其真实回执验证执行链。
|
||
|
||
可从后端仓库复现聚焦验证：
|
||
|
||
```bash
|
||
conda run -n server python -m pytest tests/unit/test_operational_action_adapters.py tests/unit/test_operational_worker.py tests/unit/test_peripheral_task_sync.py tests/unit/test_operational_data.py tests/unit/test_import_agent_investigation.py -q
|
||
OPERATIONAL_TASKS_STORAGE_TEST=1 conda run -n server python -m pytest tests/integration/test_operational_tasks_storage.py -q
|
||
```
|
||
|
||
第二条需要当前配置的 PostgreSQL 可连接及测试 DDL 权限。测试在强制回滚事务中创建表结构和隔离项目数据，结束后不保留这些修改；不代替正式迁移。外部工单调用仍使用隔离 HTTP transport，不会操作实际工单系统。该测试覆盖持久化报告、反馈重新排队、旧会话拒绝、人工确认与丢响应回执核对，而非真实模型推理质量。
|