diff --git a/CHANGELOG.md b/CHANGELOG.md index 104b68e..d809afa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ ## Unreleased - +- 新增专业软件 Job 中心：Agent 可提交、查询和停止 Windows Node 上的受控软件任务；用户可在右下角跨对话查看进度、收到完成通知，并回到原对话分析结果。 ## 0.65.2 — 2026-08-13 diff --git a/DESIGN.md b/DESIGN.md index 543c3c4..5e24c3c 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -472,7 +472,9 @@ Node 通过 `Authorization: Bearer` 与 `X-Node-Id` 建立 `/v1/software-nodes/c 第五阶段完成输出上传与发布：Node 只按固定 manifest ID 逐项流式 PUT，并携带 Node、lease、request digest 与内容摘要；云端重新绑定任务身份，不信 Node 提供的路径或媒体类型。文件先进入用户根下隐藏暂存区，固定文件名、单文件/总大小和 SHA-256 全部验证后，目录级原子移动到 `/origin//`，再登记平台 artifact UUID 并写成功终态。重复 PUT、complete 和重连均按摘要幂等；部分上传不可见，只有完整集合才能发布。 -后续仍需实现协作取消和 Token 轮换；不得以任意命令或脚本接口临时代替。当前单活与 offer 选择仍只覆盖单 Web 进程，生产启用多实例前必须增加 Redis/PG fencing 或固定路由到单一控制面实例。 +第六阶段增加用户级 Job 中心与 Agent typed tools。`software_capability_list` 只暴露固定能力及当前在线空闲节点数，`software_job_submit/status/cancel` 在构造时绑定当前 user/task，模型不能跨用户或跨对话指定归属。右下角 Job 中心按用户聚合各对话任务，活动期短轮询、空闲期降频；终态变化通知用户，成功任务可回到原对话发起分析。取消采用协作协议：未派发任务直接终止，已派发任务先进入 `cancelling`，云端通过 WebSocket 发送并在心跳时重放 `job_cancel`，Node 杀死固定 Worker 进程树后回报 `cancelled`；终态写入仍由云端账本裁决。 + +后续仍需实现 Token 轮换；不得以任意命令或脚本接口临时代替。当前 Job 中心采用轮询而非用户事件推送，单活与 offer 选择仍只覆盖单 Web 进程；生产启用多实例前必须增加 Redis/PG fencing 或固定路由到单一控制面实例。 --- diff --git a/PROGRESS.md b/PROGRESS.md index b8385f6..7e24bde 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-13(用户消息结构化附件与图片预览,bump 0.65.2) +最后更新:2026-08-13(专业软件 Job 中心开发中,未发版) --- @@ -22,6 +22,8 @@ ### 2026-08-13 +- **08-13 / Unreleased / 专业软件 Job 中心**:Agent 新增固定能力发现、提交、状态和取消工具，后端提供用户级跨对话任务列表与协作取消，Windows Node 可终止固定 Worker 进程树并幂等回报取消终态；Web 右下角展示活动/最近任务、进度、完成通知、停止与回到原对话分析入口。相关 Python 78 项、JavaScript 语法、Python 编译、.NET build 与 diff 检查通过；数据库保持在 0031，未连库、未执行 migration。 + - **08-13 / 0.65.2 / 用户消息结构化附件 + 对话内图片预览**:新增 0031 `messages.attachment_refs`，新客户端将附件作为结构化字段发送，数据库正文仅保留用户自然语言；后端按 task working_dir 校验路径并在内存模型上下文中补附件提示，旧客户端与历史正文标记继续兼容。用户消息即时态和历史态统一展示附件 chip，图片额外显示可点击缩略图；相关 Python 34 项、全部前端 Node 26 项、Python/JavaScript 语法、Alembic 单 head 与 diff 检查通过，未连接或迁移生产 DB。 ### 2026-08-12 diff --git a/RUN.md b/RUN.md index fb5acfd..e4cac07 100644 --- a/RUN.md +++ b/RUN.md @@ -1054,13 +1054,15 @@ sudo xfs_quota -x -c "limit -p bhard=10g zcbot_" /opt ### Windows Node 内网 MVP（开发中） -先执行 `alembic upgrade head` 创建 `software_node_enrollments` 和 `software_nodes`。不要在未确认目标数据库时运行迁移；本机 `.env` 的 `ZCBOT_DB_URL` 可能是生产隧道。 +先执行 `alembic upgrade head` 创建 `software_node_enrollments`、`software_nodes` 和 `software_jobs`。不要在未确认目标数据库时运行迁移；本机 `.env` 的 `ZCBOT_DB_URL` 可能是生产隧道。 云端当前提供： - 管理员 `POST /v1/admin/software-node-enrollments` 创建一次性注册码； - 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/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。 @@ -1084,6 +1086,8 @@ windows-node\install-origin-runtime.ps1 -BootstrapPython D:\programs\Python312\p 默认解释器为 `%ProgramData%\Zcbot\WindowsNode\runtimes\origin\python.exe`。如需放在其他受管目录，设置机器级 `ZCBOT_ORIGIN_PYTHON` 为绝对 `python.exe` 路径后重启 Node。运行时固定依赖见 `windows-node/origin-worker/requirements.txt`；任务请求无权选择解释器、脚本或路径。当前 Worker 支持 CSV/XLSX/JSON 输入，`line`、`scatter`、`line_scatter` 与 OPJU/PNG/SVG/PDF 输出。成功产物由 Node 流式上传，全部校验通过后发布到任务工作目录 `origin//`；上传中断会在重连时幂等续传。 +Web 用户登录后，右下角 Job 中心会聚合本人最近任务。活动任务约 4 秒刷新一次，空闲时降为约 30 秒；停止已派发任务是协作取消，状态先显示“正在停止”，Node 在线时立即接收，断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`software_job_submit`、`software_job_status` 和 `software_job_cancel`；提交工具只创建固定 schema 的持久任务，不会阻塞当前对话等待完成。 + 注册配置写入 `%ProgramData%\Zcbot\WindowsNode\node.json`；Token 使用 DPAPI `LocalMachine` 加密，ACL 仅允许注册账号和 `SYSTEM`。应始终用同一专用 Windows 账号注册并运行 Node。当前 MVP 以该账号的登录后计划任务启动，不安装 Windows Service。 直接双击 EXE 启动托盘 UI：红点为未注册/身份失效，黄点为连接中，绿点为在线；双击托盘图标打开配置窗。原 CLI 注册入口继续保留，无 UI 模式使用 `Zcbot.WindowsNode.exe run --headless`。 diff --git a/core/software_jobs.py b/core/software_jobs.py index 929bb49..5c796c4 100644 --- a/core/software_jobs.py +++ b/core/software_jobs.py @@ -8,7 +8,7 @@ from datetime import datetime, timedelta, timezone from hashlib import sha256 from uuid import UUID, uuid4 -from sqlalchemy import select +from sqlalchemy import desc, select from sqlalchemy.exc import IntegrityError from core.software_nodes import SUPPORTED_CAPABILITIES @@ -157,6 +157,121 @@ def _job_dict(row: SoftwareJob) -> dict: } +def list_jobs( + user_id: UUID, + *, + task_id: UUID | None = None, + active_only: bool = False, + limit: int = 50, +) -> list[dict]: + """列出用户的软件任务；用于全局 Job 中心和 Agent 查询。""" + limit = max(1, min(int(limit), 100)) + with session_scope() as session: + statement = ( + select(SoftwareJob, Task.name, SoftwareNode.name) + .join(Task, Task.task_id == SoftwareJob.task_id) + .outerjoin(SoftwareNode, SoftwareNode.node_id == SoftwareJob.node_id) + .where(SoftwareJob.user_id == user_id) + ) + if task_id is not None: + statement = statement.where(SoftwareJob.task_id == task_id) + if active_only: + statement = statement.where( + SoftwareJob.status.in_( + { + "queued", + "offered", + "dispatched", + "running", + "disconnected", + "cancelling", + } + ) + ) + rows = session.execute( + statement.order_by( + desc(SoftwareJob.created_at), desc(SoftwareJob.job_id) + ).limit(limit) + ).all() + results: list[dict] = [] + for job, task_name, node_name in rows: + item = _job_dict(job) + item.update( + { + "task_name": task_name, + "node_name": node_name, + "input": job.input_manifest, + "request_summary": _request_summary(job), + } + ) + results.append(item) + return results + + +def _request_summary(job: SoftwareJob) -> dict: + plot = job.request.get("plot") or {} + output = job.request.get("output") or {} + return { + "display_name": ( + "Origin 科研绘图" + if job.capability == "origin.plot@v1" + else job.capability + ), + "title": str(plot.get("title") or ""), + "formats": list(output.get("formats") or []), + } + + +def request_job_cancel(user_id: UUID, job_id: UUID) -> tuple[dict, dict | None]: + """持久化取消意图；queued 直接终止，已分派任务返回 Node 消息。""" + now = datetime.now(timezone.utc) + 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 in {"succeeded", "failed", "cancelled"}: + return _job_dict(job), None + if job.status == "queued" or job.node_id is None or job.lease_id is None: + job.status = "cancelled" + job.stage = "terminal" + job.error = {"code": "USER_CANCELLED", "detail": "Cancelled before dispatch."} + job.terminal_at = now + return _job_dict(job), None + job.status = "cancelling" + job.stage = "cancel_requested" + payload = { + "job_id": str(job.job_id), + "lease_id": str(job.lease_id), + "request_digest": job.request_digest, + } + return _job_dict(job), {"node_id": job.node_id, "payload": payload} + + +def pending_node_cancellations(node_id: UUID) -> list[dict]: + """节点重连或心跳时重放未确认的取消意图。""" + with session_scope() as session: + rows = session.execute( + select(SoftwareJob).where( + SoftwareJob.node_id == node_id, + SoftwareJob.status == "cancelling", + ) + ).scalars() + return [ + { + "job_id": str(job.job_id), + "lease_id": str(job.lease_id), + "request_digest": job.request_digest, + } + for job in rows + if job.lease_id is not None + ] + + def create_job( user_id: UUID, task_id: UUID, @@ -475,7 +590,7 @@ def respond_to_offer(node_id: UUID, *, accepted: bool, payload: dict) -> None: raise SoftwareJobError("job offer is stale or does not belong to this node") if ( accepted - and job.status in {"dispatched", "running", "succeeded", "failed", "cancelled"} + and job.status in {"dispatched", "running", "cancelling", "succeeded", "failed", "cancelled"} and payload.get("request_digest") == job.request_digest ): return @@ -518,7 +633,7 @@ def update_job_state(node_id: UUID, payload: dict) -> None: select(SoftwareJob).where(SoftwareJob.job_id == job_id).with_for_update() ).scalar_one_or_none() _assert_job_message(job, node_id, lease_id, digest) - if job.status in {"succeeded", "failed", "cancelled"}: + if job.status in {"succeeded", "failed", "cancelled", "cancelling"}: return if not _can_accept_state(job.status): raise SoftwareJobError("job state cannot advance from its current status") @@ -578,7 +693,7 @@ def record_job_terminal(node_id: UUID, payload: dict) -> None: if job.status != terminal_status: raise SoftwareJobError("job terminal status conflicts with existing terminal") return - if job.status not in {"offered", "dispatched", "running", "disconnected"}: + if job.status not in {"offered", "dispatched", "running", "disconnected", "cancelling"}: raise SoftwareJobError("job terminal cannot advance from its current status") job.status = terminal_status job.stage = "terminal" diff --git a/core/tool_registry.py b/core/tool_registry.py index 7045c39..82f2f66 100644 --- a/core/tool_registry.py +++ b/core/tool_registry.py @@ -47,6 +47,12 @@ from tools.run_python import RunPythonTool from tools.schedule import ( ScheduleCancelTool, ScheduleCreateTool, ScheduleListTool, ScheduleUpdateTool, ) +from tools.software_jobs import ( + SoftwareCapabilityListTool, + SoftwareJobCancelTool, + SoftwareJobStatusTool, + SoftwareJobSubmitTool, +) from tools.seedance import SeedanceTool from tools.seedream import SeedreamTool from tools.send_email import SendEmailTool, smtp_configured @@ -207,6 +213,14 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]: ScheduleCancelTool(ctx.uid, **base), ] + def _software_jobs() -> list: + return [ + SoftwareCapabilityListTool(ctx.uid, ctx.task_id, **base), + SoftwareJobSubmitTool(ctx.uid, ctx.task_id, **base), + SoftwareJobStatusTool(ctx.uid, ctx.task_id, **base), + SoftwareJobCancelTool(ctx.uid, ctx.task_id, **base), + ] + def _send_email() -> list: # base_dir 用 working_dir_path(宿主 task 目录):send_email 在宿主进程读附件, # docker 下 agent 给的相对路径相对容器 workdir=task_dir,翻回宿主即 working_dir_path; @@ -294,6 +308,7 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]: ("skill_authoring", lambda: True, _skill_authoring), # 定时 run 内不挂 schedule_*(防任务造任务自我繁殖);仅交互对话可建/管 job。 ("schedules", lambda: not ctx.scheduled_run, _schedules), + ("software_jobs", lambda: not ctx.scheduled_run, _software_jobs), ("send_email", smtp_configured, _send_email), ("wechat_push", wechat_push_available, _wechat_push), ("run_python", lambda: ctx.caps.enable_run_python, _run_python), diff --git a/tests/test_software_job_tools.py b/tests/test_software_job_tools.py new file mode 100644 index 0000000..e5fa4d4 --- /dev/null +++ b/tests/test_software_job_tools.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import json +import unittest +from unittest.mock import patch +from uuid import uuid4 + +from tools.software_jobs import ( + SoftwareCapabilityListTool, + SoftwareJobCancelTool, + SoftwareJobStatusTool, + SoftwareJobSubmitTool, +) + + +class SoftwareJobToolTests(unittest.TestCase): + def setUp(self): + self.user_id = uuid4() + self.task_id = uuid4() + + def test_capability_list_reports_current_capacity(self): + nodes = [ + { + "status": "online", + "capabilities": ["origin.plot@v1"], + "runtime": {"available_slots": 1}, + }, + { + "status": "offline", + "capabilities": ["origin.plot@v1"], + "runtime": {"available_slots": 1}, + }, + ] + with patch("tools.software_jobs.list_nodes", return_value=nodes): + result = json.loads( + SoftwareCapabilityListTool(self.user_id, self.task_id).execute() + ) + self.assertEqual(result["capabilities"][0]["available_nodes"], 1) + + def test_submit_injects_current_user_and_task(self): + created = {"job_id": str(uuid4()), "status": "queued"} + tool = SoftwareJobSubmitTool(self.user_id, self.task_id) + with patch("tools.software_jobs.create_job", return_value=(created, True)) as create: + result = json.loads(tool.execute("origin.plot@v1", {"input_id": str(uuid4())})) + 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@v1") + + def test_status_and_cancel_reject_cross_task_job(self): + foreign = {"job_id": str(uuid4()), "task_id": str(uuid4())} + with patch("tools.software_jobs.get_job", return_value=foreign): + status = SoftwareJobStatusTool(self.user_id, self.task_id).execute( + foreign["job_id"] + ) + cancel = SoftwareJobCancelTool(self.user_id, self.task_id).execute( + foreign["job_id"] + ) + self.assertIn("not found", status) + self.assertIn("not found", cancel) + + def test_cancel_uses_user_scoped_service(self): + job_id = uuid4() + current = {"job_id": str(job_id), "task_id": str(self.task_id)} + cancelled = {**current, "status": "cancelled"} + with ( + patch("tools.software_jobs.get_job", return_value=current), + patch( + "tools.software_jobs.request_job_cancel", + return_value=(cancelled, None), + ) as request_cancel, + ): + result = json.loads( + SoftwareJobCancelTool(self.user_id, self.task_id).execute(str(job_id)) + ) + self.assertEqual(result["status"], "cancelled") + request_cancel.assert_called_once_with(self.user_id, job_id) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_software_nodes.py b/tests/test_software_nodes.py index 9b95387..d8d9e16 100644 --- a/tests/test_software_nodes.py +++ b/tests/test_software_nodes.py @@ -25,6 +25,8 @@ from core.software_jobs import ( respond_to_offer, update_job_state, validate_output_manifest, + list_jobs, + request_job_cancel, ) from web.routers.software_nodes import NodeConnectionManager, _bearer @@ -130,6 +132,52 @@ class SoftwareNodeMigrationTests(unittest.TestCase): class SoftwareJobProtocolTests(unittest.TestCase): + @patch("core.software_jobs.session_scope") + def test_queued_job_cancels_without_node_message(self, session_scope) -> None: + session = session_scope.return_value.__enter__.return_value + job = type("Job", (), {})() + job.job_id = uuid4(); job.user_id = uuid4(); job.task_id = uuid4() + job.capability = "origin.plot@v1"; job.request_digest = "a" * 64 + job.node_id = None; job.lease_id = None; job.status = "queued"; job.stage = "" + job.progress = 0; job.metrics = {}; job.error = {}; job.artifact_manifest = [] + job.created_at = None; job.started_at = None; job.terminal_at = None + session.execute.return_value.scalar_one_or_none.return_value = job + result, node_message = request_job_cancel(job.user_id, job.job_id) + self.assertEqual(result["status"], "cancelled") + self.assertIsNone(node_message) + self.assertEqual(job.error["code"], "USER_CANCELLED") + + @patch("core.software_jobs.session_scope") + def test_running_job_persists_cancel_before_sending(self, session_scope) -> None: + session = session_scope.return_value.__enter__.return_value + job = type("Job", (), {})() + job.job_id = uuid4(); job.user_id = uuid4(); job.task_id = uuid4() + job.capability = "origin.plot@v1"; job.request_digest = "b" * 64 + job.node_id = uuid4(); job.lease_id = uuid4(); job.status = "running" + job.stage = "origin_running"; job.progress = 10; job.metrics = {}; job.error = {} + job.artifact_manifest = []; job.created_at = None; job.started_at = None; job.terminal_at = None + session.execute.return_value.scalar_one_or_none.return_value = job + result, node_message = request_job_cancel(job.user_id, job.job_id) + self.assertEqual(result["status"], "cancelling") + self.assertEqual(node_message["node_id"], job.node_id) + self.assertEqual(node_message["payload"]["lease_id"], str(job.lease_id)) + + @patch("core.software_jobs.session_scope") + def test_job_list_is_enriched_for_job_center(self, session_scope) -> None: + session = session_scope.return_value.__enter__.return_value + job = type("Job", (), {})() + job.job_id = uuid4(); job.task_id = uuid4(); job.capability = "origin.plot@v1" + job.request_digest = "c" * 64; job.node_id = uuid4(); job.status = "running" + job.stage = "origin_running"; job.progress = 20; job.metrics = {}; job.error = {} + job.artifact_manifest = []; job.input_manifest = {"filename": "input.xlsx"} + job.request = {"plot": {"title": "Test"}, "output": {"formats": ["png"]}} + job.created_at = None; job.started_at = None; job.terminal_at = None + session.execute.return_value.all.return_value = [(job, "材料仿真", "LAB-01")] + results = list_jobs(uuid4(), limit=10) + self.assertEqual(results[0]["task_name"], "材料仿真") + self.assertEqual(results[0]["node_name"], "LAB-01") + self.assertEqual(results[0]["request_summary"]["display_name"], "Origin 科研绘图") + def test_origin_request_is_canonical_and_rejects_extra_fields(self) -> None: request = { "schema_version": 1, diff --git a/tests/test_static_vendor.py b/tests/test_static_vendor.py index 22cfea1..606607a 100644 --- a/tests/test_static_vendor.py +++ b/tests/test_static_vendor.py @@ -26,6 +26,15 @@ def _frontend_source() -> str: class StaticVendorTests(unittest.TestCase): + def test_dev_console_has_software_job_center(self) -> None: + html = DEV_HTML.read_text(encoding="utf-8") + source = (JS_DIR / "software_jobs.js").read_text(encoding="utf-8") + self.assertIn('id="software-job-center"', html) + self.assertIn('id="software-job-toggle"', html) + self.assertIn('"/v1/software-jobs?limit=50"', source) + self.assertIn("/cancel`", source) + self.assertIn("分析结果", source) + def test_admin_can_create_windows_node_enrollment_code(self) -> None: html = ADMIN_HTML.read_text(encoding="utf-8") admin_js = (JS_DIR / "admin.js").read_text(encoding="utf-8") diff --git a/tests/test_web_routes_nodb.py b/tests/test_web_routes_nodb.py index f344b15..b9ee36a 100644 --- a/tests/test_web_routes_nodb.py +++ b/tests/test_web_routes_nodb.py @@ -111,6 +111,8 @@ class AuthGateTests(unittest.TestCase): ("POST", "/v1/asr/transcribe"), ("GET", "/v1/admin/overview"), ("GET", "/v1/admin/software-nodes"), + ("GET", "/v1/software-jobs"), + ("POST", "/v1/software-jobs/00000000-0000-0000-0000-000000000000/cancel"), ("DELETE", "/v1/admin/software-nodes/00000000-0000-0000-0000-000000000000"), ("GET", "/v1/admin/tool-wire-health"), ("GET", "/v1/admin/external-system-definitions"), diff --git a/tests/test_windows_node_source.py b/tests/test_windows_node_source.py index e5a4c89..a60d55c 100644 --- a/tests/test_windows_node_source.py +++ b/tests/test_windows_node_source.py @@ -169,7 +169,10 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertIn("startInfo.ArgumentList.Add(jobDirectory)", runner) self.assertIn('Path.Combine(jobDirectory, "terminal.json")', runner) self.assertIn('"NODE_RESTARTED_DURING_JOB"', runner) - self.assertNotIn("RunAsync(RecoverableJob job, CancellationToken", runner) + self.assertIn("CancellationTokenSource.CreateLinkedTokenSource", runner) + self.assertIn("process.Kill(entireProcessTree: true)", runner) + self.assertIn('type.GetString() == "job_cancel"', connection) + self.assertIn('"cancelled", "USER_CANCELLED"', connection) self.assertIn("origin-worker\\worker.py", project) self.assertIn("if op.oext:", worker) self.assertIn("op.exit()", worker) diff --git a/tools/software_jobs.py b/tools/software_jobs.py new file mode 100644 index 0000000..c8b13fa --- /dev/null +++ b/tools/software_jobs.py @@ -0,0 +1,122 @@ +"""Agent 可调用的专业软件任务工具。""" +from __future__ import annotations + +import json +from uuid import UUID, uuid4 + +from core.software_jobs import ( + SoftwareJobError, + create_job, + get_job, + list_jobs, + request_job_cancel, +) +from core.software_nodes import SUPPORTED_CAPABILITIES, list_nodes +from .base import Tool + + +class _SoftwareJobTool(Tool): + def __init__(self, user_id: UUID, task_id: UUID, **kwargs) -> None: + super().__init__(**kwargs) + self.user_id = user_id + self.task_id = task_id + + +class SoftwareCapabilityListTool(_SoftwareJobTool): + name = "software_capability_list" + description = "List professional software capabilities available through managed Windows nodes." + parameters = {"type": "object", "properties": {}, "additionalProperties": False} + + def execute(self) -> str: + nodes = list_nodes() + items = [{ + "capability": item, + "display_name": "Origin 科研绘图" if item == "origin.plot@v1" else item, + "available_nodes": sum( + 1 + for node in nodes + if node["status"] == "online" + and item in (node.get("capabilities") or []) + and (node.get("runtime") or {}).get("available_slots", 0) > 0 + ), + } for item in sorted(SUPPORTED_CAPABILITIES)] + return json.dumps({"capabilities": items}, ensure_ascii=False) + + +class SoftwareJobSubmitTool(_SoftwareJobTool): + name = "software_job_submit" + description = ( + "Submit a persistent professional software job for the current task. " + "Return immediately with job_id; do not poll continuously or wait for completion." + ) + parameters = { + "type": "object", + "properties": { + "capability": {"type": "string", "enum": sorted(SUPPORTED_CAPABILITIES)}, + "request": {"type": "object"}, + "idempotency_key": { + "type": "string", + "description": "Stable unique key for this exact submission; omit to generate one.", + }, + }, + "required": ["capability", "request"], + "additionalProperties": False, + } + + def execute(self, capability: str, request: dict, idempotency_key: str = "") -> str: + try: + job, created = create_job( + self.user_id, + self.task_id, + idempotency_key=idempotency_key.strip() or str(uuid4()), + capability=capability, + request=request, + ) + return json.dumps({**job, "created": created}, ensure_ascii=False) + except SoftwareJobError as exc: + return f"[Error] {exc}" + + +class SoftwareJobStatusTool(_SoftwareJobTool): + name = "software_job_status" + description = "Check one software job, or list recent jobs in the current task when job_id is omitted." + parameters = { + "type": "object", + "properties": {"job_id": {"type": "string"}}, + "additionalProperties": False, + } + + def execute(self, job_id: str = "") -> str: + try: + if job_id.strip(): + item = get_job(self.user_id, UUID(job_id.strip())) + if item is None or item["task_id"] != str(self.task_id): + return "[Error] software job not found" + return json.dumps(item, ensure_ascii=False) + return json.dumps( + {"results": list_jobs(self.user_id, task_id=self.task_id, limit=20)}, + ensure_ascii=False, + ) + except ValueError: + return "[Error] invalid job_id" + + +class SoftwareJobCancelTool(_SoftwareJobTool): + name = "software_job_cancel" + description = "Request cancellation of a software job in the current task after the user asks to stop it." + parameters = { + "type": "object", + "properties": {"job_id": {"type": "string"}}, + "required": ["job_id"], + "additionalProperties": False, + } + + def execute(self, job_id: str) -> str: + try: + item = get_job(self.user_id, UUID(job_id.strip())) + if item is None or item["task_id"] != str(self.task_id): + return "[Error] software job not found" + job, _ = request_job_cancel(self.user_id, UUID(job_id.strip())) + return json.dumps(job, ensure_ascii=False) + except (ValueError, SoftwareJobError) as exc: + return f"[Error] {exc}" diff --git a/web/routers/software_nodes.py b/web/routers/software_nodes.py index 24af62a..8f7ebbe 100644 --- a/web/routers/software_nodes.py +++ b/web/routers/software_nodes.py @@ -33,7 +33,10 @@ from core.software_jobs import ( get_job_output_context, mark_node_jobs_disconnected, offer_next_job, + pending_node_cancellations, record_job_terminal, + list_jobs, + request_job_cancel, respond_to_offer, update_job_state, validate_output_manifest, @@ -375,6 +378,10 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: websocket, {"type": "connected", "heartbeat_seconds": 15}, ) + for cancel in await asyncio.to_thread(pending_node_cancellations, node_id): + await node_connections.send_on( + node_id, websocket, {"type": "job_cancel", "payload": cancel} + ) while True: message = await websocket.receive_json() message_type = message.get("type") @@ -434,6 +441,10 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: await node_connections.send_on(node_id, websocket, {"type": "ack", "message_id": message.get("message_id")} ) + for cancel in await asyncio.to_thread(pending_node_cancellations, node_id): + await node_connections.send_on( + node_id, websocket, {"type": "job_cancel", "payload": cancel} + ) offer = await asyncio.to_thread( offer_next_job, await node_connections.node_ids() ) @@ -446,7 +457,13 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: await asyncio.to_thread( abandon_offer, offer["node_id"], offer["payload"] ) - except (SoftwareNodeError, WebSocketDisconnect, RuntimeError, ValueError): + except ( + SoftwareJobError, + SoftwareNodeError, + WebSocketDisconnect, + RuntimeError, + ValueError, + ): pass finally: if await node_connections.remove(node_id, websocket): @@ -479,6 +496,35 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: detail = str(exc) raise HTTPException(404 if detail == "task not found" else 400, detail) from exc + @app.get("/v1/software-jobs", tags=["software-jobs"]) + def read_software_jobs( + task_id: UUID | None = None, + active_only: bool = False, + limit: int = 50, + user_id: UUID = Depends(require_user), # noqa: B008 + ): + return { + "results": list_jobs( + user_id, task_id=task_id, active_only=active_only, limit=limit + ) + } + + @app.post("/v1/software-jobs/{job_id}/cancel", tags=["software-jobs"]) + async def cancel_software_job( + job_id: UUID, + user_id: UUID = Depends(require_user), # noqa: B008 + ): + try: + job, message = await asyncio.to_thread(request_job_cancel, user_id, job_id) + except SoftwareJobError as exc: + raise HTTPException(404, str(exc)) from exc + if message is not None: + await node_connections.send( + message["node_id"], + {"type": "job_cancel", "payload": message["payload"]}, + ) + return job + @app.get("/v1/software-jobs/{job_id}", tags=["software-jobs"]) def read_software_job( job_id: UUID, diff --git a/web/static/dev.html b/web/static/dev.html index bd9342c..85cd211 100644 --- a/web/static/dev.html +++ b/web/static/dev.html @@ -1069,6 +1069,34 @@ overflow: hidden; text-overflow: ellipsis; white-space: nowrap; margin-top: 2px; } .proc-toast.fail { border-color: rgba(192,57,43,0.5); } + #software-job-center { + position: fixed; right: 16px; bottom: 16px; z-index: 115; display: none; + width: min(380px, calc(100vw - 24px)); font-size: 13px; + } + #software-job-center.show { display: block; } + #software-job-toggle { + margin-left: auto; display: flex; align-items: center; gap: 8px; + border-radius: 999px; padding: 8px 13px; background: #fff; + border: 1px solid var(--border); box-shadow: 0 4px 16px rgba(0,0,0,.16); cursor: pointer; + } + #software-job-center.expanded #software-job-toggle { border-radius: 0 0 10px 10px; } + #software-job-panel { + max-height: min(520px, calc(100vh - 110px)); overflow: auto; background: #fff; + border: 1px solid var(--border); border-bottom: 0; border-radius: 10px 10px 0 0; + box-shadow: 0 -4px 20px rgba(0,0,0,.16); padding: 8px; + } + .sj-card { padding: 10px; border-bottom: 1px solid var(--border); } + .sj-card:last-child { border-bottom: 0; } + .sj-title { display: flex; justify-content: space-between; gap: 10px; } + .sj-title span { color: var(--muted); white-space: nowrap; } + .sj-card.failed .sj-title span { color: var(--danger); } + .sj-card.disconnected .sj-title span, .sj-card.cancelling .sj-title span { color: #a66514; } + .sj-sub, .sj-meta { margin-top: 4px; color: var(--muted); font-size: 11px; overflow-wrap: anywhere; } + .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: 8px; } + .sj-empty { padding: 22px; text-align: center; color: var(--muted); } + #software-job-center.show ~ #proc-toasts { bottom: 62px; } /* media tool 摘要 banner(model / size / cost / elapsed,折叠态也可见) */ .tool-banner { display: inline-flex; flex-wrap: wrap; gap: 6px; @@ -2174,6 +2202,13 @@ +
+ + +
+ diff --git a/web/static/js/main.js b/web/static/js/main.js index 41d4212..ab7afa3 100644 --- a/web/static/js/main.js +++ b/web/static/js/main.js @@ -19,6 +19,7 @@ import { loadFolderSuggestions } from "./newtask.js"; import { embedInit } from "./embed.js"; import { loadTaskList, loadModels, loadChannelCards, selectTask, showNewConversationDraft } from "./chat.js"; import { refreshProcs } from "./procs.js"; +import { initSoftwareJobs } from "./software_jobs.js"; // ───── enter app ───── export function enterApp(initialTaskId = "") { @@ -40,6 +41,7 @@ export function enterApp(initialTaskId = "") { loadStorage(); // 顶栏存储用量(后台扫描快照,非实时) loadRole(); // 拉 /v1/me,admin 才显「管理」入口(/static/admin.html) refreshProcs(); // 后台进程(bg proc):上个会话遗留的 running proc 恢复轮询/展示 + initSoftwareJobs(); } // 顶栏用户名:默认显 name(兜底 user_name → email → uid8),title 悬浮给完整身份。 diff --git a/web/static/js/software_jobs.js b/web/static/js/software_jobs.js new file mode 100644 index 0000000..0988809 --- /dev/null +++ b/web/static/js/software_jobs.js @@ -0,0 +1,144 @@ +// 专业软件任务中心：用户级轮询、跨对话状态、取消与终态通知。 +import { api } from "./api.js"; +import { state } from "./state.js"; +import { $ } from "./dom.js"; +import { escapeHtml } from "./format.js"; +import { selectTask } from "./chat.js"; +import { dialogConfirm, message } from "./dialog.js"; + +const ACTIVE = new Set(["queued", "offered", "dispatched", "running", "disconnected", "cancelling"]); +const TERMINAL = new Set(["succeeded", "failed", "cancelled"]); +const POLL_ACTIVE_MS = 4000; +const POLL_IDLE_MS = 30000; +let timer = null; +let known = new Map(); +let jobs = []; +let expanded = false; +let initialized = false; + +const statusLabel = { + queued: "等待计算节点", offered: "正在分配节点", dispatched: "节点已接收", + running: "正在执行", disconnected: "节点连接中断", cancelling: "正在停止", + succeeded: "已完成", failed: "失败", cancelled: "已取消", +}; +const stageLabel = { + accepted: "节点已接收", waiting_input: "正在下载输入文件", ready_to_run: "准备软件环境", + origin_running: "Origin 正在生成图表", uploading_outputs: "正在上传结果", + cancel_requested: "停止请求已发送", terminal: "任务已结束", +}; + +export function initSoftwareJobs() { + if (initialized || !$("software-job-center")) return; + initialized = true; + $("software-job-toggle").onclick = () => { expanded = !expanded; render(); }; + document.addEventListener("click", (event) => { + const center = $("software-job-center"); + if (expanded && center && !center.contains(event.target)) { expanded = false; render(); } + }); + refreshSoftwareJobs(); +} + +export async function refreshSoftwareJobs() { + if (!state.token) return; + try { + const data = await api("GET", "/v1/software-jobs?limit=50"); + 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); + }); + jobs = next; + render(); + } catch (_) { /* 后台刷新失败静默，下轮恢复 */ } + schedule(); +} + +function schedule() { + if (timer) clearTimeout(timer); + const delay = jobs.some((job) => ACTIVE.has(job.status)) ? POLL_ACTIVE_MS : POLL_IDLE_MS; + timer = setTimeout(refreshSoftwareJobs, document.hidden ? Math.max(delay, 30000) : delay); +} + +document.addEventListener("visibilitychange", () => { + if (!document.hidden && state.token) refreshSoftwareJobs(); else schedule(); +}); + +function render() { + const center = $("software-job-center"); + const panel = $("software-job-panel"); + const active = jobs.filter((job) => ACTIVE.has(job.status)); + const recent = jobs.slice(0, 10); + center.classList.toggle("show", active.length > 0 || recent.length > 0); + center.classList.toggle("expanded", expanded); + $("software-job-count").textContent = active.length ? `${active.length} 运行中` : `${jobs.length} 条`; + panel.hidden = !expanded; + if (!expanded) return; + panel.innerHTML = recent.length + ? recent.map(jobCard).join("") + : '
暂无专业软件任务
'; + panel.querySelectorAll("[data-job-action]").forEach((button) => { + button.onclick = (event) => handleAction(event, button); + }); +} + +function jobCard(job) { + const summary = job.request_summary || {}; + const input = job.input || {}; + 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); + return `
+
${escapeHtml(summary.display_name || job.capability)} + ${escapeHtml(statusLabel[job.status] || job.status)}
+
${escapeHtml(detail)}${error ? ` · ${escapeHtml(error)}` : ""}
+ ${active ? `
` : ""} +
${escapeHtml(job.task_name || "未命名对话")}${input.filename ? ` · ${escapeHtml(input.filename)}` : ""}
+
+ + ${job.status === "succeeded" ? `` : ""} + ${active && job.status !== "cancelling" ? '' : ""} +
+
`; +} + +async function handleAction(event, button) { + event.stopPropagation(); + const card = button.closest("[data-job-id]"); + const jobId = card.dataset.jobId; + const action = button.dataset.jobAction; + if (action === "cancel") { + if (!await dialogConfirm({ + title: "停止专业软件任务", + message: "确定停止这个任务？已经产生但尚未发布的中间输出可能不会保留。", + okText: "停止", danger: true, + })) return; + button.disabled = true; + try { await api("POST", `/v1/software-jobs/${jobId}/cancel`); } + catch (error) { message(error.message || "停止失败", "error"); } + refreshSoftwareJobs(); + return; + } + const taskId = button.dataset.taskId; + if (taskId) await selectTask(taskId); + expanded = false; + render(); + 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) { + const ok = job.status === "succeeded"; + const label = ok ? "已完成" : (job.status === "cancelled" ? "已取消" : "失败"); + const summary = job.request_summary || {}; + message(`${summary.display_name || "专业软件任务"}${label}`, ok ? "success" : "error", 6000); + expanded = true; +} diff --git a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs index 804f189..cf68dd7 100644 --- a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs +++ b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs @@ -251,11 +251,47 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? StartJobPipeline(socket, acceptedJob); } } + else if (type.GetString() == "job_cancel" + && document.RootElement.TryGetProperty("payload", out var cancelPayload) + && TryCancelJob(cancelPayload, out var cancelledJob)) + { + workerRunner.Cancel(cancelledJob.JobId); + jobInbox.WriteTerminal( + cancelledJob, "cancelled", "USER_CANCELLED", "Cancelled by user."); + await SendAsync(socket, "job_terminal", new + { + job_id = cancelledJob.JobId, + lease_id = cancelledJob.LeaseId, + request_digest = cancelledJob.RequestDigest, + status = "cancelled", + error = new { code = "USER_CANCELLED", detail = "Cancelled by user." }, + artifact_manifest = Array.Empty(), + }, cancellationToken); + } } message.SetLength(0); } } + private bool TryCancelJob(JsonElement payload, out RecoverableJob job) + { + job = null!; + if (!payload.TryGetProperty("job_id", out var jobIdValue) + || !Guid.TryParse(jobIdValue.GetString(), out var jobId) + || !payload.TryGetProperty("lease_id", out var leaseIdValue) + || !Guid.TryParse(leaseIdValue.GetString(), out var leaseId) + || !payload.TryGetProperty("request_digest", out var digestValue)) + { + return false; + } + job = jobInbox.ReadRecoverableJobs().SingleOrDefault(item => + item.JobId == jobId + && item.LeaseId == leaseId + && item.RequestDigest == digestValue.GetString() + && item.Terminal is null)!; + return job is not null; + } + private void StartJobPipeline(ClientWebSocket socket, RecoverableJob job) { var completion = new TaskCompletionSource( @@ -287,6 +323,13 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? if (job.Terminal is null) { await inputDownloader.DownloadAsync(job, CancellationToken.None); + var afterDownload = jobInbox.ReadRecoverableJobs() + .Single(item => item.JobId == job.JobId); + if (afterDownload.Terminal is JsonElement cancelledTerminal) + { + await TrySendAsync(socket, "job_terminal", cancelledTerminal); + return; + } await TrySendAsync(socket, "job_state", new { job_id = job.JobId, diff --git a/windows-node/Zcbot.WindowsNode/OriginWorkerRunner.cs b/windows-node/Zcbot.WindowsNode/OriginWorkerRunner.cs index d4d9d4e..513e653 100644 --- a/windows-node/Zcbot.WindowsNode/OriginWorkerRunner.cs +++ b/windows-node/Zcbot.WindowsNode/OriginWorkerRunner.cs @@ -9,11 +9,20 @@ internal sealed class OriginWorkerRunner(JobInboxStore inbox) { private static readonly TimeSpan WorkerTimeout = TimeSpan.FromMinutes(30); private readonly ConcurrentDictionary active = new(); + private readonly ConcurrentDictionary cancellations = new(); internal Task RunAsync(RecoverableJob job) => - active.GetOrAdd(job.JobId, _ => RunOnceAsync(job)); + active.GetOrAdd(job.JobId, _ => RunOnceAsync(job, CancellationFor(job.JobId).Token)); - private async Task RunOnceAsync(RecoverableJob job) + internal void Cancel(Guid jobId) + { + CancellationFor(jobId).Cancel(); + } + + private CancellationTokenSource CancellationFor(Guid jobId) => + cancellations.GetOrAdd(jobId, _ => new CancellationTokenSource()); + + private async Task RunOnceAsync(RecoverableJob job, CancellationToken cancellationToken) { try { @@ -60,7 +69,7 @@ internal sealed class OriginWorkerRunner(JobInboxStore inbox) ?? throw new InvalidOperationException("The fixed Origin worker did not start."); var stdout = process.StandardOutput.ReadToEndAsync(); var stderr = process.StandardError.ReadToEndAsync(); - using var timeout = new CancellationTokenSource(); + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); timeout.CancelAfter(WorkerTimeout); try { @@ -69,7 +78,14 @@ internal sealed class OriginWorkerRunner(JobInboxStore inbox) catch (OperationCanceledException) { process.Kill(entireProcessTree: true); - inbox.WriteTerminal(job, "failed", "ORIGIN_WORKER_TIMEOUT", "Origin worker exceeded 30 minutes."); + if (cancellationToken.IsCancellationRequested) + { + inbox.WriteTerminal(job, "cancelled", "USER_CANCELLED", "Cancelled by user."); + } + else + { + inbox.WriteTerminal(job, "failed", "ORIGIN_WORKER_TIMEOUT", "Origin worker exceeded 30 minutes."); + } return; } var output = await stdout; @@ -95,6 +111,7 @@ internal sealed class OriginWorkerRunner(JobInboxStore inbox) finally { active.TryRemove(job.JobId, out _); + if (cancellations.TryRemove(job.JobId, out var cancellation)) cancellation.Dispose(); } }