refactor(software): introduce capability contracts

This commit is contained in:
caoqianming 2026-08-14 12:55:46 +08:00
parent b77db25440
commit 25616d12cc
25 changed files with 1135 additions and 520 deletions

View File

@ -462,6 +462,10 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB)
云端控制面使用独立的 `software_node_enrollments``software_nodes`,不复用用户外部系统连接。管理员创建的一次性注册码具有 128 bit 随机熵,数据库只保存 SHA-256 摘要;节点注册在行锁事务中校验有效期、预期名称和允许能力,成功后原子消费。每个节点获得独立高熵 Token数据库只保存 bcrypt 强哈希,明文仅在注册响应出现一次。 云端控制面使用独立的 `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 固定路由到单一控制面实例。 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 表达。 第二阶段已增加 `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 表达。

View File

@ -22,6 +22,8 @@
### 2026-08-14 ### 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 / 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。 - **08-14 / Unreleased / Windows Node 输出上传收敛与诊断**:恢复中的成功任务优先向云端重放完成确认,已发布结果不再重新打开可能被 Origin 占用的 OPJU文件共享冲突按 0.5/1/2/5 秒有界退避,本地完成标记使用可跨心跳复用的固定 pending 文件,只重试重命名以避开系统程序对每个新文件的重复扫描,并按 Job 写入带阶段、产物、重试次数和 HRESULT 的 1 MiB 轮转诊断日志。同步修正本机更新时间早于接收时间的展示边界;相关 45 项专项 unittest、.NET build 与 diff 检查通过,未写入生产 DB。

2
RUN.md
View File

@ -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`,支持 116 个输入、跨输入系列和多个显式输出,只创建固定 v2 schema 的持久任务,不会阻塞当前对话等待完成。成功状态提供 `output_dir`Agent可在该目录内搜索并分析正式输出的 artifact 带 `software_job_id`供结果卡和产物详情展示来源。Node 输出上传的逐任务诊断日志位于 `%ProgramData%\Zcbot\WindowsNode\jobs\<job-id>\logs\node-output-upload.log`;日志包含上传阶段、产物文件名、重试次数和 Windows `HRESULT`,单文件达到 1 MiB 后轮转一份 `.1`,不记录 Node Token 或认证请求头。 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`,支持 116 个输入、跨输入系列和多个显式输出,只创建固定 v2 schema 的持久任务,不会阻塞当前对话等待完成。成功状态提供 `output_dir`Agent可在该目录内搜索并分析正式输出的 artifact 带 `software_job_id`供结果卡和产物详情展示来源。Node 输出上传的逐任务诊断日志位于 `%ProgramData%\Zcbot\WindowsNode\jobs\<job-id>\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。 注册配置写入 `%ProgramData%\Zcbot\WindowsNode\node.json`Token 使用 DPAPI `LocalMachine` 加密ACL 仅允许注册账号和 `SYSTEM`。应始终用同一专用 Windows 账号执行统一安装器、注册并运行 Node。当前 MVP 以该账号的登录后计划任务启动,不安装 Windows Service。
直接双击 EXE 启动托盘 UI红点为未注册/身份失效,黄点为连接中,绿点为在线;双击托盘图标打开配置窗。原 CLI 注册入口继续保留,无 UI 模式使用 `Zcbot.WindowsNode.exe run --headless` 直接双击 EXE 启动托盘 UI红点为未注册/身份失效,黄点为连接中,绿点为在线;双击托盘图标打开配置窗。原 CLI 注册入口继续保留,无 UI 模式使用 `Zcbot.WindowsNode.exe run --headless`

297
core/software_contracts.py Normal file
View File

@ -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

View File

@ -5,244 +5,45 @@ from __future__ import annotations
import json import json
import re import re
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from hashlib import sha256
from uuid import UUID, uuid4 from uuid import UUID, uuid4
from sqlalchemy import and_, desc, or_, select from sqlalchemy import and_, desc, or_, select
from sqlalchemy.exc import IntegrityError 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.engine import session_scope
from core.storage.models import Artifact, SoftwareJob, SoftwareNode, Task from core.storage.models import Artifact, SoftwareJob, SoftwareNode, Task
OFFER_SECONDS = 60 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_ARTIFACT_BYTES = 256 * 1024 * 1024
MAX_OUTPUT_TOTAL_BYTES = 512 * 1024 * 1024
class SoftwareJobError(Exception): class SoftwareJobError(Exception):
pass pass
MAX_OUTPUT_TOTAL_BYTES = 512 * 1024 * 1024 def software_job_output_path(capability: str, output_id: str) -> str:
OUTPUT_ARTIFACTS = { """返回 capability Job 输出目录内的契约路径。"""
"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: try:
UUID(str(input_spec.get("artifact_id") or "")) return get_contract(capability).output_spec(output_id).relative_path
except ValueError as exc: except SoftwareContractError as exc:
raise SoftwareJobError("inputs[].artifact_id must be an artifact UUID") from exc raise SoftwareJobError(str(exc)) 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 _canonical_request(capability: str, request: dict) -> tuple[dict, str]: def _canonical_request(capability: str, request: dict) -> tuple[dict, str]:
validator = REQUEST_VALIDATORS.get(capability) try:
if validator is None: return get_contract(capability).normalize_request(request)
raise SoftwareJobError("unsupported capability") except SoftwareContractError as exc:
return validator(request) raise SoftwareJobError(str(exc)) from exc
def _job_dict(row: SoftwareJob) -> dict: def _job_dict(row: SoftwareJob) -> dict:
contract = get_contract(row.capability)
return { return {
"job_id": str(row.job_id), "job_id": str(row.job_id),
"task_id": str(row.task_id), "task_id": str(row.task_id),
@ -255,7 +56,7 @@ def _job_dict(row: SoftwareJob) -> dict:
"metrics": row.metrics, "metrics": row.metrics,
"error": row.error, "error": row.error,
"artifact_manifest": row.artifact_manifest, "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, "created_at": row.created_at.isoformat() if row.created_at else None,
"started_at": row.started_at.isoformat() if row.started_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, "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: def _request_summary(job: SoftwareJob) -> dict:
plot = (job.request.get("operation") or {}).get("plot") or {} return get_contract(job.capability).summarize(job.request)
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)],
}
def request_job_cancel(user_id: UUID, job_id: UUID) -> tuple[dict, dict | None]: 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() key = idempotency_key.strip()
if not key or len(key) > 200: if not key or len(key) > 200:
raise SoftwareJobError("idempotency_key must contain 1 to 200 characters") raise SoftwareJobError("idempotency_key must contain 1 to 200 characters")
if capability not in SUPPORTED_CAPABILITIES: try:
raise SoftwareJobError("unsupported capability") contract = get_contract(capability)
normalized, digest = _canonical_request(capability, request) normalized, digest = contract.normalize_request(request)
except SoftwareContractError as exc:
raise SoftwareJobError(str(exc)) from exc
with session_scope() as session: with session_scope() as session:
task = session.execute( task = session.execute(
select(Task.task_id).where(Task.task_id == task_id, Task.user_id == user_id) select(Task.task_id).where(Task.task_id == task_id, Task.user_id == user_id)
).first() ).first()
if task is None: if task is None:
raise SoftwareJobError("task not found") 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( artifacts = session.execute(
select(Artifact).where( select(Artifact).where(
Artifact.artifact_id.in_(set(artifact_ids)), 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} artifacts_by_id = {artifact.artifact_id: artifact for artifact in artifacts}
input_manifest: list[dict] = [] input_manifest: list[dict] = []
total_input_bytes = 0 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) artifact = artifacts_by_id.get(artifact_id)
if artifact is None: if artifact is None:
raise SoftwareJobError("input artifact not found") raise SoftwareJobError("input artifact not found")
@ -429,12 +226,12 @@ def create_job(
"." + artifact.current_path.rsplit(".", 1)[-1].lower() "." + artifact.current_path.rsplit(".", 1)[-1].lower()
if "." in artifact.current_path else "" 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") raise SoftwareJobError("input artifact type is not supported")
if ( if (
artifact.size_bytes is None artifact.size_bytes is None
or artifact.size_bytes < 0 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 not artifact.content_sha256
or len(artifact.content_sha256) != 64 or len(artifact.content_sha256) != 64
): ):
@ -450,7 +247,7 @@ def create_job(
if binding.get("selector") is not None: if binding.get("selector") is not None:
item["selector"] = binding["selector"] item["selector"] = binding["selector"]
input_manifest.append(item) 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") raise SoftwareJobError("job inputs exceed the total size limit")
existing = session.execute( existing = session.execute(
select(SoftwareJob).where( 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: def offer_next_job(node_ids: set[UUID]) -> dict | None:
"""从当前进程实际在线的节点中选择一个,为最早 queued job 创建短租约""" """选择最早可执行的 JobNode 组合,避免跨 capability 队首阻塞"""
if not node_ids: if not node_ids:
return None return None
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
@ -531,39 +328,69 @@ def offer_next_job(node_ids: set[UUID]) -> dict | None:
item.node_id = None item.node_id = None
item.lease_id = None item.lease_id = None
item.lease_expires_at = 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( busy_node_ids = set(
session.execute( session.execute(
select(SoftwareJob.node_id).where( select(SoftwareJob.node_id).where(
SoftwareJob.node_id.is_not(None), SoftwareJob.node_id.is_not(None),
SoftwareJob.status.in_({"offered", "dispatched", "running"}), SoftwareJob.status.in_(
{"offered", "dispatched", "running", "disconnected", "cancelling"}
),
) )
).scalars() ).scalars()
) )
nodes = session.execute( nodes = list(session.execute(
select(SoftwareNode) select(SoftwareNode)
.where(SoftwareNode.node_id.in_(node_ids), SoftwareNode.status == "online") .where(SoftwareNode.node_id.in_(node_ids), SoftwareNode.status == "online")
.order_by(SoftwareNode.last_seen_at.desc()) .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() ).scalars()
selected: tuple[SoftwareJob, SoftwareNode] | None = None
for candidate in queued:
try:
contract = get_contract(candidate.capability)
except SoftwareContractError:
continue
node = next( node = next(
( (
item item for item in available_nodes
for item in nodes if candidate.capability in (item.capabilities or [])
if item.node_id not in busy_node_ids and node_supports_request(
and job.capability in item.capabilities contract, candidate.request, item.runtime or {}
and int((item.runtime or {}).get("available_slots") or 0) > 0 )
), ),
None, None,
) )
if node is 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 return None
lease_id = uuid4() lease_id = uuid4()
expires_at = now + timedelta(seconds=OFFER_SECONDS) 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: if artifact is None:
return None return None
return { return {
"capability": job.capability,
"user_id": job.user_id, "user_id": job.user_id,
"current_path": artifact.current_path, "current_path": artifact.current_path,
**manifest, **manifest,
@ -647,6 +475,7 @@ def get_job_output_context(node_id: UUID, job_id: UUID, lease_id: UUID, digest:
): ):
return None return None
return { return {
"capability": job.capability,
"user_id": job.user_id, "user_id": job.user_id,
"task_id": job.task_id, "task_id": job.task_id,
"working_dir": working_dir, "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): if not isinstance(manifest, list):
raise SoftwareJobError("job artifact manifest must be a list") raise SoftwareJobError("job artifact manifest must be a list")
expected_ids = {"plot_spec", "provenance"} try:
expected_ids.update( contract = get_contract(capability)
item.get("key") expected = contract.expected_outputs(request)
for item in request.get("outputs", []) except SoftwareContractError as exc:
if isinstance(item, dict) and isinstance(item.get("key"), str) raise SoftwareJobError(str(exc)) from exc
) expected_ids = set(expected)
if len(manifest) != len(expected_ids): if len(manifest) != len(expected_ids):
raise SoftwareJobError("job artifact manifest is incomplete") raise SoftwareJobError("job artifact manifest is incomplete")
normalized: list[dict] = [] normalized: list[dict] = []
@ -678,10 +507,10 @@ def validate_output_manifest(request: dict, manifest: object) -> list[dict]:
local_id = raw.get("artifact_id") local_id = raw.get("artifact_id")
if local_id not in expected_ids or local_id in seen: if local_id not in expected_ids or local_id in seen:
raise SoftwareJobError("job artifact manifest identity is invalid") raise SoftwareJobError("job artifact manifest identity is invalid")
filename, media_type, _ = OUTPUT_ARTIFACTS[local_id] spec = expected[local_id]
size = raw.get("size_bytes") size = raw.get("size_bytes")
digest = raw.get("sha256") 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") 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: 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") 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) _assert_job_message(job, node_id, lease_id, digest)
if terminal_status == "succeeded": if terminal_status == "succeeded":
expected = validate_output_manifest( expected = validate_output_manifest(
job.capability,
job.request, job.request,
[ [
{ {
@ -879,7 +709,7 @@ def record_job_terminal(node_id: UUID, payload: dict) -> None:
], ],
) )
if len(expected) != len(manifest) or any( 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 for item in manifest
): ):
raise SoftwareJobError("successful job artifacts have not been published") raise SoftwareJobError("successful job artifacts have not been published")
@ -905,19 +735,25 @@ def _is_uuid(value: str) -> bool:
return True 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): if not isinstance(item, dict):
return False return False
output_id = item.get("source_artifact_id") output_id = item.get("source_artifact_id")
if not isinstance(output_id, str): if not isinstance(output_id, str):
return False return False
if output_id not in OUTPUT_ARTIFACTS: try:
contract = get_contract(capability)
spec = contract.output_spec(output_id)
except SoftwareContractError:
return False 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: if item.get("path") != expected_path:
return False return False
artifact_id = item.get("artifact_id") artifact_id = item.get("artifact_id")
if output_id in SOFTWARE_JOB_METADATA_IDS: if not spec.publish:
return artifact_id is None return artifact_id is None
return isinstance(artifact_id, str) and _is_uuid(artifact_id) return isinstance(artifact_id, str) and _is_uuid(artifact_id)

View File

@ -10,10 +10,10 @@ from uuid import UUID, uuid4
import bcrypt import bcrypt
from sqlalchemy import select from sqlalchemy import select
from core.software_contracts import DEFAULT_CAPABILITIES, SUPPORTED_CAPABILITIES
from core.storage.engine import session_scope from core.storage.engine import session_scope
from core.storage.models import SoftwareNode, SoftwareNodeEnrollment from core.storage.models import SoftwareNode, SoftwareNodeEnrollment
SUPPORTED_CAPABILITIES = frozenset({"origin.plot@v2"})
MAX_ENROLLMENT_FAILURES = 5 MAX_ENROLLMENT_FAILURES = 5
@ -44,7 +44,7 @@ def create_enrollment(
capabilities: list[str] | None = None, capabilities: list[str] | None = None,
ttl_seconds: int = 600, ttl_seconds: int = 600,
) -> dict: ) -> 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): if not allowed or any(item not in SUPPORTED_CAPABILITIES for item in allowed):
raise SoftwareNodeError("unsupported capability") raise SoftwareNodeError("unsupported capability")
if not 60 <= ttl_seconds <= 3600: if not 60 <= ttl_seconds <= 3600:

View File

@ -55,6 +55,7 @@ fakeredis>=2.23
# §7 B 阶段: Storage 落 PG # §7 B 阶段: Storage 落 PG
sqlalchemy>=2.0.0 sqlalchemy>=2.0.0
jsonschema>=4.23.0 # 专业软件 capability contract 的 Draft 2020-12 校验
psycopg[binary]>=3.1.0 psycopg[binary]>=3.1.0
alembic>=1.13.0 alembic>=1.13.0

View File

@ -21,10 +21,8 @@ from sqlalchemy.orm.attributes import flag_modified
ROOT = Path(__file__).resolve().parents[1] ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT))
from core.software_jobs import ( from core.software_contracts import DEFAULT_CAPABILITIES, get_contract
SOFTWARE_JOB_METADATA_IDS, from core.software_jobs import software_job_output_path
software_job_output_path,
)
from core.storage.models import Artifact, SoftwareJob, Task from core.storage.models import Artifact, SoftwareJob, Task
from core.paths import from_db_path 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: 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() working_dir = from_db_path(task.working_dir).resolve()
user_root = _user_root(working_dir, job.user_id) 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( legacy_dir = _within(
user_root, 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() legacy_exists = legacy_dir.is_dir()
output_exists = output_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]] = [] metadata_moves: list[tuple[Path, Path]] = []
for item in job.artifact_manifest or []: for item in job.artifact_manifest or []:
output_id = str((item or {}).get("source_artifact_id") 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 continue
source = source_dir / str((item or {}).get("filename") or "") 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(): if source == destination or not source.exists():
continue continue
if destination.exists(): 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: 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 artifact_count = 0
manifest = [dict(item) for item in (job.artifact_manifest or [])] manifest = [dict(item) for item in (job.artifact_manifest or [])]
task_prefix = plan.working_dir.relative_to(plan.user_root).as_posix() 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 "") output_id = str(item.get("source_artifact_id") or "")
if not output_id: if not output_id:
continue 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 item["path"] = path
raw_artifact_id = item.get("artifact_id") raw_artifact_id = item.get("artifact_id")
if not raw_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: if artifact is None or artifact.user_id != job.user_id:
raise RuntimeError(f"job {job.job_id}: artifact {artifact_id} is missing") raise RuntimeError(f"job {job.job_id}: artifact {artifact_id} is missing")
artifact.current_path = f"{task_prefix}/{path}" 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.software_job_id = job.job_id
artifact_count += 1 artifact_count += 1
job.artifact_manifest = manifest job.artifact_manifest = manifest

View File

@ -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"}
}
}
}
}
}

View File

@ -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()

View File

@ -3,7 +3,7 @@ from __future__ import annotations
import importlib import importlib
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import AsyncMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from uuid import uuid4 from uuid import uuid4
from alembic.migration import MigrationContext from alembic.migration import MigrationContext
@ -19,6 +19,7 @@ from core.software_jobs import (
list_jobs, list_jobs,
mark_node_jobs_disconnected, mark_node_jobs_disconnected,
record_job_terminal, record_job_terminal,
offer_next_job,
replay_succeeded_outputs, replay_succeeded_outputs,
request_job_cancel, request_job_cancel,
respond_to_offer, respond_to_offer,
@ -213,17 +214,17 @@ class SoftwareJobProtocolTests(unittest.TestCase):
def test_published_output_distinguishes_artifacts_from_metadata(self) -> None: def test_published_output_distinguishes_artifacts_from_metadata(self) -> None:
job_id = uuid4() 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", "source_artifact_id": "figure_png",
"artifact_id": str(uuid4()), "artifact_id": str(uuid4()),
"path": f"origin/{job_id}/figure.png", "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", "source_artifact_id": "plot_spec",
"artifact_id": None, "artifact_id": None,
"path": f"origin/{job_id}/.meta/plot-spec.json", "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"], "source_artifact_id": ["plot_spec"],
"artifact_id": None, "artifact_id": None,
"path": f"origin/{job_id}/.meta/plot-spec.json", "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) normalized, digest = _canonical_request("origin.plot@v2", request)
self.assertEqual(normalized, request) self.assertEqual(normalized, request)
self.assertEqual(len(digest), 64) 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"}) _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": { _canonical_request("origin.plot@v2", {**request, "operation": {"plot": {
**request["operation"]["plot"], "script": "anything" **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"}]}) _canonical_request("origin.plot@v2", {**request, "inputs": [{"key": "first", "artifact_id": "C:\\data.csv"}]})
def test_origin_request_rejects_unimplemented_plot_semantics(self) -> None: 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}}], "outputs": [{"key": "figure_png", "type": "figure", "format": "png", "options": {"dpi": 300}}],
} }
base_plot = request["operation"]["plot"] base_plot = request["operation"]["plot"]
for case_plot, message in ( for case_plot in (
({**base_plot, "template": "custom"}, "unsupported origin plot template"), {**base_plot, "template": "custom"},
({**base_plot, "x_axis": {"scale": "log10"}}, "invalid x_axis"), {**base_plot, "x_axis": {"scale": "log10"}},
({**base_plot, "legend": {"enabled": False}}, "invalid plot.legend"), {**base_plot, "legend": {"enabled": False}},
({**base_plot, "series": [base_plot["series"][0], base_plot["series"][0]]}, "plot series must be unique"), {**base_plot, "series": [base_plot["series"][0], base_plot["series"][0]]},
):
with self.subTest(plot=case_plot), self.assertRaisesRegex(
Exception, "invalid origin.plot@v2 request"
): ):
with self.subTest(message=message), self.assertRaisesRegex(Exception, message):
_canonical_request("origin.plot@v2", {**request, "operation": {"plot": case_plot}}) _canonical_request("origin.plot@v2", {**request, "operation": {"plot": case_plot}})
with self.assertRaisesRegex(Exception, "unknown input"): # 跨字段引用完整性属于本机 adapter 的独立二次校验Core 只执行共享结构契约。
_canonical_request("origin.plot@v2", {**request, "operation": {"plot": { normalized, _ = _canonical_request("origin.plot@v2", {
**request,
"operation": {"plot": {
**base_plot, **base_plot,
"series": [{"input": "missing", "x": "time", "y": "a"}], "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( _canonical_request(
"origin.plot@v2", "origin.plot@v2",
{**request, "outputs": [{"key": "project", "type": "figure", "format": "png"}]} {**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": "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}, {"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"): 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"): with self.assertRaisesRegex(Exception, "metadata"):
validate_output_manifest( validate_output_manifest(
request, "origin.plot@v2", request,
[{**manifest[0], "filename": "anything.opju"}, *manifest[1:]], [{**manifest[0], "filename": "anything.opju"}, *manifest[1:]],
) )
@ -440,8 +449,49 @@ class SoftwareJobProtocolTests(unittest.TestCase):
source = ( source = (
Path(__file__).resolve().parents[1] / "core" / "software_jobs.py" Path(__file__).resolve().parents[1] / "core" / "software_jobs.py"
).read_text(encoding="utf-8") ).read_text(encoding="utf-8")
self.assertIn('{"offered", "dispatched", "running"}', source) self.assertIn('"disconnected", "cancelling"', source)
self.assertIn("item.node_id not in busy_node_ids", 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: def test_input_download_rechecks_file_digest(self) -> None:
source = ( source = (
@ -450,7 +500,7 @@ class SoftwareJobProtocolTests(unittest.TestCase):
).read_text(encoding="utf-8") ).read_text(encoding="utf-8")
self.assertIn("digest = sha256()", source) self.assertIn("digest = sha256()", source)
self.assertIn('digest.hexdigest() != item["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) self.assertIn("artifact_id not in requested_ids", source)
@patch("core.software_jobs.session_scope") @patch("core.software_jobs.session_scope")

View File

@ -26,7 +26,7 @@ class WindowsNodeSourceTests(unittest.TestCase):
'SetRequestHeader("X-Node-Id"', 'SetRequestHeader("X-Node-Id"',
"DataProtectionScope.LocalMachine", "DataProtectionScope.LocalMachine",
"SetAccessRuleProtection(isProtected: true", "SetAccessRuleProtection(isProtected: true",
'"origin.plot@v2"', '"origin.plot.v2.json"',
"NotifyIcon", "NotifyIcon",
"ConfigurationForm", "ConfigurationForm",
"TrayIconFactory.Create", "TrayIconFactory.Create",
@ -70,8 +70,8 @@ class WindowsNodeSourceTests(unittest.TestCase):
self.assertIn("CreateCard", form) self.assertIn("CreateCard", form)
self.assertIn("注册并连接", form) self.assertIn("注册并连接", form)
self.assertIn("ContentWidth = 760", form) self.assertIn("ContentWidth = 760", form)
self.assertIn("Origin 绘图", form) self.assertIn("NodeAdapterRegistry.InstalledContracts", form)
self.assertIn('CreateCapabilityRow("Origin 绘图", "origin.plot@v2")', form) self.assertIn("CreateCapabilityRow(contract.DisplayName, contract.Capability)", form)
self.assertIn('CreateButton("立即重连", 112, primary: true)', form) self.assertIn('CreateButton("立即重连", 112, primary: true)', form)
self.assertIn("ReconnectRequested?.Invoke()", form) self.assertIn("ReconnectRequested?.Invoke()", form)
self.assertIn("registrationCard.Visible = !registered", 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") project = (PROJECT / "Zcbot.WindowsNode.csproj").read_text(encoding="utf-8")
self.assertIn("..\\install-windows-node.bat", project) self.assertIn("..\\install-windows-node.bat", project)
self.assertIn("..\\origin-worker\\requirements.txt", 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-windows-node.ps1").exists())
self.assertFalse((ROOT / "install-origin-runtime.ps1").exists()) self.assertFalse((ROOT / "install-origin-runtime.ps1").exists())
self.assertFalse((ROOT / "install-startup.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('if /I "%~1"=="--self-contained"', script)
self.assertIn("dotnet publish", script) self.assertIn("dotnet publish", script)
self.assertIn('"install-windows-node.bat"', 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("tar.exe -a -c -f", script)
self.assertIn("certutil.exe -hashfile", script) self.assertIn("certutil.exe -hashfile", script)
self.assertFalse((ROOT / "package-windows-node.ps1").exists()) 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('AutomationProgId = @"Origin.ApplicationSI\\CLSID"', probe)
self.assertIn("RegistryHive.LocalMachine", probe) self.assertIn("RegistryHive.LocalMachine", probe)
self.assertIn("RegistryHive.CurrentUser", probe) self.assertIn("RegistryHive.CurrentUser", probe)
self.assertIn('new("OriginPro", version, "0.4.0", health, detail)', probe) self.assertIn("OriginPlotNodeAdapter.CurrentAdapterVersion", probe)
self.assertIn( self.assertIn(
"&& !jobInbox.HasPendingOriginJobs", "&& !jobInbox.HasPendingJobs",
connection, connection,
) )
self.assertIn("&& !workerRunner.HasActiveJobs ? 1 : 0", connection) self.assertIn("originAdapter?.HasActiveJobs", connection)
self.assertIn( self.assertIn(
"ReadRecoverableJobs().Any(item => item.Terminal is null)", "ReadRecoverableJobs().Any(item => item.Terminal is null)",
(PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8"), (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: def test_job_offer_is_persisted_before_acceptance(self) -> None:
inbox = (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8") inbox = (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8")
connection = (PROJECT / "NodeConnectionLoop.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('root.TryGetProperty("input_transfers"', inbox)
self.assertIn('"input", key, filename', inbox) self.assertIn('"input", key, filename', inbox)
self.assertIn("PlotTypes.Contains", inbox) self.assertIn("supportedPlotTypes.Contains", inbox)
self.assertIn("IsValidOutputs", inbox) self.assertIn("IsValidOutputs", inbox)
self.assertIn('("figure", "png") => "figure_png"', inbox) self.assertIn('("figure", "png") => "figure_png"', inbox)
self.assertIn("FileOptions.WriteThrough", 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('offerResult.Accepted ? "job_accept" : "job_reject"', connection)
self.assertIn("sendLock.WaitAsync", connection) self.assertIn("sendLock.WaitAsync", connection)
self.assertIn("!jobInbox.HasPendingOriginJobs", connection) self.assertIn("!jobInbox.HasPendingJobs", connection)
self.assertIn("!workerRunner.HasActiveJobs ? 1 : 0", connection) self.assertIn("capability_runtime = capabilityRuntime", connection)
self.assertIn("adapter.RunAsync(job)", connection)
self.assertIn("ReportRecoverableJobsAsync", connection) self.assertIn("ReportRecoverableJobsAsync", connection)
self.assertIn("ConcurrentDictionary<Guid, Task> jobPipelines", connection) self.assertIn("ConcurrentDictionary<Guid, Task> jobPipelines", connection)
self.assertIn("StartJobPipeline(socket, acceptedJob)", connection) self.assertIn("StartJobPipeline(socket, acceptedJob)", connection)

View File

@ -4,6 +4,12 @@ from __future__ import annotations
import json import json
from uuid import UUID, uuid4 from uuid import UUID, uuid4
from core.software_contracts import (
CONTRACTS,
SUPPORTED_CAPABILITIES,
get_contract,
node_available_slots,
)
from core.software_jobs import ( from core.software_jobs import (
SoftwareJobError, SoftwareJobError,
create_job, create_job,
@ -11,11 +17,21 @@ from core.software_jobs import (
list_jobs, list_jobs,
request_job_cancel, request_job_cancel,
) )
from core.software_nodes import SUPPORTED_CAPABILITIES, list_nodes from core.software_nodes import list_nodes
from .base import Tool 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): class _SoftwareJobTool(Tool):
def __init__(self, user_id: UUID, task_id: UUID, **kwargs) -> None: def __init__(self, user_id: UUID, task_id: UUID, **kwargs) -> None:
super().__init__(**kwargs) super().__init__(**kwargs)
@ -32,13 +48,13 @@ class SoftwareCapabilityListTool(_SoftwareJobTool):
nodes = list_nodes() nodes = list_nodes()
items = [{ items = [{
"capability": item, "capability": item,
"display_name": "Origin 科研绘图" if item == "origin.plot@v2" else item, "display_name": get_contract(item).display_name,
"available_nodes": sum( "available_nodes": sum(
1 1
for node in nodes for node in nodes
if node["status"] == "online" if node["status"] == "online"
and item in (node.get("capabilities") or []) 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)] } for item in sorted(SUPPORTED_CAPABILITIES)]
return json.dumps({"capabilities": items}, ensure_ascii=False) return json.dumps({"capabilities": items}, ensure_ascii=False)
@ -47,134 +63,17 @@ class SoftwareCapabilityListTool(_SoftwareJobTool):
class SoftwareJobSubmitTool(_SoftwareJobTool): class SoftwareJobSubmitTool(_SoftwareJobTool):
name = "software_job_submit" name = "software_job_submit"
description = ( description = (
"Submit an Origin 2D, error-bar, contour, 3D, ternary, or heatmap plot job using " "Submit a managed professional-software job using registered artifacts and a "
"one or more registered CSV, XLSX, or JSON artifacts. " "capability contract. Call register_artifact first for workspace files. Return "
"Call register_artifact first for each workspace file. " "immediately with job_id; do not poll continuously or wait for completion."
"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 = { parameters = {
"type": "object", "type": "object",
"properties": { "properties": {
"capability": {"type": "string", "enum": sorted(SUPPORTED_CAPABILITIES)}, "capability": {"type": "string", "enum": sorted(SUPPORTED_CAPABILITIES)},
"inputs": { "inputs": _contract_property_schema("inputs"),
"type": "array", "operation": _contract_property_schema("operation"),
"minItems": 1, "outputs": _contract_property_schema("outputs"),
"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."
),
},
"idempotency_key": { "idempotency_key": {
"type": "string", "type": "string",
"description": "Stable unique key for this exact submission; omit to generate one.", "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 = {**item, "artifact_id": artifact_id}
canonical_inputs.append(canonical) canonical_inputs.append(canonical)
normalized_request = { normalized_request = {
"schema_version": 2, "schema_version": get_contract(capability).request_schema[
"properties"
]["schema_version"]["const"],
"inputs": canonical_inputs, "inputs": canonical_inputs,
"operation": operation, "operation": operation,
"outputs": outputs, "outputs": outputs,

View File

@ -22,11 +22,14 @@ from fastapi.responses import FileResponse
from core.artifact_lifecycle import register_published_artifacts from core.artifact_lifecycle import register_published_artifacts
from core.paths import from_db_path from core.paths import from_db_path
from core.software_contracts import (
DEFAULT_CAPABILITIES,
SoftwareContractError,
get_contract,
)
from core.software_jobs import ( from core.software_jobs import (
MAX_OUTPUT_ARTIFACT_BYTES, MAX_OUTPUT_ARTIFACT_BYTES,
MAX_OUTPUT_TOTAL_BYTES, MAX_OUTPUT_TOTAL_BYTES,
OUTPUT_ARTIFACTS,
SOFTWARE_JOB_METADATA_IDS,
SoftwareJobError, SoftwareJobError,
abandon_offer, abandon_offer,
create_job, create_job,
@ -176,20 +179,28 @@ def _task_working_dir(root: Path, stored: str) -> Path:
return working_dir 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 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 return flat if flat.is_file() else organized
def _organize_staged_metadata(staging: Path, manifest: list[dict]) -> None: def _organize_staged_metadata(
metadata = [item for item in manifest if item["artifact_id"] in SOFTWARE_JOB_METADATA_IDS] 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: if not metadata:
return return
(staging / ".meta").mkdir(exist_ok=True) (staging / ".meta").mkdir(exist_ok=True)
for item in metadata: for item in metadata:
source = staging / item["filename"] 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 not source.is_file():
if destination.is_file(): if destination.is_file():
continue 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]: 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"]) root = load_user_root(context["user_id"])
working_dir = _task_working_dir(root, context["working_dir"]) working_dir = _task_working_dir(root, context["working_dir"])
staging = safe_join(root, f".zcbot_software_job_staging/{job_id}") 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()) destination = safe_join(working_dir, relative_output.as_posix())
source = staging if staging.is_dir() else destination source = staging if staging.is_dir() else destination
_reject_symlink_path(root, source) _reject_symlink_path(root, source)
_reject_symlink_path(root, destination) _reject_symlink_path(root, destination)
for item in manifest: for item in manifest:
path = ( path = (
_staged_output_path(staging, item["artifact_id"], item["filename"]) _staged_output_path(
staging, capability, item["artifact_id"], item["filename"]
)
if source == staging if source == staging
else destination / software_job_output_path(item["artifact_id"]) else destination / software_job_output_path(capability, item["artifact_id"])
) )
if ( if (
not path.is_file() 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']}") raise SoftwareJobError(f"uploaded artifact is missing or invalid: {item['artifact_id']}")
if source == staging: if source == staging:
_organize_staged_metadata(staging, manifest) _organize_staged_metadata(staging, capability, manifest)
destination.parent.mkdir(parents=True, exist_ok=True) destination.parent.mkdir(parents=True, exist_ok=True)
if destination.exists(): if destination.exists():
raise SoftwareJobError("software job output destination already exists unexpectedly") 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: except OSError:
pass pass
refs = tuple({ 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"], "label": item["filename"],
"media_type": item["media_type"], "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( published_refs = register_published_artifacts(
user_id=context["user_id"], user_id=context["user_id"],
task_id=context["task_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"], "source_artifact_id": item["artifact_id"],
"artifact_id": ( "artifact_id": (
refs_by_path.get( 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") ).get("artifact_id")
), ),
"path": ( "path": (
relative_output / software_job_output_path(item["artifact_id"]) relative_output / software_job_output_path(capability, item["artifact_id"])
).as_posix(), ).as_posix(),
} }
for item in manifest for item in manifest
@ -338,17 +358,15 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None:
_authenticate_output_request, _authenticate_output_request,
job_id, authorization, x_node_id, x_lease_id, x_request_digest, job_id, authorization, x_node_id, x_lease_id, x_request_digest,
) )
metadata = OUTPUT_ARTIFACTS.get(artifact_id) try:
if metadata is None: contract = get_contract(context["capability"])
raise HTTPException(400, "unsupported output artifact identity") output_spec = contract.output_spec(artifact_id)
filename, _, _ = metadata requested_ids = set(contract.expected_outputs(context["request"]))
requested_ids = { except SoftwareContractError as exc:
item.get("key") raise HTTPException(400, str(exc)) from exc
for item in context["request"].get("outputs", [])
if isinstance(item, dict)
} | SOFTWARE_JOB_METADATA_IDS
if artifact_id not in requested_ids: 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: if not 1 <= x_content_length <= MAX_OUTPUT_ARTIFACT_BYTES:
raise HTTPException(400, "output artifact size is invalid") 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): 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"]) working_dir = _task_working_dir(root, context["working_dir"])
published = safe_join( published = safe_join(
working_dir, 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.is_file():
if published.stat().st_size == x_content_length and _hash_file(published) == x_content_sha256: 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}") staging = safe_join(root, f".zcbot_software_job_staging/{job_id}")
_reject_symlink_path(root, staging) _reject_symlink_path(root, staging)
staging.mkdir(parents=True, exist_ok=True) 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.is_file():
if organized.stat().st_size == x_content_length and _hash_file(organized) == x_content_sha256: if organized.stat().st_size == x_content_length and _hash_file(organized) == x_content_sha256:
return return
@ -426,7 +447,9 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None:
if not isinstance(body, dict): if not isinstance(body, dict):
raise HTTPException(400, "output completion body must be an object") raise HTTPException(400, "output completion body must be an object")
try: 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) replayed = replay_succeeded_outputs(context, manifest)
if replayed is not None: if replayed is not None:
return {"status": "succeeded", "artifact_manifest": replayed} return {"status": "succeeded", "artifact_manifest": replayed}

View File

@ -6,6 +6,8 @@ from uuid import UUID
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from core.software_contracts import DEFAULT_CAPABILITIES
class TaskCreateRequest(BaseModel): class TaskCreateRequest(BaseModel):
# name 缺失/空值 + working_dir 有值 → 后端以“新对话”占位并自动生成标题。 # name 缺失/空值 + working_dir 有值 → 后端以“新对话”占位并自动生成标题。
@ -118,7 +120,7 @@ class ExternalSystemCredentialsRequest(BaseModel):
class SoftwareEnrollmentCreateRequest(BaseModel): class SoftwareEnrollmentCreateRequest(BaseModel):
expected_name: str = "" 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 ttl_seconds: int = 600
@ -137,5 +139,5 @@ class SoftwareNodeDisableRequest(BaseModel):
class SoftwareJobCreateRequest(BaseModel): class SoftwareJobCreateRequest(BaseModel):
idempotency_key: str idempotency_key: str
capability: str = "origin.plot@v2" capability: str = DEFAULT_CAPABILITIES[0]
request: dict = Field(default_factory=dict) request: dict = Field(default_factory=dict)

View File

@ -2,7 +2,7 @@
内网 MVP 的 Windows 执行节点,目标运行环境为 Windows 11 Enterprise + .NET 10 SDK 10.0.303。仓库根目录 `global.json` 固定 SDK patch客户端只使用 .NET Windows Desktop Framework不依赖第三方 NuGet 包。 内网 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/<key>/<filename>`,校验大小与 SHA-256 后原子保存。固定 Origin Worker 独立于单次 WebSocket 执行,断线不终止已启动绘图。成功产物按 manifest 逐项流式上传并由云端复核大小与 SHA-256全部完成后原子发布到任务工作目录的 `origin/<job_id>/`中断后按本地 `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 在仓库根目录执行一条命令生成可分发 ZIP

View File

@ -29,6 +29,7 @@ internal sealed class ConfigurationForm : Form
private readonly Label jobDetail = CreateBodyLabel(); private readonly Label jobDetail = CreateBodyLabel();
private readonly DataGridView jobGrid = CreateJobGrid(); private readonly DataGridView jobGrid = CreateJobGrid();
private readonly JobInboxStore jobInbox = new(NodePaths.ForCurrentMachine().JobsDirectory); private readonly JobInboxStore jobInbox = new(NodePaths.ForCurrentMachine().JobsDirectory);
private readonly NodeAdapterRegistry adapters;
private readonly System.Windows.Forms.Timer jobRefreshTimer = new() { Interval = 1000 }; private readonly System.Windows.Forms.Timer jobRefreshTimer = new() { Interval = 1000 };
private readonly TableLayoutPanel registrationCard; private readonly TableLayoutPanel registrationCard;
private bool changingStartup; private bool changingStartup;
@ -39,6 +40,7 @@ internal sealed class ConfigurationForm : Form
internal ConfigurationForm() internal ConfigurationForm()
{ {
adapters = NodeAdapterRegistry.CreateDefault(jobInbox);
Text = "zcbot Windows Node"; Text = "zcbot Windows Node";
AutoScaleMode = AutoScaleMode.Dpi; AutoScaleMode = AutoScaleMode.Dpi;
ClientSize = new Size(840, 680); ClientSize = new Size(840, 680);
@ -97,7 +99,10 @@ internal sealed class ConfigurationForm : Form
statusCard.Controls.Add(detail); statusCard.Controls.Add(detail);
statusCard.Controls.Add(identity); statusCard.Controls.Add(identity);
statusCard.Controls.Add(CreateDivider()); 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); statusCard.Controls.Add(capabilitySummary);
var resetActions = CreateActions(); var resetActions = CreateActions();
resetActions.Controls.Add(reconnect); resetActions.Controls.Add(reconnect);
@ -183,7 +188,7 @@ internal sealed class ConfigurationForm : Form
: $"节点:{config.NodeName}\nNode ID{config.NodeId}\n服务{config.ServerUrl}"; : $"节点:{config.NodeName}\nNode ID{config.NodeId}\n服务{config.ServerUrl}";
capabilitySummary.Text = config is null capabilitySummary.Text = config is null
? "注册后启用" ? "注册后启用"
: FormatOriginStatus(OriginRuntimeProbe.Detect()); : string.Join("\n", adapters.All.Select(FormatAdapterStatus));
var registered = config is not null; var registered = config is not null;
registrationCard.Visible = !registered; 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}"; : $"版本 {runtime.SoftwareVersion}";
var state = origin.Health == "ready" ? "可用" : "不可用"; var state = runtime.Health == "ready" ? "可用" : "不可用";
return $"{state} · {version}\n{origin.Detail}"; return $"{adapter.DisplayName}{state} · {version}\n{runtime.Detail}";
} }
private void RefreshJobs() private void RefreshJobs()

View File

@ -6,8 +6,6 @@ namespace Zcbot.WindowsNode;
internal static class EnrollmentClient internal static class EnrollmentClient
{ {
private static readonly string[] Capabilities = ["origin.plot@v2"];
internal static async Task EnrollAsync( internal static async Task EnrollAsync(
EnrollOptions options, NodeConfigStore store, CancellationToken cancellationToken) EnrollOptions options, NodeConfigStore store, CancellationToken cancellationToken)
{ {
@ -24,7 +22,7 @@ internal static class EnrollmentClient
installId, installId,
Assembly.GetExecutingAssembly().GetName().Version?.ToString(3) ?? "0.1.0", Assembly.GetExecutingAssembly().GetName().Version?.ToString(3) ?? "0.1.0",
RuntimeInformation.OSDescription, RuntimeInformation.OSDescription,
Capabilities); NodeAdapterRegistry.InstalledCapabilities);
using var client = new HttpClient { BaseAddress = options.ServerUrl, Timeout = TimeSpan.FromSeconds(30) }; using var client = new HttpClient { BaseAddress = options.ServerUrl, Timeout = TimeSpan.FromSeconds(30) };
using var response = await client.PostAsJsonAsync( using var response = await client.PostAsJsonAsync(
@ -44,7 +42,8 @@ internal static class EnrollmentClient
store.Save(new NodeConfig( store.Save(new NodeConfig(
options.ServerUrl, enrolled.NodeId, installId, options.NodeName, 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) => private static string Limit(string value, int maxLength) =>

View File

@ -6,11 +6,6 @@ namespace Zcbot.WindowsNode;
internal sealed class JobInboxStore(string jobsDirectory) internal sealed class JobInboxStore(string jobsDirectory)
{ {
private static readonly JsonSerializerOptions JsonOptions = new() { WriteIndented = true }; private static readonly JsonSerializerOptions JsonOptions = new() { WriteIndented = true };
private static readonly HashSet<string> PlotTypes =
[
"line", "scatter", "line_scatter", "column", "bar", "grouped_column",
"y_error", "contour", "surface_3d", "ternary", "heatmap",
];
private static readonly HashSet<string> XyzPlotTypes = private static readonly HashSet<string> XyzPlotTypes =
["contour", "surface_3d", "ternary", "heatmap"]; ["contour", "surface_3d", "ternary", "heatmap"];
private static readonly HashSet<string> JobStages = private static readonly HashSet<string> JobStages =
@ -27,7 +22,7 @@ internal sealed class JobInboxStore(string jobsDirectory)
// Origin 执行槽只由尚无终态的任务占用。成功但上传确认尚未落盘的任务会由 // Origin 执行槽只由尚无终态的任务占用。成功但上传确认尚未落盘的任务会由
// 心跳恢复管线继续重传;上传不使用 Origin不能反向阻塞新的绘图任务。 // 心跳恢复管线继续重传;上传不使用 Origin不能反向阻塞新的绘图任务。
internal bool HasPendingOriginJobs => Directory.Exists(jobsDirectory) internal bool HasPendingJobs => Directory.Exists(jobsDirectory)
&& ReadRecoverableJobs().Any(item => item.Terminal is null); && ReadRecoverableJobs().Any(item => item.Terminal is null);
internal IReadOnlyList<RecoverableJob> ReadRecoverableJobs() internal IReadOnlyList<RecoverableJob> ReadRecoverableJobs()
@ -47,7 +42,9 @@ internal sealed class JobInboxStore(string jobsDirectory)
if (!TryReadGuid(root, "job_id", out var jobId) if (!TryReadGuid(root, "job_id", out var jobId)
|| !TryReadGuid(root, "lease_id", out var leaseId) || !TryReadGuid(root, "lease_id", out var leaseId)
|| !root.TryGetProperty("request_digest", out var digestValue) || !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; continue;
} }
@ -56,6 +53,7 @@ internal sealed class JobInboxStore(string jobsDirectory)
jobId, jobId,
leaseId, leaseId,
requestDigest, requestDigest,
capability,
root.TryGetProperty("input_transfers", out var transfers) root.TryGetProperty("input_transfers", out var transfers)
? transfers.Clone() : null, ? transfers.Clone() : null,
ReadTerminal(Path.Combine(jobDirectory, "terminal.json")), 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) if (!TryReadGuid(payload, "job_id", out var jobId)
|| !TryReadGuid(payload, "lease_id", out var leaseId) || !TryReadGuid(payload, "lease_id", out var leaseId)
@ -200,7 +198,7 @@ internal sealed class JobInboxStore(string jobsDirectory)
|| digestValue.ValueKind != JsonValueKind.String || digestValue.ValueKind != JsonValueKind.String
|| digestValue.GetString() is not { Length: 64 } requestDigest || digestValue.GetString() is not { Length: 64 } requestDigest
|| !payload.TryGetProperty("capability", out var capabilityValue) || !payload.TryGetProperty("capability", out var capabilityValue)
|| capabilityValue.GetString() != "origin.plot@v2" || capabilityValue.GetString() is not { Length: > 0 } capability
|| !payload.TryGetProperty("request", out var request) || !payload.TryGetProperty("request", out var request)
|| request.ValueKind != JsonValueKind.Object || request.ValueKind != JsonValueKind.Object
|| !payload.TryGetProperty("input_transfers", out var inputTransfers) || !payload.TryGetProperty("input_transfers", out var inputTransfers)
@ -208,7 +206,8 @@ internal sealed class JobInboxStore(string jobsDirectory)
{ {
return JobOfferResult.Reject("invalid_offer"); 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"); return JobOfferResult.Reject("unsupported_request");
} }
@ -240,7 +239,7 @@ internal sealed class JobInboxStore(string jobsDirectory)
job_id = jobId, job_id = jobId,
lease_id = leaseId, lease_id = leaseId,
request_digest = requestDigest, request_digest = requestDigest,
capability = "origin.plot@v2", capability,
accepted_at = DateTimeOffset.UtcNow, accepted_at = DateTimeOffset.UtcNow,
request, request,
input_transfers = inputTransfers, input_transfers = inputTransfers,
@ -261,7 +260,7 @@ internal sealed class JobInboxStore(string jobsDirectory)
job_id = jobId, job_id = jobId,
lease_id = leaseId, lease_id = leaseId,
request_digest = requestDigest, request_digest = requestDigest,
capability = "origin.plot@v2", capability,
accepted_at = DateTimeOffset.UtcNow, accepted_at = DateTimeOffset.UtcNow,
request, request,
input_transfers = inputTransfers, 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<string> supportedPlotTypes)
{ {
if (!HasOnlyProperties(request, "schema_version", "inputs", "operation", "outputs") if (!HasOnlyProperties(request, "schema_version", "inputs", "operation", "outputs")
|| !request.TryGetProperty("schema_version", out var schemaVersion) || !request.TryGetProperty("schema_version", out var schemaVersion)
@ -416,7 +416,7 @@ internal sealed class JobInboxStore(string jobsDirectory)
|| operation.ValueKind != JsonValueKind.Object || operation.ValueKind != JsonValueKind.Object
|| !HasOnlyProperties(operation, "plot") || !HasOnlyProperties(operation, "plot")
|| !operation.TryGetProperty("plot", out var plot) || !operation.TryGetProperty("plot", out var plot)
|| !IsValidPlot(plot, inputs) || !IsValidPlot(plot, inputs, supportedPlotTypes)
|| !request.TryGetProperty("outputs", out var outputs)) || !request.TryGetProperty("outputs", out var outputs))
{ {
return false; return false;
@ -503,7 +503,8 @@ internal sealed class JobInboxStore(string jobsDirectory)
return true; return true;
} }
private static bool IsValidPlot(JsonElement plot, JsonElement inputs) private static bool IsValidPlot(
JsonElement plot, JsonElement inputs, IReadOnlyList<string> supportedPlotTypes)
{ {
if (plot.ValueKind != JsonValueKind.Object if (plot.ValueKind != JsonValueKind.Object
|| !HasOnlyProperties( || !HasOnlyProperties(
@ -511,7 +512,7 @@ internal sealed class JobInboxStore(string jobsDirectory)
"legend", "error_bars") "legend", "error_bars")
|| !plot.TryGetProperty("type", out var plotType) || !plot.TryGetProperty("type", out var plotType)
|| plotType.GetString() is not { } plotTypeName || plotType.GetString() is not { } plotTypeName
|| !PlotTypes.Contains(plotTypeName) || !supportedPlotTypes.Contains(plotTypeName, StringComparer.Ordinal)
|| plot.TryGetProperty("title", out var title) || plot.TryGetProperty("title", out var title)
&& (title.ValueKind != JsonValueKind.String || title.GetString()!.Length > 500) && (title.ValueKind != JsonValueKind.String || title.GetString()!.Length > 500)
|| !plot.TryGetProperty("series", out var series) || !plot.TryGetProperty("series", out var series)
@ -750,6 +751,7 @@ internal sealed record RecoverableJob(
Guid JobId, Guid JobId,
Guid LeaseId, Guid LeaseId,
string RequestDigest, string RequestDigest,
string Capability,
JsonElement? InputTransfers, JsonElement? InputTransfers,
JsonElement? Terminal, JsonElement? Terminal,
bool UploadComplete); bool UploadComplete);

View File

@ -0,0 +1,93 @@
using System.Text.Json;
namespace Zcbot.WindowsNode;
internal interface INodeAdapter
{
string Capability { get; }
string DisplayName { get; }
string AdapterVersion { get; }
IReadOnlyList<string> 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<string> 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<string> 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<string, INodeAdapter> adapters;
internal NodeAdapterRegistry(IEnumerable<INodeAdapter> values)
{
adapters = values.ToDictionary(item => item.Capability, StringComparer.Ordinal);
}
internal static NodeAdapterRegistry CreateDefault(JobInboxStore inbox) =>
new([new OriginPlotNodeAdapter(inbox)]);
internal static IReadOnlyList<NodeAdapterContract> InstalledContracts { get; } =
[
NodeAdapterContract.Load(
OriginPlotNodeAdapter.ContractFilename,
OriginPlotNodeAdapter.CurrentAdapterVersion),
];
internal static IReadOnlyList<string> InstalledCapabilities { get; } =
InstalledContracts.Select(item => item.Capability).ToArray();
internal IReadOnlyCollection<INodeAdapter> All => adapters.Values.ToArray();
internal INodeAdapter? Find(string capability) =>
adapters.TryGetValue(capability, out var adapter) ? adapter : null;
}

View File

@ -57,7 +57,9 @@ internal sealed class NodeConfigStore(NodePaths paths)
|| string.IsNullOrWhiteSpace(stored.NodeName) || string.IsNullOrWhiteSpace(stored.NodeName)
|| string.IsNullOrWhiteSpace(token) || string.IsNullOrWhiteSpace(token)
|| stored.Capabilities.Count == 0 || 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."); throw new NodeConfigurationException("Node configuration contains an invalid identity or capability.");
} }

View File

@ -14,7 +14,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
private readonly JobInboxStore jobInbox = new(NodePaths.ForCurrentMachine().JobsDirectory); private readonly JobInboxStore jobInbox = new(NodePaths.ForCurrentMachine().JobsDirectory);
private readonly JobInputDownloader inputDownloader = new( private readonly JobInputDownloader inputDownloader = new(
config, new JobInboxStore(NodePaths.ForCurrentMachine().JobsDirectory)); config, new JobInboxStore(NodePaths.ForCurrentMachine().JobsDirectory));
private readonly OriginWorkerRunner workerRunner = new( private readonly NodeAdapterRegistry adapters = NodeAdapterRegistry.CreateDefault(
new JobInboxStore(NodePaths.ForCurrentMachine().JobsDirectory)); new JobInboxStore(NodePaths.ForCurrentMachine().JobsDirectory));
private readonly JobOutputUploader outputUploader = new(config); private readonly JobOutputUploader outputUploader = new(config);
private readonly ConcurrentDictionary<Guid, Task> jobPipelines = new(); private readonly ConcurrentDictionary<Guid, Task> jobPipelines = new();
@ -215,7 +215,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
if (type.GetString() == "job_offer" if (type.GetString() == "job_offer"
&& document.RootElement.TryGetProperty("payload", out var payload)) && document.RootElement.TryGetProperty("payload", out var payload))
{ {
var offerResult = jobInbox.Accept(payload); var offerResult = jobInbox.Accept(payload, adapters);
await SendAsync( await SendAsync(
socket, socket,
offerResult.Accepted ? "job_accept" : "job_reject", offerResult.Accepted ? "job_accept" : "job_reject",
@ -255,7 +255,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
&& document.RootElement.TryGetProperty("payload", out var cancelPayload) && document.RootElement.TryGetProperty("payload", out var cancelPayload)
&& TryCancelJob(cancelPayload, out var cancelledJob)) && TryCancelJob(cancelPayload, out var cancelledJob))
{ {
workerRunner.Cancel(cancelledJob.JobId); adapters.Find(cancelledJob.Capability)?.Cancel(cancelledJob.JobId);
jobInbox.WriteTerminal( jobInbox.WriteTerminal(
cancelledJob, "cancelled", "USER_CANCELLED", "Cancelled by user."); cancelledJob, "cancelled", "USER_CANCELLED", "Cancelled by user.");
await SendAsync(socket, "job_terminal", new await SendAsync(socket, "job_terminal", new
@ -318,6 +318,8 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
private async Task RunJobPipelineAsync(ClientWebSocket socket, RecoverableJob job) 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; var recoveringOutputs = job.Terminal is JsonElement;
try try
{ {
@ -347,7 +349,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
.Sum(item => item.GetProperty("size_bytes").GetInt64()), .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 await TrySendAsync(socket, "job_state", new
{ {
job_id = job.JobId, job_id = job.JobId,
@ -357,13 +359,13 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
progress = 10, progress = 10,
metrics = new { }, metrics = new { },
}); });
await workerRunner.RunAsync(job); await adapter.RunAsync(job);
} }
var refreshed = jobInbox.ReadRecoverableJobs() var refreshed = jobInbox.ReadRecoverableJobs()
.Single(item => item.JobId == job.JobId); .Single(item => item.JobId == job.JobId);
if (refreshed.Terminal is not JsonElement terminal) 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") if (terminal.GetProperty("status").GetString() != "succeeded")
{ {
@ -482,6 +484,26 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
{ {
var root = Path.GetPathRoot(Environment.SystemDirectory) ?? "C:\\"; var root = Path.GetPathRoot(Environment.SystemDirectory) ?? "C:\\";
var origin = OriginRuntimeProbe.Detect(); 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 return new
{ {
install_id = config.InstallId, install_id = config.InstallId,
@ -490,8 +512,9 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
os_version = RuntimeInformation.OSDescription, os_version = RuntimeInformation.OSDescription,
capabilities = config.Capabilities, capabilities = config.Capabilities,
available_slots = origin.Health == "ready" available_slots = origin.Health == "ready"
&& !jobInbox.HasPendingOriginJobs && !jobInbox.HasPendingJobs
&& !workerRunner.HasActiveJobs ? 1 : 0, && !(originAdapter?.HasActiveJobs ?? false) ? 1 : 0,
capability_runtime = capabilityRuntime,
disk_free_bytes = new DriveInfo(root).AvailableFreeSpace, disk_free_bytes = new DriveInfo(root).AvailableFreeSpace,
desktop_session = Environment.UserInteractive, desktop_session = Environment.UserInteractive,
origin = new origin = new

View File

@ -3,7 +3,7 @@ using System.Security;
namespace Zcbot.WindowsNode; namespace Zcbot.WindowsNode;
internal sealed record OriginRuntimeStatus( internal sealed record AdapterRuntimeStatus(
string Software, string Software,
string? SoftwareVersion, string? SoftwareVersion,
string AdapterVersion, string AdapterVersion,
@ -12,12 +12,12 @@ internal sealed record OriginRuntimeStatus(
internal static class OriginRuntimeProbe internal static class OriginRuntimeProbe
{ {
private static readonly Lazy<OriginRuntimeStatus> Current = new(DetectCore); private static readonly Lazy<AdapterRuntimeStatus> Current = new(DetectCore);
private const string AutomationProgId = @"Origin.ApplicationSI\CLSID"; 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 try
{ {
@ -53,8 +53,8 @@ internal static class OriginRuntimeProbe
} }
} }
private static OriginRuntimeStatus Status(string? version, string health, string detail) => private static AdapterRuntimeStatus Status(string? version, string health, string detail) =>
new("OriginPro", version, "0.4.0", health, detail); new("OriginPro", version, OriginPlotNodeAdapter.CurrentAdapterVersion, health, detail);
private static string? FindInstalledVersion() private static string? FindInstalledVersion()
{ {

View File

@ -12,6 +12,11 @@
<Version>0.1.0</Version> <Version>0.1.0</Version>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<Content Include="..\..\software-contracts\*.json">
<Link>software-contracts\%(Filename)%(Extension)</Link>
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
<CopyToPublishDirectory>PreserveNewest</CopyToPublishDirectory>
</Content>
<Content Include="..\origin-worker\worker.py"> <Content Include="..\origin-worker\worker.py">
<Link>origin-worker\worker.py</Link> <Link>origin-worker\worker.py</Link>
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory> <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>

View File

@ -65,6 +65,7 @@ for %%F in (
"install-windows-node.bat" "install-windows-node.bat"
"origin-worker\worker.py" "origin-worker\worker.py"
"origin-worker\requirements.txt" "origin-worker\requirements.txt"
"software-contracts\origin.plot.v2.json"
) do ( ) do (
if not exist "!PUBLISH_DIR!\%%~F" ( if not exist "!PUBLISH_DIR!\%%~F" (
echo [ERR] Published package is incomplete: %%~F echo [ERR] Published package is incomplete: %%~F