diff --git a/INSTALL.md b/INSTALL.md index 0ed8c5eb..4f8de5f5 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -41,6 +41,38 @@ node packages/cli/lib/bin.js install subject_<32 lowercase hex characters> --hos Replace the subject id with the exact value returned by Distilly. Profile installation writes only the self-contained Profile and its digest manifest. +## Recover a briefing that exceeds the host limit + +If `distilly_pending` returns `briefing_too_large`, the research is still stored and pending. Raising the model context setting does not change a verified MCP transport limit. For a complete briefing within the engine's limits, use the explicit local file workflow from a checkout containing this command: + +```bash +node packages/cli/lib/bin.js recover job_<32 lowercase hex characters> --output /absolute/path/to/new-recovery-directory +``` + +Use the exact job ID from `distilly_pending` with `action: "list"`. Run the command with the same home directory as the installed Plugin so it opens the same `~/.distilly` store. The output directory must be new, with an existing parent. It will contain private research; choose a local location you intend to use for that data. + +Keep the command running. It writes the complete `briefing.json`, `commit-tool-schema.json`, and `README.txt`, then waits for a response. Read the entire briefing, its instructions, evidence rules, and baseline before preparing a patch. The patch schema is the `patch` property of `commit-tool-schema.json`; the command supplies all commit identity fields itself. + +Write a temporary JSON file with exactly `briefingSha256` (the digest in `README.txt`) and `patch` (your DistillPatch object). Rename the finished file to `response.json` in that directory. The command checks the digest and submits the patch through the same session and lease. It does not call a model or generate a patch. An empty patch is a deliberate decision to consume the briefing without adding or changing claims; do not use one merely to clear the error. + +The default wait is 20 minutes. `--timeout-seconds` accepts 1 through 1500 seconds, below the 30-minute lease lifetime. Timeout, Ctrl+C, SIGTERM, malformed responses, and validation errors attempt to release the lease and leave research available for another attempt. A force-killed process leaves its lease to expire. Another session's active lease is never taken over. Retry with a new directory and a fresh briefing; an older response will not match the new digest. + +A successful submission writes `submission.json` and `result.json`. The result can be current or suspended for review; a suspended result still needs the normal review workflow. If the command says the commit succeeded but its result file could not be written, use the printed version and request IDs to inspect the result. Do not resubmit blindly. An unknown commit outcome requires the same inspection before retrying. + +This path transfers a complete briefing through local files, with a maximum of 4 MiB and 999 material references. It does not raise a host's verified MCP limit or establish that a model can read that much context. Responses are limited to 256 KiB on disk; the existing 64 KiB canonical patch limit still applies. There is no truncation, automatic splitting, or deletion of stored research. Briefings above the engine limits still fail explicitly. Recovery directories are not removed automatically. + +### 超出宿主限制后的本地恢复 + +出现 `briefing_too_large` 时，调研资料仍保存在库中并等待处理。调整模型上下文不会改变已验证的 MCP 传输限制。可在包含此命令的源码构建目录中运行上面的 `recover` 命令；job ID 从 `distilly_pending` 的 `action: "list"` 结果取得。命令必须使用与 Plugin 相同的 home，才能访问同一个 `~/.distilly`。输出目录必须尚不存在，且父目录已存在；其中包含私人资料，请选择适合保存这些资料的本地位置。 + +保持命令运行，完整阅读 `briefing.json`、证据规则和已有基线，再按 `commit-tool-schema.json` 的 `patch` 字段定义准备结果。先写临时 JSON 文件，只包含 `README.txt` 中的 `briefingSha256` 和你的 `patch`，完成后再将其重命名为 `response.json`。命令会校验摘要，并通过同一会话和租约提交。它不调用模型，也不自动生成结果。空 patch 表示明确决定处理完本次资料但不增改任何 claim，请勿仅为消除报错而提交空 patch。 + +默认等待 20 分钟，`--timeout-seconds` 可设置为 1 至 1500 秒。超时、Ctrl+C、SIGTERM、响应格式错误或校验失败时，命令会尝试释放租约，供后续重试；强制杀死进程则需要等待租约过期。命令不会接管其他会话的有效租约。重试须使用新目录和新 briefing，旧响应无法通过摘要校验。 + +成功提交后会保留 `submission.json` 与 `result.json`；结果若为 suspended，仍需正常审核。若提示提交成功但回执写入失败，请用输出中的版本 ID 和请求 ID 查询结果，不要直接重复提交。提交结果不明确时也应先核实。 + +本地文件路径支持最多 4 MiB 的完整 briefing 和 999 个资料引用，不代表宿主 MCP 限制已提高，也不保证模型具备相应上下文容量。响应文件最多 256 KiB，原有 canonical patch 的 64 KiB 上限继续生效。不会裁剪、自动拆分或删除已存资料；超出引擎上限仍会明确失败。恢复目录不会自动清理。 + ## Remove the host integration ```bash diff --git a/docs/architecture.md b/docs/architecture.md index 13a84a9d..431d7e24 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -99,3 +99,5 @@ authority SQLite/WAL metadata + immutable blobs LSN projections / exports ``` Signatures, field lists, and host pitfalls stay in the design chapters. This page only orients. + +The Preview CLI also provides explicit `recover --output ` file recovery for briefings above a host transport limit. It exports the complete briefing within existing engine limits, binds a manually supplied patch to that exact export and lease, and uses the normal evidence-checked commit path. See [INSTALL.md](../INSTALL.md#recover-a-briefing-that-exceeds-the-host-limit) for deadlines, private artifacts, and failure recovery. This does not change verified host capacity. diff --git a/docs/design/system-v3.md b/docs/design/system-v3.md index d7d860e2..2b6e8df0 100644 --- a/docs/design/system-v3.md +++ b/docs/design/system-v3.md @@ -2089,6 +2089,8 @@ brief / renew / release 都是单独的 SQLite mutation： binary 升级后若仍支持 lease 固定的 grouping/prompt/draft versions，旧 lease 可正常完成；缺少 pinned implementation 返回 schema_unsupported，要求显式 release / 重新 brief，不能按当前默认值静默重算。 ### 12.5 不静默裁剪 +Preview CLI 的显式 `recover --output ` 为已录入且超出宿主传输上限的任务提供完整本地文件恢复：使用 `sdk_explicit` 文件预算，在现有引擎上限内导出完整 briefing；命令保持同一会话与 lease，接收绑定导出文件 SHA-256 的 patch 后走原有 commit 校验。默认等待 20 分钟，最多 25 分钟，不自动生成空 patch、续租、拆分或裁剪。取消或校验失败时尝试释放自己的 lease；强制终止则等待过期。回执写入失败不能被报告成提交未发生。目录须原子创建且权限为 0700，命令写入的文件为 0600；保留资料供用户处理，不自动删除。此 SDK 文件预算不修改 verified host fixture，也不证明模型上下文容量。具体操作见 INSTALL.md。 + BriefingService 只使用 ClientSessionContext 中经过可信 preflight 的 BriefCapacity；模型不能在 pending 输入里自报或放大。HostPreflight 的 success capacity 是 HostDistillBriefing 经过实际宿主 tool-result 路径后仍可完整交付给模型的**净预算**。`source=host_handshake` 只允许可信宿主 API 直接给出当前 surface 的净 input/result envelope budget；maxContextTokens、maxToolResultBytes、字符阈值、token 阈值或其它 gross field 不能靠减一个固定 wrapper 常量转换成 capacity。`source=binding_fixture` 只允许匹配 §17.1 exact host/version/surface/release/wire/skill tuple、并在真实 structured/text 双结果序列化路径上对公告的 exact net budget 通过真实宿主 transport 测试的保守净值；它不必探出宿主真实失败极限，但不得公告超过实测完整值的 capacity。当前 OpenClaw/Hermes 记录使用隔离 clean CLI home、固定 `openai-codex/gpt-5.4` 与 deterministic synthetic fixture server；“真实宿主测试”指真实 executable、模型调用和 MCP transport，不指真实产品 Engine、用户材料或所有模型/session 的剩余上下文。canonical tool descriptor、host advertised-schema projection、serializer、manifest、canonical skill 或 tuple 任一改变都使 fixture 失效；OpenClaw/Hermes 的 projection 变化必须重新运行对应真实宿主测试，即使五工具名称和 canonical descriptor digest 没变。fixture 文件保留 canonical `toolContractDigest`，并在使用 `schemaProfile` 时另外绑定实际公告面的 `advertisedToolContractDigest` 与 probe 的 `probeContractDigest`；后两者是 loader/verifier 的内部不可变元数据，不扩展 HostPreflight/MCP wire evidence。没有可信净 handshake 或完全匹配 fixture 时，preflight 返回 host_unsupported，外层不得创建 host client；普通 SDK 则必须在打开 client 时显式给 `source=sdk_explicit` 的 capacity。ClientSessionContext 没有 capacity 时 brief 同样 host_unsupported，不创建 lease。 内部常量固定为 maximumBriefingBytes=4,194,304、maximumMaterialRefs=999、maximumOutputBytes=65,536；最后一项就是 accepted DistillPatch compact canonical JSON 的 UTF-8 bytes budget，不是让模型返回任意 65,536 字节文本。commit 在打开写事务前对 schema-validated canonical patch bytes 计数；`<= 65,536`（恰好等于也允许），`65,537` 返回 invalid_input 并零写入。brief 容量算法先构造包括 limits 在内的完整 HostDistillBriefing，然后求 fixed point：令 estimatedInputTokens 从 0 开始，反复把它写回对象并计算 compact canonical JSON 的 UTF-8 byte length，直到新值等于字段值；该稳定值就是 estimatedInputTokens，采用保守的 1 UTF-8 byte = 1 token。最终**完整 briefing** 的 serializedBytes 必须同时 `<= 4,194,304`、`<= capacity.maximumToolResultBytes`，estimatedInputTokens 必须 `<= capacity.maximumInputTokens`，refs 必须 `<= 999`；等于上限允许。 diff --git a/docs/design/v3/12-briefing-and-lease.md b/docs/design/v3/12-briefing-and-lease.md index a5ca199d..c7235eea 100644 --- a/docs/design/v3/12-briefing-and-lease.md +++ b/docs/design/v3/12-briefing-and-lease.md @@ -129,6 +129,8 @@ brief / renew / release 都是单独的 SQLite mutation： binary 升级后若仍支持 lease 固定的 grouping/prompt/draft versions，旧 lease 可正常完成；缺少 pinned implementation 返回 schema_unsupported，要求显式 release / 重新 brief，不能按当前默认值静默重算。 ### 12.5 不静默裁剪 +Preview CLI 的显式 `recover --output ` 为已录入且超出宿主传输上限的任务提供完整本地文件恢复：使用 `sdk_explicit` 文件预算，在现有引擎上限内导出完整 briefing；命令保持同一会话与 lease，接收绑定导出文件 SHA-256 的 patch 后走原有 commit 校验。默认等待 20 分钟，最多 25 分钟，不自动生成空 patch、续租、拆分或裁剪。取消或校验失败时尝试释放自己的 lease；强制终止则等待过期。回执写入失败不能被报告成提交未发生。目录须原子创建且权限为 0700，命令写入的文件为 0600；保留资料供用户处理，不自动删除。此 SDK 文件预算不修改 verified host fixture，也不证明模型上下文容量。具体操作见 INSTALL.md。 + BriefingService 只使用 ClientSessionContext 中经过可信 preflight 的 BriefCapacity；模型不能在 pending 输入里自报或放大。HostPreflight 的 success capacity 是 HostDistillBriefing 经过实际宿主 tool-result 路径后仍可完整交付给模型的**净预算**。`source=host_handshake` 只允许可信宿主 API 直接给出当前 surface 的净 input/result envelope budget；maxContextTokens、maxToolResultBytes、字符阈值、token 阈值或其它 gross field 不能靠减一个固定 wrapper 常量转换成 capacity。`source=binding_fixture` 只允许匹配 §17.1 exact host/version/surface/release/wire/skill tuple、并在真实 structured/text 双结果序列化路径上对公告的 exact net budget 通过真实宿主 transport 测试的保守净值；它不必探出宿主真实失败极限，但不得公告超过实测完整值的 capacity。当前 OpenClaw/Hermes 记录使用隔离 clean CLI home、固定 `openai-codex/gpt-5.4` 与 deterministic synthetic fixture server；“真实宿主测试”指真实 executable、模型调用和 MCP transport，不指真实产品 Engine、用户材料或所有模型/session 的剩余上下文。canonical tool descriptor、host advertised-schema projection、serializer、manifest、canonical skill 或 tuple 任一改变都使 fixture 失效；OpenClaw/Hermes 的 projection 变化必须重新运行对应真实宿主测试，即使五工具名称和 canonical descriptor digest 没变。fixture 文件保留 canonical `toolContractDigest`，并在使用 `schemaProfile` 时另外绑定实际公告面的 `advertisedToolContractDigest` 与 probe 的 `probeContractDigest`；后两者是 loader/verifier 的内部不可变元数据，不扩展 HostPreflight/MCP wire evidence。没有可信净 handshake 或完全匹配 fixture 时，preflight 返回 host_unsupported，外层不得创建 host client；普通 SDK 则必须在打开 client 时显式给 `source=sdk_explicit` 的 capacity。ClientSessionContext 没有 capacity 时 brief 同样 host_unsupported，不创建 lease。 内部常量固定为 maximumBriefingBytes=4,194,304、maximumMaterialRefs=999、maximumOutputBytes=65,536；最后一项就是 accepted DistillPatch compact canonical JSON 的 UTF-8 bytes budget，不是让模型返回任意 65,536 字节文本。commit 在打开写事务前对 schema-validated canonical patch bytes 计数；`<= 65,536`（恰好等于也允许），`65,537` 返回 invalid_input 并零写入。brief 容量算法先构造包括 limits 在内的完整 HostDistillBriefing，然后求 fixed point：令 estimatedInputTokens 从 0 开始，反复把它写回对象并计算 compact canonical JSON 的 UTF-8 byte length，直到新值等于字段值；该稳定值就是 estimatedInputTokens，采用保守的 1 UTF-8 byte = 1 token。最终**完整 briefing** 的 serializedBytes 必须同时 `<= 4,194,304`、`<= capacity.maximumToolResultBytes`，estimatedInputTokens 必须 `<= capacity.maximumInputTokens`，refs 必须 `<= 999`；等于上限允许。 diff --git a/packages/cli/src/file-recovery.test.ts b/packages/cli/src/file-recovery.test.ts new file mode 100644 index 00000000..b5577fa5 --- /dev/null +++ b/packages/cli/src/file-recovery.test.ts @@ -0,0 +1,359 @@ +import { createHash, randomBytes } from "node:crypto"; +import { mkdir, mkdtemp, readFile, rename, rm, stat, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; + +import { isoDateTimeSchema, requestIdSchema, type HostDistillBriefing } from "@distilly/protocol"; +import { openPreviewLocalRuntime } from "@distilly/runtime/preview"; +import { afterEach, describe, expect, it } from "vitest"; + +import { + parseRecoveryArguments, + recoverFromFiles, + type FileRecoveryOptions, +} from "./file-recovery.js"; + +import { runPreviewCli } from "./main.js"; + +const roots: string[] = []; +const request = () => requestIdSchema.parse(`req_${randomBytes(16).toString("hex")}`); +const capacity = { + maximumInputTokens: 4_194_304, + maximumToolResultBytes: 4_194_304, + source: "sdk_explicit" as const, +}; +const content = + "Mira builds reliable systems.\n" + "Full source evidence must remain available.\n".repeat(2000); + +const seed = async () => { + const directory = await mkdtemp(join(tmpdir(), "distilly-file-recovery-")); + roots.push(directory); + const root = join(directory, ".distilly"); + const runtime = await openPreviewLocalRuntime({ root }); + try { + const client = await runtime.connectTrusted({ + actor: { kind: "sdk", id: "seed" }, + capacity: { ...capacity, maximumToolResultBytes: 65_536 }, + }); + const result = await client.call( + "materials.ingest", + { + subject: { kind: "create", input: { displayName: "Mira", identityHints: [] } }, + materials: [ + { + clientRef: "full-source", + kind: "document", + content, + source: { + medium: "document", + access: "private", + capturedAt: isoDateTimeSchema.parse("2026-09-01T00:00:00.000Z"), + }, + derivation: { kind: "native_text" }, + sensitivity: "private", + }, + ], + enqueue: "now", + }, + { requestId: request() }, + ); + if (result.kind !== "ingested" || result.job === undefined) throw new Error("Missing job"); + await expect( + client.call("distill.brief", { jobId: result.job.id }, { requestId: request() }), + ).rejects.toMatchObject({ code: "briefing_too_large" }); + return { + root, + jobId: result.job.id, + outputDirectory: join(directory, "recovery"), + timeoutMs: 5000, + }; + } finally { + await runtime.close(); + } +}; + +const assertPending = async (root: string) => { + const runtime = await openPreviewLocalRuntime({ root }); + try { + const client = await runtime.connectTrusted({ actor: { kind: "sdk", id: "verify" }, capacity }); + const jobs = await client.call("distill.pending", {}); + expect(jobs).toHaveLength(1); + const job = jobs[0]; + if (job === undefined) throw new Error("Missing job"); + const brief = await client.call("distill.brief", { jobId: job.id }, { requestId: request() }); + expect(brief.materials[0]?.content).toBe(content); + await client.call( + "distill.release", + { jobId: job.id, leaseId: brief.lease.id }, + { requestId: request() }, + ); + } finally { + await runtime.close(); + } +}; + +const responseFor = async (directory: string) => { + const bytes = await readFile(join(directory, "briefing.json")); + const briefing = JSON.parse(bytes.toString()) as HostDistillBriefing; + expect(bytes.length).toBeGreaterThan(65_536); + expect(briefing.materials[0]?.content).toBe(content); + return { + briefingSha256: createHash("sha256").update(bytes).digest("hex"), + patch: { + operations: [ + { + op: "add", + claim: { + facet: "identity", + text: "Mira builds reliable systems.", + evidence: [ + { + kind: "brief_material", + materialRef: briefing.materials[0]?.ref, + quote: "Mira builds reliable systems.", + }, + ], + }, + }, + ], + }, + }; +}; + +const runWithResponse = async (options: FileRecoveryOptions, respond: () => Promise) => { + let ready!: () => void; + const exported = new Promise((resolve) => { + ready = resolve; + }); + const output: string[] = []; + const running = recoverFromFiles(options, { + stdout: { + write: (value) => { + output.push(value); + ready(); + }, + }, + stderr: { write: (value) => output.push(value) }, + }); + // Attach a rejection handler immediately while the response is prepared. + const outcome = running.then( + () => ({ error: undefined }), + (error: unknown) => ({ error }), + ); + await exported; + await respond(); + return { ...(await outcome), output }; +}; + +afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); +}); + +describe("file briefing recovery", () => { + it("recovers a rejected 85 KiB source with exact evidence through the same lease", async () => { + const options = await seed(); + const result = await runWithResponse(options, async () => { + const response = await responseFor(options.outputDirectory); + const temporary = join(options.outputDirectory, "response.tmp"); + await writeFile(temporary, JSON.stringify(response)); + await rename(temporary, join(options.outputDirectory, "response.json")); + }); + expect(result.error).toBeUndefined(); + expect(result.output.join("")).toContain("Recovery committed current version"); + expect(result.output.join("")).not.toContain(content); + const receipt = JSON.parse( + await readFile(join(options.outputDirectory, "result.json"), "utf8"), + ) as { result: { kind: string } }; + expect(receipt.result.kind).toBe("current"); + expect((await stat(options.outputDirectory)).mode & 0o777).toBe(0o700); + for (const name of [ + "briefing.json", + "README.txt", + "commit-tool-schema.json", + "submission.json", + "result.json", + ]) { + expect((await stat(join(options.outputDirectory, name))).mode & 0o777).toBe(0o600); + } + const runtime = await openPreviewLocalRuntime({ root: options.root }); + try { + const client = await runtime.connectTrusted({ + actor: { kind: "sdk", id: "verify" }, + capacity, + }); + expect(await client.call("distill.pending", {})).toEqual([]); + } finally { + await runtime.close(); + } + }); + + it.each([ + "wrong binding", + "invalid quote", + "partial JSON", + "oversized", + "symlink", + "directory", + "null", + "extra identity", + ])("rejects %s without consuming pending research", async (mode) => { + const options = await seed(); + const result = await runWithResponse(options, async () => { + const response = await responseFor(options.outputDirectory); + const path = join(options.outputDirectory, "response.json"); + if (mode === "symlink") { + await symlink(join(options.outputDirectory, "briefing.json"), path); + return; + } + if (mode === "directory") { + await mkdir(path); + return; + } + if (mode === "wrong binding") response.briefingSha256 = "0".repeat(64); + if (mode === "invalid quote") + response.patch.operations[0]!.claim.evidence[0]!.quote = "This quote is not present."; + const serialized = + mode === "partial JSON" + ? "{" + : mode === "oversized" + ? " ".repeat(262_145) + : mode === "null" + ? "null" + : JSON.stringify( + mode === "extra identity" ? { ...response, generation: 999 } : response, + ); + await writeFile(path, serialized); + }); + expect(result.error).toBeInstanceOf(Error); + await assertPending(options.root); + }); + + it("times out with no automatic empty commit and releases the lease", async () => { + const options = await seed(); + await expect( + recoverFromFiles( + { ...options, timeoutMs: 20 }, + { stdout: { write: () => {} }, stderr: { write: () => {} } }, + ), + ).rejects.toThrow("timed out"); + await expect(stat(join(options.outputDirectory, "submission.json"))).rejects.toMatchObject({ + code: "ENOENT", + }); + await assertPending(options.root); + }); + + it("releases the lease on cancellation", async () => { + const options = await seed(); + const controller = new AbortController(); + const result = await runWithResponse({ ...options, signal: controller.signal }, () => { + controller.abort(); + return Promise.resolve(); + }); + expect(result.error).toBeInstanceOf(Error); + await assertPending(options.root); + }); + + it("does not overwrite an existing output directory", async () => { + const options = await seed(); + await mkdir(options.outputDirectory); + await writeFile(join(options.outputDirectory, "briefing.json"), "keep me"); + await expect( + recoverFromFiles(options, { stdout: { write: () => {} }, stderr: { write: () => {} } }), + ).rejects.toMatchObject({ code: "EEXIST" }); + expect(await readFile(join(options.outputDirectory, "briefing.json"), "utf8")).toBe("keep me"); + await assertPending(options.root); + }); + + it("reports a durable commit even when the result receipt cannot be written", async () => { + const options = await seed(); + const result = await runWithResponse(options, async () => { + await writeFile(join(options.outputDirectory, "result.json"), "keep me"); + await writeFile( + join(options.outputDirectory, "response.json"), + JSON.stringify(await responseFor(options.outputDirectory)), + ); + }); + expect(result.error).toBeInstanceOf(Error); + expect((result.error as Error).message).toContain("Commit succeeded:"); + expect(await readFile(join(options.outputDirectory, "result.json"), "utf8")).toBe("keep me"); + }); + + it.each([ + { args: [] }, + { args: ["bad", "--output", "/tmp/new"] }, + { args: ["job_" + "a".repeat(32), "--output", "/tmp/new", "--timeout-seconds", "1800"] }, + ])("rejects invalid arguments before opening storage: $args", ({ args }) => { + expect(() => parseRecoveryArguments(args)).toThrow(); + }); + + it.each(["SIGINT", "SIGTERM"] as const)( + "routes the real CLI command and cleans up %s listeners", + async (signal) => { + const options = await seed(); + const beforeInt = process.listenerCount("SIGINT"); + const beforeTerm = process.listenerCount("SIGTERM"); + await expect( + runPreviewCli( + ["recover", options.jobId, "--output", options.outputDirectory], + { + lifecycle: { + homeDirectory: dirname(options.root), + nodePath: process.execPath, + entryPath: "/unused", + pluginSourcesPath: "/unused", + pathValue: "", + }, + panelAssetsPath: "/unused", + }, + { + stdout: { + write: () => { + process.emit(signal); + }, + }, + stderr: { write: () => {} }, + }, + ), + ).rejects.toThrow("cancelled"); + expect(process.listenerCount("SIGINT")).toBe(beforeInt); + expect(process.listenerCount("SIGTERM")).toBe(beforeTerm); + await assertPending(options.root); + }, + ); + + it("does not take over or release another session's active lease", async () => { + const options = await seed(); + const runtime = await openPreviewLocalRuntime({ root: options.root }); + try { + const client = await runtime.connectTrusted({ + actor: { kind: "sdk", id: "other-owner" }, + capacity, + }); + const brief = await client.call( + "distill.brief", + { jobId: options.jobId }, + { requestId: request() }, + ); + await runtime.close(); + await expect( + recoverFromFiles(options, { stdout: { write: () => {} }, stderr: { write: () => {} } }), + ).rejects.toMatchObject({ code: "lease_conflict" }); + await expect(stat(options.outputDirectory)).rejects.toMatchObject({ code: "ENOENT" }); + const reopened = await openPreviewLocalRuntime({ root: options.root }); + try { + const observer = await reopened.connectTrusted({ + actor: { kind: "sdk", id: "observer" }, + capacity, + }); + await expect( + observer.call("distill.brief", { jobId: options.jobId }, { requestId: request() }), + ).rejects.toMatchObject({ code: "lease_conflict" }); + expect(brief.lease.owner).toBeDefined(); + } finally { + await reopened.close(); + } + } finally { + await runtime.close(); + } + }); +}); diff --git a/packages/cli/src/file-recovery.ts b/packages/cli/src/file-recovery.ts new file mode 100644 index 00000000..4cf5f12b --- /dev/null +++ b/packages/cli/src/file-recovery.ts @@ -0,0 +1,265 @@ +import { createHash, randomBytes } from "node:crypto"; +import { constants } from "node:fs"; +import { mkdir, open, writeFile } from "node:fs/promises"; +import { join, resolve } from "node:path"; +import { setTimeout } from "node:timers/promises"; + +import { + DistillyError, + distillPatchSchema, + distillyMcpTools, + jobIdSchema, + requestIdSchema, + type CommitInput, + type HostDistillBriefing, + type JobId, +} from "@distilly/protocol"; +import { openPreviewLocalRuntime } from "@distilly/runtime/preview"; + +import type { PreviewCliIo } from "./main.js"; + +const FILE_CAPACITY = { + maximumInputTokens: 4_194_304, + maximumToolResultBytes: 4_194_304, + source: "sdk_explicit" as const, +}; +const RESPONSE_MAXIMUM_BYTES = 262_144; +const request = () => requestIdSchema.parse(`req_${randomBytes(16).toString("hex")}`); +const privateWrite = (path: string, content: string) => + writeFile(path, content, { flag: "wx", mode: 0o600 }); + +/** One bounded, explicitly selected local recovery session. */ +export interface FileRecoveryOptions { + readonly root: string; + readonly jobId: JobId; + readonly outputDirectory: string; + readonly timeoutMs: number; + readonly signal?: AbortSignal; +} + +/** + * Parses the recovery command without opening storage or touching output paths. + * + * @param args - Arguments after recover. + * @returns Validated local recovery arguments. + */ +export const parseRecoveryArguments = ( + args: readonly string[], +): Omit => { + if ( + (args.length !== 3 && args.length !== 5) || + args[1] !== "--output" || + !args[2] || + (args.length === 5 && args[3] !== "--timeout-seconds") + ) { + throw new Error( + "Expected recover --output [--timeout-seconds 1..1500].", + ); + } + const seconds = args.length === 5 ? Number(args[4]) : 1200; + if (!Number.isInteger(seconds) || seconds < 1 || seconds > 1500) { + throw new Error("Recovery timeout must be an integer from 1 to 1500 seconds."); + } + return { + jobId: jobIdSchema.parse(args[0]), + outputDirectory: resolve(args[2]), + timeoutMs: seconds * 1000, + }; +}; + +const readResponse = async (path: string): Promise => { + const file = await open( + path, + constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK, + ).catch((error: unknown) => { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; + throw new Error("Cannot open response.json as an ordinary local file."); + }); + if (file === undefined) return undefined; + try { + const before = await file.stat(); + if (!before.isFile() || before.size > RESPONSE_MAXIMUM_BYTES) { + throw new Error("response.json must be an ordinary file of at most 262144 bytes."); + } + const bytes = Buffer.alloc(RESPONSE_MAXIMUM_BYTES + 1); + let length = 0; + while (length < bytes.length) { + const read = await file.read(bytes, length, bytes.length - length, null); + if (read.bytesRead === 0) break; + length += read.bytesRead; + } + const after = await file.stat(); + if ( + length > RESPONSE_MAXIMUM_BYTES || + length !== before.size || + after.mtimeMs !== before.mtimeMs || + after.size !== before.size + ) { + throw new Error("response.json changed while reading or exceeds the size limit."); + } + try { + // A null JSON value must be rejected, not mistaken for a missing file. + return JSON.parse(bytes.subarray(0, length).toString("utf8")) as unknown; + } catch { + throw new Error( + "Invalid response.json. Write a complete temporary file, then rename it to response.json.", + ); + } + } finally { + await file.close(); + } +}; + +const responsePatch = (value: unknown, digest: string): CommitInput["patch"] => { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw new Error("Expected a response object with briefingSha256 and patch."); + } + const response = value as Record; + if ( + Object.keys(response).length !== 2 || + !Object.hasOwn(response, "patch") || + response.briefingSha256 !== digest + ) { + throw new Error( + "Response does not match this briefing. Use its exact briefingSha256 and patch only.", + ); + } + const parsed = distillPatchSchema.safeParse(response.patch); + if (!parsed.success) + throw new Error( + "Invalid patch: follow the briefing contract and the 65536-byte canonical patch limit.", + ); + return parsed.data as CommitInput["patch"]; +}; + +/** + * Exports one complete briefing and commits only an explicitly supplied matching response. + * + * @param options - Private artifact path, job, deadline, and optional cancellation. + * @param io - Content-free progress and error output. + * @returns Completion after commit, lease cleanup, and runtime closure. + */ +export const recoverFromFiles = async ( + options: FileRecoveryOptions, + io: PreviewCliIo, +): Promise => { + if ( + !Number.isInteger(options.timeoutMs) || + options.timeoutMs < 1 || + options.timeoutMs > 1_500_000 + ) { + throw new Error("Recovery timeout must be positive and no longer than 25 minutes."); + } + const runtime = await openPreviewLocalRuntime({ root: options.root }); + let briefing: HostDistillBriefing | undefined; + let committed = false; + let client: Awaited> | undefined; + try { + options.signal?.throwIfAborted(); + client = await runtime.connectTrusted({ + actor: { kind: "sdk", id: `file-recovery-${randomBytes(16).toString("hex")}` }, + capacity: FILE_CAPACITY, + }); + briefing = await client.call( + "distill.brief", + { jobId: options.jobId }, + { requestId: request() }, + ); + const deadline = Date.now() + options.timeoutMs; + const serialized = `${JSON.stringify(briefing)}\n`; + const digest = createHash("sha256").update(serialized).digest("hex"); + await mkdir(options.outputDirectory, { mode: 0o700 }); + await privateWrite(join(options.outputDirectory, "briefing.json"), serialized); + const commitTool = distillyMcpTools.find((tool) => tool.name === "distilly_commit"); + if (commitTool === undefined) throw new Error("Missing commit tool schema."); + await privateWrite( + join(options.outputDirectory, "commit-tool-schema.json"), + `${JSON.stringify(commitTool.inputSchema)}\n`, + ); + await privateWrite( + join(options.outputDirectory, "README.txt"), + `Read the complete briefing.json, including its contract and all materials, before preparing a patch.\nSHA-256 of the exact briefing.json bytes: ${digest}\nWrite a temporary JSON file with exactly these two fields:\n{"briefingSha256":"${digest}","patch":}\nThen rename it to response.json in this directory. Do not edit briefing.json.\nThe patch must follow the patch property in commit-tool-schema.json and cite exact evidence from this briefing.\nDo not submit an empty patch unless you have reviewed all material and deliberately found no claims to add or change.\nThis command waits up to ${String(options.timeoutMs / 1000)} seconds and keeps the same lease.\nThe file transport supports a complete briefing up to 4 MiB and 999 material refs; this does not establish a model's context capacity.\nCancellation or an invalid response attempts to release the lease. Force-killing the process leaves it to expire.\n`, + ); + io.stdout.write( + `Recovery briefing: ${join(options.outputDirectory, "briefing.json")}\nRead README.txt and submit response.json within ${String(options.timeoutMs / 1000)} seconds.\n`, + ); + let response: unknown; + while (response === undefined) { + options.signal?.throwIfAborted(); + if (Date.now() >= deadline) + throw new Error("Recovery timed out before submission; no commit was attempted."); + response = await readResponse(join(options.outputDirectory, "response.json")); + if (response === undefined) + await setTimeout(Math.min(250, deadline - Date.now()), undefined, { + signal: options.signal, + }); + } + const patch = responsePatch(response, digest); + const input: CommitInput = { + jobId: briefing.job.id, + generation: briefing.job.generation, + leaseId: briefing.lease.id, + briefContractDigest: briefing.contract.digest, + materialSetHash: briefing.job.materialSetHash, + ...(briefing.job.baseVersionId === undefined + ? {} + : { baseVersionId: briefing.job.baseVersionId }), + patch, + }; + const requestId = request(); + await privateWrite( + join(options.outputDirectory, "submission.json"), + `${JSON.stringify({ requestId, input })}\n`, + ); + options.signal?.throwIfAborted(); + if (Date.now() >= deadline) + throw new Error("Recovery timed out before submission; no commit was attempted."); + const result = await client + .call("distill.commit", input, { requestId }) + .catch((error: unknown) => { + if (error instanceof DistillyError && error.code !== "internal_error") throw error; + throw new Error( + `Commit outcome is unknown for ${requestId}. Keep submission.json and inspect the subject's current version and pending work before retrying.`, + { cause: error }, + ); + }); + committed = true; + const versionId = result.kind === "current" ? result.version.id : result.candidate.id; + io.stdout.write( + `Recovery committed ${result.kind} version ${versionId} (request ${requestId}).\n`, + ); + try { + await privateWrite( + join(options.outputDirectory, "result.json"), + `${JSON.stringify({ requestId, result })}\n`, + ); + } catch { + throw new Error( + `Commit succeeded: ${result.kind} version ${versionId}, request ${requestId}. Could not write result.json; do not resubmit this patch.`, + ); + } + } finally { + try { + if (briefing !== undefined && client !== undefined && !committed) { + await client + .call( + "distill.release", + { jobId: briefing.job.id, leaseId: briefing.lease.id }, + { requestId: request() }, + ) + .catch((error: unknown) => { + if ( + error instanceof DistillyError && + ["stale_job", "lease_expired", "lease_conflict", "not_found"].includes(error.code) + ) + return; + io.stderr.write( + "Could not confirm lease release; wait for lease expiry before retrying.\n", + ); + }); + } + } finally { + await runtime.close(); + } + } +}; diff --git a/packages/cli/src/main.ts b/packages/cli/src/main.ts index e09081dc..0c821d6a 100644 --- a/packages/cli/src/main.ts +++ b/packages/cli/src/main.ts @@ -121,6 +121,7 @@ Usage: distilly doctor [--host ] distilly install --host distilly uninstall --host + distilly recover --output [--timeout-seconds 1..1500] # : codex | claude-code | openclaw | hermes The four host bindings share the same five-tool MCP contract. Setup remains @@ -147,6 +148,28 @@ export const runPreviewCli = async ( io.stdout.write(help); return 0; } + if (command === "recover") { + const { parseRecoveryArguments, recoverFromFiles } = await import("./file-recovery.js"); + const options = parseRecoveryArguments(args); + const controller = new AbortController(); + const cancel = () => controller.abort(new Error("Recovery cancelled before completion.")); + process.once("SIGINT", cancel); + process.once("SIGTERM", cancel); + try { + await recoverFromFiles( + { + ...options, + root: join(environment.lifecycle.homeDirectory, ".distilly"), + signal: controller.signal, + }, + io, + ); + } finally { + process.removeListener("SIGINT", cancel); + process.removeListener("SIGTERM", cancel); + } + return 0; + } if (command === "setup") { const host = hostOption(args, true); if (host === undefined) throw new Error("This command requires --host."); diff --git a/packages/engine/src/internal-errors.ts b/packages/engine/src/internal-errors.ts index 8db68d47..1ee77b9f 100644 --- a/packages/engine/src/internal-errors.ts +++ b/packages/engine/src/internal-errors.ts @@ -245,7 +245,7 @@ export const briefingTooLarge = (details: JsonObject): DistillyError => message: "The complete distillation briefing exceeds a verified session limit.", retryable: false, remediation: - "Use a larger-capacity host or reduce the new research batch; Distilly will not truncate it.", + "Use a larger-capacity host, or run distilly recover --output for local file recovery within the engine limits. See INSTALL.md; Distilly will not truncate the briefing.", details, });