diff --git a/CHANGELOG.md b/CHANGELOG.md index 60c8513..8ca2103 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ - 专业软件生成的正式文件会保存在当前对话的正确目录并记录来源；运行参数和溯源信息收纳到隐藏元数据目录，减少产物列表噪声。 - 专业软件任务即使在完成回执时短暂断线，也不会持续占用 Windows Node；后续任务可继续执行，已完成结果会在后台自动确认。 - Windows Node 窗口新增本机任务列表，可查看任务标题、输入文件、执行阶段、进度、时间、Job ID 和失败详情；任务状态会在本机持久保存，断线或重启后仍可查看。 +- Windows Node 会自动重试被 Origin 或系统程序短暂占用的输出文件；云端已确认的结果可在重启后直接恢复本机完成状态，并为每个任务保留不含凭据的上传诊断日志。 - Origin 绘图任务支持同时绑定多个 CSV、XLSX 或 JSON 输入、从不同输入中选择数据系列，并显式选择需要生成的工程文件和多种图形格式。 ## 0.65.2 — 2026-08-13 diff --git a/PROGRESS.md b/PROGRESS.md index b844f61..b091d4f 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -22,6 +22,8 @@ ### 2026-08-14 +- **08-14 / Unreleased / Windows Node 输出上传收敛与诊断**:恢复中的成功任务优先向云端重放完成确认，已发布结果不再重新打开可能被 Origin 占用的 OPJU；文件共享冲突按 0.5/1/2/5 秒有界退避，本地完成标记独立重试，并按 Job 写入带阶段、产物、重试次数和 HRESULT 的 1 MiB 轮转诊断日志。同步修正本机更新时间早于接收时间的展示边界；相关 45 项专项 unittest、.NET build 与 diff 检查通过，未写入生产 DB。 + - **08-14 / Unreleased / Windows Node 本机任务监控 + Origin 多输入多输出 v2**:配置窗口新增只读本机任务列表与详情，每秒从 `%ProgramData%/Zcbot/WindowsNode/jobs` 合并 request/state/terminal/upload-complete，展示标题、全部输入、阶段、执行耗时、时间、Job ID 与错误，重启后可恢复且不查询云端未派发 Job。按用户明确授权直接切换到 `origin.plot@v2`：通用 `inputs[] + operation + outputs[]` 支持 1–16 个输入、跨输入系列和多个显式产物，Node 下载到 keyed 目录，Origin Worker 建立多工作表后合并绘图；本地阶段统一为 `downloading_inputs/software_running`，不兼容 v1 请求或旧本地任务。相关专项 71 项 unittest、Python/JavaScript 语法、Ruff 致命规则、.NET build 与 diff 检查通过；完整测试 593 项中非数据库测试通过，3 个数据库集成模块因显式测试库未迁移、缺少 `users` 表而未通过，未连接或写入生产 DB。 - **08-14 / Unreleased / 专业软件执行槽与上传恢复解耦**:修正服务端已完成但 Node 本地 `upload-complete.json` 未落盘时整节点长期上报 0 slot 的活性故障；Origin 终态且 Worker 退出后即释放执行槽，成功输出继续后台重传，服务端对 succeeded Job 的重复 PUT/complete 以已持久化 manifest 幂等确认并兼容旧目录布局。相关 43 项 unittest 与 .NET build 通过；只读核查生产任务状态，未写入生产 DB。 diff --git a/RUN.md b/RUN.md index 548a137..53084a5 100644 --- a/RUN.md +++ b/RUN.md @@ -1100,7 +1100,7 @@ install-windows-node.bat 若 Python 未加入 PATH，可把绝对路径作为第一个参数，例如 `install-windows-node.bat "C:\Python312\python.exe"`。默认解释器为 `%ProgramData%\Zcbot\WindowsNode\runtimes\origin\Scripts\python.exe`。如需使用其他受管解释器，设置机器级 `ZCBOT_ORIGIN_PYTHON` 为绝对 `python.exe` 路径后重启 Node。`node.json`、可恢复任务和 runtime 集中保存在 `%ProgramData%\Zcbot\WindowsNode\`，不会因替换程序目录而丢失。运行时固定依赖见发布目录的 `origin-worker/requirements.txt`；任务请求无权选择解释器、脚本或路径。当前 Worker 支持 CSV/XLSX/JSON 输入，`line`、`scatter`、`line_scatter` 与 OPJU/PNG/SVG/PDF 输出。成功产物由 Node 流式上传，全部校验通过后发布到任务工作目录 `origin//`，plot spec 与 provenance 位于其 `.meta/`；上传中断会在重连时幂等续传。 -Web 用户登录后，文件栏 Job 中心会聚合本人最近任务。活动任务约 4 秒刷新一次，空闲时降为约 30 秒；停止已派发任务是协作取消，状态先显示“正在停止”，Node 在线时立即接收，断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`register_artifact`、`software_job_submit`、`software_job_status` 和 `software_job_cancel`。Origin 输入必须是 artifact：已有 UUID 可直接提交，普通 task 文件先逐个用相对路径登记；登记不会发布聊天交付卡片。提交工具接收 `inputs`、`operation`、`outputs`，支持 1–16 个输入、跨输入系列和多个显式输出，只创建固定 v2 schema 的持久任务，不会阻塞当前对话等待完成。成功状态提供 `output_dir`，Agent可在该目录内搜索并分析；正式输出的 artifact 带 `software_job_id`，供结果卡和产物详情展示来源。 +Web 用户登录后，文件栏 Job 中心会聚合本人最近任务。活动任务约 4 秒刷新一次，空闲时降为约 30 秒；停止已派发任务是协作取消，状态先显示“正在停止”，Node 在线时立即接收，断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`register_artifact`、`software_job_submit`、`software_job_status` 和 `software_job_cancel`。Origin 输入必须是 artifact：已有 UUID 可直接提交，普通 task 文件先逐个用相对路径登记；登记不会发布聊天交付卡片。提交工具接收 `inputs`、`operation`、`outputs`，支持 1–16 个输入、跨输入系列和多个显式输出，只创建固定 v2 schema 的持久任务，不会阻塞当前对话等待完成。成功状态提供 `output_dir`，Agent可在该目录内搜索并分析；正式输出的 artifact 带 `software_job_id`，供结果卡和产物详情展示来源。Node 输出上传的逐任务诊断日志位于 `%ProgramData%\Zcbot\WindowsNode\jobs\\logs\node-output-upload.log`；日志包含上传阶段、产物文件名、重试次数和 Windows `HRESULT`，单文件达到 1 MiB 后轮转一份 `.1`，不记录 Node Token 或认证请求头。 注册配置写入 `%ProgramData%\Zcbot\WindowsNode\node.json`；Token 使用 DPAPI `LocalMachine` 加密，ACL 仅允许注册账号和 `SYSTEM`。应始终用同一专用 Windows 账号执行统一安装器、注册并运行 Node。当前 MVP 以该账号的登录后计划任务启动，不安装 Windows Service。 diff --git a/tests/test_windows_node_source.py b/tests/test_windows_node_source.py index 2d502ff..06f590b 100644 --- a/tests/test_windows_node_source.py +++ b/tests/test_windows_node_source.py @@ -265,8 +265,23 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertIn('DefaultRequestHeaders.Add("X-Lease-Id"', uploader) self.assertIn("SHA256.HashDataAsync", uploader) self.assertIn("upload-complete.json", connection + uploader) + self.assertIn("UploadAsync(refreshed, recoveringOutputs)", connection) + self.assertIn("manifest, allowConflict: true", uploader) + self.assertIn("completeResponse.StatusCode == HttpStatusCode.Conflict", uploader) + self.assertIn("OpenOutputWithRetryAsync", uploader) + self.assertIn("IsSharingViolation", uploader) + self.assertIn("(exception.HResult & 0xFFFF) is 32 or 33", uploader) + self.assertIn("artifact={artifactId}", uploader) + self.assertIn("hresult=0x{exception.HResult:X8}", connection + uploader) + self.assertIn('"node-output-upload.log"', uploader) + self.assertIn("MaxDiagnosticLogBytes", uploader) + self.assertIn("outputUploader.RecordDeferred(current, exception)", connection) self.assertNotIn("Process.Start", uploader) + def test_local_job_monitor_never_shows_update_before_acceptance(self) -> None: + inbox = (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8") + self.assertIn("state.UpdatedAt > acceptedAt", inbox) + if __name__ == "__main__": unittest.main() diff --git a/windows-node/README.md b/windows-node/README.md index f0a0983..8c8e45c 100644 --- a/windows-node/README.md +++ b/windows-node/README.md @@ -31,7 +31,7 @@ windows-node/Zcbot.WindowsNode/bin/Debug/net10.0-windows/Zcbot.WindowsNode.exe e windows-node/Zcbot.WindowsNode/bin/Debug/net10.0-windows/Zcbot.WindowsNode.exe ``` -直接双击 EXE 默认启动托盘 UI；红点表示未注册或身份失效，黄点表示正在连接，绿点表示在线。双击托盘图标打开窗口，可在“本机任务”中查看最近 50 条已接收任务及执行详情。无界面运行使用： +直接双击 EXE 默认启动托盘 UI；红点表示未注册或身份失效，黄点表示正在连接，绿点表示在线。双击托盘图标打开窗口，可在“本机任务”中查看最近 50 条已接收任务及执行详情。每个任务的输出上传诊断日志保存在 `%ProgramData%\Zcbot\WindowsNode\jobs\\logs\node-output-upload.log`，记录阶段、产物文件名、重试次数和 Windows `HRESULT`，不记录 Node Token 或认证请求头；单文件达到 1 MiB 后轮转一份 `.1`。无界面运行使用： ```powershell Zcbot.WindowsNode.exe run --headless diff --git a/windows-node/Zcbot.WindowsNode/JobInboxStore.cs b/windows-node/Zcbot.WindowsNode/JobInboxStore.cs index 6a488f3..7b6fbfb 100644 --- a/windows-node/Zcbot.WindowsNode/JobInboxStore.cs +++ b/windows-node/Zcbot.WindowsNode/JobInboxStore.cs @@ -106,9 +106,11 @@ internal sealed class JobInboxStore(string jobsDirectory) var stage = state?.Stage ?? "accepted"; var progress = state?.Progress ?? 0; var detail = state?.Detail ?? "任务已由本机接收"; - var updatedAt = state is not null && state.UpdatedAt != DateTimeOffset.MinValue - ? state.UpdatedAt - : acceptedAt; + var updatedAt = state is not null + && state.UpdatedAt != DateTimeOffset.MinValue + && state.UpdatedAt > acceptedAt + ? state.UpdatedAt + : acceptedAt; if (terminal is JsonElement terminalValue) { var terminalStatus = ReadString(terminalValue, "status", "failed"); diff --git a/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs b/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs index 8fd114c..c29a91e 100644 --- a/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs +++ b/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs @@ -1,3 +1,4 @@ +using System.Net; using System.Net.Http.Headers; using System.Security.Cryptography; using System.Text; @@ -7,9 +8,18 @@ namespace Zcbot.WindowsNode; internal sealed class JobOutputUploader(NodeConfig config) { + private const long MaxDiagnosticLogBytes = 1024 * 1024; + private static readonly object DiagnosticLogLock = new(); private static readonly JsonSerializerOptions JsonOptions = new() { WriteIndented = true }; + private static readonly TimeSpan[] SharingViolationBackoff = + [ + TimeSpan.FromMilliseconds(500), + TimeSpan.FromSeconds(1), + TimeSpan.FromSeconds(2), + TimeSpan.FromSeconds(5), + ]; - internal async Task UploadAsync(RecoverableJob job) + internal async Task UploadAsync(RecoverableJob job, bool recovery) { var jobDirectory = Path.Combine( NodePaths.ForCurrentMachine().JobsDirectory, job.JobId.ToString("D")); @@ -26,6 +36,18 @@ internal sealed class JobOutputUploader(NodeConfig config) client.DefaultRequestHeaders.Add("X-Node-Id", config.NodeId.ToString()); client.DefaultRequestHeaders.Add("X-Lease-Id", job.LeaseId.ToString()); client.DefaultRequestHeaders.Add("X-Request-Digest", job.RequestDigest); + if (recovery) + { + var replay = await TryCompleteAsync( + client, jobDirectory, job, manifest, allowConflict: true); + if (replay is not null) + { + await WriteCompletionWithRetryAsync(job, completionPath, replay); + Log(jobDirectory, "OK", + $"Output upload reconciled job={job.JobId:D} source=cloud"); + return; + } + } foreach (var artifact in manifest.EnumerateArray()) { var localId = artifact.GetProperty("artifact_id").GetString()!; @@ -38,9 +60,11 @@ internal sealed class JobOutputUploader(NodeConfig config) { throw new InvalidDataException($"Output artifact is missing or changed: {localId}."); } - await using (var verify = new FileStream( - path, FileMode.Open, FileAccess.Read, FileShare.Read, 64 * 1024, - FileOptions.Asynchronous | FileOptions.SequentialScan)) + Log(jobDirectory, "INFO", + $"Output upload phase=verify job={job.JobId:D} " + + $"artifact={localId} file={filename}"); + await using (var verify = await OpenOutputWithRetryAsync( + jobDirectory, job, localId, filename, path, "verify")) { var digest = Convert.ToHexString( await SHA256.HashDataAsync(verify)).ToLowerInvariant(); @@ -49,9 +73,11 @@ internal sealed class JobOutputUploader(NodeConfig config) throw new InvalidDataException($"Output artifact digest changed: {localId}."); } } - await using var stream = new FileStream( - path, FileMode.Open, FileAccess.Read, FileShare.Read, 64 * 1024, - FileOptions.Asynchronous | FileOptions.SequentialScan); + Log(jobDirectory, "INFO", + $"Output upload phase=transfer job={job.JobId:D} " + + $"artifact={localId} file={filename}"); + await using var stream = await OpenOutputWithRetryAsync( + jobDirectory, job, localId, filename, path, "transfer"); using var content = new StreamContent(stream); content.Headers.ContentType = new MediaTypeHeaderValue("application/octet-stream"); content.Headers.ContentLength = expectedSize; @@ -63,15 +89,158 @@ internal sealed class JobOutputUploader(NodeConfig config) response.EnsureSuccessStatusCode(); } + var responseBody = await TryCompleteAsync( + client, jobDirectory, job, manifest, allowConflict: false) + ?? throw new InvalidDataException("Output completion response is unavailable."); + await WriteCompletionWithRetryAsync(job, completionPath, responseBody); + Log(jobDirectory, "OK", $"Output upload completed job={job.JobId:D}"); + } + + internal void RecordDeferred(RecoverableJob job, Exception exception) + { + var jobDirectory = Path.Combine( + NodePaths.ForCurrentMachine().JobsDirectory, job.JobId.ToString("D")); + Log(jobDirectory, "WARN", + $"Output upload deferred job={job.JobId:D} " + + $"exception={exception.GetType().Name} " + + $"hresult=0x{exception.HResult:X8} detail={exception.Message}", + error: true); + } + + private static async Task TryCompleteAsync( + HttpClient client, + string jobDirectory, + RecoverableJob job, + JsonElement manifest, + bool allowConflict) + { + Log(jobDirectory, "INFO", + $"Output upload phase=complete job={job.JobId:D} " + + $"mode={(allowConflict ? "reconcile" : "publish")}"); using var completeContent = new StringContent( JsonSerializer.Serialize(new { artifact_manifest = manifest }), Encoding.UTF8, "application/json"); using var completeResponse = await client.PostAsync( $"/v1/software-jobs/{job.JobId:D}/outputs/complete", completeContent); + if (allowConflict && completeResponse.StatusCode == HttpStatusCode.Conflict) + { + Log(jobDirectory, "INFO", + $"Output upload reconciliation pending job={job.JobId:D} " + + "reason=cloud_not_complete"); + return null; + } completeResponse.EnsureSuccessStatusCode(); - var responseBody = await completeResponse.Content.ReadAsByteArrayAsync(); - AtomicWrite(completionPath, responseBody); + return await completeResponse.Content.ReadAsByteArrayAsync(); + } + + private static async Task OpenOutputWithRetryAsync( + string jobDirectory, + RecoverableJob job, + string artifactId, + string filename, + string path, + string phase) + { + for (var attempt = 0; ; attempt++) + { + try + { + return new FileStream( + path, FileMode.Open, FileAccess.Read, FileShare.Read, 64 * 1024, + FileOptions.Asynchronous | FileOptions.SequentialScan); + } + catch (IOException exception) when ( + IsSharingViolation(exception) && attempt < SharingViolationBackoff.Length) + { + var delay = SharingViolationBackoff[attempt]; + Log(jobDirectory, "WARN", + $"Output file busy job={job.JobId:D} artifact={artifactId} " + + $"file={filename} phase={phase} attempt={attempt + 1} " + + $"retry_ms={(int)delay.TotalMilliseconds} " + + $"hresult=0x{exception.HResult:X8}", error: true); + await Task.Delay(delay); + } + catch (IOException exception) when (IsSharingViolation(exception)) + { + Log(jobDirectory, "WARN", + $"Output file busy job={job.JobId:D} artifact={artifactId} " + + $"file={filename} phase={phase} retries=exhausted " + + $"hresult=0x{exception.HResult:X8}", error: true); + throw new IOException( + $"输出文件 {filename} 在{PhaseLabel(phase)}阶段持续被占用" + + $"（artifact={artifactId}, HRESULT=0x{exception.HResult:X8}）。", + exception); + } + } + } + + private static async Task WriteCompletionWithRetryAsync( + RecoverableJob job, + string completionPath, + byte[] responseBody) + { + for (var attempt = 0; ; attempt++) + { + try + { + AtomicWrite(completionPath, responseBody); + return; + } + catch (IOException exception) when ( + IsSharingViolation(exception) && attempt < SharingViolationBackoff.Length) + { + var delay = SharingViolationBackoff[attempt]; + Log(Path.GetDirectoryName(completionPath)!, "WARN", + $"Completion marker busy job={job.JobId:D} " + + $"attempt={attempt + 1} retry_ms={(int)delay.TotalMilliseconds} " + + $"hresult=0x{exception.HResult:X8}", error: true); + await Task.Delay(delay); + } + } + } + + private static bool IsSharingViolation(IOException exception) => + (exception.HResult & 0xFFFF) is 32 or 33; + + private static string PhaseLabel(string phase) => phase == "verify" ? "校验" : "上传"; + + private static void Log( + string jobDirectory, + string level, + string message, + bool error = false) + { + var consoleMessage = $"[{level}] {message}"; + if (error) + { + Console.Error.WriteLine(consoleMessage); + } + else + { + Console.WriteLine(consoleMessage); + } + try + { + lock (DiagnosticLogLock) + { + var logsDirectory = Path.Combine(jobDirectory, "logs"); + Directory.CreateDirectory(logsDirectory); + var path = Path.Combine(logsDirectory, "node-output-upload.log"); + if (File.Exists(path) && new FileInfo(path).Length >= MaxDiagnosticLogBytes) + { + File.Move(path, path + ".1", overwrite: true); + } + File.AppendAllText( + path, + $"{DateTimeOffset.UtcNow:O} [{level}] {message}{Environment.NewLine}", + new UTF8Encoding(false)); + } + } + catch (Exception exception) when ( + exception is IOException or UnauthorizedAccessException) + { + } } private static void AtomicWrite(string path, byte[] responseBody) diff --git a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs index 32446a3..837d94a 100644 --- a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs +++ b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs @@ -318,6 +318,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? private async Task RunJobPipelineAsync(ClientWebSocket socket, RecoverableJob job) { + var recoveringOutputs = job.Terminal is JsonElement; try { if (job.Terminal is null) @@ -386,7 +387,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? progress = 90, metrics = new { }, }); - await outputUploader.UploadAsync(refreshed); + await outputUploader.UploadAsync(refreshed, recoveringOutputs); jobInbox.WriteState(refreshed, "succeeded", 100, "结果已上传并由云端确认"); } catch (Exception exception) when ( @@ -406,7 +407,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? "uploading_outputs", 90, $"结果上传暂缓：{exception.Message}"); - Console.Error.WriteLine($"[WARN] Output upload deferred: {exception.Message}"); + outputUploader.RecordDeferred(current, exception); return; } jobInbox.WriteTerminal(