From 927d05938864e1f1b8d0877108f2df3f1830d063 Mon Sep 17 00:00:00 2001 From: caoqianming Date: Mon, 17 Aug 2026 12:12:27 +0800 Subject: [PATCH] feat(software): automate job completion follow-ups --- CHANGELOG.md | 2 + DESIGN.md | 2 + PROGRESS.md | 4 +- RUN.md | 6 +- core/software_jobs.py | 32 +++ core/storage/models.py | 9 + ...260817_1000_0034_software_job_followups.py | 41 +++ tests/test_software_followups.py | 124 +++++++++ tests/test_software_job_tools.py | 21 ++ tests/test_software_nodes.py | 54 ++++ tests/test_static_vendor.py | 9 +- tools/software_jobs.py | 11 + web/app.py | 11 +- web/routers/messages.py | 3 + web/routers/software_nodes.py | 23 +- web/schemas.py | 3 +- web/software_followups.py | 239 ++++++++++++++++++ web/static/dev.html | 24 +- web/static/js/chat.js | 29 +++ web/static/js/software_jobs.js | 147 +++++++++-- 20 files changed, 759 insertions(+), 35 deletions(-) create mode 100644 db/migrations/versions/20260817_1000_0034_software_job_followups.py create mode 100644 tests/test_software_followups.py create mode 100644 web/software_followups.py diff --git a/CHANGELOG.md b/CHANGELOG.md index abff507..8018b30 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ ## Unreleased +- 专业软件任务完成后可按用户意图仅报告产物，或自动读取图表和数据并给出分析；Job 中心重新整理了状态、输入输出、耗时和执行详情，支持查看结果、复制 Job ID、深入分析及重新分析。 + - 对话步骤进度改为按每轮任务保存完整计划；长任务、刷新或网络重连后可恢复当前步骤，不再因历史分页或首个实时事件错过而出现进度消失、串到上一轮或无法完成。 - 改进 Origin 多面板图排版：由 Origin 统一排列图层，共享横轴时仅在底行显示横轴标题和刻度标签，图例可自动避让数据；同时消除中文标题和坐标轴文字在导出图片中的异常横线。 diff --git a/DESIGN.md b/DESIGN.md index 21ccaf6..fc8cd30 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -484,6 +484,8 @@ Recipe 是专业软件的声明式目标状态，不是任意脚本或逐次鼠 第六阶段增加用户级 Job 中心与 Agent typed tools。`software_capability_list` 只暴露固定能力及当前在线空闲节点数，`software_job_submit/status/cancel` 在构造时绑定当前 user/task，模型不能跨用户或跨对话指定归属。`software_job_submit` 的唯一入口为通用 `inputs[] + operation + outputs[]`；输入和输出使用任务内稳定 key，具体 selector、type、format 和 options 由 capability 校验。`register_artifact` 是普通文件获得输入身份的唯一入口，并明确返回 UUID。右下角 Job 中心按用户聚合各对话任务，活动期短轮询、空闲期降频；终态变化通知用户，成功任务可回到原对话发起分析。取消采用协作协议：未派发任务直接终止，已派发任务先进入 `cancelling`，云端通过 WebSocket 发送并在心跳时重放 `job_cancel`，Node 杀死固定 Worker 进程树后回报 `cancelled`；终态写入仍由云端账本裁决。 +Job 完成后的对话闭环复用 `software_jobs` 账本，不增加通用事件表。提交时用 `completion_action=report|analyze` 区分“只报告产物”和“报告并分析”，未明确分析时默认 `report`；成功终态将 `followup_status` 置为 `pending`。分发器只在原 task 空闲时领取：`report` 直接写入带 artifact refs 的固定 assistant 消息，不产生模型费用；`analyze` 写入前端隐藏的内部完成事件，并复用 task 单活锁、持久消息和 SSE run 自动续跑 Agent。`pending/running/completed/failed` 状态使服务重启、对话繁忙和重复完成回执均不会造成重复回复；Job 中心的手动深入分析通过同一入口重新排入分析，而不是模拟点击或只预填输入框。 + 后续仍需实现 Token 轮换；不得以任意命令或脚本接口临时代替。当前 Job 中心采用轮询而非用户事件推送，单活与 offer 选择仍只覆盖单 Web 进程；生产启用多实例前必须增加 Redis/PG fencing 或固定路由到单一控制面实例。 --- diff --git a/PROGRESS.md b/PROGRESS.md index ed49c9c..ea90296 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -2,7 +2,7 @@ > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。 -最后更新:2026-08-17(Origin 多面板自动布局与导出基线修复完成,未发版) +最后更新:2026-08-17(专业软件 Job 完成自动报告/分析与结果中心优化完成,未发版) --- @@ -20,6 +20,8 @@ --- ## 已完成关键能力 +- **08-17 / Unreleased / Software Job 完成闭环与结果中心优化**：`software_job_submit` 新增默认 `report`、可选 `analyze` 的完成策略，0034 在原 Job 账本加入可恢复 follow-up 状态；成功任务可直接向原对话报告产物，或在 task 空闲后复用单活锁和 SSE 自动续跑分析，内部完成事件不冒充用户消息，手动深入/重新分析走同一幂等入口。Job 中心改为结果优先布局，显示短 Job ID、输入输出、完成方式和耗时，折叠展示完整 ID/输出目录/执行版本，并支持复制、查看结果及状态化分析按钮。相关专项 unittest、Python/JavaScript 语法、Ruff 致命规则和 diff 检查通过；未连接生产 DB、未执行 migration。 + ### 2026-08-17 - **08-17 / Unreleased / Origin 多面板自动布局与导出基线修复**：adapter 提升至 0.9.2，多面板图改用 Origin `layarrange` 根据行列、页边距和间距统一排列图层，并读取 Origin 计算后的实际图层位置放置标题与图例；共享 X 轴时仅底行保留标题和刻度标签，图例新增 `auto` 位置以调用 Origin 智能避让。每次任务显式设置 `@U=1`，消除 Origin 默认打印基线在中文标题、轴标题和图例上形成的异常横线，且不修改用户机器的持久配置。 diff --git a/RUN.md b/RUN.md index 0139228..37298e2 100644 --- a/RUN.md +++ b/RUN.md @@ -1054,7 +1054,7 @@ sudo xfs_quota -x -c "limit -p bhard=10g zcbot_" /opt ### Windows Node 内网 MVP（开发中） -先执行 `.venv/Scripts/python.exe main.py db upgrade head` 创建 `software_node_enrollments`、`software_nodes`、`software_jobs`，并为 artifact 增加专业软件来源字段。不要在未确认目标数据库时运行迁移；本机 `.env` 的 `ZCBOT_DB_URL` 可能是生产隧道。 +先执行 `.venv/Scripts/python.exe main.py db upgrade head` 创建 `software_node_enrollments`、`software_nodes`、`software_jobs`，为 artifact 增加专业软件来源字段，并加入 Job 完成策略与续跑状态。不要在未确认目标数据库时运行迁移；本机 `.env` 的 `ZCBOT_DB_URL` 可能是生产隧道。 0032 旧版曾把 `tasks.working_dir` 二次拼进 user root，已完成 Job 的文件和 artifact 路径可能位于重复的 `workspace/users//...` 子树。升级到 0033 后，用显式迁移库地址先 dry-run；确认所有目标目录无冲突后再加 `--apply`。脚本不加载 `.env`，文件和数据库更新均按 Job manifest 精确处理，目标目录同时存在时会停止，不能覆盖。 @@ -1070,7 +1070,7 @@ $env:ZCBOT_MIGRATION_DB_URL="postgresql+psycopg://user:pass@host:5432/zcbot" - Node `POST /v1/software-nodes/enroll` 注册并一次性取得 `node_id`、`node_token`； - Node 携带 `Authorization: Bearer ` 和 `X-Node-Id` 连接 `WS /v1/software-nodes/connect`； - 用户或 Agent 通过 `POST /v1/tasks/{task_id}/software-jobs` 提交专业软件任务； -- 用户通过 `GET /v1/software-jobs` 查看本人跨对话任务，可用 `task_id`、`active_only` 和 `limit` 筛选，`POST /v1/software-jobs/{job_id}/cancel` 请求停止； +- 用户通过 `GET /v1/software-jobs` 查看本人跨对话任务，可用 `task_id`、`active_only` 和 `limit` 筛选，`POST /v1/software-jobs/{job_id}/cancel` 请求停止，`POST /v1/software-jobs/{job_id}/analyze` 对成功结果发起或重新发起分析； - 管理员 `GET /v1/admin/software-nodes` 查看节点，`PATCH /v1/admin/software-nodes/{node_id}` 启停节点，`DELETE /v1/admin/software-nodes/{node_id}` 永久删除节点身份。 Node API 只能绑定受控内网地址并由安全组限制来源 IP。当前 HTTP/WS 链路不加密；跨安全域、公网或不可信终端接入前，必须先升级 HTTPS/WSS。多 Web 实例部署时，Node API 暂时固定路由到单一实例，直至 Connection Manager 增加跨实例 fencing。 @@ -1100,7 +1100,7 @@ install-windows-node.bat 若 Python 未加入 PATH，可把绝对路径作为第一个参数，例如 `install-windows-node.bat "C:\Python312\python.exe"`。默认解释器为 `%ProgramData%\Zcbot\WindowsNode\runtimes\origin\Scripts\python.exe`。如需使用其他受管解释器，优先设置机器级 `ZCBOT_ADAPTER_ORIGIN_PYTHON`；旧名 `ZCBOT_ORIGIN_PYTHON` 暂时兼容。`node.json`、可恢复任务和 runtime 集中保存在 `%ProgramData%\Zcbot\WindowsNode\`，不会因替换程序目录而丢失。运行时固定依赖见发布目录的 `adapters/origin.plot@v2/requirements.txt`；任务请求无权选择解释器、脚本或路径。当前 Worker 支持 CSV/XLSX/JSON 输入及 OPJU/PNG/SVG/PDF 输出。成功产物由 Node 流式上传，全部校验通过后发布到任务工作目录 `origin//`，plot spec 与 provenance 位于其 `.meta/`；上传中断会在重连时幂等续传。 -Web 用户登录后，文件栏 Job 中心会聚合本人最近任务。活动任务约 4 秒刷新一次，空闲时降为约 30 秒；停止已派发任务是协作取消，状态先显示“正在停止”，Node 在线时立即接收，断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`register_artifact`、`software_job_submit`、`software_job_status`、`software_job_revise` 和 `software_job_cancel`。Origin 输入必须是 artifact：已有 UUID 可直接提交，普通 task 文件先逐个用相对路径登记；登记不会发布聊天交付卡片。提交工具接收 `inputs`、`operation`、`outputs`，支持 1–16 个输入、跨输入系列和多个显式输出；`plot.canvas` 可指定毫米画布，轴可指定范围、步长、尺度、刻度角度/字号、标题字号和网格，`legend` 可控制显隐、位置和字号，`series[].style` 可控制颜色、线宽/线型、点型/点大小和透明度。所有排版字段可选，旧请求保持默认样式。任务只创建固定 v2 schema 的持久任务，不会阻塞当前对话等待完成。成功状态提供 `output_dir`，Agent可在该目录内搜索并分析；正式输出的 artifact 带 `software_job_id`，供结果卡和产物详情展示来源。Node 输出上传的逐任务诊断日志位于 `%ProgramData%\Zcbot\WindowsNode\jobs\\logs\node-output-upload.log`；日志包含上传阶段、产物文件名、重试次数和 Windows `HRESULT`，单文件达到 1 MiB 后轮转一份 `.1`，不记录 Node Token 或认证请求头。 +Web 用户登录后，文件栏 Job 中心会聚合本人最近任务。活动任务或完成后的自动分析约 4 秒刷新一次，空闲时降为约 30 秒；卡片优先显示状态、所属对话、输入输出、完成策略和耗时，Job ID、输出目录、执行节点及软件版本收在“任务详情”中。成功任务可查看结果目录、复制完整 Job ID、深入分析或重新分析。停止已派发任务是协作取消，状态先显示“正在停止”，Node 在线时立即接收，断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`register_artifact`、`software_job_submit`、`software_job_status`、`software_job_revise` 和 `software_job_cancel`。Origin 输入必须是 artifact：已有 UUID 可直接提交，普通 task 文件先逐个用相对路径登记；登记不会发布聊天交付卡片。提交工具接收 `inputs`、`operation`、`outputs` 和可选的 `completion_action=report|analyze`：只要求生成或保存时用默认 `report`，任务成功后平台向原对话写入固定产物报告；明确要求解释、总结或结论时用 `analyze`，平台在原 task 空闲后自动续跑 Agent。两者均不阻塞提交轮次。请求支持 1–16 个输入、跨输入系列和多个显式输出；`plot.canvas` 可指定毫米画布，轴可指定范围、步长、尺度、刻度角度/字号、标题字号和网格，`legend` 可控制显隐、位置和字号，`series[].style` 可控制颜色、线宽/线型、点型/点大小和透明度。所有排版字段可选，旧请求保持默认样式。成功状态提供 `output_dir`，正式输出的 artifact 带 `software_job_id`，供结果卡和产物详情展示来源。Node 输出上传的逐任务诊断日志位于 `%ProgramData%\Zcbot\WindowsNode\jobs\\logs\node-output-upload.log`；日志包含上传阶段、产物文件名、重试次数和 Windows `HRESULT`，单文件达到 1 MiB 后轮转一份 `.1`，不记录 Node Token 或认证请求头。 二维复合图使用 `plot.type=multi_panel`。`layout` 支持 1×1、1×2、2×1 和 2×2，`panels` 数量为 1–4；两个 panel 必须选择横排或竖排，三个和四个 panel 使用 2×2。每个 panel 的 `series[]` 必须显式给出 `kind`（`line/scatter/line_scatter/column/bar`），可用 `y_axis=right` 绑定右 Y 轴，并用 `x_error`、`y_error` 指定对称误差列；顶层 `x_axis/y_axis/legend` 作为全部 panel 的默认值，panel 内同名设置可单独覆盖，`right_y_axis` 仍按 panel 设置。一个请求最多仍为 16 条系列，且每个 panel 至少有一条左轴系列。`share_x/share_y` 会统一各主图层自动缩放后的范围。复合图不接受 Origin 模板名，也不支持把等高线、3D 曲面、三元图或热图混入 panel。 diff --git a/core/software_jobs.py b/core/software_jobs.py index 0836d27..f3545a4 100644 --- a/core/software_jobs.py +++ b/core/software_jobs.py @@ -92,6 +92,8 @@ def _job_dict(row: SoftwareJob) -> dict: "execution_runtime": _job_execution_runtime(row), "error": row.error, "artifact_manifest": row.artifact_manifest, + "completion_action": getattr(row, "completion_action", "report"), + "followup_status": getattr(row, "followup_status", "none"), "output_dir": f"{contract.output_namespace}/{row.job_id}", "created_at": row.created_at.isoformat() if row.created_at else None, "started_at": row.started_at.isoformat() if row.started_at else None, @@ -224,10 +226,13 @@ def create_job( idempotency_key: str, capability: str, request: dict, + completion_action: str = "report", ) -> tuple[dict, bool]: key = idempotency_key.strip() if not key or len(key) > 200: raise SoftwareJobError("idempotency_key must contain 1 to 200 characters") + if completion_action not in {"report", "analyze"}: + raise SoftwareJobError("completion_action must be report or analyze") try: contract = get_contract(capability) normalized, digest = contract.normalize_request(request) @@ -296,6 +301,7 @@ def create_job( existing.task_id != task_id or existing.capability != capability or existing.request_digest != digest + or getattr(existing, "completion_action", "report") != completion_action ): raise SoftwareJobError("idempotency key was already used for a different request") return _job_dict(existing), False @@ -313,6 +319,8 @@ def create_job( metrics={}, error={}, artifact_manifest=[], + completion_action=completion_action, + followup_status="none", ) try: with session.begin_nested(): @@ -330,6 +338,7 @@ def create_job( existing.task_id != task_id or existing.capability != capability or existing.request_digest != digest + or getattr(existing, "completion_action", "report") != completion_action ): raise SoftwareJobError( "idempotency key was already used for a different request" @@ -381,6 +390,7 @@ def revise_job( if source is None: raise SoftwareJobError("source software job not found") capability = source.capability + completion_action = getattr(source, "completion_action", "report") source_request = source.request inputs = deepcopy(source_request.get("inputs") or []) schema_version = get_contract(capability).request_schema[ @@ -391,6 +401,7 @@ def revise_job( task_id, idempotency_key=idempotency_key, capability=capability, + completion_action=completion_action, request={ "schema_version": schema_version, "inputs": inputs, @@ -400,6 +411,25 @@ def revise_job( ) +def request_job_analysis(user_id: UUID, job_id: UUID) -> dict: + """为已成功任务排入一次分析续跑；pending/running 请求保持幂等。""" + with session_scope() as session: + job = session.execute( + select(SoftwareJob).where( + SoftwareJob.job_id == job_id, + SoftwareJob.user_id == user_id, + ).with_for_update() + ).scalar_one_or_none() + if job is None: + raise SoftwareJobError("job not found") + if job.status != "succeeded": + raise SoftwareJobError("only a succeeded job can be analyzed") + job.completion_action = "analyze" + if job.followup_status not in {"pending", "running"}: + job.followup_status = "pending" + return _job_dict(job) + + def offer_next_job(node_ids: set[UUID]) -> dict | None: """选择最早可执行的 Job–Node 组合，避免跨 capability 队首阻塞。""" if not node_ids: @@ -827,6 +857,8 @@ def record_job_terminal(node_id: UUID, payload: dict) -> None: job.error = error job.artifact_manifest = manifest job.terminal_at = now + if terminal_status == "succeeded": + job.followup_status = "pending" def _is_uuid(value: str) -> bool: diff --git a/core/storage/models.py b/core/storage/models.py index d8ba776..9f126dd 100644 --- a/core/storage/models.py +++ b/core/storage/models.py @@ -478,6 +478,7 @@ class SoftwareJob(Base): UniqueConstraint("user_id", "idempotency_key", name="uq_software_jobs_user_idempotency"), Index("ix_software_jobs_status_created", "status", "created_at"), Index("ix_software_jobs_node_status", "node_id", "status"), + Index("ix_software_jobs_followup", "followup_status", "terminal_at"), ) job_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True, default=uuid4) @@ -503,6 +504,14 @@ class SoftwareJob(Base): metrics: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) error: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) artifact_manifest: Mapped[list[Any]] = mapped_column(JSONB, nullable=False, default=list) + # 0034:专业软件完成后的用户侧动作。report 只发布固定结果通知；analyze 会在 + # 原 task 空闲后自动续跑 Agent。followup_status 同时充当可恢复的轻量 outbox。 + completion_action: Mapped[str] = mapped_column( + Text, nullable=False, default="report", server_default="report" + ) + followup_status: Mapped[str] = mapped_column( + Text, nullable=False, default="none", server_default="none" + ) started_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) terminal_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) created_at: Mapped[datetime] = mapped_column( diff --git a/db/migrations/versions/20260817_1000_0034_software_job_followups.py b/db/migrations/versions/20260817_1000_0034_software_job_followups.py new file mode 100644 index 0000000..e56acca --- /dev/null +++ b/db/migrations/versions/20260817_1000_0034_software_job_followups.py @@ -0,0 +1,41 @@ +"""Add durable completion actions for professional software jobs. + +Revision ID: 0034 +Revises: 0033 +Create Date: 2026-08-17 +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0034" +down_revision: str | None = "0033" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "software_jobs", + sa.Column( + "completion_action", sa.Text(), nullable=False, server_default="report" + ), + ) + op.add_column( + "software_jobs", + sa.Column( + "followup_status", sa.Text(), nullable=False, server_default="none" + ), + ) + op.create_index( + "ix_software_jobs_followup", + "software_jobs", + ["followup_status", "terminal_at"], + ) + + +def downgrade() -> None: + op.drop_index("ix_software_jobs_followup", table_name="software_jobs") + op.drop_column("software_jobs", "followup_status") + op.drop_column("software_jobs", "completion_action") diff --git a/tests/test_software_followups.py b/tests/test_software_followups.py new file mode 100644 index 0000000..92a7c89 --- /dev/null +++ b/tests/test_software_followups.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import unittest +from contextlib import contextmanager +from types import SimpleNamespace +from unittest.mock import patch +from uuid import uuid4 + +from web.software_followups import claim_followup + + +class _Result: + def __init__(self, value=None): + self.value = value + + def scalar_one_or_none(self): + return self.value + + +class _Session: + def __init__(self, results): + self.results = list(results) + self.added = [] + self.statements = [] + + def execute(self, statement): + self.statements.append(statement) + return _Result(self.results.pop(0) if self.results else None) + + def add(self, row): + self.added.append(row) + + +def _job(action: str): + job_id = uuid4() + task_id = uuid4() + return SimpleNamespace( + job_id=job_id, + task_id=task_id, + user_id=uuid4(), + capability="origin.plot@v2", + status="succeeded", + followup_status="pending", + completion_action=action, + artifact_manifest=[ + { + "artifact_id": str(uuid4()), + "filename": "figure.png", + "path": f"origin/{job_id}/figure.png", + }, + { + "artifact_id": None, + "filename": "plot-spec.json", + "path": f"origin/{job_id}/.meta/plot-spec.json", + }, + ], + ) + + +class SoftwareFollowupTests(unittest.TestCase): + def test_report_claim_persists_fixed_assistant_message_with_artifacts(self): + job = _job("report") + task = SimpleNamespace(task_id=job.task_id, run_status="idle") + session = _Session([job.task_id, task, job]) + + @contextmanager + def scope(): + yield session + + with ( + patch("web.software_followups.session_scope", scope), + patch("web.software_followups.allocate_message_idx", return_value=7), + ): + claim = claim_followup(job.job_id) + + self.assertEqual(claim.action, "report") + self.assertEqual(job.followup_status, "completed") + self.assertEqual(len(session.added), 1) + message = session.added[0] + self.assertEqual(message.payload["role"], "assistant") + self.assertIn(str(job.job_id), message.payload["content"]) + self.assertEqual(len(message.artifact_refs), 1) + self.assertEqual(message.artifact_refs[0]["label"], "figure.png") + + def test_analyze_claim_persists_internal_turn_and_locks_task(self): + job = _job("analyze") + task = SimpleNamespace(task_id=job.task_id, run_status="idle") + session = _Session([job.task_id, task, job, None]) + + @contextmanager + def scope(): + yield session + + with ( + patch("web.software_followups.session_scope", scope), + patch("web.software_followups.allocate_message_idx", return_value=8), + ): + claim = claim_followup(job.job_id) + + self.assertEqual(claim.action, "analyze") + self.assertEqual(job.followup_status, "running") + self.assertEqual(session.added[0].payload["role"], "user") + self.assertIn("software_job_status", session.added[0].payload["content"]) + self.assertGreaterEqual(len(session.statements), 4) + + def test_busy_task_leaves_followup_pending(self): + job = _job("analyze") + task = SimpleNamespace(task_id=job.task_id, run_status="running") + session = _Session([job.task_id, task]) + + @contextmanager + def scope(): + yield session + + with patch("web.software_followups.session_scope", scope): + claim = claim_followup(job.job_id) + + self.assertIsNone(claim) + self.assertEqual(job.followup_status, "pending") + self.assertFalse(session.added) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_software_job_tools.py b/tests/test_software_job_tools.py index 00fe752..b55cace 100644 --- a/tests/test_software_job_tools.py +++ b/tests/test_software_job_tools.py @@ -61,6 +61,7 @@ class SoftwareJobToolTests(unittest.TestCase): self.assertTrue(result["created"]) self.assertEqual(create.call_args.args[:2], (self.user_id, self.task_id)) self.assertEqual(create.call_args.kwargs["capability"], "origin.plot@v2") + self.assertEqual(create.call_args.kwargs["completion_action"], "report") self.assertEqual( create.call_args.kwargs["request"], { @@ -96,6 +97,25 @@ class SoftwareJobToolTests(unittest.TestCase): self.assertIn("outputs", required) self.assertNotIn("input_path", SoftwareJobSubmitTool.parameters["properties"]) self.assertNotIn("request", SoftwareJobSubmitTool.parameters["properties"]) + self.assertEqual( + SoftwareJobSubmitTool.parameters["properties"]["completion_action"]["enum"], + ["report", "analyze"], + ) + + def test_submit_can_request_automatic_analysis(self): + artifact_id = uuid4() + with patch( + "tools.software_jobs.create_job", + return_value=({"job_id": str(uuid4())}, True), + ) as create: + SoftwareJobSubmitTool(self.user_id, self.task_id).execute( + "origin.plot@v2", + inputs=[{"key": "sample", "artifact_id": str(artifact_id)}], + operation={"plot": {"type": "line", "series": []}}, + outputs=[{"key": "figure_png", "type": "figure", "format": "png"}], + completion_action="analyze", + ) + self.assertEqual(create.call_args.kwargs["completion_action"], "analyze") def test_status_and_cancel_reject_cross_task_job(self): foreign = {"job_id": str(uuid4()), "task_id": str(uuid4())} @@ -201,6 +221,7 @@ class SoftwareJobToolTests(unittest.TestCase): ) self.assertEqual(result, (created, True)) self.assertEqual(create.call_args.kwargs["capability"], "origin.plot@v2") + self.assertEqual(create.call_args.kwargs["completion_action"], "report") self.assertEqual(create.call_args.kwargs["request"], { "schema_version": 2, "inputs": source.request["inputs"], diff --git a/tests/test_software_nodes.py b/tests/test_software_nodes.py index 43efb44..ea680f5 100644 --- a/tests/test_software_nodes.py +++ b/tests/test_software_nodes.py @@ -155,6 +155,25 @@ class SoftwareNodeMigrationTests(unittest.TestCase): self.assertIn("ix_artifacts_software_job_id", rendered) self.assertIn("jsonb_array_elements", rendered) + def test_0034_adds_software_job_followup_state(self) -> None: + statements: list[str] = [] + + def capture(sql, *multiparams, **params): + statements.append(str(sql.compile(dialect=postgresql.dialect()))) + + engine = create_mock_engine("postgresql+psycopg://", capture) + operations = Operations(MigrationContext.configure(engine.connect())) + migration = importlib.import_module( + "db.migrations.versions.20260817_1000_0034_software_job_followups" + ) + with patch.object(migration, "op", operations): + migration.upgrade() + + rendered = "\n".join(statements) + self.assertIn("completion_action", rendered) + self.assertIn("followup_status", rendered) + self.assertIn("ix_software_jobs_followup", rendered) + class SoftwareJobProtocolTests(unittest.TestCase): def test_succeeded_replay_uses_persisted_manifest_across_layout_versions(self) -> None: @@ -738,6 +757,41 @@ class SoftwareJobProtocolTests(unittest.TestCase): }) self.assertEqual(job.status, "failed") + @patch("core.software_jobs._published_output_is_valid", return_value=True) + @patch("core.software_jobs.validate_output_manifest") + @patch("core.software_jobs.session_scope") + def test_successful_terminal_queues_completion_followup( + self, session_scope, validate_manifest, _published + ) -> None: + session = session_scope.return_value.__enter__.return_value + node_id = uuid4() + lease_id = uuid4() + digest = "d" * 64 + manifest = [{"source_artifact_id": "figure_png"}] + validate_manifest.return_value = manifest + job = type("Job", (), {})() + job.job_id = uuid4() + job.node_id = node_id + job.lease_id = lease_id + job.request_digest = digest + job.status = "running" + job.capability = "origin.plot@v2" + job.request = {} + job.progress = 80 + session.execute.return_value.scalar_one_or_none.return_value = job + + record_job_terminal(node_id, { + "job_id": str(uuid4()), + "lease_id": str(lease_id), + "request_digest": digest, + "status": "succeeded", + "error": {}, + "artifact_manifest": manifest, + }) + + self.assertEqual(job.status, "succeeded") + self.assertEqual(job.followup_status, "pending") + @patch("core.software_jobs.session_scope") def test_disconnect_does_not_requeue_active_jobs(self, session_scope) -> None: session = session_scope.return_value.__enter__.return_value diff --git a/tests/test_static_vendor.py b/tests/test_static_vendor.py index dd9f742..4595e95 100644 --- a/tests/test_static_vendor.py +++ b/tests/test_static_vendor.py @@ -43,16 +43,19 @@ class StaticVendorTests(unittest.TestCase): self.assertIn("before_job_id", source) self.assertIn("onListScroll", source) self.assertIn("/cancel`", source) - self.assertIn("分析结果", source) - self.assertIn("refreshCurrentTaskFiles(job.task_id)", source) + self.assertIn("深入分析", source) + self.assertIn("查看结果", source) + self.assertIn("Job ID", source) + self.assertIn("refreshCurrentTaskAfterSoftwareJob(job.task_id)", source) self.assertIn("job.execution_runtime", source) self.assertIn("执行版本未记录", source) - self.assertIn("版本：${escapeHtml(jobRuntimeText(job))}", source) + self.assertIn("执行版本
${escapeHtml(jobRuntimeText(job))}", source) self.assertIn( 'document.addEventListener("software-job-submitted", refreshSoftwareJobs)', source, ) self.assertIn("export function refreshCurrentTaskFiles(taskId)", chat_source) + self.assertIn("export async function openSoftwareJobResults", chat_source) self.assertIn( 'document.dispatchEvent(new Event("software-job-submitted"))', chat_source, diff --git a/tools/software_jobs.py b/tools/software_jobs.py index 5c50ce1..1897ab0 100644 --- a/tools/software_jobs.py +++ b/tools/software_jobs.py @@ -80,6 +80,15 @@ class SoftwareJobSubmitTool(_SoftwareJobTool): "inputs": _contract_property_schema("inputs"), "operation": _contract_property_schema("operation"), "outputs": _contract_property_schema("outputs"), + "completion_action": { + "type": "string", + "enum": ["report", "analyze"], + "default": "report", + "description": ( + "Use report to notify with output files only. Use analyze when the " + "user also asked for interpretation, conclusions, or data analysis." + ), + }, "idempotency_key": { "type": "string", "description": ( @@ -106,6 +115,7 @@ class SoftwareJobSubmitTool(_SoftwareJobTool): inputs: list[dict] | None = None, operation: dict | None = None, outputs: list[dict] | None = None, + completion_action: str = "report", idempotency_key: str = "", ) -> str: try: @@ -134,6 +144,7 @@ class SoftwareJobSubmitTool(_SoftwareJobTool): self.task_id, idempotency_key=idempotency_key.strip() or str(uuid4()), capability=capability, + completion_action=completion_action, request=normalized_request, ) return json.dumps({**job, "created": created}, ensure_ascii=False) diff --git a/web/app.py b/web/app.py index a2107db..e39e7ff 100644 --- a/web/app.py +++ b/web/app.py @@ -49,7 +49,6 @@ from .background import ( from .broker import broker from .routers.asr import register_asr_routes from .routers.authroutes import register_auth_routes -from .routers.software_nodes import register_software_node_routes from .routers.external_systems import register_external_system_routes from .routers.files import register_file_routes from .routers.kb import register_kb_routes @@ -58,9 +57,14 @@ from .routers.misc import register_misc_routes from .routers.models import register_model_routes from .routers.schedules import register_schedule_routes from .routers.skills_memory import register_skill_memory_routes +from .routers.software_nodes import register_software_node_routes from .routers.tasks import register_task_routes from .routers.wechat import register_wechat_routes from .scheduler_runner import start_scheduler +from .software_followups import ( + requeue_stale_analysis_followups, + start_software_followup_dispatcher, +) from .static_files import NoCacheStaticFiles from .wechat_runner import start_wechat_inbound @@ -113,6 +117,9 @@ def create_app() -> FastAPI: # 启动钩子 + 后台协程群(本体见 web/background.py 等;None=该项未启用) reap_stale_runs() + requeued_followups = requeue_stale_analysis_followups() + if requeued_followups: + print(f"[startup] requeued {requeued_followups} software followup(s)") disk_scanner_task = start_disk_scanner(_cfg) stats_logger_task = start_stats_logger(app, run_max_workers) toolfail_task = start_toolfail_scanner() @@ -120,12 +127,14 @@ def create_app() -> FastAPI: wechat_task, wechat_stop = start_wechat_inbound(app) sandbox_reaper_task = init_sandbox(app, _cfg) proc_sweeper_task = start_proc_sweeper(_cfg) + software_followup_task = start_software_followup_dispatcher(app) try: yield finally: # 先拒新 run + drain in-flight(细节见 background.drain_inflight) app.state.draining.set() + await cancel_and_wait(software_followup_task) await drain_inflight(app, drain_timeout, cancel_grace) await cancel_and_wait(disk_scanner_task) diff --git a/web/routers/messages.py b/web/routers/messages.py index 0def393..c764a2a 100644 --- a/web/routers/messages.py +++ b/web/routers/messages.py @@ -102,6 +102,7 @@ def register_message_routes(app, *, require_user) -> None: Message.tokens_out, Message.model_profile, Message.created_at, Message.artifact_refs, Message.attachment_refs, + Message.kind, ) if limit is None: # 旧行为:升序全量 @@ -153,6 +154,7 @@ def register_message_routes(app, *, require_user) -> None: "created_at": iso(r.created_at), "artifact_refs": r.artifact_refs, "attachment_refs": r.attachment_refs, + "kind": r.kind, } for r in rows ] @@ -177,6 +179,7 @@ def register_message_routes(app, *, require_user) -> None: .where( Message.task_id == tid, Message.payload["role"].astext == "user", + Message.kind.is_distinct_from("software_job_followup"), ) .order_by(Message.idx) ).all() diff --git a/web/routers/software_nodes.py b/web/routers/software_nodes.py index 9efb122..794eddb 100644 --- a/web/routers/software_nodes.py +++ b/web/routers/software_nodes.py @@ -22,7 +22,11 @@ from fastapi.responses import FileResponse from core.artifact_lifecycle import register_published_artifacts from core.paths import from_db_path -from core.software_contracts import SoftwareContractError, default_capabilities, get_contract +from core.software_contracts import ( + SoftwareContractError, + default_capabilities, + get_contract, +) from core.software_jobs import ( MAX_OUTPUT_ARTIFACT_BYTES, MAX_OUTPUT_TOTAL_BYTES, @@ -38,6 +42,7 @@ from core.software_jobs import ( pending_node_cancellations, record_job_terminal, replay_succeeded_outputs, + request_job_analysis, request_job_cancel, respond_to_offer, software_job_output_path, @@ -62,6 +67,7 @@ from web.schemas import ( SoftwareNodeDisableRequest, SoftwareNodeEnrollRequest, ) +from web.software_followups import dispatch_followup from web.userfiles import load_user_root, safe_join @@ -461,6 +467,7 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: "artifact_manifest": published, } await asyncio.to_thread(record_job_terminal, node_id, terminal) + await dispatch_followup(request.app, job_id) except (SoftwareJobError, KeyError, TypeError) as exc: raise HTTPException(409, str(exc)) from exc return {"status": "succeeded", "artifact_manifest": published} @@ -654,6 +661,20 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: ) return job + @app.post("/v1/software-jobs/{job_id}/analyze", tags=["software-jobs"]) + async def analyze_software_job( + job_id: UUID, + request: Request, + user_id: UUID = Depends(require_user), # noqa: B008 + ): + try: + await asyncio.to_thread(request_job_analysis, user_id, job_id) + except SoftwareJobError as exc: + detail = str(exc) + raise HTTPException(404 if detail == "job not found" else 409, detail) from exc + await dispatch_followup(request.app, job_id) + return get_job(user_id, job_id) + @app.get("/v1/software-jobs/{job_id}", tags=["software-jobs"]) def read_software_job( job_id: UUID, diff --git a/web/schemas.py b/web/schemas.py index fbc8a4b..086e48f 100644 --- a/web/schemas.py +++ b/web/schemas.py @@ -1,7 +1,7 @@ """/v1 请求体 Pydantic 模型(从 app.py 析出,2026-07-23 拆分)。""" from __future__ import annotations -from typing import Optional +from typing import Literal, Optional from uuid import UUID from pydantic import BaseModel, Field @@ -141,3 +141,4 @@ class SoftwareJobCreateRequest(BaseModel): idempotency_key: str capability: str = Field(default_factory=lambda: default_capabilities()[0]) request: dict = Field(default_factory=dict) + completion_action: Literal["report", "analyze"] = "report" diff --git a/web/software_followups.py b/web/software_followups.py new file mode 100644 index 0000000..399482f --- /dev/null +++ b/web/software_followups.py @@ -0,0 +1,239 @@ +"""Software Job 完成后的固定报告与 Agent 自动续跑。""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from uuid import UUID + +from sqlalchemy import select, update + +from core.software_contracts import get_contract +from core.storage import session_scope +from core.storage.message_index import allocate_message_idx +from core.storage.models import Message, SoftwareJob, Task + +from .common import INSTANCE +from .run_lifecycle import RunScheduleError, schedule_claimed_run + + +@dataclass(frozen=True) +class FollowupClaim: + job_id: UUID + task_id: UUID + user_id: UUID + action: str + prompt: str = "" + + +def _artifact_refs(manifest: list) -> list[dict]: + refs: list[dict] = [] + for item in manifest: + if not isinstance(item, dict) or not item.get("artifact_id") or not item.get("path"): + continue + refs.append({ + "path": item["path"], + "label": item.get("filename") or item["path"].rsplit("/", 1)[-1], + "artifact_id": item["artifact_id"], + "version": 2, + }) + return refs + + +def _report_text(job: SoftwareJob) -> str: + contract = get_contract(job.capability) + output_dir = f"{contract.output_namespace}/{job.job_id}" + names = [ + str(item.get("filename")) + for item in job.artifact_manifest + if isinstance(item, dict) and item.get("artifact_id") and item.get("filename") + ] + files = "、".join(names) if names else "无可发布文件" + return ( + f"专业软件任务已完成：{contract.display_name}\n\n" + f"- Job ID：`{job.job_id}`\n" + f"- 输出目录：`{output_dir}`\n" + f"- 结果文件：{files}" + ) + + +def _analysis_prompt(job: SoftwareJob) -> str: + contract = get_contract(job.capability) + output_dir = f"{contract.output_namespace}/{job.job_id}" + return ( + "[专业软件任务完成事件]\n" + f"任务 {job.job_id}（{contract.display_name}）已成功完成，" + f"输出目录为 {output_dir}。请调用 software_job_status 获取完整产物清单，" + "读取输出图表和输入数据，向用户报告结果并分析主要趋势、结论及必要的限制。" + ) + + +def pending_followup_ids(limit: int = 20) -> list[UUID]: + with session_scope() as session: + return list(session.execute( + select(SoftwareJob.job_id) + .where( + SoftwareJob.status == "succeeded", + SoftwareJob.followup_status == "pending", + ) + .order_by(SoftwareJob.terminal_at, SoftwareJob.job_id) + .limit(limit) + ).scalars()) + + +def requeue_stale_analysis_followups() -> int: + """启动 reaper 已确认 run 随进程丢失时，把自动分析退回可重试状态。""" + with session_scope() as session: + job_ids = list(session.execute( + select(SoftwareJob.job_id) + .join(Task, Task.task_id == SoftwareJob.task_id) + .where( + SoftwareJob.followup_status == "running", + Task.run_status == "error", + Task.run_error == "server restarted before run finished", + ) + ).scalars()) + if job_ids: + session.execute( + update(SoftwareJob) + .where(SoftwareJob.job_id.in_(job_ids)) + .values(followup_status="pending") + ) + return len(job_ids) + + +def claim_followup(job_id: UUID) -> FollowupClaim | None: + """task 空闲时原子领取回调；report 在事务内直接落固定 assistant 消息。""" + with session_scope() as session: + candidate = session.execute( + select(SoftwareJob.task_id).where(SoftwareJob.job_id == job_id) + ).scalar_one_or_none() + if candidate is None: + return None + task = session.execute( + select(Task).where(Task.task_id == candidate).with_for_update() + ).scalar_one_or_none() + if task is None or task.run_status in {"running", "cancelling"}: + return None + job = session.execute( + select(SoftwareJob).where(SoftwareJob.job_id == job_id).with_for_update() + ).scalar_one_or_none() + if ( + job is None + or job.status != "succeeded" + or job.followup_status != "pending" + ): + return None + + next_idx = allocate_message_idx(session, task.task_id, locked_task=task) + if job.completion_action == "report": + session.add(Message( + task_id=task.task_id, + idx=next_idx, + payload={"role": "assistant", "content": _report_text(job)}, + artifact_refs=_artifact_refs(job.artifact_manifest), + kind="software_job_report", + )) + job.followup_status = "completed" + return FollowupClaim( + job_id=job.job_id, + task_id=task.task_id, + user_id=job.user_id, + action="report", + ) + + prompt = _analysis_prompt(job) + session.add(Message( + task_id=task.task_id, + idx=next_idx, + payload={"role": "user", "content": prompt}, + kind="software_job_followup", + )) + session.execute(update(Task).where(Task.task_id == task.task_id).values( + run_status="running", + run_error=None, + run_owner=INSTANCE or None, + )) + job.followup_status = "running" + return FollowupClaim( + job_id=job.job_id, + task_id=task.task_id, + user_id=job.user_id, + action="analyze", + prompt=prompt, + ) + + +def mark_followup_failed(job_id: UUID) -> None: + with session_scope() as session: + session.execute( + update(SoftwareJob) + .where( + SoftwareJob.job_id == job_id, + SoftwareJob.followup_status.in_({"pending", "running"}), + ) + .values(followup_status="failed") + ) + + +def finish_analysis_followup(job_id: UUID) -> None: + with session_scope() as session: + job = session.execute( + select(SoftwareJob).where(SoftwareJob.job_id == job_id).with_for_update() + ).scalar_one_or_none() + if job is None or job.followup_status != "running": + return + task_status = session.execute( + select(Task.run_status).where(Task.task_id == job.task_id) + ).scalar_one_or_none() + job.followup_status = ( + "failed" if task_status in {"error", "cancelled"} else "completed" + ) + + +def _track(app, task: asyncio.Task) -> None: + app.state.aux_tasks.add(task) + task.add_done_callback(app.state.aux_tasks.discard) + + +async def dispatch_followup(app, job_id: UUID) -> bool: + claim = await asyncio.to_thread(claim_followup, job_id) + if claim is None: + return False + if claim.action == "report": + return True + try: + run_task = schedule_claimed_run( + app, + claim.task_id, + claim.user_id, + claim.prompt, + scheduled=False, + ) + except RunScheduleError: + await asyncio.to_thread(mark_followup_failed, claim.job_id) + return False + + async def finish() -> None: + try: + await run_task + finally: + await asyncio.to_thread(finish_analysis_followup, claim.job_id) + + _track(app, asyncio.create_task(finish())) + return True + + +async def dispatch_pending_followups(app) -> None: + for job_id in await asyncio.to_thread(pending_followup_ids): + if app.state.draining.is_set(): + return + await dispatch_followup(app, job_id) + + +def start_software_followup_dispatcher(app) -> asyncio.Task: + async def loop() -> None: + while True: + await dispatch_pending_followups(app) + await asyncio.sleep(3) + + return asyncio.create_task(loop(), name="software-followups") diff --git a/web/static/dev.html b/web/static/dev.html index 577aa4c..751aaeb 100644 --- a/web/static/dev.html +++ b/web/static/dev.html @@ -1096,7 +1096,7 @@ #software-job-backdrop.open { opacity: 1; visibility: visible; } #software-job-panel { position: fixed; z-index: 130; top: 0; right: 0; bottom: 0; - width: min(420px, 92vw); display: flex; flex-direction: column; background: #fff; + width: min(480px, 94vw); display: flex; flex-direction: column; background: #fff; box-shadow: -8px 0 28px rgba(0,0,0,.16); transform: translateX(105%); visibility: hidden; transition: transform .2s ease, visibility .2s ease; } @@ -1111,6 +1111,9 @@ .sj-title { display: flex; align-items: center; justify-content: space-between; gap: 10px; } .sj-title-main { min-width: 0; display: flex; align-items: center; gap: 7px; } .sj-title-main strong { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } + .sj-context { display: flex; align-items: center; justify-content: space-between; gap: 8px; margin-top: 5px; color: #555; font-size: 11px; } + .sj-context span { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } + .sj-context code { flex-shrink: 0; color: var(--muted); font-size: 10px; } .sj-status { flex-shrink: 0; display: inline-flex; align-items: center; gap: 4px; padding: 2px 7px; border-radius: 999px; color: #6b7280; background: #f3f4f6; font-size: 11px; white-space: nowrap; } .sj-status svg { width: 12px; height: 12px; } .sj-card.running .sj-status, .sj-card.dispatched .sj-status, .sj-card.offered .sj-status, .sj-card.queued .sj-status { color: #9b4316; background: #fff3e7; } @@ -1125,11 +1128,28 @@ .sj-sub svg, .sj-meta svg { width: 12px; height: 12px; flex-shrink: 0; vertical-align: -2px; } .sj-progress { height: 4px; margin-top: 7px; border-radius: 4px; background: var(--panel-muted); overflow: hidden; } .sj-progress i { display: block; height: 100%; background: var(--accent); transition: width .25s ease; } - .sj-actions { display: flex; justify-content: flex-end; gap: 6px; margin-top: 9px; } + .sj-facts { display: grid; grid-template-columns: minmax(0, 1fr) minmax(0, 1fr); gap: 4px 12px; margin-top: 7px; padding: 7px 8px; border-radius: 6px; background: #f8f8f8; color: #62666d; font-size: 11px; } + .sj-facts span { min-width: 0; overflow-wrap: anywhere; } + .sj-details { margin-top: 6px; font-size: 11px; } + .sj-details > summary { width: fit-content; color: var(--muted); cursor: pointer; user-select: none; } + .sj-details dl { margin: 7px 0 0; padding: 7px 8px; border: 1px solid var(--border-soft); border-radius: 6px; background: #fcfcfc; } + .sj-details dl > div { display: grid; grid-template-columns: 72px minmax(0, 1fr); gap: 8px; padding: 3px 0; } + .sj-details dt { color: var(--muted); } + .sj-details dd { min-width: 0; margin: 0; overflow-wrap: anywhere; } + .sj-details code { font-size: 10px; } + .sj-copy { margin-left: 6px; padding: 1px 5px; border: 0; border-radius: 4px; background: transparent; color: var(--muted); cursor: pointer; font-size: 10px; } + .sj-copy:hover { color: var(--accent); background: var(--accent-soft); } + .sj-copy svg { width: 10px; height: 10px; vertical-align: -2px; } + .sj-actions { display: flex; flex-wrap: wrap; justify-content: flex-end; gap: 6px; margin-top: 9px; } .sj-actions button { display: inline-flex; align-items: center; gap: 4px; } + .sj-actions button:disabled { cursor: default; opacity: .62; } .sj-actions svg { width: 12px; height: 12px; } .sj-empty { padding: 22px; text-align: center; color: var(--muted); } .sj-loading { padding: 14px; text-align: center; color: var(--muted); font-size: 11px; } + @media (max-width: 520px) { + .sj-facts { grid-template-columns: 1fr; } + .sj-actions button { flex: 1 1 auto; justify-content: center; } + } /* media tool 摘要 banner(model / size / cost / elapsed,折叠态也可见) */ .tool-banner { display: inline-flex; flex-wrap: wrap; gap: 6px; diff --git a/web/static/js/chat.js b/web/static/js/chat.js index 77619c1..d68402c 100644 --- a/web/static/js/chat.js +++ b/web/static/js/chat.js @@ -1654,6 +1654,7 @@ function renderMessages(msgs, { stickBottom = true } = {}) { const m = msgs[mi]; const p = m.payload || {}; const role = p.role || "?"; + if (m.kind === "software_job_followup") continue; if (role === "system") continue; // 不显示 system if (role === "assistant" && m.model_profile && m.model_profile !== lastAsstModel) { const dn = (state.models.find(x => x.profile === m.model_profile) || {}).display_name || m.model_profile; @@ -1880,6 +1881,34 @@ export function refreshCurrentTaskFiles(taskId) { if (taskId && state.taskId === taskId) scheduleFilesRefresh(); } +// Software Job 终态可能新增一条固定报告，也可能启动自动分析 run。刷新当前 task +// 并在后者场景主动接上 SSE，用户无需切走再切回来才能看到回复。 +export async function refreshCurrentTaskAfterSoftwareJob(taskId) { + if (!taskId || state.taskId !== taskId) return; + try { + const meta = await api("GET", "/v1/tasks/" + taskId); + if (state.taskId !== taskId) return; + state.taskMeta = meta; + renderChatMeta(); + if (meta.run_status === "running" || meta.run_status === "cancelling") { + ensureRunningTaskSubscribed(taskId, `/v1/tasks/${taskId}/events`, meta); + } else { + await loadMessages(); + refreshOutline(); + } + scheduleFilesRefresh(); + } catch (_) { /* Job 中心下一轮轮询或切换 task 时会恢复 */ } +} + +export async function openSoftwareJobResults(taskId, outputDir) { + if (!taskId) return; + await selectTask(taskId); + const workingDir = state.taskMeta?.working_dir || ""; + const base = workingDir.split("/").filter(Boolean).pop() || ""; + state.filesPath = [base, outputDir].filter(Boolean).join("/"); + await loadFiles(); +} + function chatAction() { if (isCurrentTaskStreaming()) { cancelCurrentTask(); return; } if (hasRunningProc(state.taskId)) { killTaskProcs(state.taskId); return; } diff --git a/web/static/js/software_jobs.js b/web/static/js/software_jobs.js index 5a9ba00..04c88d3 100644 --- a/web/static/js/software_jobs.js +++ b/web/static/js/software_jobs.js @@ -3,7 +3,11 @@ import { api } from "./api.js"; import { state } from "./state.js"; import { $ } from "./dom.js"; import { escapeHtml } from "./format.js"; -import { refreshCurrentTaskFiles, selectTask } from "./chat.js"; +import { + openSoftwareJobResults, + refreshCurrentTaskAfterSoftwareJob, + selectTask, +} from "./chat.js"; import { dialogConfirm, message } from "./dialog.js"; const ACTIVE = new Set(["queued", "offered", "dispatched", "running", "disconnected", "cancelling"]); @@ -38,6 +42,7 @@ const icons = { clock: '', failed: '', open: '', + copy: '', pending: '', running: '', succeeded: '', @@ -81,8 +86,26 @@ export async function refreshSoftwareJobs() { const next = data.results || []; next.forEach((job) => { const previous = known.get(job.job_id); - if (previous && ACTIVE.has(previous) && TERMINAL.has(job.status)) notifyTerminal(job); - known.set(job.job_id, job.status); + if (previous && ACTIVE.has(previous.status) && TERMINAL.has(job.status)) { + notifyTerminal(job); + } else if ( + previous + && previous.followup_status !== job.followup_status + && ["running", "completed", "failed"].includes(job.followup_status) + ) { + void refreshCurrentTaskAfterSoftwareJob(job.task_id); + } + if ( + job.followup_status === "running" + && state.taskId === job.task_id + && !state.liveRuns.has(job.task_id) + ) { + void refreshCurrentTaskAfterSoftwareJob(job.task_id); + } + known.set(job.job_id, { + status: job.status, + followup_status: job.followup_status, + }); }); const previousJobs = jobs; const hadLoadedMore = previousJobs.length > PAGE_SIZE; @@ -113,7 +136,9 @@ async function loadMore() { function schedule() { if (timer) clearTimeout(timer); - const delay = jobs.some((job) => ACTIVE.has(job.status)) ? POLL_ACTIVE_MS : POLL_IDLE_MS; + const delay = jobs.some((job) => ( + ACTIVE.has(job.status) || ["pending", "running"].includes(job.followup_status) + )) ? POLL_ACTIVE_MS : POLL_IDLE_MS; timer = setTimeout(refreshSoftwareJobs, document.hidden ? Math.max(delay, 30000) : delay); } @@ -178,28 +203,86 @@ function jobRuntimeText(job) { : "执行版本未记录"; } +function fmtDate(value) { + return value ? new Date(value).toLocaleString("zh-CN", { hour12: false }) : "—"; +} + +function elapsedText(job) { + const start = job.started_at || job.created_at; + const end = job.terminal_at; + if (!start || !end) return ""; + const seconds = Math.max(0, Math.round((new Date(end) - new Date(start)) / 1000)); + if (seconds < 60) return `${seconds} 秒`; + const minutes = Math.floor(seconds / 60); + return seconds % 60 ? `${minutes} 分 ${seconds % 60} 秒` : `${minutes} 分钟`; +} + +function followupLabel(job) { + if (job.completion_action === "analyze") { + if (job.followup_status === "pending") return "等待自动分析"; + if (job.followup_status === "running") return "正在自动分析"; + if (job.followup_status === "completed") return "已报告并分析"; + if (job.followup_status === "failed") return "自动分析失败"; + return "报告并分析"; + } + return job.followup_status === "completed" ? "已报告结果" : "仅报告结果"; +} + +function jobActions(job, active) { + const task = escapeHtml(job.task_id); + if (active) return ` + + ${job.status !== "cancelling" ? `` : ""}`; + if (job.status !== "succeeded") { + return ``; + } + const view = ``; + if (["pending", "running"].includes(job.followup_status) && job.completion_action === "analyze") { + const label = job.followup_status === "pending" ? "等待分析" : "分析中"; + return `${view}`; + } + if (job.completion_action === "analyze" && job.followup_status === "completed") { + return `${view}`; + } + const label = job.followup_status === "failed" ? "重试分析" : "深入分析"; + return `${view}`; +} + function jobCard(job) { const summary = job.request_summary || {}; const inputs = Array.isArray(job.input) ? job.input : []; - const inputNames = inputs.map((item) => item.filename).filter(Boolean).join("、"); + const allInputNames = inputs.map((item) => item.filename).filter(Boolean); + const inputNames = allInputNames.slice(0, 2).join("、") + + (allInputNames.length > 2 ? ` 等 ${allInputNames.length} 个文件` : ""); const active = ACTIVE.has(job.status); const progress = Math.max(0, Math.min(100, Number(job.progress || 0))); const detail = stageLabel[job.stage] || statusLabel[job.status] || "正在处理"; const error = job.error && (job.error.detail || job.error.code); - const opened = job.created_at; - const openedText = opened ? new Date(opened).toLocaleString("zh-CN", { hour12: false }) : ""; + const artifacts = Array.isArray(job.artifact_manifest) + ? job.artifact_manifest.filter((item) => item && item.artifact_id) : []; + const formats = (summary.formats || []).filter(Boolean).join("、"); + const elapsed = elapsedText(job); + const shortId = String(job.job_id || "").slice(0, 8); return `
${escapeHtml(summary.display_name || job.capability)}
${statusIcon(job.status)}${escapeHtml(statusLabel[job.status] || job.status)}
+
${escapeHtml(job.task_name || "未命名对话")}Job ${escapeHtml(shortId)}
${icons.activity}${escapeHtml(detail)}${error ? ` · ${escapeHtml(error)}` : ""}
${active ? `
` : ""} -
${icons.clock}${escapeHtml(job.task_name || "未命名对话")}${inputNames ? ` · ${escapeHtml(inputNames)}` : ""}${openedText ? ` · 开启于 ${escapeHtml(openedText)}` : ""}
-
版本：${escapeHtml(jobRuntimeText(job))}
-
- - ${job.status === "succeeded" ? `` : ""} - ${active && job.status !== "cancelling" ? `` : ""} +
+ ${inputNames ? `输入：${escapeHtml(inputNames)}` : ""} + ${job.status === "succeeded" ? `输出：${artifacts.length} 个文件${formats ? ` · ${escapeHtml(formats)}` : ""}` : ""} + 完成方式：${escapeHtml(followupLabel(job))} + ${active ? `创建于 ${escapeHtml(fmtDate(job.created_at))}` : `${elapsed ? `耗时 ${escapeHtml(elapsed)} · ` : ""}${job.terminal_at ? `完成于 ${escapeHtml(fmtDate(job.terminal_at))}` : `创建于 ${escapeHtml(fmtDate(job.created_at))}`}`}
+
任务详情
+
Job ID
${escapeHtml(job.job_id)}
+
Capability
${escapeHtml(job.capability)}
+
输出目录
${escapeHtml(job.output_dir || "—")}
+
执行版本
${escapeHtml(jobRuntimeText(job))}
+
时间
创建 ${escapeHtml(fmtDate(job.created_at))}
开始 ${escapeHtml(fmtDate(job.started_at))}
结束 ${escapeHtml(fmtDate(job.terminal_at))}
+
+
${jobActions(job, active)}
`; } @@ -208,6 +291,13 @@ async function handleAction(event, button) { const card = button.closest("[data-job-id]"); const jobId = card.dataset.jobId; const action = button.dataset.jobAction; + if (action === "copy") { + try { + await navigator.clipboard.writeText(jobId); + message("Job ID 已复制", "success"); + } catch (_) { message("复制失败，请手动复制", "error"); } + return; + } if (action === "cancel") { if (!await dialogConfirm({ title: "停止专业软件任务", @@ -221,17 +311,28 @@ async function handleAction(event, button) { return; } const taskId = button.dataset.taskId; + if (action === "results") { + await openSoftwareJobResults(taskId, jobs.find((item) => item.job_id === jobId)?.output_dir || ""); + closeDrawer(); + return; + } + if (action === "analyze") { + button.disabled = true; + try { + await api("POST", `/v1/software-jobs/${jobId}/analyze`); + if (taskId) await selectTask(taskId); + await refreshCurrentTaskAfterSoftwareJob(taskId); + message("已排队分析结果", "success"); + closeDrawer(); + refreshSoftwareJobs(); + } catch (error) { + button.disabled = false; + message(error.message || "启动分析失败", "error"); + } + return; + } if (taskId) await selectTask(taskId); closeDrawer(); - if (action === "analyze") { - setTimeout(() => { - const input = $("chat-input"); - if (!input) return; - input.value = `请分析专业软件任务 ${jobId} 的结果，结合输出图表和输入数据总结主要结论。`; - input.focus(); - input.dispatchEvent(new Event("input", { bubbles: true })); - }, 400); - } } function notifyTerminal(job) { @@ -239,5 +340,5 @@ function notifyTerminal(job) { const label = ok ? "已完成" : (job.status === "cancelled" ? "已取消" : "失败"); const summary = job.request_summary || {}; message(`${summary.display_name || "专业软件任务"}${label}`, ok ? "success" : "error", 6000); - refreshCurrentTaskFiles(job.task_id); + void refreshCurrentTaskAfterSoftwareJob(job.task_id); }