diff --git a/CHANGELOG.md b/CHANGELOG.md index 4656868..ecd9fb9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ - 新增专业软件 Job 中心：Agent 可将当前对话目录内的数据登记为稳定输入，提交、查询和停止 Windows Node 上的受控软件任务；文件栏底部固定展示任务状态，点击可从右侧打开按开启时间倒序、滚动加载的跨对话任务列表，并可收到完成通知或回到原对话分析结果。 - 专业软件任务完成后，输出文件会立即显示在当前对话的文件面板；任务中心收起时也不会再遮挡发送按钮。 - 专业软件生成的正式文件会保存在当前对话的正确目录并记录来源；运行参数和溯源信息收纳到隐藏元数据目录，减少产物列表噪声。 +- 专业软件任务即使在完成回执时短暂断线，也不会持续占用 Windows Node；后续任务可继续执行，已完成结果会在后台自动确认。 ## 0.65.2 — 2026-08-13 diff --git a/DESIGN.md b/DESIGN.md index ca3fd70..4a7d4a1 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -470,7 +470,7 @@ Node 通过 `Authorization: Bearer` 与 `X-Node-Id` 建立 `/v1/software-nodes/c 第四阶段落地固定 Origin Worker：Node 仅从管理员安装的固定 Python 运行时启动随程序发布的 `worker.py`，参数只有本机 job 目录；请求不能指定脚本、解释器或文件路径。Worker 使用 `originpro` 生成 OPJU、PNG、SVG、PDF、plot spec 和 provenance，校验产物签名并原子写入终态；当前受控图形仅含 line、scatter、line_scatter 和双栏出版布局。进程内 pipeline 按 job 去重，并脱离单次 WebSocket 的取消令牌运行；连接中断只延迟状态/终态上报。Node 进程若在 Worker 启动后重启，则保守失败而不重复驱动 Origin，避免无法证明的双执行。 -第五阶段完成输出上传与发布：Node 只按固定 manifest ID 逐项流式 PUT，并携带 Node、lease、request digest 与内容摘要；云端重新绑定任务身份，不信 Node 提供的路径或媒体类型。文件先进入用户根下隐藏暂存区，固定文件名、单文件/总大小和 SHA-256 全部验证后，把 plot spec、provenance 整理进 `.meta/`，再将完整目录原子移动到 `/origin//`。PNG/SVG/PDF/OPJU 等正式输出登记平台 artifact UUID 和 `software_job_id`，`.meta/` 只落真实文件；成功状态返回 task-relative `output_dir`，Agent 以该目录为起点按需搜索。重复 PUT、complete 和重连均按摘要幂等；部分上传不可见，只有完整集合才能发布。 +第五阶段完成输出上传与发布：Node 只按固定 manifest ID 逐项流式 PUT，并携带 Node、lease、request digest 与内容摘要；云端重新绑定任务身份，不信 Node 提供的路径或媒体类型。文件先进入用户根下隐藏暂存区，固定文件名、单文件/总大小和 SHA-256 全部验证后，把 plot spec、provenance 整理进 `.meta/`，再将完整目录原子移动到 `/origin//`。PNG/SVG/PDF/OPJU 等正式输出登记平台 artifact UUID 和 `software_job_id`，`.meta/` 只落真实文件；成功状态返回 task-relative `output_dir`，Agent 以该目录为起点按需搜索。重复 PUT、complete 和重连均按摘要幂等；部分上传不可见，只有完整集合才能发布。Origin 执行槽与上传确认是两个正交状态：本地已有终态且固定 Worker 已退出时即释放软件执行槽，成功但尚无 `upload-complete.json` 的任务继续后台补传；若云端已经是 succeeded，重复 PUT/complete 必须按数据库持久化 manifest 校验并直接确认，不得按新版本目录规则重新发布旧 Job。 第六阶段增加用户级 Job 中心与 Agent typed tools。`software_capability_list` 只暴露固定能力及当前在线空闲节点数，`software_job_submit/status/cancel` 在构造时绑定当前 user/task，模型不能跨用户或跨对话指定归属。`software_job_submit` 的模型可见唯一入口为 `input_id + plot + output`；`register_artifact` 是普通文件获得输入身份的唯一入口，并明确返回 UUID。内部完整 `request` 形式仅保留执行层兼容，不进入工具 schema。右下角 Job 中心按用户聚合各对话任务，活动期短轮询、空闲期降频；终态变化通知用户，成功任务可回到原对话发起分析。取消采用协作协议：未派发任务直接终止，已派发任务先进入 `cancelling`，云端通过 WebSocket 发送并在心跳时重放 `job_cancel`，Node 杀死固定 Worker 进程树后回报 `cancelled`；终态写入仍由云端账本裁决。 diff --git a/PROGRESS.md b/PROGRESS.md index f657c6d..7b66b9a 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -22,6 +22,8 @@ ### 2026-08-14 +- **08-14 / Unreleased / 专业软件执行槽与上传恢复解耦**:修正服务端已完成但 Node 本地 `upload-complete.json` 未落盘时整节点长期上报 0 slot 的活性故障；Origin 终态且 Worker 退出后即释放执行槽，成功输出继续后台重传，服务端对 succeeded Job 的重复 PUT/complete 以已持久化 manifest 幂等确认并兼容旧目录布局。相关 43 项 unittest 与 .NET build 通过；只读核查生产任务状态，未写入生产 DB。 + - **08-14 / Unreleased / 专业软件输出目录归一 + artifact 来源**:修正 `tasks.working_dir` 被二次拼入 user root 导致 Job 输出错位的问题，正式输出原子发布到 `/origin//` 并以 nullable `software_job_id` 记录来源；plot spec/provenance 改入 `.meta/` 且不再登记 artifact。新增 0033 migration 和只认显式迁移库地址、默认 dry-run、冲突即停的存量目录修复脚本；相关 Python 63 项、Node 前端 26 项、编译、Alembic 单 head、Ruff 致命规则及 diff 检查通过，未连接或写入生产 DB。 ### 2026-08-13 diff --git a/core/software_jobs.py b/core/software_jobs.py index f90ac49..fa1e0c5 100644 --- a/core/software_jobs.py +++ b/core/software_jobs.py @@ -576,6 +576,60 @@ def validate_output_manifest(request: dict, manifest: object) -> list[dict]: return normalized +def replay_succeeded_outputs(context: dict, manifest: list[dict]) -> list[dict] | None: + """确认成功 Job 的 Node 重放，并返回云端已持久化的发布清单。 + + Node 可能在服务端完成发布后、写本地 ``upload-complete.json`` 前断线。 + 此时数据库中的 published manifest 是事实源；不能再按当前版本的目录规则 + 重做发布，否则跨版本布局调整会让已经成功的旧 Job 永久无法确认。 + """ + if context.get("status") != "succeeded": + return None + published = context.get("artifact_manifest") + if not isinstance(published, list): + raise SoftwareJobError("successful job artifact manifest is invalid") + by_source = { + item.get("source_artifact_id"): item + for item in published + if isinstance(item, dict) and isinstance(item.get("source_artifact_id"), str) + } + if len(by_source) != len(published) or len(by_source) != len(manifest): + raise SoftwareJobError("successful job artifact manifest does not match replay") + identity_fields = ("filename", "media_type", "size_bytes", "sha256") + for submitted in manifest: + stored = by_source.get(submitted["artifact_id"]) + if stored is None or any(stored.get(field) != submitted.get(field) for field in identity_fields): + raise SoftwareJobError("successful job artifact manifest does not match replay") + return [dict(item) for item in published] + + +def succeeded_output_upload_matches( + context: dict, + output_id: str, + *, + size_bytes: int, + digest: str, +) -> bool: + """判断单个重复 PUT 是否已包含在成功 Job 的持久化清单中。""" + if context.get("status") != "succeeded": + return False + published = context.get("artifact_manifest") + if not isinstance(published, list): + raise SoftwareJobError("successful job artifact manifest is invalid") + matches = [ + item + for item in published + if isinstance(item, dict) and item.get("source_artifact_id") == output_id + ] + if ( + len(matches) != 1 + or matches[0].get("size_bytes") != size_bytes + or matches[0].get("sha256") != digest + ): + raise SoftwareJobError("successful job output does not match replay") + return True + + def abandon_offer(node_id: UUID, payload: dict) -> None: """WebSocket 发送失败时只回滚仍属于该连接租约的 offer。""" try: diff --git a/tests/test_software_nodes.py b/tests/test_software_nodes.py index d85c195..4f7868a 100644 --- a/tests/test_software_nodes.py +++ b/tests/test_software_nodes.py @@ -18,8 +18,10 @@ from core.software_jobs import ( list_jobs, mark_node_jobs_disconnected, record_job_terminal, + replay_succeeded_outputs, request_job_cancel, respond_to_offer, + succeeded_output_upload_matches, update_job_state, validate_output_manifest, ) @@ -152,6 +154,62 @@ class SoftwareNodeMigrationTests(unittest.TestCase): class SoftwareJobProtocolTests(unittest.TestCase): + def test_succeeded_replay_uses_persisted_manifest_across_layout_versions(self) -> None: + content_digest = "a" * 64 + submitted = [{ + "artifact_id": "plot_spec", + "filename": "plot-spec.json", + "media_type": "application/json", + "size_bytes": 42, + "sha256": content_digest, + }] + persisted = [{ + **submitted[0], + "source_artifact_id": "plot_spec", + "artifact_id": str(uuid4()), + "path": "origin/old-job/plot-spec.json", + }] + context = {"status": "succeeded", "artifact_manifest": persisted} + + self.assertEqual(replay_succeeded_outputs(context, submitted), persisted) + self.assertTrue(succeeded_output_upload_matches( + context, + "plot_spec", + size_bytes=42, + digest=content_digest, + )) + + def test_succeeded_replay_rejects_changed_output(self) -> None: + context = { + "status": "succeeded", + "artifact_manifest": [{ + "source_artifact_id": "figure_png", + "filename": "figure.png", + "media_type": "image/png", + "size_bytes": 10, + "sha256": "a" * 64, + "artifact_id": str(uuid4()), + "path": "origin/job/figure.png", + }], + } + submitted = [{ + "artifact_id": "figure_png", + "filename": "figure.png", + "media_type": "image/png", + "size_bytes": 10, + "sha256": "b" * 64, + }] + + with self.assertRaisesRegex(Exception, "does not match replay"): + replay_succeeded_outputs(context, submitted) + with self.assertRaisesRegex(Exception, "does not match replay"): + succeeded_output_upload_matches( + context, + "figure_png", + size_bytes=10, + digest="b" * 64, + ) + def test_published_output_distinguishes_artifacts_from_metadata(self) -> None: job_id = uuid4() self.assertTrue(_published_output_is_valid(job_id, { diff --git a/tests/test_windows_node_source.py b/tests/test_windows_node_source.py index c0de4f1..0ac4773 100644 --- a/tests/test_windows_node_source.py +++ b/tests/test_windows_node_source.py @@ -143,9 +143,14 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertIn("RegistryHive.CurrentUser", probe) self.assertIn('new("OriginPro", version, "0.2.0", health, detail)', probe) self.assertIn( - 'available_slots = origin.Health == "ready" && !jobInbox.HasPendingJobs ? 1 : 0', + "&& !jobInbox.HasPendingOriginJobs", connection, ) + self.assertIn("&& !workerRunner.HasActiveJobs ? 1 : 0", connection) + self.assertIn( + "ReadRecoverableJobs().Any(item => item.Terminal is null)", + (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8"), + ) self.assertNotIn("CreateInstance", probe) self.assertNotIn("Process.Start", probe) for marker in ( @@ -171,7 +176,8 @@ class WindowsNodeSourceTests(unittest.TestCase): ) self.assertIn('offerResult.Accepted ? "job_accept" : "job_reject"', connection) self.assertIn("sendLock.WaitAsync", connection) - self.assertIn("!jobInbox.HasPendingJobs ? 1 : 0", connection) + self.assertIn("!jobInbox.HasPendingOriginJobs", connection) + self.assertIn("!workerRunner.HasActiveJobs ? 1 : 0", connection) self.assertIn("ReportRecoverableJobsAsync", connection) self.assertIn("ConcurrentDictionary jobPipelines", connection) self.assertIn("StartJobPipeline(socket, acceptedJob)", connection) diff --git a/web/routers/software_nodes.py b/web/routers/software_nodes.py index 90ce8cb..954a5e9 100644 --- a/web/routers/software_nodes.py +++ b/web/routers/software_nodes.py @@ -38,9 +38,11 @@ from core.software_jobs import ( offer_next_job, pending_node_cancellations, record_job_terminal, + replay_succeeded_outputs, request_job_cancel, respond_to_offer, software_job_output_path, + succeeded_output_upload_matches, update_job_state, validate_output_manifest, ) @@ -346,6 +348,16 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: 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): raise HTTPException(400, "output artifact digest is invalid") + try: + if succeeded_output_upload_matches( + context, + artifact_id, + size_bytes=x_content_length, + digest=x_content_sha256, + ): + return + except SoftwareJobError as exc: + raise HTTPException(409, str(exc)) from exc root = load_user_root(context["user_id"]) working_dir = _task_working_dir(root, context["working_dir"]) published = safe_join( @@ -354,7 +366,7 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: ) if published.is_file(): if published.stat().st_size == x_content_length and _hash_file(published) == x_content_sha256: - return None + return raise HTTPException(409, "published output conflicts with uploaded artifact") staging = safe_join(root, f".zcbot_software_job_staging/{job_id}") _reject_symlink_path(root, staging) @@ -362,12 +374,12 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: organized = staging / software_job_output_path(artifact_id) if organized.is_file(): if organized.stat().st_size == x_content_length and _hash_file(organized) == x_content_sha256: - return None + return raise HTTPException(409, "staged output conflicts with uploaded artifact") destination = staging / filename if destination.is_file(): if destination.stat().st_size == x_content_length and _hash_file(destination) == x_content_sha256: - return None + return raise HTTPException(409, "uploaded output conflicts with existing staging file") staged_total = sum(item.stat().st_size for item in staging.rglob("*") if item.is_file()) if staged_total + x_content_length > MAX_OUTPUT_TOTAL_BYTES: @@ -390,7 +402,7 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: os.replace(temporary, destination) finally: temporary.unlink(missing_ok=True) - return None + return @app.post("/v1/software-jobs/{job_id}/outputs/complete", tags=["software-nodes"]) async def complete_software_job_outputs( @@ -410,6 +422,9 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: raise HTTPException(400, "output completion body must be an object") try: manifest = validate_output_manifest(context["request"], body.get("artifact_manifest")) + replayed = replay_succeeded_outputs(context, manifest) + if replayed is not None: + return {"status": "succeeded", "artifact_manifest": replayed} published = await asyncio.to_thread( _publish_software_job_outputs, job_id, context, manifest ) diff --git a/windows-node/Zcbot.WindowsNode/JobInboxStore.cs b/windows-node/Zcbot.WindowsNode/JobInboxStore.cs index 87252ed..5a69959 100644 --- a/windows-node/Zcbot.WindowsNode/JobInboxStore.cs +++ b/windows-node/Zcbot.WindowsNode/JobInboxStore.cs @@ -10,11 +10,10 @@ internal sealed class JobInboxStore(string jobsDirectory) ["line", "scatter", "line_scatter"]; private static readonly HashSet OutputFormats = ["opju", "png", "svg", "pdf"]; - internal bool HasPendingJobs => Directory.Exists(jobsDirectory) - && ReadRecoverableJobs().Any(item => - item.Terminal is null - || item.Terminal.Value.GetProperty("status").GetString() == "succeeded" - && !item.UploadComplete); + // Origin 执行槽只由尚无终态的任务占用。成功但上传确认尚未落盘的任务会由 + // 心跳恢复管线继续重传；上传不使用 Origin，不能反向阻塞新的绘图任务。 + internal bool HasPendingOriginJobs => Directory.Exists(jobsDirectory) + && ReadRecoverableJobs().Any(item => item.Terminal is null); internal IReadOnlyList ReadRecoverableJobs() { diff --git a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs index cf68dd7..ec0e292 100644 --- a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs +++ b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs @@ -465,7 +465,9 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? node_version = Assembly.GetExecutingAssembly().GetName().Version?.ToString(3) ?? "0.1.0", os_version = RuntimeInformation.OSDescription, capabilities = config.Capabilities, - available_slots = origin.Health == "ready" && !jobInbox.HasPendingJobs ? 1 : 0, + available_slots = origin.Health == "ready" + && !jobInbox.HasPendingOriginJobs + && !workerRunner.HasActiveJobs ? 1 : 0, disk_free_bytes = new DriveInfo(root).AvailableFreeSpace, desktop_session = Environment.UserInteractive, origin = new diff --git a/windows-node/Zcbot.WindowsNode/OriginWorkerRunner.cs b/windows-node/Zcbot.WindowsNode/OriginWorkerRunner.cs index 75dcf2f..8162579 100644 --- a/windows-node/Zcbot.WindowsNode/OriginWorkerRunner.cs +++ b/windows-node/Zcbot.WindowsNode/OriginWorkerRunner.cs @@ -11,6 +11,8 @@ internal sealed class OriginWorkerRunner(JobInboxStore inbox) private readonly ConcurrentDictionary active = new(); private readonly ConcurrentDictionary cancellations = new(); + internal bool HasActiveJobs => !active.IsEmpty; + internal Task RunAsync(RecoverableJob job) => active.GetOrAdd(job.JobId, _ => RunOnceAsync(job, CancellationFor(job.JobId).Token));