feat(software): require artifact inputs for Origin jobs

This commit is contained in:
caoqianming 2026-08-13 17:25:21 +08:00
parent e6128c7e5b
commit 40f8899cc1
11 changed files with 309 additions and 20 deletions

View File

@ -8,7 +8,7 @@
## Unreleased
- 新增专业软件 Job 中心Agent 可提交、查询和停止 Windows Node 上的受控软件任务;用户可在右下角跨对话查看进度、收到完成通知,并回到原对话分析结果。
- 新增专业软件 Job 中心Agent 可将当前对话目录内的数据登记为稳定输入,提交、查询和停止 Windows Node 上的受控软件任务;用户可在右下角跨对话查看进度、收到完成通知,并回到原对话分析结果。
## 0.65.2 — 2026-08-13

View File

@ -466,13 +466,13 @@ Node 通过 `Authorization: Bearer` 与 `X-Node-Id` 建立 `/v1/software-nodes/c
第二阶段已增加 `software_jobs`(专业软件任务)账本与 `origin.plot@v1` 的 offer/accept 骨架。用户只能在本人 task 下以幂等键提交固定 schema云端规范化请求并记录 SHA-256按当前进程真实在线、能力匹配、健康且有空闲 slot 的 Node 创建短期 offer。Node 再次校验 schema、图形类型和输出格式使用 write-through、flush 与原子 rename 先落本机任务目录,再回 `job_accept`;重复 job 只有 digest 一致才接受。过期或发送失败的 offer 回到队列lease、Node 和 digest 不匹配的响应被拒绝。Node 接收后云端进入 `dispatched` 而非 `running`,并将 slot 降为 0只有固定 Worker 真正启动后才进入 `origin_running`
第三阶段补齐输入下载与恢复状态协议:`input_id` 固定为用户已有 artifact UUID提交时快照文件名、大小和 SHA-256只允许 CSV/XLSX/JSON 且不超过 100 MiB。Node 以自身 Bearer 身份访问任务绑定的只读下载端点,流式写入本 job 的 `input/`,同时限制声明大小并校验 SHA-256完成后原子 rename不暴露工作区路径。Node 会原子读取/补报 `terminal.json`,断线后云端把活动任务标记 `disconnected` 并保留 Node/lease重连按 job、lease、digest 恢复下载或幂等补报终态,不自动重派。
第三阶段补齐输入下载与恢复状态协议:`input_id` 固定为 artifact UUID。已有 artifact 可直接提交;普通 task 文件先调用 `register_artifact(path)` 登记稳定身份该动作不把文件发布为聊天交付物。提交时快照文件名、大小和 SHA-256只允许 CSV/XLSX/JSON 且不超过 100 MiB。Node 以自身 Bearer 身份访问任务绑定的只读下载端点,流式写入本 job 的 `input/`,同时限制声明大小并校验 SHA-256完成后原子 rename不暴露工作区路径。Node 会原子读取/补报 `terminal.json`,断线后云端把活动任务标记 `disconnected` 并保留 Node/lease重连按 job、lease、digest 恢复下载或幂等补报终态,不自动重派。
第四阶段落地固定 Origin WorkerNode 仅从管理员安装的固定 Python 运行时启动随程序发布的 `worker.py`,参数只有本机 job 目录请求不能指定脚本、解释器或文件路径。Worker 使用 `originpro` 生成 OPJU、PNG、SVG、PDF、plot spec 和 provenance校验产物签名并原子写入终态当前受控图形仅含 line、scatter、line_scatter 和双栏出版布局。进程内 pipeline 按 job 去重,并脱离单次 WebSocket 的取消令牌运行;连接中断只延迟状态/终态上报。Node 进程若在 Worker 启动后重启,则保守失败而不重复驱动 Origin避免无法证明的双执行。
第五阶段完成输出上传与发布Node 只按固定 manifest ID 逐项流式 PUT并携带 Node、lease、request digest 与内容摘要;云端重新绑定任务身份,不信 Node 提供的路径或媒体类型。文件先进入用户根下隐藏暂存区,固定文件名、单文件/总大小和 SHA-256 全部验证后,目录级原子移动到 `<working_dir>/origin/<job_id>/`,再登记平台 artifact UUID 并写成功终态。重复 PUT、complete 和重连均按摘要幂等;部分上传不可见,只有完整集合才能发布。
第六阶段增加用户级 Job 中心与 Agent typed tools。`software_capability_list` 只暴露固定能力及当前在线空闲节点数,`software_job_submit/status/cancel` 在构造时绑定当前 user/task模型不能跨用户或跨对话指定归属。右下角 Job 中心按用户聚合各对话任务,活动期短轮询、空闲期降频;终态变化通知用户,成功任务可回到原对话发起分析。取消采用协作协议:未派发任务直接终止,已派发任务先进入 `cancelling`,云端通过 WebSocket 发送并在心跳时重放 `job_cancel`Node 杀死固定 Worker 进程树后回报 `cancelled`;终态写入仍由云端账本裁决。
第六阶段增加用户级 Job 中心与 Agent typed tools。`software_capability_list` 只暴露固定能力及当前在线空闲节点数,`software_job_submit/status/cancel` 在构造时绑定当前 user/task模型不能跨用户或跨对话指定归属。`software_job_submit` 的模型可见唯一入口为 `input_id + plot + output``register_artifact` 是普通文件获得输入身份的唯一入口,并明确返回 UUID。内部完整 `request` 形式仅保留执行层兼容,不进入工具 schema。右下角 Job 中心按用户聚合各对话任务,活动期短轮询、空闲期降频;终态变化通知用户,成功任务可回到原对话发起分析。取消采用协作协议:未派发任务直接终止,已派发任务先进入 `cancelling`,云端通过 WebSocket 发送并在心跳时重放 `job_cancel`Node 杀死固定 Worker 进程树后回报 `cancelled`;终态写入仍由云端账本裁决。
后续仍需实现 Token 轮换;不得以任意命令或脚本接口临时代替。当前 Job 中心采用轮询而非用户事件推送,单活与 offer 选择仍只覆盖单 Web 进程;生产启用多实例前必须增加 Redis/PG fencing 或固定路由到单一控制面实例。

View File

@ -22,6 +22,8 @@
### 2026-08-13
- **08-13 / Unreleased / 专业软件输入 artifact 契约**:新增 `register_artifact`,将 task 内普通文件登记为稳定输入但不发布聊天交付卡片;`software_job_submit` 改为模型可见的完整 Origin schema并强制使用 artifact UUID已有 artifact 可直接复用,内部完整 request 仅保留兼容。同步校准当前三种图形类型与运行/设计文档;相关 51 项 unittest、Python 编译和 diff 检查通过,未连接或写入生产 DB。
- **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。

2
RUN.md
View File

@ -1092,7 +1092,7 @@ install-windows-node.bat
若 Python 未加入 PATH可把绝对路径作为第一个参数例如 `install-windows-node.bat "C:\Python312\python.exe"`。默认解释器为 `%ProgramData%\Zcbot\WindowsNode\runtimes\origin\Scripts\python.exe`。如需使用其他受管解释器,设置机器级 `ZCBOT_ORIGIN_PYTHON` 为绝对 `python.exe` 路径后重启 Node。`node.json`、可恢复任务和 runtime 集中保存在 `%ProgramData%\Zcbot\WindowsNode\`,不会因替换程序目录而丢失。运行时固定依赖见发布目录的 `origin-worker/requirements.txt`;任务请求无权选择解释器、脚本或路径。当前 Worker 支持 CSV/XLSX/JSON 输入,`line`、`scatter`、`line_scatter` 与 OPJU/PNG/SVG/PDF 输出。成功产物由 Node 流式上传,全部校验通过后发布到任务工作目录 `origin/<job_id>/`;上传中断会在重连时幂等续传。
Web 用户登录后,右下角 Job 中心会聚合本人最近任务。活动任务约 4 秒刷新一次,空闲时降为约 30 秒停止已派发任务是协作取消状态先显示“正在停止”Node 在线时立即接收断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`software_job_submit`、`software_job_status` 和 `software_job_cancel`;提交工具只创建固定 schema 的持久任务,不会阻塞当前对话等待完成。
Web 用户登录后,右下角 Job 中心会聚合本人最近任务。活动任务约 4 秒刷新一次,空闲时降为约 30 秒停止已派发任务是协作取消状态先显示“正在停止”Node 在线时立即接收断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`register_artifact`、`software_job_submit`、`software_job_status` 和 `software_job_cancel`。Origin 输入必须是 artifact已有 UUID 可直接提交,普通 task 文件先用相对路径登记登记不会发布聊天交付卡片。提交工具接收 `input_id`、`plot`、`output`只创建固定 schema 的持久任务,不会阻塞当前对话等待完成。
注册配置写入 `%ProgramData%\Zcbot\WindowsNode\node.json`Token 使用 DPAPI `LocalMachine` 加密ACL 仅允许注册账号和 `SYSTEM`。应始终用同一专用 Windows 账号执行统一安装器、注册并运行 Node。当前 MVP 以该账号的登录后计划任务启动,不安装 Windows Service。

View File

@ -45,6 +45,7 @@ def register_published_artifacts(
user_root: Path,
working_dir: Path,
refs: tuple[dict, ...],
preserve_existing_labels: bool = False,
) -> tuple[dict, ...]:
"""Upsert active artifact identities and return version-2 message refs."""
root = Path(user_root).resolve()
@ -62,6 +63,14 @@ def register_published_artifacts(
media_type = str(ref.get("media_type") or "") or mimetypes.guess_type(path.name)[0]
size_bytes = path.stat().st_size
content_sha256 = _hash_file(path)
update_values = {
"media_type": media_type,
"size_bytes": size_bytes,
"content_sha256": content_sha256,
"updated_at": datetime.now(timezone.utc),
}
if not preserve_existing_labels:
update_values["label"] = label
statement = pg_insert(Artifact).values(
user_id=user_id,
origin_task_id=task_id,
@ -73,13 +82,7 @@ def register_published_artifacts(
).on_conflict_do_update(
index_elements=[Artifact.user_id, Artifact.current_path],
index_where=Artifact.status == "active",
set_={
"label": label,
"media_type": media_type,
"size_bytes": size_bytes,
"content_sha256": content_sha256,
"updated_at": datetime.now(timezone.utc),
},
set_=update_values,
).returning(Artifact.artifact_id)
artifact_id = session.execute(statement).scalar_one()
output.append(ArtifactRef(
@ -91,6 +94,34 @@ def register_published_artifacts(
return tuple(output)
def register_workspace_artifact(
*,
user_id: UUID,
task_id: UUID,
user_root: Path,
working_dir: Path,
path: str,
label: str = "",
) -> dict:
"""Register one task-relative file for an internal platform workflow.
Unlike ``publish_artifacts``, this only creates or refreshes the stable artifact
identity. It does not attach the file to an assistant message or present it as a
final deliverable.
"""
refs = register_published_artifacts(
user_id=user_id,
task_id=task_id,
user_root=user_root,
working_dir=working_dir,
refs=({"path": path, "label": label},),
preserve_existing_labels=True,
)
if len(refs) != 1 or not refs[0].get("artifact_id"):
raise ValueError("workspace artifact could not be registered")
return refs[0]
def rename_active_artifacts(
*,
user_id: UUID,

View File

@ -42,6 +42,7 @@ from tools.materials_project import (
)
from tools.office_to_pdf import OfficeToPdfTool
from tools.read_document import ReadDocumentTool
from tools.register_artifact import RegisterArtifactTool
from tools.rename_working_dir import RenameWorkingDirTool
from tools.run_python import RunPythonTool
from tools.schedule import (
@ -145,6 +146,12 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
working_dir=ctx.working_dir_path,
**wd_base,
),
RegisterArtifactTool(
ctx.uid,
ctx.task_id,
working_dir=ctx.working_dir_path,
**wd_base,
),
]
def _document_search() -> list:

View File

@ -440,7 +440,7 @@ C# COM 只作未覆盖能力或旧版本备用,并与实际 Origin 版本匹
```json
{
"schema_version": 1,
"input": {"input_id": "data-01", "sheet": "Sheet1"},
"input": {"input_id": "5af18255-c859-43f3-a424-220bc68b7db2", "sheet": "Sheet1"},
"plot": {
"type": "line_scatter",
"x": "Temperature",
@ -461,7 +461,13 @@ C# COM 只作未覆盖能力或旧版本备用,并与实际 Origin 版本匹
}
```
首期图形类型line、scatter、line_scatter、grouped_bar、box、histogram、heatmap。后续增加误差棒组合、三元图、等高线、三维曲面、XRD 堆叠图、热分析联图和多面板布局。
当前图形类型line、scatter、line_scatter。后续增加 grouped_bar、box、histogram、heatmap、误差棒组合、三元图、等高线、三维曲面、XRD 堆叠图、热分析联图和多面板布局。
上述 JSON 同时是提交工具与云端账本、Windows Node 之间的 artifact 输入契约。
Agent 优先复用已有 artifact UUID普通 task 文件先调用
`register_artifact(path)` 获得 UUID再以 `input_id`、`plot`、`output` 调用
`software_job_submit`。登记输入不会把文件发布为聊天交付物。旧的完整 `request`
参数仅作为执行层兼容入口保留,不暴露在模型可见 schema 中。
### 12.3 模板与产物

View File

@ -1,3 +1,4 @@
import json
import tempfile
import unittest
from contextlib import contextmanager
@ -16,6 +17,7 @@ from core.artifacts import (
from core.executor import ExecCtx
from core.executor_host import HostExecutor
from tools.publish_artifacts import PublishArtifactsTool
from tools.register_artifact import RegisterArtifactTool
from web.routers.files import _task_file_target
@ -133,6 +135,43 @@ class ArtifactPathTests(unittest.TestCase):
self.assertEqual(executed.content, "[OK] published 1 artifact(s): report.pdf")
self.assertEqual(executed.artifacts[0]["path"], "report.pdf")
def test_register_artifact_returns_identity_without_publishing(self) -> None:
artifact_id = uuid4()
tool = RegisterArtifactTool(
uuid4(),
uuid4(),
working_dir=self.wd,
base_dir=self.wd,
user_root=self.root,
)
with patch(
"tools.register_artifact.register_workspace_artifact",
return_value={
"version": 2,
"artifact_id": str(artifact_id),
"scope": "working_dir",
"path": "report.pdf",
},
) as register:
result = json.loads(tool.execute("report.pdf"))
self.assertEqual(result["artifact_id"], str(artifact_id))
self.assertEqual(register.call_args.kwargs["path"], "report.pdf")
def test_register_artifact_rejects_path_escape(self) -> None:
tool = RegisterArtifactTool(
uuid4(),
uuid4(),
working_dir=self.wd,
base_dir=self.wd,
user_root=self.root,
)
self.assertIn("cannot register artifact", tool.execute("../outside.csv"))
def test_register_artifact_schema_is_distinct_from_publish(self) -> None:
self.assertEqual(RegisterArtifactTool.parameters["required"], ["path"])
self.assertNotIn("artifacts", RegisterArtifactTool.parameters["properties"])
def test_publish_path_is_strictly_task_relative_when_names_repeat(self) -> None:
nested = self.wd / "技术讨论" / "nested.html"
nested.parent.mkdir()

View File

@ -39,12 +39,42 @@ class SoftwareJobToolTests(unittest.TestCase):
def test_submit_injects_current_user_and_task(self):
created = {"job_id": str(uuid4()), "status": "queued"}
artifact_id = uuid4()
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())}))
result = json.loads(tool.execute(
"origin.plot@v1",
input_id=str(artifact_id),
plot={"type": "scatter", "x": "x", "y": ["y"]},
output={"formats": ["opju", "png"], "dpi": 300},
))
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")
self.assertEqual(
create.call_args.kwargs["request"],
{
"schema_version": 1,
"input": {"input_id": str(artifact_id)},
"plot": {"type": "scatter", "x": "x", "y": ["y"]},
"output": {"formats": ["opju", "png"], "dpi": 300},
},
)
def test_submit_requires_artifact_uuid(self):
result = SoftwareJobSubmitTool(self.user_id, self.task_id).execute(
"origin.plot@v1",
input_id="data/input.csv",
plot={"type": "scatter", "x": "x", "y": ["y"]},
output={"formats": ["png"]},
)
self.assertIn("call register_artifact first", result)
def test_submit_schema_requires_artifact_input(self):
required = SoftwareJobSubmitTool.parameters["required"]
self.assertIn("input_id", required)
self.assertNotIn("input_path", SoftwareJobSubmitTool.parameters["properties"])
self.assertNotIn("request", SoftwareJobSubmitTool.parameters["properties"])
def test_status_and_cancel_reject_cross_task_job(self):
foreign = {"job_id": str(uuid4()), "task_id": str(uuid4())}

View File

@ -0,0 +1,70 @@
"""Register a task file as a stable artifact without publishing it to chat."""
from __future__ import annotations
import json
from pathlib import Path
from uuid import UUID
from core.artifact_lifecycle import register_workspace_artifact
from core.artifacts import ArtifactPathError, resolve_artifact_path
from .base import Tool
class RegisterArtifactTool(Tool):
name = "register_artifact"
description = (
"Register one existing file in the current task as a stable artifact and return "
"its artifact_id. Use this before a tool that requires an artifact input. "
"Registration does not publish the file as a final chat deliverable."
)
parameters = {
"type": "object",
"properties": {
"path": {
"type": "string",
"minLength": 1,
"maxLength": 1000,
"description": "File path relative to the current task working directory.",
}
},
"required": ["path"],
"additionalProperties": False,
}
def __init__(
self,
user_id: UUID,
task_id: UUID,
*,
working_dir: Path,
**kwargs,
) -> None:
super().__init__(**kwargs)
self.user_id = user_id
self.task_id = task_id
self.working_dir = Path(working_dir)
def execute(self, path: str) -> str:
if self.user_root is None:
return "[Error] artifact registration requires a user workspace"
if len(str(path or "")) > 1000:
return "[Error] artifact path is too long"
try:
_, relative_path = resolve_artifact_path(
path,
working_dir=self.working_dir,
user_root=self.user_root,
require_file=True,
allow_legacy_user_relative=False,
)
artifact = register_workspace_artifact(
user_id=self.user_id,
task_id=self.task_id,
user_root=self.user_root,
working_dir=self.working_dir,
path=relative_path,
)
except (ArtifactPathError, ValueError) as exc:
return f"[Error] cannot register artifact: {exc}"
return json.dumps(artifact, ensure_ascii=False)

View File

@ -46,34 +46,138 @@ class SoftwareCapabilityListTool(_SoftwareJobTool):
class SoftwareJobSubmitTool(_SoftwareJobTool):
name = "software_job_submit"
description = (
"Submit a persistent professional software job for the current task. "
"Submit an Origin plot job using a registered CSV, XLSX, or JSON artifact. "
"Call register_artifact first when the input is only a workspace file. "
"Return immediately with job_id; do not poll continuously or wait for completion."
)
_axis_schema = {
"type": "object",
"properties": {
"title": {"type": "string"},
"unit": {"type": "string"},
"scale": {"type": "string", "enum": ["linear"]},
},
"additionalProperties": False,
}
_plot_schema = {
"type": "object",
"properties": {
"type": {"type": "string", "enum": ["line", "scatter", "line_scatter"]},
"x": {
"type": "string",
"minLength": 1,
"maxLength": 128,
"description": "Exact X column name in the input file.",
},
"y": {
"type": "array",
"minItems": 1,
"maxItems": 16,
"uniqueItems": True,
"items": {"type": "string", "minLength": 1, "maxLength": 128},
"description": "One to sixteen unique Y column names.",
},
"template": {"type": "string", "enum": ["publication_double_column"]},
"title": {"type": "string", "maxLength": 500},
"x_axis": _axis_schema,
"y_axis": _axis_schema,
"legend": {
"type": "object",
"properties": {
"enabled": {"type": "boolean", "enum": [True]},
"position": {"type": "string", "enum": ["top_right"]},
},
"additionalProperties": False,
},
"error_bars": {"type": "null"},
},
"required": ["type", "x", "y"],
"additionalProperties": False,
}
_output_schema = {
"type": "object",
"properties": {
"formats": {
"type": "array",
"minItems": 1,
"uniqueItems": True,
"items": {"type": "string", "enum": ["opju", "png", "svg", "pdf"]},
},
"dpi": {"type": "integer", "minimum": 72, "maximum": 1200},
"capture_screenshots": {"type": "boolean"},
"record_video": {"type": "boolean", "enum": [False]},
},
"required": ["formats"],
"additionalProperties": False,
}
parameters = {
"type": "object",
"properties": {
"capability": {"type": "string", "enum": sorted(SUPPORTED_CAPABILITIES)},
"request": {"type": "object"},
"input_id": {
"type": "string",
"description": "Artifact UUID returned by register_artifact or another artifact-producing flow.",
},
"sheet": {
"type": "string",
"minLength": 1,
"maxLength": 128,
"description": "Optional XLSX worksheet name.",
},
"plot": _plot_schema,
"output": _output_schema,
"idempotency_key": {
"type": "string",
"description": "Stable unique key for this exact submission; omit to generate one.",
},
},
"required": ["capability", "request"],
"required": ["capability", "input_id", "plot", "output"],
"additionalProperties": False,
}
def execute(self, capability: str, request: dict, idempotency_key: str = "") -> str:
def execute(
self,
capability: str,
input_id: str = "",
plot: dict | None = None,
output: dict | None = None,
sheet: str = "",
request: dict | None = None,
idempotency_key: str = "",
) -> str:
try:
if isinstance(input_id, dict) and request is None:
request = input_id
input_id = ""
if request is not None:
if input_id or plot is not None or output is not None or sheet:
return "[Error] request cannot be combined with input_id, sheet, plot, or output"
normalized_request = request
else:
if not input_id or plot is None or output is None:
return "[Error] input_id, plot, and output are required for an Origin job"
try:
canonical_input_id = str(UUID(str(input_id)))
except ValueError:
return "[Error] input_id must be an artifact UUID; call register_artifact first"
input_spec = {"input_id": canonical_input_id}
if sheet:
input_spec["sheet"] = sheet
normalized_request = {
"schema_version": 1,
"input": input_spec,
"plot": plot,
"output": output,
}
job, created = create_job(
self.user_id,
self.task_id,
idempotency_key=idempotency_key.strip() or str(uuid4()),
capability=capability,
request=request,
request=normalized_request,
)
return json.dumps({**job, "created": created}, ensure_ascii=False)
except SoftwareJobError as exc:
except (SoftwareJobError, ValueError) as exc:
return f"[Error] {exc}"