From 25616d12cc174b0499f7da75306d849cee8d35d9 Mon Sep 17 00:00:00 2001 From: caoqianming Date: Fri, 14 Aug 2026 12:55:46 +0800 Subject: [PATCH] refactor(software): introduce capability contracts --- DESIGN.md | 4 + PROGRESS.md | 2 + RUN.md | 2 + core/software_contracts.py | 297 ++++++++++++++ core/software_jobs.py | 386 +++++------------- core/software_nodes.py | 4 +- requirements.txt | 1 + scripts/repair_software_job_outputs.py | 27 +- software-contracts/origin.plot.v2.json | 287 +++++++++++++ tests/test_software_contracts.py | 69 ++++ tests/test_software_nodes.py | 96 +++-- tests/test_windows_node_source.py | 23 +- tools/software_jobs.py | 155 ++----- web/routers/software_nodes.py | 79 ++-- web/schemas.py | 6 +- windows-node/README.md | 2 +- .../Zcbot.WindowsNode/ConfigurationForm.cs | 20 +- .../Zcbot.WindowsNode/EnrollmentClient.cs | 7 +- .../Zcbot.WindowsNode/JobInboxStore.cs | 34 +- .../Zcbot.WindowsNode/NodeAdapters.cs | 93 +++++ .../Zcbot.WindowsNode/NodeConfigStore.cs | 4 +- .../Zcbot.WindowsNode/NodeConnectionLoop.cs | 39 +- .../Zcbot.WindowsNode/OriginRuntimeProbe.cs | 12 +- .../Zcbot.WindowsNode.csproj | 5 + windows-node/package-windows-node.bat | 1 + 25 files changed, 1135 insertions(+), 520 deletions(-) create mode 100644 core/software_contracts.py create mode 100644 software-contracts/origin.plot.v2.json create mode 100644 tests/test_software_contracts.py create mode 100644 windows-node/Zcbot.WindowsNode/NodeAdapters.cs diff --git a/DESIGN.md b/DESIGN.md index 7247269..e77816b 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -462,6 +462,10 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB) 云端控制面使用独立的 `software_node_enrollments` 与 `software_nodes`，不复用用户外部系统连接。管理员创建的一次性注册码具有 128 bit 随机熵，数据库只保存 SHA-256 摘要；节点注册在行锁事务中校验有效期、预期名称和允许能力，成功后原子消费。每个节点获得独立高熵 Token，数据库只保存 bcrypt 强哈希，明文仅在注册响应出现一次。 +专业软件采用“共享能力契约 + Node adapter”边界。仓库根目录 `software-contracts/*.json` 是语言无关的声明式事实源，描述 capability、请求 JSON Schema、输入额度、输出 manifest、feature 与最低 adapter 版本；Core 启动时自动发现契约，只负责身份、账本、调度、传输、摘要与最终发布，不包含 Origin 或其他软件的操作分支。Windows Node 是可信宿主，负责持久化 job 目录、下载/上传、恢复、取消和 adapter 注册；adapter 才负责探测具体软件、二次语义校验并执行。adapter 可以是 Node 内置 .NET 实现，也可以由固定 runner 启动任意语言的受信进程；进程只接收 job 目录并通过 `state.json`、`terminal.json` 与固定输出目录交接，不把 Python、COM 或某个 SDK 写入通用协议。当前 Origin adapter 的执行体恰好是固定 Python Worker，这是实现选择而非平台契约。 + +扩展现有 capability 的 feature 时修改共享契约、对应 adapter/Worker 和测试，不修改 Core 调度与 Job 生命周期；增加新专业软件时新增契约，并在目标 Node 安装包的单一 adapter registry 注册实现。Cloud 会自动获得校验、工具 schema、输出发布和能力发现；Node 的注册能力、配置校验与界面展示也从 registry/契约派生。当前 Node 仍按整机单执行槽保守串行，未来只有真实并行软件需求出现时，才把 slot 账本升级为 per-capability 租约，而不改变 Job 协议。 + Node 通过 `Authorization: Bearer` 与 `X-Node-Id` 建立 `/v1/software-nodes/connect` WebSocket。进程内 Connection Manager 保证同一节点单活，新连接关闭旧连接；`hello`/`heartbeat` 更新版本、容量、软件健康与最后在线时间。管理员禁用节点时先持久化禁用态，再关闭现有连接；断线收尾不得覆盖禁用态。当前单活只覆盖单 Web 进程，生产启用多实例前必须增加 Redis/PG fencing 或将 Node API 固定路由到单一控制面实例。 第二阶段已增加 `software_jobs`（专业软件任务）账本与 `origin.plot@v2` 的 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 真正启动后才进入软件无关的 `software_running`，具体软件和操作由 capability/request 表达。 diff --git a/PROGRESS.md b/PROGRESS.md index 692e4a8..2733925 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -22,6 +22,8 @@ ### 2026-08-14 +- **08-14 / Unreleased / 专业软件契约与语言无关 adapter 边界**:将 capability 请求 schema、输入额度、输出 manifest、feature/adapter 版本要求和兼容运行时集中到 `software-contracts/*.json`；Core 自动发现契约并通用完成校验、工具 schema、调度、上传发布，不再包含 Origin 分支。Windows Node 以单一 adapter registry 派生注册能力、配置校验、界面和运行时上报，通用 Host 只处理 job 目录协议；adapter 可由 .NET 内置或固定进程以任意语言实现，Origin 的 Python Worker 仅是当前实现。调度同时按 feature/adapter 版本选节点，并跳过队首暂不可执行 Job。专项 82 项 unittest、Python 编译、Ruff 致命规则、diff 检查与 .NET build 通过；全量 605 项仅 3 个数据库集成模块因显式测试库未迁移、缺少 `users` 表而未通过，未连接或写入生产 DB。 + - **08-14 / Unreleased / Origin 统一图型契约扩展**:`origin.plot@v2` 保持单一入口，`series[]` 增加 `z/y_error` 数据角色并由 `plot.type` 判别校验，在兼容既有 XY 请求的同时新增 column、bar、grouped_column、y_error、contour、surface_3d、ternary、heatmap；规则热图在 Worker 内拒绝缺格、重复坐标、非等间距和非有限数值。云端、Agent 工具、Node 二次校验和固定 Worker 已同步，专项 61 项 unittest、Python 编译与 .NET build 通过，未连接或写入生产 DB。 - **08-14 / Unreleased / Windows Node 输出上传收敛与诊断**:恢复中的成功任务优先向云端重放完成确认，已发布结果不再重新打开可能被 Origin 占用的 OPJU；文件共享冲突按 0.5/1/2/5 秒有界退避，本地完成标记使用可跨心跳复用的固定 pending 文件，只重试重命名以避开系统程序对每个新文件的重复扫描，并按 Job 写入带阶段、产物、重试次数和 HRESULT 的 1 MiB 轮转诊断日志。同步修正本机更新时间早于接收时间的展示边界；相关 45 项专项 unittest、.NET build 与 diff 检查通过，未写入生产 DB。 diff --git a/RUN.md b/RUN.md index 53084a5..d725952 100644 --- a/RUN.md +++ b/RUN.md @@ -1102,6 +1102,8 @@ install-windows-node.bat Web 用户登录后，文件栏 Job 中心会聚合本人最近任务。活动任务约 4 秒刷新一次，空闲时降为约 30 秒；停止已派发任务是协作取消，状态先显示“正在停止”，Node 在线时立即接收，断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`register_artifact`、`software_job_submit`、`software_job_status` 和 `software_job_cancel`。Origin 输入必须是 artifact：已有 UUID 可直接提交，普通 task 文件先逐个用相对路径登记；登记不会发布聊天交付卡片。提交工具接收 `inputs`、`operation`、`outputs`，支持 1–16 个输入、跨输入系列和多个显式输出，只创建固定 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 或认证请求头。 +专业软件契约位于 `software-contracts/*.json`，Core 会在启动时自动发现。新增 feature 时更新对应契约与 Node adapter；新增软件时新增契约，并只在 Windows Node 的 `NodeAdapterRegistry.CreateDefault` 注册已安装实现。通用协议不要求 adapter 使用 Python：实现可为 .NET 内置代码或由固定 runner 启动的任意语言受信进程；Origin 安装器创建 Python 3.12 runtime 只服务当前 Origin adapter。 + 注册配置写入 `%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_contracts.py b/core/software_contracts.py new file mode 100644 index 0000000..3d45efc --- /dev/null +++ b/core/software_contracts.py @@ -0,0 +1,297 @@ +"""语言无关的专业软件 capability 契约加载与查询。""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from hashlib import sha256 +from pathlib import Path, PurePosixPath +from typing import Any + +from jsonschema import Draft202012Validator, FormatChecker + + +CONTRACT_ROOT = Path(__file__).resolve().parents[1] / "software-contracts" + + +class SoftwareContractError(ValueError): + pass + + +@dataclass(frozen=True) +class OutputSpec: + output_id: str + filename: str + media_type: str + relative_path: str + publish: bool + required: bool + + +@dataclass(frozen=True) +class CapabilityContract: + capability: str + display_name: str + default_enrollment: bool + output_namespace: str + request_schema: dict[str, Any] + input_policy: dict[str, Any] + outputs: dict[str, OutputSpec] + feature_path: tuple[str, ...] + features: dict[str, str] + summary: dict[str, Any] + legacy_runtime: dict[str, Any] | None + + def normalize_request(self, request: object) -> tuple[dict[str, Any], str]: + errors = sorted( + Draft202012Validator( + self.request_schema, format_checker=FormatChecker() + ).iter_errors(request), + key=lambda item: list(item.absolute_path), + ) + if errors: + first = errors[0] + location = ".".join(str(item) for item in first.absolute_path) + prefix = f"{location}: " if location else "" + raise SoftwareContractError(f"invalid {self.capability} request: {prefix}{first.message}") + encoded = json.dumps(request, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + max_bytes = int(self.request_schema.get("x-maxBytes") or 256 * 1024) + if len(encoded.encode("utf-8")) > max_bytes: + raise SoftwareContractError(f"{self.capability} request is too large") + normalized = json.loads(encoded) + return normalized, sha256(encoded.encode("utf-8")).hexdigest() + + def input_bindings(self, request: dict[str, Any]) -> list[dict[str, Any]]: + return list(request.get("inputs") or []) + + def expected_outputs(self, request: dict[str, Any]) -> dict[str, OutputSpec]: + requested = { + item.get("key") + for item in request.get("outputs") or [] + if isinstance(item, dict) and isinstance(item.get("key"), str) + } + expected = { + output_id: spec + for output_id, spec in self.outputs.items() + if spec.required or output_id in requested + } + if requested - self.outputs.keys(): + raise SoftwareContractError(f"{self.capability} request contains unknown outputs") + return expected + + def output_spec(self, output_id: str) -> OutputSpec: + try: + return self.outputs[output_id] + except KeyError as exc: + raise SoftwareContractError("unsupported output artifact identity") from exc + + def feature(self, request: dict[str, Any]) -> str: + value: Any = request + for part in self.feature_path: + if not isinstance(value, dict): + return "" + value = value.get(part) + return value if isinstance(value, str) else "" + + def required_adapter_version(self, request: dict[str, Any]) -> str: + return self.features.get(self.feature(request), "0.0.0") + + def summarize(self, request: dict[str, Any]) -> dict[str, Any]: + return { + "display_name": self.display_name, + "title": str(_value_at(request, self.summary.get("title_path") or []) or ""), + "formats": [ + item.get("format") + for item in request.get("outputs") or [] + if isinstance(item, dict) + ], + } + + def submission_schema(self) -> dict[str, Any]: + definitions = self.request_schema.get("$defs") or {} + properties = _resolve_local_refs( + dict(self.request_schema.get("properties") or {}), definitions + ) + properties.pop("schema_version", None) + required = [ + item for item in self.request_schema.get("required") or [] + if item != "schema_version" + ] + return { + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": False, + } + + +def _value_at(value: object, path: list[str]) -> object: + current = value + for part in path: + if not isinstance(current, dict): + return None + current = current.get(part) + return current + + +def _resolve_local_refs(value: Any, definitions: dict[str, Any]) -> Any: + if isinstance(value, list): + return [_resolve_local_refs(item, definitions) for item in value] + if not isinstance(value, dict): + return value + if set(value) == {"$ref"} and isinstance(value["$ref"], str): + prefix = "#/$defs/" + if not value["$ref"].startswith(prefix): + raise RuntimeError("software contract contains a non-local schema reference") + name = value["$ref"][len(prefix):] + if name not in definitions: + raise RuntimeError("software contract contains an unknown schema reference") + return _resolve_local_refs(definitions[name], definitions) + return { + key: _resolve_local_refs(item, definitions) + for key, item in value.items() + } + + +def _load_contract(path: Path) -> CapabilityContract: + raw = json.loads(path.read_text(encoding="utf-8")) + required = { + "capability", "display_name", "default_enrollment", "output_namespace", "request_schema", + "input_policy", "outputs", "feature_path", "features", "summary", "legacy_runtime", + } + if not isinstance(raw, dict) or set(raw) != required: + raise RuntimeError(f"invalid software contract fields: {path.name}") + capability = raw["capability"] + namespace = raw["output_namespace"] + if not isinstance(capability, str) or not re.fullmatch(r"[a-z][a-z0-9_.-]+@v[1-9][0-9]*", capability): + raise RuntimeError(f"invalid software capability: {path.name}") + if not isinstance(namespace, str) or not re.fullmatch(r"[a-z][a-z0-9_-]{0,31}", namespace): + raise RuntimeError(f"invalid output namespace: {path.name}") + outputs: dict[str, OutputSpec] = {} + for output_id, item in raw["outputs"].items(): + if not isinstance(item, dict) or set(item) != { + "filename", "media_type", "relative_path", "publish", "required" + }: + raise RuntimeError(f"invalid output spec: {path.name}:{output_id}") + filename = item["filename"] + relative_path = item["relative_path"] + parsed_path = PurePosixPath(relative_path) if isinstance(relative_path, str) else None + if ( + not isinstance(output_id, str) + or not re.fullmatch(r"[a-z][a-z0-9_]{0,63}", output_id) + or not isinstance(filename, str) + or PurePosixPath(filename).name != filename + or parsed_path is None + or parsed_path.is_absolute() + or ".." in parsed_path.parts + or parsed_path.name != filename + or parsed_path.as_posix() != relative_path + or not isinstance(item["media_type"], str) + or not item["media_type"] + or not isinstance(item["publish"], bool) + or not isinstance(item["required"], bool) + ): + raise RuntimeError(f"unsafe output spec: {path.name}:{output_id}") + outputs[output_id] = OutputSpec(output_id=output_id, **item) + if len({item.relative_path for item in outputs.values()}) != len(outputs): + raise RuntimeError(f"duplicate output paths: {path.name}") + Draft202012Validator.check_schema(raw["request_schema"]) + return CapabilityContract( + capability=capability, + display_name=raw["display_name"], + default_enrollment=raw["default_enrollment"], + output_namespace=namespace, + request_schema=raw["request_schema"], + input_policy=raw["input_policy"], + outputs=outputs, + feature_path=tuple(raw["feature_path"]), + features=raw["features"], + summary=raw["summary"], + legacy_runtime=raw["legacy_runtime"], + ) + + +_loaded_contracts = [_load_contract(path) for path in sorted(CONTRACT_ROOT.glob("*.json"))] +CONTRACTS = {contract.capability: contract for contract in _loaded_contracts} +if not CONTRACTS or len(CONTRACTS) != len(_loaded_contracts): + raise RuntimeError("software contracts are missing or contain duplicate capabilities") +SUPPORTED_CAPABILITIES = frozenset(CONTRACTS) +DEFAULT_CAPABILITIES = tuple( + item.capability for item in CONTRACTS.values() if item.default_enrollment +) + + +def get_contract(capability: str) -> CapabilityContract: + try: + return CONTRACTS[capability] + except KeyError as exc: + raise SoftwareContractError("unsupported capability") from exc + + +def version_at_least(actual: str, required: str) -> bool: + def parts(value: str) -> tuple[int, ...]: + match = re.match(r"^(\d+)(?:\.(\d+))?(?:\.(\d+))?", value.strip()) + return tuple(int(item or 0) for item in match.groups(default="0")) if match else (0, 0, 0) + + return parts(actual) >= parts(required) + + +def node_supports_request( + contract: CapabilityContract, + request: dict[str, Any], + runtime: dict[str, Any], +) -> bool: + capability_runtime = (runtime.get("capability_runtime") or {}).get(contract.capability) + if isinstance(capability_runtime, dict): + if capability_runtime.get("health") != "ready": + return False + if int(capability_runtime.get("available_slots") or 0) <= 0: + return False + actual_version = str(capability_runtime.get("adapter_version") or "0.0.0") + advertised_features = capability_runtime.get("features") + feature = contract.feature(request) + if isinstance(advertised_features, list) and feature not in advertised_features: + return False + return version_at_least(actual_version, contract.required_adapter_version(request)) + # 兼容尚未升级 capability_runtime 的 Node；兼容路径由版本化 contract 声明。 + legacy_config = contract.legacy_runtime or {} + if not legacy_config: + return False + legacy = _value_at(runtime, legacy_config.get("detail_path") or []) + assumed_version = str(legacy_config.get("assumed_adapter_version") or "0.0.0") + slots = int(_value_at(runtime, legacy_config.get("slots_path") or []) or 0) + if legacy is None: + return ( + slots > 0 + and version_at_least(assumed_version, contract.required_adapter_version(request)) + ) + return ( + isinstance(legacy, dict) + and legacy.get("health") == "ready" + and slots > 0 + and version_at_least( + str(legacy.get("adapter_version") or "0.0.0"), + contract.required_adapter_version(request), + ) + ) + + +def node_available_slots(capability: str, runtime: dict[str, Any]) -> int: + contract = get_contract(capability) + item = (runtime.get("capability_runtime") or {}).get(capability) + if isinstance(item, dict) and item.get("health") == "ready": + return max(0, int(item.get("available_slots") or 0)) + legacy_config = contract.legacy_runtime or {} + if not legacy_config: + return 0 + legacy = _value_at(runtime, legacy_config.get("detail_path") or []) + slots = max(0, int(_value_at(runtime, legacy_config.get("slots_path") or []) or 0)) + if legacy is None: + return slots + if ( + isinstance(legacy, dict) + and legacy.get("health") == "ready" + ): + return slots + return 0 diff --git a/core/software_jobs.py b/core/software_jobs.py index 451b0fc..524b497 100644 --- a/core/software_jobs.py +++ b/core/software_jobs.py @@ -5,244 +5,45 @@ from __future__ import annotations import json import re from datetime import datetime, timedelta, timezone -from hashlib import sha256 from uuid import UUID, uuid4 from sqlalchemy import and_, desc, or_, select from sqlalchemy.exc import IntegrityError -from core.software_nodes import SUPPORTED_CAPABILITIES +from core.software_contracts import ( + SoftwareContractError, + get_contract, + node_supports_request, +) from core.storage.engine import session_scope from core.storage.models import Artifact, SoftwareJob, SoftwareNode, Task OFFER_SECONDS = 60 -ALLOWED_PLOT_TYPES = frozenset( - { - "line", "scatter", "line_scatter", "column", "bar", "grouped_column", - "y_error", "contour", "surface_3d", "ternary", "heatmap", - } -) -XYZ_PLOT_TYPES = frozenset({"contour", "surface_3d", "ternary", "heatmap"}) -ALLOWED_INPUT_SUFFIXES = frozenset({".csv", ".xlsx", ".json"}) -MAX_INPUT_BYTES = 100 * 1024 * 1024 -MAX_INPUTS = 16 -MAX_INPUT_TOTAL_BYTES = 512 * 1024 * 1024 MAX_OUTPUT_ARTIFACT_BYTES = 256 * 1024 * 1024 +MAX_OUTPUT_TOTAL_BYTES = 512 * 1024 * 1024 + + class SoftwareJobError(Exception): pass -MAX_OUTPUT_TOTAL_BYTES = 512 * 1024 * 1024 -OUTPUT_ARTIFACTS = { - "project": ("project.opju", "application/x-origin-project", "opju"), - "figure_png": ("figure.png", "image/png", "png"), - "figure_svg": ("figure.svg", "image/svg+xml", "svg"), - "figure_pdf": ("figure.pdf", "application/pdf", "pdf"), - "plot_spec": ("plot-spec.json", "application/json", None), - "provenance": ("provenance.json", "application/json", None), -} -SOFTWARE_JOB_METADATA_IDS = frozenset({"plot_spec", "provenance"}) -ORIGIN_OUTPUT_IDENTITIES = { - ("project", "opju"): "project", - ("figure", "png"): "figure_png", - ("figure", "svg"): "figure_svg", - ("figure", "pdf"): "figure_pdf", -} - - -def software_job_output_path(output_id: str) -> str: - """返回 Job 输出目录内路径；技术元数据固定进入隐藏 `.meta/`。""" - item = OUTPUT_ARTIFACTS.get(output_id) - if item is None: - raise SoftwareJobError("unsupported output artifact identity") - filename = item[0] - return f".meta/{filename}" if output_id in SOFTWARE_JOB_METADATA_IDS else filename - - -def _has_only(value: dict, fields: set[str]) -> bool: - return set(value).issubset(fields) - - -def _canonical_origin_plot_request(request: dict) -> tuple[dict, str]: - if not isinstance(request, dict) or set(request) != { - "schema_version", "inputs", "operation", "outputs" - }: - raise SoftwareJobError("invalid origin plot request fields") - if request.get("schema_version") != 2: - raise SoftwareJobError("unsupported origin plot schema version") - inputs = request.get("inputs") - operation = request.get("operation") - outputs = request.get("outputs") - if not isinstance(inputs, list) or not 1 <= len(inputs) <= MAX_INPUTS: - raise SoftwareJobError("inputs must contain 1 to 16 artifact bindings") - if not isinstance(operation, dict) or set(operation) != {"plot"}: - raise SoftwareJobError("origin operation must contain exactly plot") - plot = operation.get("plot") - if not isinstance(plot, dict): - raise SoftwareJobError("origin plot request sections must be objects") - input_keys: list[str] = [] - for input_spec in inputs: - if not isinstance(input_spec, dict) or set(input_spec) not in ( - {"key", "artifact_id"}, {"key", "artifact_id", "selector"} - ): - raise SoftwareJobError("invalid origin input binding fields") - input_key = input_spec.get("key") - if not isinstance(input_key, str) or not re.fullmatch(r"[a-z][a-z0-9_]{0,31}", input_key): - raise SoftwareJobError("input key must match [a-z][a-z0-9_]{0,31}") - try: - UUID(str(input_spec.get("artifact_id") or "")) - except ValueError as exc: - raise SoftwareJobError("inputs[].artifact_id must be an artifact UUID") from exc - selector = input_spec.get("selector") - if selector is not None and ( - not isinstance(selector, dict) - or set(selector) != {"sheet"} - or not isinstance(selector.get("sheet"), str) - or not 1 <= len(selector["sheet"]) <= 128 - ): - raise SoftwareJobError("origin input selector must contain a valid sheet") - input_keys.append(input_key) - if len(input_keys) != len(set(input_keys)): - raise SoftwareJobError("input keys must be unique") - if not _has_only( - plot, - { - "type", "series", "template", "title", "x_axis", "y_axis", "z_axis", - "legend", "error_bars", - }, - ): - raise SoftwareJobError("unsupported origin plot fields") - if plot.get("type") not in ALLOWED_PLOT_TYPES: - raise SoftwareJobError("unsupported origin plot type") - if "title" in plot and ( - not isinstance(plot["title"], str) or len(plot["title"]) > 500 - ): - raise SoftwareJobError("plot.title must be a string") - if plot.get("template", "publication_double_column") != "publication_double_column": - raise SoftwareJobError("unsupported origin plot template") - series = plot.get("series") - if not isinstance(series, list) or not 1 <= len(series) <= 16: - raise SoftwareJobError("plot.series must contain 1 to 16 series") - plot_type = plot["type"] - if plot_type == "grouped_column" and len(series) < 2: - raise SoftwareJobError("grouped_column requires at least two series") - if plot_type in XYZ_PLOT_TYPES and len(series) != 1: - raise SoftwareJobError(f"{plot_type} requires exactly one XYZ series") - required_roles = ( - {"x", "y", "z"} if plot_type in XYZ_PLOT_TYPES - else {"x", "y", "y_error"} if plot_type == "y_error" - else {"x", "y"} - ) - identities: list[tuple[str, ...]] = [] - used_input_keys: set[str] = set() - series_labels: dict[tuple[str, str], str] = {} - for item in series: - if not isinstance(item, dict) or not _has_only( - item, {"input", "x", "y", "z", "y_error", "label"} - ): - raise SoftwareJobError("invalid plot series fields") - if not {"input", *required_roles}.issubset(item): - roles = ", ".join(sorted(required_roles)) - raise SoftwareJobError(f"{plot_type} series requires input, {roles}") - if {"x", "y", "z", "y_error"}.intersection(item) - required_roles: - raise SoftwareJobError(f"{plot_type} series contains unsupported data roles") - input_key = item.get("input") - if input_key not in input_keys: - raise SoftwareJobError("plot series references an unknown input") - if any( - not isinstance(value, str) or not 1 <= len(value) <= 128 - for value in (item.get(role) for role in required_roles) - ): - raise SoftwareJobError("plot series data roles must be column names") - if "label" in item and ( - not isinstance(item["label"], str) or not 1 <= len(item["label"]) <= 200 - ): - raise SoftwareJobError("plot series label must be a string") - y_column = item["y"] - label_key = (input_key, y_column) - effective_label = item.get("label", y_column) - if label_key in series_labels and series_labels[label_key] != effective_label: - raise SoftwareJobError("series sharing an input Y column must use one label") - series_labels[label_key] = effective_label - used_input_keys.add(input_key) - identities.append((input_key, *(item[role] for role in sorted(required_roles)))) - if len(identities) != len(set(identities)): - raise SoftwareJobError("plot series must be unique") - if used_input_keys != set(input_keys): - raise SoftwareJobError("every input must be referenced by a plot series") - for axis_name in ("x_axis", "y_axis", "z_axis"): - axis = plot.get(axis_name) - if axis is not None and ( - not isinstance(axis, dict) - or not _has_only(axis, {"title", "unit", "scale"}) - or axis.get("scale", "linear") != "linear" - or any( - name in axis and not isinstance(axis[name], str) - for name in ("title", "unit") - ) - ): - raise SoftwareJobError(f"invalid {axis_name}") - legend = plot.get("legend") - if legend is not None and ( - not isinstance(legend, dict) - or not _has_only(legend, {"enabled", "position"}) - or ("enabled" in legend and not isinstance(legend["enabled"], bool)) - or legend.get("enabled", True) is not True - or legend.get("position", "top_right") != "top_right" - ): - raise SoftwareJobError("invalid plot.legend") - if plot.get("error_bars") is not None: - raise SoftwareJobError("error bars are not supported") - if not isinstance(outputs, list) or not 1 <= len(outputs) <= 16: - raise SoftwareJobError("outputs must contain 1 to 16 declarations") - output_keys: list[str] = [] - output_identities: list[tuple[str, str]] = [] - for output in outputs: - if not isinstance(output, dict) or set(output) not in ( - {"key", "type", "format"}, {"key", "type", "format", "options"} - ): - raise SoftwareJobError("invalid origin output declaration fields") - output_type = output.get("type") - output_format = output.get("format") - expected_key = ORIGIN_OUTPUT_IDENTITIES.get((output_type, output_format)) - if output.get("key") != expected_key: - raise SoftwareJobError("origin output key, type, and format do not match") - options = output.get("options") - if output_format == "png": - if options is not None and ( - not isinstance(options, dict) - or set(options) != {"dpi"} - or not isinstance(options.get("dpi"), int) - or isinstance(options.get("dpi"), bool) - or not 72 <= options["dpi"] <= 1200 - ): - raise SoftwareJobError("PNG output options must contain a valid dpi") - elif options is not None: - raise SoftwareJobError("output options are only supported for PNG") - output_keys.append(expected_key) - output_identities.append((output_type, output_format)) - if len(output_keys) != len(set(output_keys)) or len(output_identities) != len(set(output_identities)): - raise SoftwareJobError("outputs must be unique") - encoded = json.dumps(request, ensure_ascii=False, sort_keys=True, separators=(",", ":")) - if len(encoded.encode("utf-8")) > 256 * 1024: - raise SoftwareJobError("origin plot request is too large") - normalized = json.loads(encoded) - return normalized, sha256(encoded.encode("utf-8")).hexdigest() - - -REQUEST_VALIDATORS = { - "origin.plot@v2": _canonical_origin_plot_request, -} +def software_job_output_path(capability: str, output_id: str) -> str: + """返回 capability Job 输出目录内的契约路径。""" + try: + return get_contract(capability).output_spec(output_id).relative_path + except SoftwareContractError as exc: + raise SoftwareJobError(str(exc)) from exc def _canonical_request(capability: str, request: dict) -> tuple[dict, str]: - validator = REQUEST_VALIDATORS.get(capability) - if validator is None: - raise SoftwareJobError("unsupported capability") - return validator(request) + try: + return get_contract(capability).normalize_request(request) + except SoftwareContractError as exc: + raise SoftwareJobError(str(exc)) from exc def _job_dict(row: SoftwareJob) -> dict: + contract = get_contract(row.capability) return { "job_id": str(row.job_id), "task_id": str(row.task_id), @@ -255,7 +56,7 @@ def _job_dict(row: SoftwareJob) -> dict: "metrics": row.metrics, "error": row.error, "artifact_manifest": row.artifact_manifest, - "output_dir": f"origin/{row.job_id}", + "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, "terminal_at": row.terminal_at.isoformat() if row.terminal_at else None, @@ -327,17 +128,7 @@ def list_jobs( def _request_summary(job: SoftwareJob) -> dict: - plot = (job.request.get("operation") or {}).get("plot") or {} - outputs = job.request.get("outputs") or [] - return { - "display_name": ( - "Origin 科研绘图" - if job.capability == "origin.plot@v2" - else job.capability - ), - "title": str(plot.get("title") or ""), - "formats": [item.get("format") for item in outputs if isinstance(item, dict)], - } + return get_contract(job.capability).summarize(job.request) def request_job_cancel(user_id: UUID, job_id: UUID) -> tuple[dict, dict | None]: @@ -401,16 +192,19 @@ def create_job( key = idempotency_key.strip() if not key or len(key) > 200: raise SoftwareJobError("idempotency_key must contain 1 to 200 characters") - if capability not in SUPPORTED_CAPABILITIES: - raise SoftwareJobError("unsupported capability") - normalized, digest = _canonical_request(capability, request) + try: + contract = get_contract(capability) + normalized, digest = contract.normalize_request(request) + except SoftwareContractError as exc: + raise SoftwareJobError(str(exc)) from exc with session_scope() as session: task = session.execute( select(Task.task_id).where(Task.task_id == task_id, Task.user_id == user_id) ).first() if task is None: raise SoftwareJobError("task not found") - artifact_ids = [UUID(item["artifact_id"]) for item in normalized["inputs"]] + bindings = contract.input_bindings(normalized) + artifact_ids = [UUID(item["artifact_id"]) for item in bindings] artifacts = session.execute( select(Artifact).where( Artifact.artifact_id.in_(set(artifact_ids)), @@ -421,7 +215,10 @@ def create_job( artifacts_by_id = {artifact.artifact_id: artifact for artifact in artifacts} input_manifest: list[dict] = [] total_input_bytes = 0 - for binding, artifact_id in zip(normalized["inputs"], artifact_ids, strict=True): + policy = contract.input_policy + allowed_suffixes = frozenset(policy.get("suffixes") or []) + max_input_bytes = int(policy.get("max_bytes") or 0) + for binding, artifact_id in zip(bindings, artifact_ids, strict=True): artifact = artifacts_by_id.get(artifact_id) if artifact is None: raise SoftwareJobError("input artifact not found") @@ -429,12 +226,12 @@ def create_job( "." + artifact.current_path.rsplit(".", 1)[-1].lower() if "." in artifact.current_path else "" ) - if suffix not in ALLOWED_INPUT_SUFFIXES: + if suffix not in allowed_suffixes: raise SoftwareJobError("input artifact type is not supported") if ( artifact.size_bytes is None or artifact.size_bytes < 0 - or artifact.size_bytes > MAX_INPUT_BYTES + or artifact.size_bytes > max_input_bytes or not artifact.content_sha256 or len(artifact.content_sha256) != 64 ): @@ -450,7 +247,7 @@ def create_job( if binding.get("selector") is not None: item["selector"] = binding["selector"] input_manifest.append(item) - if total_input_bytes > MAX_INPUT_TOTAL_BYTES: + if total_input_bytes > int(policy.get("max_total_bytes") or 0): raise SoftwareJobError("job inputs exceed the total size limit") existing = session.execute( select(SoftwareJob).where( @@ -513,7 +310,7 @@ def get_job(user_id: UUID, job_id: UUID) -> dict | None: def offer_next_job(node_ids: set[UUID]) -> dict | None: - """从当前进程实际在线的节点中选择一个，为最早 queued job 创建短租约。""" + """选择最早可执行的 Job–Node 组合，避免跨 capability 队首阻塞。""" if not node_ids: return None now = datetime.now(timezone.utc) @@ -531,39 +328,69 @@ def offer_next_job(node_ids: set[UUID]) -> dict | None: item.node_id = None item.lease_id = None item.lease_expires_at = None - job = session.execute( - select(SoftwareJob) - .where(SoftwareJob.status == "queued") - .order_by(SoftwareJob.created_at, SoftwareJob.job_id) - .with_for_update(skip_locked=True) - .limit(1) - ).scalar_one_or_none() - if job is None: - return None busy_node_ids = set( session.execute( select(SoftwareJob.node_id).where( SoftwareJob.node_id.is_not(None), - SoftwareJob.status.in_({"offered", "dispatched", "running"}), + SoftwareJob.status.in_( + {"offered", "dispatched", "running", "disconnected", "cancelling"} + ), ) ).scalars() ) - nodes = session.execute( + nodes = list(session.execute( select(SoftwareNode) .where(SoftwareNode.node_id.in_(node_ids), SoftwareNode.status == "online") .order_by(SoftwareNode.last_seen_at.desc()) + .with_for_update(skip_locked=True) + ).scalars()) + available_nodes = [item for item in nodes if item.node_id not in busy_node_ids] + available_capabilities = { + capability + for node in available_nodes + for capability in (node.capabilities or []) + } + if not available_capabilities: + return None + queued = session.execute( + select(SoftwareJob) + .where( + SoftwareJob.status == "queued", + SoftwareJob.capability.in_(available_capabilities), + ) + .order_by(SoftwareJob.created_at, SoftwareJob.job_id) ).scalars() - node = next( - ( - item - for item in nodes - if item.node_id not in busy_node_ids - and job.capability in item.capabilities - and int((item.runtime or {}).get("available_slots") or 0) > 0 - ), - None, - ) - if node is None: + selected: tuple[SoftwareJob, SoftwareNode] | None = None + for candidate in queued: + try: + contract = get_contract(candidate.capability) + except SoftwareContractError: + continue + node = next( + ( + item for item in available_nodes + if candidate.capability in (item.capabilities or []) + and node_supports_request( + contract, candidate.request, item.runtime or {} + ) + ), + None, + ) + if node is not None: + selected = candidate, node + break + if selected is None: + return None + candidate, node = selected + job = session.execute( + select(SoftwareJob) + .where( + SoftwareJob.job_id == candidate.job_id, + SoftwareJob.status == "queued", + ) + .with_for_update(skip_locked=True) + ).scalar_one_or_none() + if job is None: return None lease_id = uuid4() expires_at = now + timedelta(seconds=OFFER_SECONDS) @@ -622,6 +449,7 @@ def get_job_input(node_id: UUID, job_id: UUID, input_key: str) -> dict | None: if artifact is None: return None return { + "capability": job.capability, "user_id": job.user_id, "current_path": artifact.current_path, **manifest, @@ -647,6 +475,7 @@ def get_job_output_context(node_id: UUID, job_id: UUID, lease_id: UUID, digest: ): return None return { + "capability": job.capability, "user_id": job.user_id, "task_id": job.task_id, "working_dir": working_dir, @@ -656,15 +485,15 @@ def get_job_output_context(node_id: UUID, job_id: UUID, lease_id: UUID, digest: } -def validate_output_manifest(request: dict, manifest: object) -> list[dict]: +def validate_output_manifest(capability: str, request: dict, manifest: object) -> list[dict]: if not isinstance(manifest, list): raise SoftwareJobError("job artifact manifest must be a list") - expected_ids = {"plot_spec", "provenance"} - expected_ids.update( - item.get("key") - for item in request.get("outputs", []) - if isinstance(item, dict) and isinstance(item.get("key"), str) - ) + try: + contract = get_contract(capability) + expected = contract.expected_outputs(request) + except SoftwareContractError as exc: + raise SoftwareJobError(str(exc)) from exc + expected_ids = set(expected) if len(manifest) != len(expected_ids): raise SoftwareJobError("job artifact manifest is incomplete") normalized: list[dict] = [] @@ -678,10 +507,10 @@ def validate_output_manifest(request: dict, manifest: object) -> list[dict]: local_id = raw.get("artifact_id") if local_id not in expected_ids or local_id in seen: raise SoftwareJobError("job artifact manifest identity is invalid") - filename, media_type, _ = OUTPUT_ARTIFACTS[local_id] + spec = expected[local_id] size = raw.get("size_bytes") digest = raw.get("sha256") - if raw.get("filename") != filename or raw.get("media_type") != media_type: + if raw.get("filename") != spec.filename or raw.get("media_type") != spec.media_type: raise SoftwareJobError("job artifact manifest metadata does not match its identity") if not isinstance(size, int) or isinstance(size, bool) or not 1 <= size <= MAX_OUTPUT_ARTIFACT_BYTES: raise SoftwareJobError("job output artifact size is invalid") @@ -865,6 +694,7 @@ def record_job_terminal(node_id: UUID, payload: dict) -> None: _assert_job_message(job, node_id, lease_id, digest) if terminal_status == "succeeded": expected = validate_output_manifest( + job.capability, job.request, [ { @@ -879,7 +709,7 @@ def record_job_terminal(node_id: UUID, payload: dict) -> None: ], ) if len(expected) != len(manifest) or any( - not _published_output_is_valid(job.job_id, item) + not _published_output_is_valid(job.capability, job.job_id, item) for item in manifest ): raise SoftwareJobError("successful job artifacts have not been published") @@ -905,19 +735,25 @@ def _is_uuid(value: str) -> bool: return True -def _published_output_is_valid(job_id: UUID, item: object) -> bool: +def _published_output_is_valid(capability: str, job_id: UUID, item: object) -> bool: if not isinstance(item, dict): return False output_id = item.get("source_artifact_id") if not isinstance(output_id, str): return False - if output_id not in OUTPUT_ARTIFACTS: + try: + contract = get_contract(capability) + spec = contract.output_spec(output_id) + except SoftwareContractError: return False - expected_path = f"origin/{job_id}/{software_job_output_path(output_id)}" + expected_path = ( + f"{contract.output_namespace}/{job_id}/" + f"{software_job_output_path(capability, output_id)}" + ) if item.get("path") != expected_path: return False artifact_id = item.get("artifact_id") - if output_id in SOFTWARE_JOB_METADATA_IDS: + if not spec.publish: return artifact_id is None return isinstance(artifact_id, str) and _is_uuid(artifact_id) diff --git a/core/software_nodes.py b/core/software_nodes.py index 595c0cb..f24cccf 100644 --- a/core/software_nodes.py +++ b/core/software_nodes.py @@ -10,10 +10,10 @@ from uuid import UUID, uuid4 import bcrypt from sqlalchemy import select +from core.software_contracts import DEFAULT_CAPABILITIES, SUPPORTED_CAPABILITIES from core.storage.engine import session_scope from core.storage.models import SoftwareNode, SoftwareNodeEnrollment -SUPPORTED_CAPABILITIES = frozenset({"origin.plot@v2"}) MAX_ENROLLMENT_FAILURES = 5 @@ -44,7 +44,7 @@ def create_enrollment( capabilities: list[str] | None = None, ttl_seconds: int = 600, ) -> dict: - allowed = list(dict.fromkeys(capabilities or ["origin.plot@v2"])) + allowed = list(dict.fromkeys(capabilities or DEFAULT_CAPABILITIES)) if not allowed or any(item not in SUPPORTED_CAPABILITIES for item in allowed): raise SoftwareNodeError("unsupported capability") if not 60 <= ttl_seconds <= 3600: diff --git a/requirements.txt b/requirements.txt index c1114a9..53cb950 100644 --- a/requirements.txt +++ b/requirements.txt @@ -55,6 +55,7 @@ fakeredis>=2.23 # §7 B 阶段: Storage 落 PG sqlalchemy>=2.0.0 +jsonschema>=4.23.0 # 专业软件 capability contract 的 Draft 2020-12 校验 psycopg[binary]>=3.1.0 alembic>=1.13.0 diff --git a/scripts/repair_software_job_outputs.py b/scripts/repair_software_job_outputs.py index 66cef82..532c4f8 100644 --- a/scripts/repair_software_job_outputs.py +++ b/scripts/repair_software_job_outputs.py @@ -21,10 +21,8 @@ from sqlalchemy.orm.attributes import flag_modified ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) -from core.software_jobs import ( - SOFTWARE_JOB_METADATA_IDS, - software_job_output_path, -) +from core.software_contracts import DEFAULT_CAPABILITIES, get_contract +from core.software_jobs import software_job_output_path from core.storage.models import Artifact, SoftwareJob, Task from core.paths import from_db_path @@ -55,12 +53,16 @@ def _within(root: Path, target: Path) -> Path: def build_plan(job: SoftwareJob, task: Task) -> RepairPlan | None: + capability = getattr(job, "capability", DEFAULT_CAPABILITIES[0]) + contract = get_contract(capability) working_dir = from_db_path(task.working_dir).resolve() user_root = _user_root(working_dir, job.user_id) - output_dir = _within(user_root, working_dir / "origin" / str(job.job_id)) + output_dir = _within( + user_root, working_dir / contract.output_namespace / str(job.job_id) + ) legacy_dir = _within( user_root, - user_root / Path(task.working_dir) / "origin" / str(job.job_id), + user_root / Path(task.working_dir) / contract.output_namespace / str(job.job_id), ) legacy_exists = legacy_dir.is_dir() output_exists = output_dir.is_dir() @@ -73,10 +75,10 @@ def build_plan(job: SoftwareJob, task: Task) -> RepairPlan | None: metadata_moves: list[tuple[Path, Path]] = [] for item in job.artifact_manifest or []: output_id = str((item or {}).get("source_artifact_id") or "") - if output_id not in SOFTWARE_JOB_METADATA_IDS: + if not output_id or contract.output_spec(output_id).publish: continue source = source_dir / str((item or {}).get("filename") or "") - destination = source_dir / software_job_output_path(output_id) + destination = source_dir / software_job_output_path(capability, output_id) if source == destination or not source.exists(): continue if destination.exists(): @@ -104,6 +106,8 @@ def apply_files(plan: RepairPlan) -> None: def update_rows(session: Session, job: SoftwareJob, task: Task, plan: RepairPlan) -> int: + capability = getattr(job, "capability", DEFAULT_CAPABILITIES[0]) + contract = get_contract(capability) artifact_count = 0 manifest = [dict(item) for item in (job.artifact_manifest or [])] task_prefix = plan.working_dir.relative_to(plan.user_root).as_posix() @@ -111,7 +115,10 @@ def update_rows(session: Session, job: SoftwareJob, task: Task, plan: RepairPlan output_id = str(item.get("source_artifact_id") or "") if not output_id: continue - path = f"origin/{job.job_id}/{software_job_output_path(output_id)}" + path = ( + f"{contract.output_namespace}/{job.job_id}/" + f"{software_job_output_path(capability, output_id)}" + ) item["path"] = path raw_artifact_id = item.get("artifact_id") if not raw_artifact_id: @@ -124,7 +131,7 @@ def update_rows(session: Session, job: SoftwareJob, task: Task, plan: RepairPlan if artifact is None or artifact.user_id != job.user_id: raise RuntimeError(f"job {job.job_id}: artifact {artifact_id} is missing") artifact.current_path = f"{task_prefix}/{path}" - if output_id not in SOFTWARE_JOB_METADATA_IDS: + if contract.output_spec(output_id).publish: artifact.software_job_id = job.job_id artifact_count += 1 job.artifact_manifest = manifest diff --git a/software-contracts/origin.plot.v2.json b/software-contracts/origin.plot.v2.json new file mode 100644 index 0000000..e0727ac --- /dev/null +++ b/software-contracts/origin.plot.v2.json @@ -0,0 +1,287 @@ +{ + "capability": "origin.plot@v2", + "display_name": "Origin 科研绘图", + "default_enrollment": true, + "output_namespace": "origin", + "input_policy": { + "suffixes": [".csv", ".xlsx", ".json"], + "max_count": 16, + "max_bytes": 104857600, + "max_total_bytes": 536870912 + }, + "outputs": { + "project": { + "filename": "project.opju", + "media_type": "application/x-origin-project", + "relative_path": "project.opju", + "publish": true, + "required": false + }, + "figure_png": { + "filename": "figure.png", + "media_type": "image/png", + "relative_path": "figure.png", + "publish": true, + "required": false + }, + "figure_svg": { + "filename": "figure.svg", + "media_type": "image/svg+xml", + "relative_path": "figure.svg", + "publish": true, + "required": false + }, + "figure_pdf": { + "filename": "figure.pdf", + "media_type": "application/pdf", + "relative_path": "figure.pdf", + "publish": true, + "required": false + }, + "plot_spec": { + "filename": "plot-spec.json", + "media_type": "application/json", + "relative_path": ".meta/plot-spec.json", + "publish": false, + "required": true + }, + "provenance": { + "filename": "provenance.json", + "media_type": "application/json", + "relative_path": ".meta/provenance.json", + "publish": false, + "required": true + } + }, + "feature_path": ["operation", "plot", "type"], + "features": { + "line": "0.3.0", + "scatter": "0.3.0", + "line_scatter": "0.3.0", + "column": "0.4.0", + "bar": "0.4.0", + "grouped_column": "0.4.0", + "y_error": "0.4.0", + "contour": "0.4.0", + "surface_3d": "0.4.0", + "ternary": "0.4.0", + "heatmap": "0.4.0" + }, + "summary": { + "title_path": ["operation", "plot", "title"] + }, + "legacy_runtime": { + "detail_path": ["origin"], + "slots_path": ["available_slots"], + "assumed_adapter_version": "0.3.0" + }, + "request_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "x-maxBytes": 262144, + "required": ["schema_version", "inputs", "operation", "outputs"], + "additionalProperties": false, + "properties": { + "schema_version": {"const": 2}, + "inputs": { + "type": "array", + "minItems": 1, + "maxItems": 16, + "items": { + "type": "object", + "required": ["key", "artifact_id"], + "additionalProperties": false, + "properties": { + "key": {"type": "string", "pattern": "^[a-z][a-z0-9_]{0,31}$"}, + "artifact_id": {"type": "string", "format": "uuid"}, + "selector": { + "type": "object", + "required": ["sheet"], + "additionalProperties": false, + "properties": { + "sheet": {"type": "string", "minLength": 1, "maxLength": 128} + } + } + } + } + }, + "operation": { + "type": "object", + "required": ["plot"], + "additionalProperties": false, + "properties": { + "plot": { + "type": "object", + "required": ["type", "series"], + "additionalProperties": false, + "properties": { + "type": { + "enum": [ + "line", "scatter", "line_scatter", "column", "bar", + "grouped_column", "y_error", "contour", "surface_3d", + "ternary", "heatmap" + ] + }, + "series": { + "type": "array", + "minItems": 1, + "maxItems": 16, + "uniqueItems": true, + "items": { + "type": "object", + "required": ["input"], + "additionalProperties": false, + "properties": { + "input": {"type": "string", "pattern": "^[a-z][a-z0-9_]{0,31}$"}, + "x": {"type": "string", "minLength": 1, "maxLength": 128}, + "y": {"type": "string", "minLength": 1, "maxLength": 128}, + "z": {"type": "string", "minLength": 1, "maxLength": 128}, + "y_error": {"type": "string", "minLength": 1, "maxLength": 128}, + "label": {"type": "string", "minLength": 1, "maxLength": 200} + } + } + }, + "template": {"const": "publication_double_column"}, + "title": {"type": "string", "maxLength": 500}, + "x_axis": {"$ref": "#/$defs/axis"}, + "y_axis": {"$ref": "#/$defs/axis"}, + "z_axis": {"$ref": "#/$defs/axis"}, + "legend": { + "type": "object", + "additionalProperties": false, + "properties": { + "enabled": {"const": true}, + "position": {"const": "top_right"} + } + }, + "error_bars": {"type": "null"} + }, + "allOf": [ + { + "if": {"properties": {"type": {"enum": ["contour", "surface_3d", "ternary", "heatmap"]}}}, + "then": { + "properties": { + "series": { + "maxItems": 1, + "items": { + "required": ["input", "x", "y", "z"], + "not": {"required": ["y_error"]} + } + } + } + } + }, + { + "if": {"properties": {"type": {"const": "y_error"}}}, + "then": { + "properties": { + "series": { + "items": { + "required": ["input", "x", "y", "y_error"], + "not": {"required": ["z"]} + } + } + } + } + }, + { + "if": {"properties": {"type": {"const": "grouped_column"}}}, + "then": { + "properties": { + "series": { + "minItems": 2, + "items": { + "required": ["input", "x", "y"], + "not": {"anyOf": [{"required": ["z"]}, {"required": ["y_error"]}]} + } + } + } + } + }, + { + "if": {"properties": {"type": {"enum": ["line", "scatter", "line_scatter", "column", "bar"]}}}, + "then": { + "properties": { + "series": { + "items": { + "required": ["input", "x", "y"], + "not": {"anyOf": [{"required": ["z"]}, {"required": ["y_error"]}]} + } + } + } + } + } + ] + } + } + }, + "outputs": { + "type": "array", + "minItems": 1, + "maxItems": 4, + "uniqueItems": true, + "items": { + "oneOf": [ + { + "type": "object", + "required": ["key", "type", "format"], + "additionalProperties": false, + "properties": { + "key": {"const": "project"}, + "type": {"const": "project"}, + "format": {"const": "opju"} + } + }, + { + "type": "object", + "required": ["key", "type", "format"], + "additionalProperties": false, + "properties": { + "key": {"const": "figure_png"}, + "type": {"const": "figure"}, + "format": {"const": "png"}, + "options": { + "type": "object", + "required": ["dpi"], + "additionalProperties": false, + "properties": {"dpi": {"type": "integer", "minimum": 72, "maximum": 1200}} + } + } + }, + { + "type": "object", + "required": ["key", "type", "format"], + "additionalProperties": false, + "properties": { + "key": {"const": "figure_svg"}, + "type": {"const": "figure"}, + "format": {"const": "svg"} + } + }, + { + "type": "object", + "required": ["key", "type", "format"], + "additionalProperties": false, + "properties": { + "key": {"const": "figure_pdf"}, + "type": {"const": "figure"}, + "format": {"const": "pdf"} + } + } + ] + } + } + }, + "$defs": { + "axis": { + "type": "object", + "additionalProperties": false, + "properties": { + "title": {"type": "string"}, + "unit": {"type": "string"}, + "scale": {"const": "linear"} + } + } + } + } +} diff --git a/tests/test_software_contracts.py b/tests/test_software_contracts.py new file mode 100644 index 0000000..bba4037 --- /dev/null +++ b/tests/test_software_contracts.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +import unittest + +from core.software_contracts import ( + DEFAULT_CAPABILITIES, + get_contract, + node_supports_request, + version_at_least, +) + + +def _request(plot_type: str) -> dict: + roles = {"input": "sample", "x": "x", "y": "y"} + if plot_type in {"contour", "surface_3d", "ternary", "heatmap"}: + roles["z"] = "z" + return { + "schema_version": 2, + "inputs": [{ + "key": "sample", + "artifact_id": "f4186347-65cc-4f07-9c26-bf11992beef8", + }], + "operation": {"plot": {"type": plot_type, "series": [roles]}}, + "outputs": [{"key": "figure_png", "type": "figure", "format": "png"}], + } + + +class SoftwareContractTests(unittest.TestCase): + def test_origin_contract_drives_schema_outputs_and_summary(self) -> None: + contract = get_contract("origin.plot@v2") + request = _request("line") + normalized, digest = contract.normalize_request(request) + self.assertEqual(normalized, request) + self.assertEqual(len(digest), 64) + self.assertEqual(contract.output_namespace, "origin") + self.assertEqual( + set(contract.expected_outputs(request)), + {"figure_png", "plot_spec", "provenance"}, + ) + self.assertEqual(DEFAULT_CAPABILITIES, ("origin.plot@v2",)) + + def test_adapter_version_and_features_gate_additive_requests(self) -> None: + contract = get_contract("origin.plot@v2") + legacy_runtime = { + "available_slots": 1, + "origin": {"health": "ready", "adapter_version": "0.3.0"}, + } + self.assertTrue(node_supports_request(contract, _request("line"), legacy_runtime)) + self.assertFalse(node_supports_request(contract, _request("heatmap"), legacy_runtime)) + current_runtime = { + "capability_runtime": { + "origin.plot@v2": { + "health": "ready", + "available_slots": 1, + "adapter_version": "0.4.0", + "features": ["line", "heatmap"], + } + } + } + self.assertTrue(node_supports_request(contract, _request("heatmap"), current_runtime)) + self.assertFalse(node_supports_request(contract, _request("bar"), current_runtime)) + + def test_semantic_version_comparison_is_numeric(self) -> None: + self.assertTrue(version_at_least("0.10.0", "0.4.0")) + self.assertFalse(version_at_least("0.3.9", "0.4.0")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_software_nodes.py b/tests/test_software_nodes.py index c740880..8e1b64b 100644 --- a/tests/test_software_nodes.py +++ b/tests/test_software_nodes.py @@ -3,7 +3,7 @@ from __future__ import annotations import importlib import unittest from pathlib import Path -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch from uuid import uuid4 from alembic.migration import MigrationContext @@ -19,6 +19,7 @@ from core.software_jobs import ( list_jobs, mark_node_jobs_disconnected, record_job_terminal, + offer_next_job, replay_succeeded_outputs, request_job_cancel, respond_to_offer, @@ -213,17 +214,17 @@ class SoftwareJobProtocolTests(unittest.TestCase): def test_published_output_distinguishes_artifacts_from_metadata(self) -> None: job_id = uuid4() - self.assertTrue(_published_output_is_valid(job_id, { + self.assertTrue(_published_output_is_valid("origin.plot@v2", job_id, { "source_artifact_id": "figure_png", "artifact_id": str(uuid4()), "path": f"origin/{job_id}/figure.png", })) - self.assertTrue(_published_output_is_valid(job_id, { + self.assertTrue(_published_output_is_valid("origin.plot@v2", job_id, { "source_artifact_id": "plot_spec", "artifact_id": None, "path": f"origin/{job_id}/.meta/plot-spec.json", })) - self.assertFalse(_published_output_is_valid(job_id, { + self.assertFalse(_published_output_is_valid("origin.plot@v2", job_id, { "source_artifact_id": ["plot_spec"], "artifact_id": None, "path": f"origin/{job_id}/.meta/plot-spec.json", @@ -301,13 +302,13 @@ class SoftwareJobProtocolTests(unittest.TestCase): normalized, digest = _canonical_request("origin.plot@v2", request) self.assertEqual(normalized, request) self.assertEqual(len(digest), 64) - with self.assertRaisesRegex(Exception, "invalid origin plot request fields"): + with self.assertRaisesRegex(Exception, "invalid origin.plot@v2 request"): _canonical_request("origin.plot@v2", {**request, "script": "anything"}) - with self.assertRaisesRegex(Exception, "unsupported origin plot fields"): + with self.assertRaisesRegex(Exception, "invalid origin.plot@v2 request"): _canonical_request("origin.plot@v2", {**request, "operation": {"plot": { **request["operation"]["plot"], "script": "anything" }}}) - with self.assertRaisesRegex(Exception, "artifact UUID"): + with self.assertRaisesRegex(Exception, "not a 'uuid'"): _canonical_request("origin.plot@v2", {**request, "inputs": [{"key": "first", "artifact_id": "C:\\data.csv"}]}) def test_origin_request_rejects_unimplemented_plot_semantics(self) -> None: @@ -321,20 +322,26 @@ class SoftwareJobProtocolTests(unittest.TestCase): "outputs": [{"key": "figure_png", "type": "figure", "format": "png", "options": {"dpi": 300}}], } base_plot = request["operation"]["plot"] - for case_plot, message in ( - ({**base_plot, "template": "custom"}, "unsupported origin plot template"), - ({**base_plot, "x_axis": {"scale": "log10"}}, "invalid x_axis"), - ({**base_plot, "legend": {"enabled": False}}, "invalid plot.legend"), - ({**base_plot, "series": [base_plot["series"][0], base_plot["series"][0]]}, "plot series must be unique"), + for case_plot in ( + {**base_plot, "template": "custom"}, + {**base_plot, "x_axis": {"scale": "log10"}}, + {**base_plot, "legend": {"enabled": False}}, + {**base_plot, "series": [base_plot["series"][0], base_plot["series"][0]]}, ): - with self.subTest(message=message), self.assertRaisesRegex(Exception, message): + with self.subTest(plot=case_plot), self.assertRaisesRegex( + Exception, "invalid origin.plot@v2 request" + ): _canonical_request("origin.plot@v2", {**request, "operation": {"plot": case_plot}}) - with self.assertRaisesRegex(Exception, "unknown input"): - _canonical_request("origin.plot@v2", {**request, "operation": {"plot": { + # 跨字段引用完整性属于本机 adapter 的独立二次校验；Core 只执行共享结构契约。 + normalized, _ = _canonical_request("origin.plot@v2", { + **request, + "operation": {"plot": { **base_plot, "series": [{"input": "missing", "x": "time", "y": "a"}], - }}}) - with self.assertRaisesRegex(Exception, "key, type, and format do not match"): + }}, + }) + self.assertEqual(normalized["operation"]["plot"]["series"][0]["input"], "missing") + with self.assertRaisesRegex(Exception, "invalid origin.plot@v2 request"): _canonical_request( "origin.plot@v2", {**request, "outputs": [{"key": "project", "type": "figure", "format": "png"}]} @@ -395,12 +402,14 @@ class SoftwareJobProtocolTests(unittest.TestCase): {"artifact_id": "plot_spec", "filename": "plot-spec.json", "media_type": "application/json", "size_bytes": 30, "sha256": "c" * 64}, {"artifact_id": "provenance", "filename": "provenance.json", "media_type": "application/json", "size_bytes": 40, "sha256": "d" * 64}, ] - self.assertEqual(validate_output_manifest(request, manifest), manifest) + self.assertEqual( + validate_output_manifest("origin.plot@v2", request, manifest), manifest + ) with self.assertRaisesRegex(Exception, "incomplete"): - validate_output_manifest(request, manifest[:-1]) + validate_output_manifest("origin.plot@v2", request, manifest[:-1]) with self.assertRaisesRegex(Exception, "metadata"): validate_output_manifest( - request, + "origin.plot@v2", request, [{**manifest[0], "filename": "anything.opju"}, *manifest[1:]], ) @@ -440,8 +449,49 @@ class SoftwareJobProtocolTests(unittest.TestCase): source = ( Path(__file__).resolve().parents[1] / "core" / "software_jobs.py" ).read_text(encoding="utf-8") - self.assertIn('{"offered", "dispatched", "running"}', source) - self.assertIn("item.node_id not in busy_node_ids", source) + self.assertIn('"disconnected", "cancelling"', source) + self.assertIn("node_supports_request", source) + self.assertIn("SoftwareJob.capability.in_(available_capabilities)", source) + + @patch("core.software_jobs.session_scope") + def test_dispatcher_skips_job_that_requires_newer_adapter(self, session_scope) -> None: + session = session_scope.return_value.__enter__.return_value + node = type("Node", (), {})() + node.node_id = uuid4() + node.capabilities = ["origin.plot@v2"] + node.runtime = { + "available_slots": 1, + "origin": {"health": "ready", "adapter_version": "0.3.0"}, + } + jobs = [] + for plot_type in ("heatmap", "line"): + job = type("Job", (), {})() + job.job_id = uuid4() + job.capability = "origin.plot@v2" + roles = {"input": "sample", "x": "x", "y": "y"} + if plot_type == "heatmap": + roles["z"] = "z" + job.request = { + "operation": {"plot": {"type": plot_type, "series": [roles]}}, + "outputs": [], + } + job.input_manifest = [] + job.request_digest = plot_type[0] * 64 + job.status = "queued" + jobs.append(job) + results = [MagicMock() for _ in range(5)] + results[0].scalars.return_value = [] + results[1].scalars.return_value = [] + results[2].scalars.return_value = [node] + results[3].scalars.return_value = jobs + results[4].scalar_one_or_none.return_value = jobs[1] + session.execute.side_effect = results + + offer = offer_next_job({node.node_id}) + + self.assertEqual(offer["node_id"], node.node_id) + self.assertEqual(offer["payload"]["job_id"], str(jobs[1].job_id)) + self.assertEqual(offer["payload"]["request"]["operation"]["plot"]["type"], "line") def test_input_download_rechecks_file_digest(self) -> None: source = ( @@ -450,7 +500,7 @@ class SoftwareJobProtocolTests(unittest.TestCase): ).read_text(encoding="utf-8") self.assertIn("digest = sha256()", source) self.assertIn('digest.hexdigest() != item["sha256"]', source) - self.assertIn('context["request"].get("outputs", [])', source) + self.assertIn('contract.expected_outputs(context["request"])', source) self.assertIn("artifact_id not in requested_ids", source) @patch("core.software_jobs.session_scope") diff --git a/tests/test_windows_node_source.py b/tests/test_windows_node_source.py index 198f3ac..9444595 100644 --- a/tests/test_windows_node_source.py +++ b/tests/test_windows_node_source.py @@ -26,7 +26,7 @@ class WindowsNodeSourceTests(unittest.TestCase): 'SetRequestHeader("X-Node-Id"', "DataProtectionScope.LocalMachine", "SetAccessRuleProtection(isProtected: true", - '"origin.plot@v2"', + '"origin.plot.v2.json"', "NotifyIcon", "ConfigurationForm", "TrayIconFactory.Create", @@ -70,8 +70,8 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertIn("CreateCard", form) self.assertIn("注册并连接", form) self.assertIn("ContentWidth = 760", form) - self.assertIn("Origin 绘图", form) - self.assertIn('CreateCapabilityRow("Origin 绘图", "origin.plot@v2")', form) + self.assertIn("NodeAdapterRegistry.InstalledContracts", form) + self.assertIn("CreateCapabilityRow(contract.DisplayName, contract.Capability)", form) self.assertIn('CreateButton("立即重连", 112, primary: true)', form) self.assertIn("ReconnectRequested?.Invoke()", form) self.assertIn("registrationCard.Visible = !registered", form) @@ -135,6 +135,7 @@ class WindowsNodeSourceTests(unittest.TestCase): project = (PROJECT / "Zcbot.WindowsNode.csproj").read_text(encoding="utf-8") self.assertIn("..\\install-windows-node.bat", project) self.assertIn("..\\origin-worker\\requirements.txt", project) + self.assertIn("..\\..\\software-contracts\\*.json", project) self.assertFalse((ROOT / "install-windows-node.ps1").exists()) self.assertFalse((ROOT / "install-origin-runtime.ps1").exists()) self.assertFalse((ROOT / "install-startup.ps1").exists()) @@ -146,6 +147,7 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertIn('if /I "%~1"=="--self-contained"', script) self.assertIn("dotnet publish", script) self.assertIn('"install-windows-node.bat"', script) + self.assertIn('"software-contracts\\origin.plot.v2.json"', script) self.assertIn("tar.exe -a -c -f", script) self.assertIn("certutil.exe -hashfile", script) self.assertFalse((ROOT / "package-windows-node.ps1").exists()) @@ -167,12 +169,12 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertIn('AutomationProgId = @"Origin.ApplicationSI\\CLSID"', probe) self.assertIn("RegistryHive.LocalMachine", probe) self.assertIn("RegistryHive.CurrentUser", probe) - self.assertIn('new("OriginPro", version, "0.4.0", health, detail)', probe) + self.assertIn("OriginPlotNodeAdapter.CurrentAdapterVersion", probe) self.assertIn( - "&& !jobInbox.HasPendingOriginJobs", + "&& !jobInbox.HasPendingJobs", connection, ) - self.assertIn("&& !workerRunner.HasActiveJobs ? 1 : 0", connection) + self.assertIn("originAdapter?.HasActiveJobs", connection) self.assertIn( "ReadRecoverableJobs().Any(item => item.Terminal is null)", (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8"), @@ -190,10 +192,10 @@ class WindowsNodeSourceTests(unittest.TestCase): def test_job_offer_is_persisted_before_acceptance(self) -> None: inbox = (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8") connection = (PROJECT / "NodeConnectionLoop.cs").read_text(encoding="utf-8") - self.assertIn('capabilityValue.GetString() != "origin.plot@v2"', inbox) + self.assertIn("adapters.Find(capability)", inbox) self.assertIn('root.TryGetProperty("input_transfers"', inbox) self.assertIn('"input", key, filename', inbox) - self.assertIn("PlotTypes.Contains", inbox) + self.assertIn("supportedPlotTypes.Contains", inbox) self.assertIn("IsValidOutputs", inbox) self.assertIn('("figure", "png") => "figure_png"', inbox) self.assertIn("FileOptions.WriteThrough", inbox) @@ -205,8 +207,9 @@ class WindowsNodeSourceTests(unittest.TestCase): ) self.assertIn('offerResult.Accepted ? "job_accept" : "job_reject"', connection) self.assertIn("sendLock.WaitAsync", connection) - self.assertIn("!jobInbox.HasPendingOriginJobs", connection) - self.assertIn("!workerRunner.HasActiveJobs ? 1 : 0", connection) + self.assertIn("!jobInbox.HasPendingJobs", connection) + self.assertIn("capability_runtime = capabilityRuntime", connection) + self.assertIn("adapter.RunAsync(job)", connection) self.assertIn("ReportRecoverableJobsAsync", connection) self.assertIn("ConcurrentDictionary jobPipelines", connection) self.assertIn("StartJobPipeline(socket, acceptedJob)", connection) diff --git a/tools/software_jobs.py b/tools/software_jobs.py index 99c3b1b..c18b7b4 100644 --- a/tools/software_jobs.py +++ b/tools/software_jobs.py @@ -4,6 +4,12 @@ from __future__ import annotations import json from uuid import UUID, uuid4 +from core.software_contracts import ( + CONTRACTS, + SUPPORTED_CAPABILITIES, + get_contract, + node_available_slots, +) from core.software_jobs import ( SoftwareJobError, create_job, @@ -11,11 +17,21 @@ from core.software_jobs import ( list_jobs, request_job_cancel, ) -from core.software_nodes import SUPPORTED_CAPABILITIES, list_nodes +from core.software_nodes import list_nodes from .base import Tool +def _contract_property_schema(name: str) -> dict: + schemas = [ + item.submission_schema()["properties"][name] + for item in CONTRACTS.values() + ] + unique = {json.dumps(item, ensure_ascii=False, sort_keys=True): item for item in schemas} + values = list(unique.values()) + return values[0] if len(values) == 1 else {"anyOf": values} + + class _SoftwareJobTool(Tool): def __init__(self, user_id: UUID, task_id: UUID, **kwargs) -> None: super().__init__(**kwargs) @@ -32,13 +48,13 @@ class SoftwareCapabilityListTool(_SoftwareJobTool): nodes = list_nodes() items = [{ "capability": item, - "display_name": "Origin 科研绘图" if item == "origin.plot@v2" else item, + "display_name": get_contract(item).display_name, "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 + and node_available_slots(item, node.get("runtime") or {}) > 0 ), } for item in sorted(SUPPORTED_CAPABILITIES)] return json.dumps({"capabilities": items}, ensure_ascii=False) @@ -47,134 +63,17 @@ class SoftwareCapabilityListTool(_SoftwareJobTool): class SoftwareJobSubmitTool(_SoftwareJobTool): name = "software_job_submit" description = ( - "Submit an Origin 2D, error-bar, contour, 3D, ternary, or heatmap plot job using " - "one or more registered CSV, XLSX, or JSON artifacts. " - "Call register_artifact first for each workspace file. " - "Return immediately with job_id; do not poll continuously or wait for completion." + "Submit a managed professional-software job using registered artifacts and a " + "capability contract. Call register_artifact first for workspace files. 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", "column", "bar", "grouped_column", - "y_error", "contour", "surface_3d", "ternary", "heatmap", - ]}, - "series": { - "type": "array", - "minItems": 1, - "maxItems": 16, - "items": { - "type": "object", - "properties": { - "input": {"type": "string", "pattern": "^[a-z][a-z0-9_]{0,31}$"}, - "x": {"type": "string", "minLength": 1, "maxLength": 128}, - "y": {"type": "string", "minLength": 1, "maxLength": 128}, - "z": {"type": "string", "minLength": 1, "maxLength": 128}, - "y_error": {"type": "string", "minLength": 1, "maxLength": 128}, - "label": {"type": "string", "minLength": 1, "maxLength": 200}, - }, - "required": ["input"], - "additionalProperties": False, - }, - "description": ( - "Typed data roles. XY plots require x/y; y_error requires x/y/y_error; " - "contour, surface_3d, ternary, and heatmap require x/y/z." - ), - }, - "template": {"type": "string", "enum": ["publication_double_column"]}, - "title": {"type": "string", "maxLength": 500}, - "x_axis": _axis_schema, - "y_axis": _axis_schema, - "z_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", "series"], - "additionalProperties": False, - } - _input_schema = { - "type": "object", - "properties": { - "key": {"type": "string", "pattern": "^[a-z][a-z0-9_]{0,31}$"}, - "artifact_id": { - "type": "string", - "description": "Artifact UUID returned by register_artifact.", - }, - "selector": { - "type": "object", - "properties": { - "sheet": {"type": "string", "minLength": 1, "maxLength": 128}, - }, - "required": ["sheet"], - "additionalProperties": False, - }, - }, - "required": ["key", "artifact_id"], - "additionalProperties": False, - } - _output_schema = { - "type": "object", - "properties": { - "key": { - "type": "string", - "enum": ["project", "figure_png", "figure_svg", "figure_pdf"], - }, - "type": {"type": "string", "enum": ["project", "figure"]}, - "format": {"type": "string", "enum": ["opju", "png", "svg", "pdf"]}, - "options": { - "type": "object", - "properties": {"dpi": {"type": "integer", "minimum": 72, "maximum": 1200}}, - "required": ["dpi"], - "additionalProperties": False, - "description": "Only valid for the PNG figure output.", - }, - }, - "required": ["key", "type", "format"], - "additionalProperties": False, - } parameters = { "type": "object", "properties": { "capability": {"type": "string", "enum": sorted(SUPPORTED_CAPABILITIES)}, - "inputs": { - "type": "array", - "minItems": 1, - "maxItems": 16, - "items": _input_schema, - }, - "operation": { - "type": "object", - "properties": {"plot": _plot_schema}, - "required": ["plot"], - "additionalProperties": False, - }, - "outputs": { - "type": "array", - "minItems": 1, - "maxItems": 16, - "uniqueItems": True, - "items": _output_schema, - "description": ( - "Requested deliverables. Use project/project/opju, " - "figure_png/figure/png, figure_svg/figure/svg, or figure_pdf/figure/pdf." - ), - }, + "inputs": _contract_property_schema("inputs"), + "operation": _contract_property_schema("operation"), + "outputs": _contract_property_schema("outputs"), "idempotency_key": { "type": "string", "description": "Stable unique key for this exact submission; omit to generate one.", @@ -206,7 +105,9 @@ class SoftwareJobSubmitTool(_SoftwareJobTool): canonical = {**item, "artifact_id": artifact_id} canonical_inputs.append(canonical) normalized_request = { - "schema_version": 2, + "schema_version": get_contract(capability).request_schema[ + "properties" + ]["schema_version"]["const"], "inputs": canonical_inputs, "operation": operation, "outputs": outputs, diff --git a/web/routers/software_nodes.py b/web/routers/software_nodes.py index 82e2f96..860c33f 100644 --- a/web/routers/software_nodes.py +++ b/web/routers/software_nodes.py @@ -22,11 +22,14 @@ 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 ( + DEFAULT_CAPABILITIES, + SoftwareContractError, + get_contract, +) from core.software_jobs import ( MAX_OUTPUT_ARTIFACT_BYTES, MAX_OUTPUT_TOTAL_BYTES, - OUTPUT_ARTIFACTS, - SOFTWARE_JOB_METADATA_IDS, SoftwareJobError, abandon_offer, create_job, @@ -176,20 +179,28 @@ def _task_working_dir(root: Path, stored: str) -> Path: return working_dir -def _staged_output_path(staging: Path, output_id: str, filename: str) -> Path: +def _staged_output_path( + staging: Path, capability: str, output_id: str, filename: str +) -> Path: flat = staging / filename - organized = staging / software_job_output_path(output_id) + organized = staging / software_job_output_path(capability, output_id) return flat if flat.is_file() else organized -def _organize_staged_metadata(staging: Path, manifest: list[dict]) -> None: - metadata = [item for item in manifest if item["artifact_id"] in SOFTWARE_JOB_METADATA_IDS] +def _organize_staged_metadata( + staging: Path, capability: str, manifest: list[dict] +) -> None: + contract = get_contract(capability) + metadata = [ + item for item in manifest + if not contract.output_spec(item["artifact_id"]).publish + ] if not metadata: return (staging / ".meta").mkdir(exist_ok=True) for item in metadata: source = staging / item["filename"] - destination = staging / software_job_output_path(item["artifact_id"]) + destination = staging / software_job_output_path(capability, item["artifact_id"]) if not source.is_file(): if destination.is_file(): continue @@ -207,19 +218,23 @@ def _organize_staged_metadata(staging: Path, manifest: list[dict]) -> None: def _publish_software_job_outputs(job_id: UUID, context: dict, manifest: list[dict]) -> list[dict]: + capability = context.get("capability", DEFAULT_CAPABILITIES[0]) + contract = get_contract(capability) root = load_user_root(context["user_id"]) working_dir = _task_working_dir(root, context["working_dir"]) staging = safe_join(root, f".zcbot_software_job_staging/{job_id}") - relative_output = Path("origin") / str(job_id) + relative_output = Path(contract.output_namespace) / str(job_id) destination = safe_join(working_dir, relative_output.as_posix()) source = staging if staging.is_dir() else destination _reject_symlink_path(root, source) _reject_symlink_path(root, destination) for item in manifest: path = ( - _staged_output_path(staging, item["artifact_id"], item["filename"]) + _staged_output_path( + staging, capability, item["artifact_id"], item["filename"] + ) if source == staging - else destination / software_job_output_path(item["artifact_id"]) + else destination / software_job_output_path(capability, item["artifact_id"]) ) if ( not path.is_file() @@ -228,7 +243,7 @@ def _publish_software_job_outputs(job_id: UUID, context: dict, manifest: list[di ): raise SoftwareJobError(f"uploaded artifact is missing or invalid: {item['artifact_id']}") if source == staging: - _organize_staged_metadata(staging, manifest) + _organize_staged_metadata(staging, capability, manifest) destination.parent.mkdir(parents=True, exist_ok=True) if destination.exists(): raise SoftwareJobError("software job output destination already exists unexpectedly") @@ -238,10 +253,12 @@ def _publish_software_job_outputs(job_id: UUID, context: dict, manifest: list[di except OSError: pass refs = tuple({ - "path": (relative_output / software_job_output_path(item["artifact_id"])).as_posix(), + "path": ( + relative_output / software_job_output_path(capability, item["artifact_id"]) + ).as_posix(), "label": item["filename"], "media_type": item["media_type"], - } for item in manifest if item["artifact_id"] not in SOFTWARE_JOB_METADATA_IDS) + } for item in manifest if contract.output_spec(item["artifact_id"]).publish) published_refs = register_published_artifacts( user_id=context["user_id"], task_id=context["task_id"], @@ -257,12 +274,15 @@ def _publish_software_job_outputs(job_id: UUID, context: dict, manifest: list[di "source_artifact_id": item["artifact_id"], "artifact_id": ( refs_by_path.get( - (relative_output / software_job_output_path(item["artifact_id"])).as_posix(), + ( + relative_output + / software_job_output_path(capability, item["artifact_id"]) + ).as_posix(), {}, ).get("artifact_id") ), "path": ( - relative_output / software_job_output_path(item["artifact_id"]) + relative_output / software_job_output_path(capability, item["artifact_id"]) ).as_posix(), } for item in manifest @@ -338,17 +358,15 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: _authenticate_output_request, job_id, authorization, x_node_id, x_lease_id, x_request_digest, ) - metadata = OUTPUT_ARTIFACTS.get(artifact_id) - if metadata is None: - raise HTTPException(400, "unsupported output artifact identity") - filename, _, _ = metadata - requested_ids = { - item.get("key") - for item in context["request"].get("outputs", []) - if isinstance(item, dict) - } | SOFTWARE_JOB_METADATA_IDS + try: + contract = get_contract(context["capability"]) + output_spec = contract.output_spec(artifact_id) + requested_ids = set(contract.expected_outputs(context["request"])) + except SoftwareContractError as exc: + raise HTTPException(400, str(exc)) from exc if artifact_id not in requested_ids: - raise HTTPException(400, "output artifact was not requested") + raise HTTPException(400, "unsupported output artifact identity") + filename = output_spec.filename if not 1 <= x_content_length <= MAX_OUTPUT_ARTIFACT_BYTES: raise HTTPException(400, "output artifact size is invalid") if len(x_content_sha256) != 64 or any(c not in "0123456789abcdef" for c in x_content_sha256): @@ -367,7 +385,10 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: working_dir = _task_working_dir(root, context["working_dir"]) published = safe_join( working_dir, - f"origin/{job_id}/{software_job_output_path(artifact_id)}", + ( + f"{contract.output_namespace}/{job_id}/" + f"{software_job_output_path(context['capability'], artifact_id)}" + ), ) if published.is_file(): if published.stat().st_size == x_content_length and _hash_file(published) == x_content_sha256: @@ -376,7 +397,7 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: staging = safe_join(root, f".zcbot_software_job_staging/{job_id}") _reject_symlink_path(root, staging) staging.mkdir(parents=True, exist_ok=True) - organized = staging / software_job_output_path(artifact_id) + organized = staging / software_job_output_path(context["capability"], artifact_id) if organized.is_file(): if organized.stat().st_size == x_content_length and _hash_file(organized) == x_content_sha256: return @@ -426,7 +447,9 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: if not isinstance(body, dict): raise HTTPException(400, "output completion body must be an object") try: - manifest = validate_output_manifest(context["request"], body.get("artifact_manifest")) + manifest = validate_output_manifest( + context["capability"], context["request"], body.get("artifact_manifest") + ) replayed = replay_succeeded_outputs(context, manifest) if replayed is not None: return {"status": "succeeded", "artifact_manifest": replayed} diff --git a/web/schemas.py b/web/schemas.py index 982d7df..2f206cb 100644 --- a/web/schemas.py +++ b/web/schemas.py @@ -6,6 +6,8 @@ from uuid import UUID from pydantic import BaseModel, Field +from core.software_contracts import DEFAULT_CAPABILITIES + class TaskCreateRequest(BaseModel): # name 缺失/空值 + working_dir 有值 → 后端以“新对话”占位并自动生成标题。 @@ -118,7 +120,7 @@ class ExternalSystemCredentialsRequest(BaseModel): class SoftwareEnrollmentCreateRequest(BaseModel): expected_name: str = "" - capabilities: list[str] = Field(default_factory=lambda: ["origin.plot@v2"]) + capabilities: list[str] = Field(default_factory=lambda: list(DEFAULT_CAPABILITIES)) ttl_seconds: int = 600 @@ -137,5 +139,5 @@ class SoftwareNodeDisableRequest(BaseModel): class SoftwareJobCreateRequest(BaseModel): idempotency_key: str - capability: str = "origin.plot@v2" + capability: str = DEFAULT_CAPABILITIES[0] request: dict = Field(default_factory=dict) diff --git a/windows-node/README.md b/windows-node/README.md index 3f0f0b0..49e4554 100644 --- a/windows-node/README.md +++ b/windows-node/README.md @@ -2,7 +2,7 @@ 内网 MVP 的 Windows 执行节点，目标运行环境为 Windows 11 Enterprise + .NET 10 SDK 10.0.303。仓库根目录 `global.json` 固定 SDK patch；客户端只使用 .NET Windows Desktop Framework，不依赖第三方 NuGet 包。 -当前实现托盘状态角标、配置与本机任务窗口、注册、DPAPI/ACL 配置保存、WebSocket `hello`/心跳和退避重连，并只读探测 Origin/OriginPro 安装版本、COM 自动化组件与桌面会话状态。本机任务列表只读取已派发到该 Node 的持久化目录，展示标题、全部输入、通用执行阶段、进度、时间、Job ID 和错误，不查询云端未派发队列。Node 可以接收受控的 `origin.plot@v2` offer，在本机任务目录原子保存请求后回报 accept/reject；随后以 Node 身份逐个流式下载任务绑定的 CSV/XLSX/JSON 到 `input//`，校验大小与 SHA-256 后原子保存。固定 Origin Worker 独立于单次 WebSocket 执行，断线不终止已启动绘图。成功产物按 manifest 逐项流式上传并由云端复核大小与 SHA-256，全部完成后原子发布到任务工作目录的 `origin//`；中断后按本地 `upload-complete.json` 幂等续传。 +当前实现托盘状态角标、配置与本机任务窗口、注册、DPAPI/ACL 配置保存、WebSocket `hello`/心跳和退避重连。Node Host 以语言无关的 job 目录协议负责持久化、下载、恢复、取消和上传，具体软件由 `NodeAdapterRegistry` 中的 adapter 负责探测、校验和执行；adapter 可为 .NET 内置实现，也可由固定 runner 启动任意语言的受信进程。注册能力、配置校验和界面从 registry 与随包发布的 `software-contracts/*.json` 派生。当前安装包注册 `origin.plot@v2`，其 adapter 使用 Python Worker 驱动 Origin/OriginPro，但 Python 不是 Node 通用协议的一部分。本机任务列表只读取已派发到该 Node 的持久化目录，不查询云端未派发队列；成功产物按 manifest 上传并由云端复核，全部完成后原子发布，中断后按本地 `upload-complete.json` 幂等续传。 在仓库根目录执行一条命令生成可分发 ZIP： diff --git a/windows-node/Zcbot.WindowsNode/ConfigurationForm.cs b/windows-node/Zcbot.WindowsNode/ConfigurationForm.cs index af2fff4..dfa1a7c 100644 --- a/windows-node/Zcbot.WindowsNode/ConfigurationForm.cs +++ b/windows-node/Zcbot.WindowsNode/ConfigurationForm.cs @@ -29,6 +29,7 @@ internal sealed class ConfigurationForm : Form private readonly Label jobDetail = CreateBodyLabel(); private readonly DataGridView jobGrid = CreateJobGrid(); private readonly JobInboxStore jobInbox = new(NodePaths.ForCurrentMachine().JobsDirectory); + private readonly NodeAdapterRegistry adapters; private readonly System.Windows.Forms.Timer jobRefreshTimer = new() { Interval = 1000 }; private readonly TableLayoutPanel registrationCard; private bool changingStartup; @@ -39,6 +40,7 @@ internal sealed class ConfigurationForm : Form internal ConfigurationForm() { + adapters = NodeAdapterRegistry.CreateDefault(jobInbox); Text = "zcbot Windows Node"; AutoScaleMode = AutoScaleMode.Dpi; ClientSize = new Size(840, 680); @@ -97,7 +99,10 @@ internal sealed class ConfigurationForm : Form statusCard.Controls.Add(detail); statusCard.Controls.Add(identity); statusCard.Controls.Add(CreateDivider()); - statusCard.Controls.Add(CreateCapabilityRow("Origin 绘图", "origin.plot@v2")); + foreach (var contract in NodeAdapterRegistry.InstalledContracts) + { + statusCard.Controls.Add(CreateCapabilityRow(contract.DisplayName, contract.Capability)); + } statusCard.Controls.Add(capabilitySummary); var resetActions = CreateActions(); resetActions.Controls.Add(reconnect); @@ -183,7 +188,7 @@ internal sealed class ConfigurationForm : Form : $"节点：{config.NodeName}\nNode ID：{config.NodeId}\n服务：{config.ServerUrl}"; capabilitySummary.Text = config is null ? "注册后启用" - : FormatOriginStatus(OriginRuntimeProbe.Detect()); + : string.Join("\n", adapters.All.Select(FormatAdapterStatus)); var registered = config is not null; registrationCard.Visible = !registered; @@ -198,13 +203,14 @@ internal sealed class ConfigurationForm : Form } } - private static string FormatOriginStatus(OriginRuntimeStatus origin) + private static string FormatAdapterStatus(INodeAdapter adapter) { - var version = string.IsNullOrWhiteSpace(origin.SoftwareVersion) + var runtime = adapter.DetectRuntime(); + var version = string.IsNullOrWhiteSpace(runtime.SoftwareVersion) ? "版本未知" - : $"版本 {origin.SoftwareVersion}"; - var state = origin.Health == "ready" ? "可用" : "不可用"; - return $"{state} · {version}\n{origin.Detail}"; + : $"版本 {runtime.SoftwareVersion}"; + var state = runtime.Health == "ready" ? "可用" : "不可用"; + return $"{adapter.DisplayName}：{state} · {version}\n{runtime.Detail}"; } private void RefreshJobs() diff --git a/windows-node/Zcbot.WindowsNode/EnrollmentClient.cs b/windows-node/Zcbot.WindowsNode/EnrollmentClient.cs index 60728b7..1d13465 100644 --- a/windows-node/Zcbot.WindowsNode/EnrollmentClient.cs +++ b/windows-node/Zcbot.WindowsNode/EnrollmentClient.cs @@ -6,8 +6,6 @@ namespace Zcbot.WindowsNode; internal static class EnrollmentClient { - private static readonly string[] Capabilities = ["origin.plot@v2"]; - internal static async Task EnrollAsync( EnrollOptions options, NodeConfigStore store, CancellationToken cancellationToken) { @@ -24,7 +22,7 @@ internal static class EnrollmentClient installId, Assembly.GetExecutingAssembly().GetName().Version?.ToString(3) ?? "0.1.0", RuntimeInformation.OSDescription, - Capabilities); + NodeAdapterRegistry.InstalledCapabilities); using var client = new HttpClient { BaseAddress = options.ServerUrl, Timeout = TimeSpan.FromSeconds(30) }; using var response = await client.PostAsJsonAsync( @@ -44,7 +42,8 @@ internal static class EnrollmentClient store.Save(new NodeConfig( options.ServerUrl, enrolled.NodeId, installId, options.NodeName, - enrolled.NodeToken, Math.Clamp(enrolled.HeartbeatSeconds, 5, 300), Capabilities)); + enrolled.NodeToken, Math.Clamp(enrolled.HeartbeatSeconds, 5, 300), + NodeAdapterRegistry.InstalledCapabilities)); } private static string Limit(string value, int maxLength) => diff --git a/windows-node/Zcbot.WindowsNode/JobInboxStore.cs b/windows-node/Zcbot.WindowsNode/JobInboxStore.cs index c919248..f404b03 100644 --- a/windows-node/Zcbot.WindowsNode/JobInboxStore.cs +++ b/windows-node/Zcbot.WindowsNode/JobInboxStore.cs @@ -6,11 +6,6 @@ namespace Zcbot.WindowsNode; internal sealed class JobInboxStore(string jobsDirectory) { private static readonly JsonSerializerOptions JsonOptions = new() { WriteIndented = true }; - private static readonly HashSet PlotTypes = - [ - "line", "scatter", "line_scatter", "column", "bar", "grouped_column", - "y_error", "contour", "surface_3d", "ternary", "heatmap", - ]; private static readonly HashSet XyzPlotTypes = ["contour", "surface_3d", "ternary", "heatmap"]; private static readonly HashSet JobStages = @@ -27,7 +22,7 @@ internal sealed class JobInboxStore(string jobsDirectory) // Origin 执行槽只由尚无终态的任务占用。成功但上传确认尚未落盘的任务会由 // 心跳恢复管线继续重传；上传不使用 Origin，不能反向阻塞新的绘图任务。 - internal bool HasPendingOriginJobs => Directory.Exists(jobsDirectory) + internal bool HasPendingJobs => Directory.Exists(jobsDirectory) && ReadRecoverableJobs().Any(item => item.Terminal is null); internal IReadOnlyList ReadRecoverableJobs() @@ -47,7 +42,9 @@ internal sealed class JobInboxStore(string jobsDirectory) if (!TryReadGuid(root, "job_id", out var jobId) || !TryReadGuid(root, "lease_id", out var leaseId) || !root.TryGetProperty("request_digest", out var digestValue) - || digestValue.GetString() is not { Length: 64 } requestDigest) + || digestValue.GetString() is not { Length: 64 } requestDigest + || !root.TryGetProperty("capability", out var capabilityValue) + || capabilityValue.GetString() is not { Length: > 0 } capability) { continue; } @@ -56,6 +53,7 @@ internal sealed class JobInboxStore(string jobsDirectory) jobId, leaseId, requestDigest, + capability, root.TryGetProperty("input_transfers", out var transfers) ? transfers.Clone() : null, ReadTerminal(Path.Combine(jobDirectory, "terminal.json")), @@ -192,7 +190,7 @@ internal sealed class JobInboxStore(string jobsDirectory) } } - internal JobOfferResult Accept(JsonElement payload) + internal JobOfferResult Accept(JsonElement payload, NodeAdapterRegistry adapters) { if (!TryReadGuid(payload, "job_id", out var jobId) || !TryReadGuid(payload, "lease_id", out var leaseId) @@ -200,7 +198,7 @@ internal sealed class JobInboxStore(string jobsDirectory) || digestValue.ValueKind != JsonValueKind.String || digestValue.GetString() is not { Length: 64 } requestDigest || !payload.TryGetProperty("capability", out var capabilityValue) - || capabilityValue.GetString() != "origin.plot@v2" + || capabilityValue.GetString() is not { Length: > 0 } capability || !payload.TryGetProperty("request", out var request) || request.ValueKind != JsonValueKind.Object || !payload.TryGetProperty("input_transfers", out var inputTransfers) @@ -208,7 +206,8 @@ internal sealed class JobInboxStore(string jobsDirectory) { return JobOfferResult.Reject("invalid_offer"); } - if (!IsValidRequest(request)) + var adapter = adapters.Find(capability); + if (adapter is null || !adapter.ValidateRequest(request)) { return JobOfferResult.Reject("unsupported_request"); } @@ -240,7 +239,7 @@ internal sealed class JobInboxStore(string jobsDirectory) job_id = jobId, lease_id = leaseId, request_digest = requestDigest, - capability = "origin.plot@v2", + capability, accepted_at = DateTimeOffset.UtcNow, request, input_transfers = inputTransfers, @@ -261,7 +260,7 @@ internal sealed class JobInboxStore(string jobsDirectory) job_id = jobId, lease_id = leaseId, request_digest = requestDigest, - capability = "origin.plot@v2", + capability, accepted_at = DateTimeOffset.UtcNow, request, input_transfers = inputTransfers, @@ -404,7 +403,8 @@ internal sealed class JobInboxStore(string jobsDirectory) } } - private static bool IsValidRequest(JsonElement request) + internal static bool IsValidOriginRequest( + JsonElement request, IReadOnlyList supportedPlotTypes) { if (!HasOnlyProperties(request, "schema_version", "inputs", "operation", "outputs") || !request.TryGetProperty("schema_version", out var schemaVersion) @@ -416,7 +416,7 @@ internal sealed class JobInboxStore(string jobsDirectory) || operation.ValueKind != JsonValueKind.Object || !HasOnlyProperties(operation, "plot") || !operation.TryGetProperty("plot", out var plot) - || !IsValidPlot(plot, inputs) + || !IsValidPlot(plot, inputs, supportedPlotTypes) || !request.TryGetProperty("outputs", out var outputs)) { return false; @@ -503,7 +503,8 @@ internal sealed class JobInboxStore(string jobsDirectory) return true; } - private static bool IsValidPlot(JsonElement plot, JsonElement inputs) + private static bool IsValidPlot( + JsonElement plot, JsonElement inputs, IReadOnlyList supportedPlotTypes) { if (plot.ValueKind != JsonValueKind.Object || !HasOnlyProperties( @@ -511,7 +512,7 @@ internal sealed class JobInboxStore(string jobsDirectory) "legend", "error_bars") || !plot.TryGetProperty("type", out var plotType) || plotType.GetString() is not { } plotTypeName - || !PlotTypes.Contains(plotTypeName) + || !supportedPlotTypes.Contains(plotTypeName, StringComparer.Ordinal) || plot.TryGetProperty("title", out var title) && (title.ValueKind != JsonValueKind.String || title.GetString()!.Length > 500) || !plot.TryGetProperty("series", out var series) @@ -750,6 +751,7 @@ internal sealed record RecoverableJob( Guid JobId, Guid LeaseId, string RequestDigest, + string Capability, JsonElement? InputTransfers, JsonElement? Terminal, bool UploadComplete); diff --git a/windows-node/Zcbot.WindowsNode/NodeAdapters.cs b/windows-node/Zcbot.WindowsNode/NodeAdapters.cs new file mode 100644 index 0000000..cb732a9 --- /dev/null +++ b/windows-node/Zcbot.WindowsNode/NodeAdapters.cs @@ -0,0 +1,93 @@ +using System.Text.Json; + +namespace Zcbot.WindowsNode; + +internal interface INodeAdapter +{ + string Capability { get; } + string DisplayName { get; } + string AdapterVersion { get; } + IReadOnlyList Features { get; } + bool HasActiveJobs { get; } + string RunningDetail { get; } + AdapterRuntimeStatus DetectRuntime(); + bool ValidateRequest(JsonElement request); + Task RunAsync(RecoverableJob job); + void Cancel(Guid jobId); +} + +internal sealed class OriginPlotNodeAdapter(JobInboxStore inbox) : INodeAdapter +{ + internal const string CurrentAdapterVersion = "0.4.0"; + internal const string ContractFilename = "origin.plot.v2.json"; + private static readonly NodeAdapterContract Contract = + NodeAdapterContract.Load(ContractFilename, CurrentAdapterVersion); + private readonly OriginWorkerRunner runner = new(inbox); + + internal static string CapabilityName => Contract.Capability; + + public string Capability => Contract.Capability; + public string DisplayName => Contract.DisplayName; + public string AdapterVersion => CurrentAdapterVersion; + public IReadOnlyList Features => Contract.Features; + public bool HasActiveJobs => runner.HasActiveJobs; + public string RunningDetail => "Origin 正在生成图形"; + public AdapterRuntimeStatus DetectRuntime() => OriginRuntimeProbe.Detect(); + public bool ValidateRequest(JsonElement request) => + JobInboxStore.IsValidOriginRequest(request, Features); + public Task RunAsync(RecoverableJob job) => runner.RunAsync(job); + public void Cancel(Guid jobId) => runner.Cancel(jobId); +} + +internal sealed record NodeAdapterContract( + string Capability, string DisplayName, IReadOnlyList Features) +{ + internal static NodeAdapterContract Load(string filename, string adapterVersion) + { + var path = Path.GetFullPath( + Path.Combine(AppContext.BaseDirectory, "software-contracts", filename)); + using var document = JsonDocument.Parse(File.ReadAllBytes(path)); + var root = document.RootElement; + var capability = root.GetProperty("capability").GetString() + ?? throw new InvalidDataException("Adapter contract capability is missing."); + var displayName = root.GetProperty("display_name").GetString() + ?? throw new InvalidDataException("Adapter contract display name is missing."); + var current = ParseVersion(adapterVersion); + var features = root.GetProperty("features").EnumerateObject() + .Where(item => ParseVersion(item.Value.GetString() ?? "0.0.0") <= current) + .Select(item => item.Name) + .ToArray(); + return new NodeAdapterContract(capability, displayName, features); + } + + private static Version ParseVersion(string value) => + Version.TryParse(value, out var version) ? version : new Version(0, 0, 0); +} + +internal sealed class NodeAdapterRegistry +{ + private readonly IReadOnlyDictionary adapters; + + internal NodeAdapterRegistry(IEnumerable values) + { + adapters = values.ToDictionary(item => item.Capability, StringComparer.Ordinal); + } + + internal static NodeAdapterRegistry CreateDefault(JobInboxStore inbox) => + new([new OriginPlotNodeAdapter(inbox)]); + + internal static IReadOnlyList InstalledContracts { get; } = + [ + NodeAdapterContract.Load( + OriginPlotNodeAdapter.ContractFilename, + OriginPlotNodeAdapter.CurrentAdapterVersion), + ]; + + internal static IReadOnlyList InstalledCapabilities { get; } = + InstalledContracts.Select(item => item.Capability).ToArray(); + + internal IReadOnlyCollection All => adapters.Values.ToArray(); + + internal INodeAdapter? Find(string capability) => + adapters.TryGetValue(capability, out var adapter) ? adapter : null; +} diff --git a/windows-node/Zcbot.WindowsNode/NodeConfigStore.cs b/windows-node/Zcbot.WindowsNode/NodeConfigStore.cs index 1fc13ea..0659522 100644 --- a/windows-node/Zcbot.WindowsNode/NodeConfigStore.cs +++ b/windows-node/Zcbot.WindowsNode/NodeConfigStore.cs @@ -57,7 +57,9 @@ internal sealed class NodeConfigStore(NodePaths paths) || string.IsNullOrWhiteSpace(stored.NodeName) || string.IsNullOrWhiteSpace(token) || stored.Capabilities.Count == 0 - || stored.Capabilities.Any(item => item != "origin.plot@v2")) + || stored.Capabilities.Any( + item => !NodeAdapterRegistry.InstalledCapabilities.Contains( + item, StringComparer.Ordinal))) { throw new NodeConfigurationException("Node configuration contains an invalid identity or capability."); } diff --git a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs index 837d94a..b19182f 100644 --- a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs +++ b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs @@ -14,7 +14,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? private readonly JobInboxStore jobInbox = new(NodePaths.ForCurrentMachine().JobsDirectory); private readonly JobInputDownloader inputDownloader = new( config, new JobInboxStore(NodePaths.ForCurrentMachine().JobsDirectory)); - private readonly OriginWorkerRunner workerRunner = new( + private readonly NodeAdapterRegistry adapters = NodeAdapterRegistry.CreateDefault( new JobInboxStore(NodePaths.ForCurrentMachine().JobsDirectory)); private readonly JobOutputUploader outputUploader = new(config); private readonly ConcurrentDictionary jobPipelines = new(); @@ -215,7 +215,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? if (type.GetString() == "job_offer" && document.RootElement.TryGetProperty("payload", out var payload)) { - var offerResult = jobInbox.Accept(payload); + var offerResult = jobInbox.Accept(payload, adapters); await SendAsync( socket, offerResult.Accepted ? "job_accept" : "job_reject", @@ -255,7 +255,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? && document.RootElement.TryGetProperty("payload", out var cancelPayload) && TryCancelJob(cancelPayload, out var cancelledJob)) { - workerRunner.Cancel(cancelledJob.JobId); + adapters.Find(cancelledJob.Capability)?.Cancel(cancelledJob.JobId); jobInbox.WriteTerminal( cancelledJob, "cancelled", "USER_CANCELLED", "Cancelled by user."); await SendAsync(socket, "job_terminal", new @@ -318,6 +318,8 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? private async Task RunJobPipelineAsync(ClientWebSocket socket, RecoverableJob job) { + var adapter = adapters.Find(job.Capability) + ?? throw new InvalidDataException($"No local adapter for {job.Capability}."); var recoveringOutputs = job.Terminal is JsonElement; try { @@ -347,7 +349,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? .Sum(item => item.GetProperty("size_bytes").GetInt64()), }, }); - jobInbox.WriteState(job, "software_running", 10, "Origin 正在生成图形"); + jobInbox.WriteState(job, "software_running", 10, adapter.RunningDetail); await TrySendAsync(socket, "job_state", new { job_id = job.JobId, @@ -357,13 +359,13 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? progress = 10, metrics = new { }, }); - await workerRunner.RunAsync(job); + await adapter.RunAsync(job); } var refreshed = jobInbox.ReadRecoverableJobs() .Single(item => item.JobId == job.JobId); if (refreshed.Terminal is not JsonElement terminal) { - throw new InvalidDataException("Origin worker did not create a terminal record."); + throw new InvalidDataException("Software adapter did not create a terminal record."); } if (terminal.GetProperty("status").GetString() != "succeeded") { @@ -482,6 +484,26 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? { var root = Path.GetPathRoot(Environment.SystemDirectory) ?? "C:\\"; var origin = OriginRuntimeProbe.Detect(); + var capabilityRuntime = adapters.All + .Where(item => config.Capabilities.Contains(item.Capability, StringComparer.Ordinal)) + .ToDictionary( + item => item.Capability, + item => + { + var runtime = item.DetectRuntime(); + var slots = runtime.Health == "ready" + && !jobInbox.HasPendingJobs + && !item.HasActiveJobs ? 1 : 0; + return new + { + adapter_version = item.AdapterVersion, + features = item.Features, + available_slots = slots, + health = runtime.Health, + detail = runtime.Detail, + }; + }); + var originAdapter = adapters.Find(OriginPlotNodeAdapter.CapabilityName); return new { install_id = config.InstallId, @@ -490,8 +512,9 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? os_version = RuntimeInformation.OSDescription, capabilities = config.Capabilities, available_slots = origin.Health == "ready" - && !jobInbox.HasPendingOriginJobs - && !workerRunner.HasActiveJobs ? 1 : 0, + && !jobInbox.HasPendingJobs + && !(originAdapter?.HasActiveJobs ?? false) ? 1 : 0, + capability_runtime = capabilityRuntime, disk_free_bytes = new DriveInfo(root).AvailableFreeSpace, desktop_session = Environment.UserInteractive, origin = new diff --git a/windows-node/Zcbot.WindowsNode/OriginRuntimeProbe.cs b/windows-node/Zcbot.WindowsNode/OriginRuntimeProbe.cs index 8bfb861..535f0c1 100644 --- a/windows-node/Zcbot.WindowsNode/OriginRuntimeProbe.cs +++ b/windows-node/Zcbot.WindowsNode/OriginRuntimeProbe.cs @@ -3,7 +3,7 @@ using System.Security; namespace Zcbot.WindowsNode; -internal sealed record OriginRuntimeStatus( +internal sealed record AdapterRuntimeStatus( string Software, string? SoftwareVersion, string AdapterVersion, @@ -12,12 +12,12 @@ internal sealed record OriginRuntimeStatus( internal static class OriginRuntimeProbe { - private static readonly Lazy Current = new(DetectCore); + private static readonly Lazy Current = new(DetectCore); private const string AutomationProgId = @"Origin.ApplicationSI\CLSID"; - internal static OriginRuntimeStatus Detect() => Current.Value; + internal static AdapterRuntimeStatus Detect() => Current.Value; - private static OriginRuntimeStatus DetectCore() + private static AdapterRuntimeStatus DetectCore() { try { @@ -53,8 +53,8 @@ internal static class OriginRuntimeProbe } } - private static OriginRuntimeStatus Status(string? version, string health, string detail) => - new("OriginPro", version, "0.4.0", health, detail); + private static AdapterRuntimeStatus Status(string? version, string health, string detail) => + new("OriginPro", version, OriginPlotNodeAdapter.CurrentAdapterVersion, health, detail); private static string? FindInstalledVersion() { diff --git a/windows-node/Zcbot.WindowsNode/Zcbot.WindowsNode.csproj b/windows-node/Zcbot.WindowsNode/Zcbot.WindowsNode.csproj index f8f88c2..61a1d91 100644 --- a/windows-node/Zcbot.WindowsNode/Zcbot.WindowsNode.csproj +++ b/windows-node/Zcbot.WindowsNode/Zcbot.WindowsNode.csproj @@ -12,6 +12,11 @@ 0.1.0 + + software-contracts\%(Filename)%(Extension) + PreserveNewest + PreserveNewest + origin-worker\worker.py PreserveNewest diff --git a/windows-node/package-windows-node.bat b/windows-node/package-windows-node.bat index b694f5a..893b3dc 100644 --- a/windows-node/package-windows-node.bat +++ b/windows-node/package-windows-node.bat @@ -65,6 +65,7 @@ for %%F in ( "install-windows-node.bat" "origin-worker\worker.py" "origin-worker\requirements.txt" + "software-contracts\origin.plot.v2.json" ) do ( if not exist "!PUBLISH_DIR!\%%~F" ( echo [ERR] Published package is incomplete: %%~F