fix(software): release slots after worker completion

CHANGELOG.md

@@ -11,6 +11,7 @@
 - 新增专业软件 Job 中心：Agent 可将当前对话目录内的数据登记为稳定输入，提交、查询和停止 Windows Node 上的受控软件任务；文件栏底部固定展示任务状态，点击可从右侧打开按开启时间倒序、滚动加载的跨对话任务列表，并可收到完成通知或回到原对话分析结果。
 - 专业软件任务完成后，输出文件会立即显示在当前对话的文件面板；任务中心收起时也不会再遮挡发送按钮。
 - 专业软件生成的正式文件会保存在当前对话的正确目录并记录来源；运行参数和溯源信息收纳到隐藏元数据目录，减少产物列表噪声。
+- 专业软件任务即使在完成回执时短暂断线，也不会持续占用 Windows Node；后续任务可继续执行，已完成结果会在后台自动确认。
 
 ## 0.65.2 — 2026-08-13
 

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/`，再将完整目录原子移动到 `<working_dir>/origin/<job_id>/`。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/`，再将完整目录原子移动到 `<working_dir>/origin/<job_id>/`。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`；终态写入仍由云端账本裁决。
 

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 输出错位的问题，正式输出原子发布到 `<working_dir>/origin/<job_id>/` 并以 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

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:

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, {

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<Guid, Task> jobPipelines", connection)
         self.assertIn("StartJobPipeline(socket, acceptedJob)", connection)

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
             )

windows-node/Zcbot.WindowsNode/JobInboxStore.cs

@@ -10,11 +10,10 @@ internal sealed class JobInboxStore(string jobsDirectory)
     ["line", "scatter", "line_scatter"];
     private static readonly HashSet<string> 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<RecoverableJob> ReadRecoverableJobs()
     {

windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs

@@ -465,7 +465,9 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
             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

windows-node/Zcbot.WindowsNode/OriginWorkerRunner.cs

@@ -11,6 +11,8 @@ internal sealed class OriginWorkerRunner(JobInboxStore inbox)
     private readonly ConcurrentDictionary<Guid, Task> active = new();
     private readonly ConcurrentDictionary<Guid, CancellationTokenSource> cancellations = new();
 
+    internal bool HasActiveJobs => !active.IsEmpty;
+
     internal Task RunAsync(RecoverableJob job) =>
         active.GetOrAdd(job.JobId, _ => RunOnceAsync(job, CancellationFor(job.JobId).Token));