refactor(ptc-runtime-node): unify temporary and retained processes
This commit is contained in:
_Kerman committed
1 parent
399a74e808
commit
c3473d1ea6
11 files changed
+419
-693
No files matched your search
@@ -24,8 +24,8 @@
|
||||
en: b7c41598e68edd14
|
||||
zh: eee2f2392c2b59ce
|
||||
/deepseek-ai-dsh-ptc-runtime-node/use-this-package/in-memory-modes:
|
||||
en: 0023ae7fa3b5d0bd
|
||||
zh: 3a4ab42af9302dca
|
||||
en: 0901d0bd388a4034
|
||||
zh: 73b80f267d2dfc31
|
||||
/deepseek-ai-dsh-ptc-runtime-node/use-this-package/deadlines-and-cancellation:
|
||||
en: ad72b8bdba6a3966
|
||||
zh: 28a20fb5fb7c622d
|
||||
@@ -33,17 +33,17 @@
|
||||
en: 7682e50f9dcfa25c
|
||||
zh: ddcb3502e12be06f
|
||||
/deepseek-ai-dsh-ptc-runtime-node/understand-the-implementation:
|
||||
en: 3e4a87c9b90692c1
|
||||
zh: 74195f0bf7e04368
|
||||
en: 50e6cd38e298da31
|
||||
zh: c8b252b3acf59147
|
||||
/deepseek-ai-dsh-ptc-runtime-node/understand-the-implementation/launch-and-control:
|
||||
en: 894cf1c1fdbe0092
|
||||
zh: c1e1179391e89a3f
|
||||
en: 53e24b6ba2614d4c
|
||||
zh: 0fe4d5ace9d2ef75
|
||||
/deepseek-ai-dsh-ptc-runtime-node/understand-the-implementation/source-and-built-bootstraps:
|
||||
en: 1c8c0109032cc729
|
||||
zh: f85bc632d616afd4
|
||||
en: 3e30860547f88777
|
||||
zh: d7ba84b1f88c3fb2
|
||||
/deepseek-ai-dsh-ptc-runtime-node/understand-the-implementation/source-map:
|
||||
en: aa80cd23ba951724
|
||||
zh: b6eaaa36466e7f15
|
||||
en: a797a875fbfb0598
|
||||
zh: 0cc0b19f8241f6bc
|
||||
/deepseek-ai-dsh-ptc-runtime-node/further-exploration:
|
||||
en: ac2e2ae13d3f1394
|
||||
zh: 5d08b9c658fcf83a
|
||||
|
||||
@@ -87,7 +87,7 @@ The `persistent-in-memory` mode and Script mode with `scriptPersistence: true` r
|
||||
executionMode: persistent-in-memory
|
||||
```
|
||||
|
||||
Save the patch and pass its path to `dsh headless --patch <path>` in a composition with PTC tool presentation. The `per-call` mode and Script mode with `scriptPersistence: false` retain no program memory between calls. Changing the configuration reloads the provider and releases its execution resources. Packaged single-executable hosts require a `node-script` launch with `launch.executable` pointing to Node and `launch.bootstrapPath` pointing to this package's built `persistent-process.js` for in-memory execution.
|
||||
Save the patch and pass its path to `dsh headless --patch <path>` in a composition with PTC tool presentation. The `per-call` mode and Script mode with `scriptPersistence: false` retain no program memory between calls. Changing the configuration reloads the provider and releases its execution resources. Packaged single-executable hosts require a `node-script` launch with `launch.executable` pointing to Node and `launch.bootstrapPath` pointing to this package's built `process.js` for in-memory execution.
|
||||
|
||||
In REPL mode, each live Agent keeps its own process and top-level variables, functions, and imported modules across `run_code` calls and turns. Use `await import(...)`, `console.log(...)`, and a final JSON expression; top-level `return` is not supported. Assignments to undeclared names create retained globals under Node REPL’s non-strict evaluation. `_` and `_error` are ordinary retained globals, without interactive REPL result setters. Node REPL declaration rules apply, including errors on redeclaring existing `let`, `const`, or `class` names; assign to a mutable binding or choose a new name. A leading `{ ... }` is a statement block and can contain top-level await and temporary declarations; parenthesize an object expression to return it. Retained functions and saved tool methods call the current evaluation's bindings. All tool calls must be awaited; callbacks from a completed evaluation cannot invoke tools in a later evaluation.
|
||||
|
||||
@@ -117,17 +117,17 @@ Program parse errors and thrown exceptions are `exception`; deadline expiry is `
|
||||
|
||||
The host owns policy, deadlines, binding lookup and process cleanup. The child owns program evaluation and binding proxies; model-written code is an untrusted peer even when its messages use the expected control descriptor.
|
||||
|
||||
`createExecution()` resolves the configured syntax, type erasure, model guidance and process lifetime once when the provider loads. The provider handles admission and cancellation, then delegates to the one-shot or retained process owner. Both child entry points use `createProgramEvaluator()`, whose async-function, REPL and Script implementations share globals, evaluation and cleanup operations. Process ownership does not select or interpret program syntax.
|
||||
`createExecution()` resolves the configured syntax, type erasure, model guidance and process lifetime once when the provider loads. The provider handles admission and cancellation. One execution owner creates temporary processes or reuses an enabled state owner's process; both use `ProgramWorker` for launch, deadlines, binding dispatch and cleanup. The shared child loop selects `createProgramEvaluator()` from the evaluation data; async-function, REPL and Script implementations expose globals, evaluation and cleanup operations independently of process lifetime.
|
||||
|
||||
### Launch and control
|
||||
|
||||
For per-call execution, the host strips erasable types, resolves the executable and bootstrap in the configured execution world, awaits argv confinement through `ctx.sandbox`, then spawns through `ctx.subprocess`. Cancellation is checked again after confinement, so a provider returning after cancellation cannot start the program. After adopting the inherited control channel, the child retains only executable-search, Windows system, and temporary paths in its OS environment and replaces the program-visible `process.env` with an empty dictionary. Windows ACL setup receives the parent's distinct `TEMP` and `TMP` values for shared grant locks, then replaces both with its private directory before starting the program. These native paths keep nested process creation and native temporary-file APIs functional. The host preserves `ELECTRON_RUN_AS_NODE` only for child startup so the Desktop executable runs the Node bootstrap; the bootstrap removes the selector before evaluating model code. The heap limit uses Node argv or a provider-created `NODE_OPTIONS` value for packaged executables; ambient loader and inspector flags are discarded.
|
||||
The host strips erasable types and, when starting a process, resolves the executable and bootstrap in the configured execution world, awaits argv confinement through `ctx.sandbox`, then spawns through `ctx.subprocess`. Cancellation is checked again after confinement, so a provider returning after cancellation cannot start the program. After adopting the inherited control channel, the child retains only executable-search, Windows system, and temporary paths in its OS environment and replaces the program-visible `process.env` with an empty dictionary. Windows ACL setup receives the parent's distinct `TEMP` and `TMP` values for shared grant locks, then replaces both with its private directory before starting the program. These native paths keep nested process creation and native temporary-file APIs functional. The host preserves `ELECTRON_RUN_AS_NODE` only for child startup so the Desktop executable runs the Node bootstrap; the bootstrap removes the selector before evaluating model code. The heap limit uses Node argv or a provider-created `NODE_OPTIONS` value for packaged executables; ambient loader and inspector flags are discarded.
|
||||
|
||||
Length-framed JSON travels separately from stdout/stderr. The host bounds frames and queued writes, validates call identity and declared binding names before dispatch, and refuses invalid traffic. The child flushes its terminal frame and keeps the control channel open until the host closes it. After submitting that frame, it ignores later binding replies and sends no further program control messages. Output capture meters serialized logs plus the completion or diagnostic; fixed result-envelope fields and sandbox metadata are outside that ledger.
|
||||
Length-framed JSON travels separately from stdout/stderr. Every evaluation carries its identity, cwd, syntax and binding declarations through the same run protocol. The host bounds frames and queued writes, validates call identity and binding lookup before dispatch, and refuses invalid traffic. The child keeps the channel open between evaluations until the host closes it. Later binding replies are ignored after an evaluation's terminal frame; its callbacks cannot emit further binding calls or captured logs. `JsonChannel` owns write tracking and drainage; decoded replies settle binding calls directly. Output capture meters serialized logs plus the completion or diagnostic; fixed result-envelope fields and sandbox metadata are outside that ledger.
|
||||
|
||||
### Source and built bootstraps
|
||||
|
||||
Source execution loads an erasable-only bootstrap closure without relying on sibling built exports. Built script execution uses `process.js` for one-shot programs or `persistent-process.js` for retained REPL and Script environments. Persistent mode requires a `node-script` launch. A remote script deployment supplies the matching `launch.bootstrapPath`. Embedded carriers launch their private worker through `launch.executable`. `resolveLaunch()` selects the local default during construction; an explicit remote invocation never depends on whether the client itself is packaged. A host path is never assumed to name the same remote file.
|
||||
Source execution loads an erasable-only bootstrap closure without relying on sibling built exports. Built script execution uses `process.js` for every mode; `persistent-process.js` delegates to the same loop with a REPL default for hosts that omit program syntax. Persistent mode requires a `node-script` launch. A remote script deployment supplies the matching `launch.bootstrapPath`. Embedded carriers launch their private worker through `launch.executable`. `resolveLaunch()` selects the local default during construction; an explicit remote invocation never depends on whether the client itself is packaged. A host path is never assumed to name the same remote file.
|
||||
|
||||
Scripts accepted by the native compiler execute unchanged. When compilation requires top-level await support, Acorn parses an async wrapper around the source. The transform follows Node's REPL visitors, declaration hoisting and character-position edits, including its boxed final-expression result; the evaluator unwraps that result and awaits its value. A local AST walker keeps Acorn as the only added dependency. Dynamic `import()` expressions remain native and use the compiling Script's `importModuleDynamically` option. Both private process bundles include Acorn; source execution imports the package directly.
|
||||
|
||||
@@ -137,10 +137,9 @@ Scripts accepted by the native compiler execute unchanged. When compilation requ
|
||||
|---|---|
|
||||
| [`src/index.ts`](src/index.ts) | Node provider: common admission, cancellation and service delegation |
|
||||
| [`src/config.ts`](src/config.ts) | Execution mode, validated limits and defaults |
|
||||
| [`src/execution.ts`](src/execution.ts) | Mode selection, type erasure, model guidance and process-owner selection |
|
||||
| [`src/per-call.ts`](src/per-call.ts) | One-shot process execution and cleanup |
|
||||
| [`src/persistent.ts`](src/persistent.ts) | In-memory process ownership, evaluation deadlines and cleanup |
|
||||
| [`src/persistent-process.ts`](src/persistent-process.ts) | Retained child lifecycle and current binding dispatch |
|
||||
| [`src/execution.ts`](src/execution.ts) | Mode selection, request resolution and temporary or retained process ownership |
|
||||
| [`src/worker.ts`](src/worker.ts) | Shared process launch, evaluation deadlines, binding dispatch and cleanup |
|
||||
| [`src/persistent-process.ts`](src/persistent-process.ts) | Installed REPL-default entry delegating to the common child loop |
|
||||
| [`src/evaluators/index.ts`](src/evaluators/index.ts) | Shared selection entry for the three evaluators |
|
||||
| [`src/evaluators/async-function.ts`](src/evaluators/async-function.ts) | AsyncFunction: strict async function bodies |
|
||||
| [`src/evaluators/repl.ts`](src/evaluators/repl.ts) | Node REPL: interactive declarations and retained context |
|
||||
@@ -148,7 +147,7 @@ Scripts accepted by the native compiler execute unchanged. When compilation requ
|
||||
| [`src/evaluators/vm-script-await.ts`](src/evaluators/vm-script-await.ts) | VM Script top-level await lowering and retained declarations |
|
||||
| [`src/output-stream.ts`](src/output-stream.ts) | Bounded stream capture and process-output draining |
|
||||
| [`src/launch.ts`](src/launch.ts) | Executable/bootstrap arguments and execution-world asset mapping |
|
||||
| [`src/process.ts`](src/process.ts) | Child handshake, environment clearing and program lifecycle |
|
||||
| [`src/process.ts`](src/process.ts) | Common child handshake, evaluator lifetime and direct binding replies |
|
||||
| [`src/bootstrap.ts`](src/bootstrap.ts) | Program evaluation, binding proxies and output capture |
|
||||
| [`src/channel.ts`](src/channel.ts) | Framing, bounded writes and protocol failures |
|
||||
| [`src/output-ledger.ts`](src/output-ledger.ts) | Host accounting for the outer result |
|
||||
|
||||
@@ -87,7 +87,7 @@ kind: "package-reference"
|
||||
executionMode: persistent-in-memory
|
||||
```
|
||||
|
||||
保存补丁，在使用 PTC 工具展示的配置中通过 `dsh headless --patch <path>` 加载。`per-call` 模式和 `scriptPersistence: false` 的 Script 模式不跨调用保留程序内存。修改配置会重载提供方并释放其执行资源。单文件打包宿主使用内存保留模式时，需要使用 `node-script` 启动方式，将 `launch.executable` 指向 Node，并将 `launch.bootstrapPath` 指向本包构建出的 `persistent-process.js`。
|
||||
保存补丁，在使用 PTC 工具展示的配置中通过 `dsh headless --patch <path>` 加载。`per-call` 模式和 `scriptPersistence: false` 的 Script 模式不跨调用保留程序内存。修改配置会重载提供方并释放其执行资源。单文件打包宿主使用内存保留模式时，需要使用 `node-script` 启动方式，将 `launch.executable` 指向 Node，并将 `launch.bootstrapPath` 指向本包构建出的 `process.js`。
|
||||
|
||||
在 REPL 模式下，每个存活 Agent 保留独立进程，跨 `run_code` 调用和轮次保留顶层变量、函数及导入模块。使用 `await import(...)`、`console.log(...)` 和末尾 JSON 表达式；不支持顶层 `return`。Node REPL 使用非严格模式，给未声明的名称赋值会创建保留的全局变量。`_` 和 `_error` 是普通的保留全局变量，不使用交互式 REPL 的结果设置器。声明遵循 Node REPL 规则，包括重复声明已有 `let`、`const` 或 `class` 名称时报错；应给可变绑定赋值或选择新名称。开头的 `{ ... }` 按语句块执行，可以包含顶层 await 和临时声明；需要返回对象表达式时，用圆括号包裹。保留的函数和工具方法引用调用当前执行的绑定。必须等待所有工具调用；已完成执行的回调不能在后续执行中调用工具。
|
||||
|
||||
@@ -117,17 +117,17 @@ PTC 消费方按 [dsh-tools](../../core/tools/README.zh.md#ptc-mode) 的说明
|
||||
|
||||
Host 负责策略、截止时间、绑定查找和进程清理。子进程负责程序求值与绑定代理；即使使用预期的控制描述符，模型编写的代码仍是不可信对端。
|
||||
|
||||
`createExecution()` 在提供方加载时一次性确定配置对应的语法、类型擦除、模型指导和进程生命周期。提供方处理执行准入与取消，再委托单次或保留进程的所有者执行。两个子进程入口都使用 `createProgramEvaluator()`；异步函数、REPL 和 Script 实现共用全局变量、求值与清理操作。进程所有权代码不选择或解释程序语法。
|
||||
`createExecution()` 在提供方加载时一次性确定配置对应的语法、类型擦除、模型指导和进程生命周期。提供方处理执行准入与取消。统一的执行所有者创建临时进程，或在启用状态保留时复用所有者的进程；二者都由 `ProgramWorker` 负责启动、截止时间、绑定分派和清理。共用的子进程循环根据本次执行数据选择 `createProgramEvaluator()`；异步函数、REPL 和 Script 实现提供全局变量、求值与清理操作，与进程生命周期独立。
|
||||
|
||||
### 启动与控制
|
||||
|
||||
Host 擦除可擦除类型，在配置的执行世界中解析可执行文件与 bootstrap，通过 `ctx.sandbox` 等待 argv 限制准备完成，再通过 `ctx.subprocess` 启动。限制准备完成后会再次检查取消状态，因此提供方在取消后返回也无法启动程序。接管继承的控制通道后，子进程在 OS 环境中只保留可执行文件搜索路径、Windows 系统路径和临时路径，并将程序可见的 `process.env` 替换为空字典。Windows ACL 初始化接收父进程各自的 `TEMP` 和 `TMP` 值以使用共享授权锁，然后在启动程序前将二者替换为私有目录。这些原生路径使嵌套进程创建和原生临时文件 API 仍可正常工作。Host 仅在子进程启动时保留 `ELECTRON_RUN_AS_NODE`，使桌面端可执行文件运行 Node bootstrap；bootstrap 在求值模型代码前删除该选择变量。堆上限通过 Node argv 或为打包可执行文件由提供方构造的 `NODE_OPTIONS` 值传递；环境中的加载器和调试器标志会被丢弃。
|
||||
Host 擦除可擦除类型；需要启动进程时，在配置的执行世界中解析可执行文件与 bootstrap，通过 `ctx.sandbox` 等待 argv 限制准备完成，再通过 `ctx.subprocess` 启动。限制准备完成后会再次检查取消状态，因此提供方在取消后返回也无法启动程序。接管继承的控制通道后，子进程在 OS 环境中只保留可执行文件搜索路径、Windows 系统路径和临时路径，并将程序可见的 `process.env` 替换为空字典。Windows ACL 初始化接收父进程各自的 `TEMP` 和 `TMP` 值以使用共享授权锁，然后在启动程序前将二者替换为私有目录。这些原生路径使嵌套进程创建和原生临时文件 API 仍可正常工作。Host 仅在子进程启动时保留 `ELECTRON_RUN_AS_NODE`，使桌面端可执行文件运行 Node bootstrap；bootstrap 在求值模型代码前删除该选择变量。堆上限通过 Node argv 或为打包可执行文件由提供方构造的 `NODE_OPTIONS` 值传递；环境中的加载器和调试器标志会被丢弃。
|
||||
|
||||
带长度分帧的 JSON 与 stdout/stderr 分开传输。Host 限制帧与排队写入，在分派前验证调用身份和已声明的绑定名，并拒绝无效通信。子进程刷新终态帧后仍保持控制通道打开，直到 Host 关闭通道。提交终态帧后，子进程忽略后续绑定回复，不再发送程序控制消息。输出捕获计量序列化日志加完成值或诊断；固定结果信封字段与沙箱元数据不计入该账本。
|
||||
带长度分帧的 JSON 与 stdout/stderr 分开传输。每次执行都通过同一种 run 协议携带标识、cwd、语法和绑定声明。Host 限制帧与排队写入，在分派前验证调用身份和绑定查找，并拒绝无效通信。子进程在执行间保持通道打开，直到 Host 关闭通道。提交某次执行的终态帧后，子进程忽略该次执行后续的绑定回复，其回调也不能再发送绑定调用或捕获的日志。`JsonChannel` 统一跟踪写入并等待排空；解码后的回复直接结算绑定调用。输出捕获计量序列化日志加完成值或诊断；固定结果信封字段与沙箱元数据不计入该账本。
|
||||
|
||||
### 源代码与构建后 bootstrap
|
||||
|
||||
源代码执行加载仅含可擦除语法的 bootstrap 依赖，不依赖同级包的构建后导出。构建后脚本通过 `process.js` 运行单次程序，通过 `persistent-process.js` 运行保留状态的 REPL 和 Script 环境。保留模式要求 `node-script` 启动方式。远端脚本部署提供对应的 `launch.bootstrapPath`。内嵌载体通过 `launch.executable` 启动私有 worker。`resolveLaunch()` 在构造时选择本地默认值；显式的远端调用不依赖客户端自身是否打包。不会假设 Host 路径对应同一个远程文件。
|
||||
源代码执行加载仅含可擦除语法的 bootstrap 依赖，不依赖同级包的构建后导出。构建后脚本在所有模式下都使用 `process.js`；`persistent-process.js` 委托同一个循环，供未指定程序语法的宿主默认使用 REPL。保留模式要求 `node-script` 启动方式。远端脚本部署提供对应的 `launch.bootstrapPath`。内嵌载体通过 `launch.executable` 启动私有 worker。`resolveLaunch()` 在构造时选择本地默认值；显式的远端调用不依赖客户端自身是否打包。不会假设 Host 路径对应同一个远程文件。
|
||||
|
||||
原生编译器接受的脚本直接执行。当编译需要顶层 await 支持时，Acorn 解析包住源码的异步函数。转换沿用 Node REPL 的 visitor、声明提升和按字符位置修改源码的方式，包括对末尾表达式结果的包装；求值器解包后等待该值。局部 AST 遍历器使新增依赖仍只有 Acorn。动态 `import()` 表达式保持原生形式，使用编译该 Script 时的 `importModuleDynamically` 选项。两个私有进程 bundle 都包含 Acorn；源码执行直接导入该包。
|
||||
|
||||
@@ -137,10 +137,9 @@ Host 擦除可擦除类型，在配置的执行世界中解析可执行文件与
|
||||
|---|---|
|
||||
| [`src/index.ts`](src/index.ts) | Node 提供方：统一执行准入、取消与服务委托 |
|
||||
| [`src/config.ts`](src/config.ts) | 执行模式、限制校验与默认值 |
|
||||
| [`src/execution.ts`](src/execution.ts) | 模式选择、类型擦除、模型指导与进程所有者选择 |
|
||||
| [`src/per-call.ts`](src/per-call.ts) | 单次进程执行与清理 |
|
||||
| [`src/persistent.ts`](src/persistent.ts) | 内存保留模式的进程所有权、执行截止与清理 |
|
||||
| [`src/persistent-process.ts`](src/persistent-process.ts) | 保留状态的子进程生命周期与当前绑定分派 |
|
||||
| [`src/execution.ts`](src/execution.ts) | 模式选择、请求解析及临时或保留进程的所有权 |
|
||||
| [`src/worker.ts`](src/worker.ts) | 共用的进程启动、执行截止、绑定分派与清理 |
|
||||
| [`src/persistent-process.ts`](src/persistent-process.ts) | 默认使用 REPL 的已安装入口，委托共用子进程循环 |
|
||||
| [`src/evaluators/index.ts`](src/evaluators/index.ts) | 三种求值器的统一选择入口 |
|
||||
| [`src/evaluators/async-function.ts`](src/evaluators/async-function.ts) | AsyncFunction：严格模式异步函数体 |
|
||||
| [`src/evaluators/repl.ts`](src/evaluators/repl.ts) | Node REPL：交互式声明与保留的 context |
|
||||
@@ -148,7 +147,7 @@ Host 擦除可擦除类型，在配置的执行世界中解析可执行文件与
|
||||
| [`src/evaluators/vm-script-await.ts`](src/evaluators/vm-script-await.ts) | VM Script 的顶层 await 转换与声明保留 |
|
||||
| [`src/output-stream.ts`](src/output-stream.ts) | 有界流捕获与进程输出排空 |
|
||||
| [`src/launch.ts`](src/launch.ts) | 可执行文件／bootstrap 参数与执行世界资源映射 |
|
||||
| [`src/process.ts`](src/process.ts) | 子进程握手、环境清空与程序生命周期 |
|
||||
| [`src/process.ts`](src/process.ts) | 共用的子进程握手、求值器生命周期与直接绑定回复 |
|
||||
| [`src/bootstrap.ts`](src/bootstrap.ts) | 程序求值、绑定代理与输出捕获 |
|
||||
| [`src/channel.ts`](src/channel.ts) | 分帧、有界写入与协议失败 |
|
||||
| [`src/output-ledger.ts`](src/output-ledger.ts) | Host 外层结果计量 |
|
||||
|
||||
@@ -1,13 +1,15 @@
|
||||
/** Resolve evaluation syntax and process lifetime once for the Node provider. */
|
||||
/** Select program syntax and own temporary or retained Node processes. */
|
||||
import { stripTypeScriptTypes } from 'node:module'
|
||||
import { isAbsolute } from 'node:path'
|
||||
import type { Context } from '@deepseek-ai/cordis'
|
||||
import type { PtcRunRequest, PtcRunResult, PtcRunSpec } from '@deepseek-ai/dsh-ptc-runtime'
|
||||
import { assertNever } from '@deepseek-ai/dsh-util-values'
|
||||
import { clampTimeout } from '@deepseek-ai/dsh-timeout'
|
||||
import type { ResolvedConfig } from './config.ts'
|
||||
import type { ProgramFormat } from './evaluators/types.ts'
|
||||
import { PerCallExecution } from './per-call.ts'
|
||||
import { InMemoryExecution } from './persistent.ts'
|
||||
import { validateBindings } from './bindings.ts'
|
||||
import type { PolicyProcesses } from './policy-processes.ts'
|
||||
import { ProgramWorker } from './worker.ts'
|
||||
|
||||
/** Syntax, state retention and model guidance selected during provider construction. */
|
||||
export interface ExecutionMode {
|
||||
@@ -25,53 +27,124 @@ export interface ExecutionMode {
|
||||
stripTypes(program: string): string
|
||||
}
|
||||
|
||||
/** A resolved run carrying provider cancellation as well as caller cancellation. */
|
||||
export type AdmittedRun = PtcRunSpec & { signal: AbortSignal }
|
||||
/**
|
||||
* Select program semantics and create their shared process owner.
|
||||
* @param ctx - Provider lifecycle and execution services.
|
||||
* @param config - Validated configuration with loader defaults.
|
||||
* @param processes - Policy invalidation and process cleanup coordination.
|
||||
* @returns Mode selection and temporary or retained execution through one process implementation.
|
||||
*/
|
||||
export function createExecution(ctx: Context, config: ResolvedConfig, processes: PolicyProcesses): NodeExecution {
|
||||
return new NodeExecution(ctx, config, processes, resolveMode(config))
|
||||
}
|
||||
|
||||
/** Temporary processes and optional object-owned reuse share one process implementation. */
|
||||
class NodeExecution {
|
||||
private readonly owners = new WeakMap<object, ProgramWorker>()
|
||||
private readonly workers = new Set<ProgramWorker>()
|
||||
|
||||
/** Process ownership shared by one-shot and retained execution. */
|
||||
export interface ExecutionRunner {
|
||||
/**
|
||||
* Resolve policy, directory and elapsed budget.
|
||||
* @param request - Program and optional execution settings.
|
||||
* @returns Complete inputs for run.
|
||||
* Own process instances and optional owner lookup for one provider.
|
||||
* @param ctx - Provider lifecycle and execution services.
|
||||
* @param config - Validated limits and selected executable.
|
||||
* @param processes - Provider-owned policy invalidation and cleanup.
|
||||
* @param mode - Program syntax and TypeScript erasure.
|
||||
*/
|
||||
resolve(request: PtcRunRequest): PtcRunSpec
|
||||
constructor(
|
||||
private readonly ctx: Context, private readonly config: ResolvedConfig,
|
||||
private readonly processes: PolicyProcesses, readonly mode: ExecutionMode,
|
||||
) {
|
||||
if (mode.retainsState && this.config.launch.kind === 'embedded') throw new Error('Persistent PTC requires a node-script launch')
|
||||
}
|
||||
|
||||
/** Stop every owned process, retaining failed cleanup for a later retry. */
|
||||
async dispose(): Promise<void> {
|
||||
const results = await Promise.allSettled([...this.workers].map(async (worker) => {
|
||||
await worker.dispose()
|
||||
this.workers.delete(worker)
|
||||
}))
|
||||
const failed = results.find(result => result.status === 'rejected')
|
||||
if (failed?.status === 'rejected') throw failed.reason
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute an admitted call under the selected process lifetime.
|
||||
* @param spec - Resolved inputs with combined cancellation.
|
||||
* @param controller - Provider-owned cancellation, also used for one-shot deadlines.
|
||||
* @returns Captured output or a structured execution failure after required cleanup.
|
||||
* Resolve the common file policy, directory and deadline defaults.
|
||||
* @param request - Program, bindings, owner, directory and elapsed budget.
|
||||
* @returns Resolved inputs with the selected file policy.
|
||||
*/
|
||||
run(spec: AdmittedRun, controller: AbortController): Promise<PtcRunResult>
|
||||
resolve(request: PtcRunRequest): PtcRunSpec {
|
||||
const sandboxPolicy = request.sandboxPolicy ?? this.ctx.sandboxPolicy.resolve()
|
||||
const cwd = request.cwd ?? sandboxPolicy.workspaceRoot
|
||||
if (!isAbsolute(cwd)) throw new Error(`${this.mode.retainsState ? 'persistent PTC' : 'ptc-runtime-node'}: cwd must be absolute`)
|
||||
return { ...request, sandboxPolicy, cwd, timeoutMs: request.timeoutMs === null ? null : clampTimeout(request.timeoutMs, this.config.timeoutMs, this.config.maxTimeoutMs, this.mode.retainsState ? 'persistent PTC timeoutMs' : 'ptc-runtime-node: timeoutMs') }
|
||||
}
|
||||
|
||||
/**
|
||||
* Release resources retained for a caller-owned object.
|
||||
* Evaluate in a temporary process or reuse an enabled state owner's process.
|
||||
* @param spec - Resolved request; concurrent calls for one owner are rejected.
|
||||
* @returns Captured logs, JSON completion, or a structured execution failure.
|
||||
*/
|
||||
async run(spec: PtcRunSpec): Promise<PtcRunResult> {
|
||||
if (spec.sandboxPolicy === undefined) throw new Error('ptc-runtime-node: run requires a resolved sandbox policy')
|
||||
if (!isAbsolute(spec.cwd) || (spec.timeoutMs !== null && (!Number.isFinite(spec.timeoutMs) || spec.timeoutMs <= 0 || spec.timeoutMs > this.config.maxTimeoutMs))) throw new Error('ptc-runtime-node: run requires resolved cwd and timeout')
|
||||
const bindings = validateBindings(spec)
|
||||
const owner = this.mode.retainsState ? spec.stateOwner : undefined
|
||||
let worker = owner === undefined ? undefined : this.owners.get(owner)
|
||||
worker?.assertIdle()
|
||||
if (worker && !worker.closed && (worker.policy.mode !== spec.sandboxPolicy.mode
|
||||
|| worker.policy.workspaceRoot !== spec.sandboxPolicy.workspaceRoot
|
||||
|| worker.policy.sessionId !== spec.sandboxPolicy.sessionId
|
||||
|| !sameAdditionalRoots(worker.policy.additionalWritableRoots, spec.sandboxPolicy.additionalWritableRoots))) {
|
||||
throw new Error('persistent PTC: release the retained environment before changing its sandbox policy')
|
||||
}
|
||||
if (!worker || worker.closed) {
|
||||
if (worker) this.workers.delete(worker)
|
||||
// Direct callers can own mutable policy arrays; the retained process owns its grant snapshot.
|
||||
worker = new ProgramWorker(this.ctx, this.config, {
|
||||
...spec.sandboxPolicy,
|
||||
...spec.sandboxPolicy.additionalWritableRoots === undefined
|
||||
? {}
|
||||
: { additionalWritableRoots: [...spec.sandboxPolicy.additionalWritableRoots] },
|
||||
}, this.processes, this.mode)
|
||||
this.workers.add(worker)
|
||||
if (owner !== undefined) this.owners.set(owner, worker)
|
||||
}
|
||||
try { return await worker.run(spec, bindings) } finally {
|
||||
if (owner === undefined && worker.cleanupComplete) this.workers.delete(worker)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Release retained state and await its process cleanup.
|
||||
* @param owner - Exact object supplied to run.
|
||||
* @returns Completion after owned processes stop; immediate for one-shot execution.
|
||||
* @returns Completion after the owned process and output streams stop.
|
||||
*/
|
||||
release(owner: object): Promise<void>
|
||||
async release(owner: object): Promise<void> {
|
||||
const worker = this.owners.get(owner)
|
||||
if (!worker) return
|
||||
await worker.dispose()
|
||||
if (this.owners.get(owner) === worker) this.owners.delete(owner)
|
||||
this.workers.delete(worker)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare requested extra grants without resolving paths in another execution world.
|
||||
* @param left - one policy's additional roots.
|
||||
* @param right - the other policy's additional roots.
|
||||
* @returns whether both contain the same paths, ignoring order and duplicates.
|
||||
*/
|
||||
function sameAdditionalRoots(left: readonly string[] | undefined, right: readonly string[] | undefined): boolean {
|
||||
const effective = (roots: readonly string[] | undefined): string[] => [...new Set(roots ?? [])].sort()
|
||||
const a = effective(left)
|
||||
const b = effective(right)
|
||||
return a.length === b.length && a.every((root, index) => root === b[index])
|
||||
}
|
||||
|
||||
const STRIP_PREFIX = 'async function __dsh_program__() {\n'
|
||||
const STRIP_SUFFIX = '\n}'
|
||||
const RETAINED_STATE_INSTRUCTIONS = 'Changing the session sandbox mode clears retained variables. An approved sandbox escalation runs in a separate temporary environment: it cannot access retained variables, and its declarations are discarded afterward. Restart, Agent disposal, cancellation, timeout, output overflow, or process failure clears the environment. If a variable is undefined, recreate it and reacquire its resources. State is not restored from the conversation, and failed evaluations can leave earlier mutations in memory. Redeclaring an existing let, const, or class name fails; assign to a mutable binding or use a new name. Await all tool calls; background callbacks cannot call tools after their evaluation ends. Await child-process completion before ending an evaluation; raw subprocess output belongs to whichever evaluation is active when it arrives.'
|
||||
|
||||
/**
|
||||
* Select the configured evaluator and its process owner.
|
||||
* @param ctx - Provider lifecycle and execution services.
|
||||
* @param config - Validated configuration with loader defaults.
|
||||
* @param processes - Policy invalidation and process cleanup coordination.
|
||||
* @returns Immutable mode selection and the corresponding execution runner.
|
||||
*/
|
||||
export function createExecution(
|
||||
ctx: Context, config: ResolvedConfig, processes: PolicyProcesses,
|
||||
): { mode: ExecutionMode; runner: ExecutionRunner } {
|
||||
const mode = resolveMode(config)
|
||||
const runner = mode.retainsState
|
||||
? new InMemoryExecution(ctx, config, processes, mode)
|
||||
: new PerCallExecution(ctx, config, processes, mode)
|
||||
return { mode, runner }
|
||||
}
|
||||
|
||||
function resolveMode(config: ResolvedConfig): ExecutionMode {
|
||||
switch (config.executionMode) {
|
||||
case 'per-call': return {
|
||||
|
||||
@@ -41,14 +41,16 @@ export class NodePtcRuntime extends PtcRuntime {
|
||||
this.disposed = true
|
||||
const active = [...this.live]
|
||||
for (const run of active) run.controller.abort('runtime disposed')
|
||||
await Promise.all(active.map(run => run.finished))
|
||||
const results = await Promise.allSettled([this.execution.dispose(), ...active.map(run => run.finished)])
|
||||
const failed = results.find(result => result.status === 'rejected')
|
||||
if (failed?.status === 'rejected') throw failed.reason
|
||||
}, 'Node ptc-runtime cleanup')
|
||||
}
|
||||
|
||||
override get sandboxMode(): SandboxMode | undefined { return this.ctx.sandboxPolicy.defaultMode }
|
||||
|
||||
override release(owner: object): Promise<void> {
|
||||
return this.execution.runner.release(owner)
|
||||
return this.execution.release(owner)
|
||||
}
|
||||
|
||||
override get timeout(): { defaultMs: number; maxMs: number } {
|
||||
@@ -65,7 +67,7 @@ export class NodePtcRuntime extends PtcRuntime {
|
||||
const source = this.retainsState ? 'persistent PTC' : 'ptc-runtime-node'
|
||||
throw new Error(`${source}: resolve after disposal`)
|
||||
}
|
||||
return this.execution.runner.resolve(request)
|
||||
return this.execution.resolve(request)
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -86,8 +88,11 @@ export class NodePtcRuntime extends PtcRuntime {
|
||||
this.live.add(live)
|
||||
try {
|
||||
await this.processes.wait(spec.sandboxPolicy?.sessionId)
|
||||
// Disposal can begin while a previous policy's process cleanup is pending.
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition
|
||||
if (this.disposed) throw new Error(`${this.retainsState ? 'persistent PTC' : 'ptc-runtime-node'}: run after disposal`)
|
||||
const signal = spec.signal === undefined ? controller.signal : AbortSignal.any([spec.signal, controller.signal])
|
||||
return await this.execution.runner.run({ ...spec, signal }, controller)
|
||||
return await this.execution.run({ ...spec, signal })
|
||||
} finally {
|
||||
unregister?.()
|
||||
this.live.delete(live)
|
||||
|
||||
@@ -3,7 +3,6 @@ import { dirname, isAbsolute, resolve } from 'node:path'
|
||||
import { fileURLToPath, pathToFileURL } from 'node:url'
|
||||
import type { FileSystem } from '@deepseek-ai/dsh-fs'
|
||||
import { assertNever } from '@deepseek-ai/dsh-util-values'
|
||||
import type { ProgramFormat } from './evaluators/types.ts'
|
||||
|
||||
/** Deployment-owned worker invocation in the subprocess execution world. */
|
||||
export type LaunchConfig = {
|
||||
@@ -39,27 +38,24 @@ export function resolveLaunch(config?: LaunchConfig): LaunchConfig {
|
||||
* @param fs - Filesystem mapping host bootstrap assets into the process world.
|
||||
* @param config - resolved worker invocation.
|
||||
* @param maxMessageBytes - Validated frame and queued-write limit.
|
||||
* @param persistent - Whether the process serves sequential evaluations.
|
||||
* @param format - Retained evaluator syntax; one-shot programs carry it in their boot payload.
|
||||
* @param persistent - Whether the selected mode requires retained-process launch support.
|
||||
* @returns Arguments following the resolved Node executable.
|
||||
*/
|
||||
export function bootstrapArgs(fs: Pick<FileSystem, 'processPathFromHostPath'>, config: LaunchConfig, maxMessageBytes: number, persistent: boolean = false, format: ProgramFormat = 'repl'): string[] {
|
||||
export function bootstrapArgs(fs: Pick<FileSystem, 'processPathFromHostPath'>, config: LaunchConfig, maxMessageBytes: number, persistent: boolean = false): string[] {
|
||||
switch (config.kind) {
|
||||
case 'embedded':
|
||||
if (persistent) throw new Error('Persistent PTC requires a node-script launch')
|
||||
return [String(maxMessageBytes)]
|
||||
case 'node-script': return scriptBootstrapArgs(fs, config.bootstrapPath, maxMessageBytes, persistent, format)
|
||||
case 'node-script': return scriptBootstrapArgs(fs, config.bootstrapPath, maxMessageBytes)
|
||||
/* v8 ignore next -- Callers supply the resolved, closed LaunchConfig union. */
|
||||
default: return assertNever(config)
|
||||
}
|
||||
}
|
||||
|
||||
function scriptBootstrapArgs(
|
||||
fs: Pick<FileSystem, 'processPathFromHostPath'>, bootstrapPath: string | undefined, maxMessageBytes: number, persistent: boolean, format: ProgramFormat,
|
||||
fs: Pick<FileSystem, 'processPathFromHostPath'>, bootstrapPath: string | undefined, maxMessageBytes: number,
|
||||
): string[] {
|
||||
const moduleName = persistent ? 'persistent-process' : 'process'
|
||||
const formatArgs = persistent && format !== 'repl' ? [format] : []
|
||||
if (bootstrapPath !== undefined) return [bootstrapPath, String(maxMessageBytes), ...formatArgs]
|
||||
if (bootstrapPath !== undefined) return [bootstrapPath, String(maxMessageBytes)]
|
||||
const mapped = (path: string): string => {
|
||||
const result = fs.processPathFromHostPath(path)
|
||||
if (result === undefined) throw new Error(`PTC runtime bootstrap is unavailable in the subprocess execution world: ${path}`)
|
||||
@@ -67,13 +63,11 @@ function scriptBootstrapArgs(
|
||||
}
|
||||
/* v8 ignore next 3 -- built-lib.e2e.ts executes the bundled provider and sibling process.js under plain Node. */
|
||||
if (!new URL(import.meta.url).pathname.endsWith('.ts')) {
|
||||
return [mapped(fileURLToPath(new URL(`./${moduleName}.js`, import.meta.url))), String(maxMessageBytes), ...formatArgs]
|
||||
return [mapped(fileURLToPath(new URL('./process.js', import.meta.url))), String(maxMessageBytes)]
|
||||
}
|
||||
const entry = mapped(fileURLToPath(new URL(`./${moduleName}.ts`, import.meta.url)))
|
||||
const entry = mapped(fileURLToPath(new URL('./process.ts', import.meta.url)))
|
||||
const subprocess = dirname(fileURLToPath(import.meta.resolve('@deepseek-ai/dsh-subprocess/package.json')))
|
||||
const helper = mapped(resolve(subprocess, 'src/control.ts'))
|
||||
const main = persistent ? 'runPersistentMain' : 'runNodeMain'
|
||||
const formatArgument = formatArgs.length > 0 ? `,${JSON.stringify(format)}` : ''
|
||||
const source = `const {openInheritedControlChannel}=await import(${JSON.stringify(pathToFileURL(helper).href)});const {${main}}=await import(${JSON.stringify(pathToFileURL(entry).href)});await ${main}(openInheritedControlChannel(),${maxMessageBytes},process${formatArgument});`
|
||||
const source = `const {openInheritedControlChannel}=await import(${JSON.stringify(pathToFileURL(helper).href)});const {runNodeMain}=await import(${JSON.stringify(pathToFileURL(entry).href)});await runNodeMain(openInheritedControlChannel(),${maxMessageBytes},process);`
|
||||
return ['--input-type=module', '--eval', source]
|
||||
}
|
||||
@@ -1,272 +0,0 @@
|
||||
/** One-shot process execution; evaluation syntax is supplied by the mode selector. */
|
||||
import { AsyncLocalStorage } from 'node:async_hooks'
|
||||
import { isAbsolute } from 'node:path'
|
||||
import type { Context } from '@deepseek-ai/cordis'
|
||||
import type { PtcBindingNamespace, PtcJsonValue, PtcRunFailure, PtcRunRequest, PtcRunResult, PtcRunSandbox, PtcRunSpec } from '@deepseek-ai/dsh-ptc-runtime'
|
||||
import { clampTimeout } from '@deepseek-ai/dsh-timeout'
|
||||
import { SandboxUnavailableError, classifyRunnerFailure, isRunnerSpawnFailure } from '@deepseek-ai/dsh-sandbox'
|
||||
import type { ConfinedArgv, SandboxExecutionPolicy } from '@deepseek-ai/dsh-sandbox'
|
||||
import type { SubprocessHandle, SubprocessOutcome } from '@deepseek-ai/dsh-subprocess'
|
||||
import { replyToBinding, validateBindings } from './bindings.ts'
|
||||
import { JsonChannel } from './channel.ts'
|
||||
import type { ResolvedConfig } from './config.ts'
|
||||
import type { AdmittedRun, ExecutionMode, ExecutionRunner } from './execution.ts'
|
||||
import { bootstrapArgs } from './launch.ts'
|
||||
import type { PolicyProcesses } from './policy-processes.ts'
|
||||
import { OutputLedger } from './output-ledger.ts'
|
||||
import { terminateProcess } from './output-stream.ts'
|
||||
import { STARTUP_ENVIRONMENT_NAMES } from './environment.ts'
|
||||
import { decodePtcJsonWire } from './json-wire.ts'
|
||||
import type { ProgramBootData } from './protocol.ts'
|
||||
|
||||
function messageOf(error: unknown): string { return error instanceof Error ? error.message : String(error) }
|
||||
function record(value: unknown): value is Record<string, unknown> { return typeof value === 'object' && value !== null && !Array.isArray(value) }
|
||||
|
||||
/** One managed process per call, released before its result is returned. */
|
||||
export class PerCallExecution implements ExecutionRunner {
|
||||
/**
|
||||
* Bind execution services and the resolved evaluation mode.
|
||||
* @param ctx - Provider execution services.
|
||||
* @param config - Validated resource limits and selected executable.
|
||||
* @param processes - Provider-owned policy invalidation and cleanup.
|
||||
* @param mode - Program syntax and TypeScript erasure.
|
||||
*/
|
||||
constructor(
|
||||
private readonly ctx: Context, private readonly config: ResolvedConfig,
|
||||
private readonly processes: PolicyProcesses, private readonly mode: ExecutionMode,
|
||||
) {}
|
||||
|
||||
resolve(request: PtcRunRequest): PtcRunSpec {
|
||||
const sandboxPolicy = request.sandboxPolicy ?? this.ctx.sandboxPolicy.resolve()
|
||||
const cwd = request.cwd ?? sandboxPolicy.workspaceRoot
|
||||
if (!isAbsolute(cwd)) throw new Error('ptc-runtime-node: cwd must be absolute')
|
||||
return {
|
||||
...request,
|
||||
cwd,
|
||||
timeoutMs: request.timeoutMs === null ? null : clampTimeout(request.timeoutMs, this.config.timeoutMs, this.config.maxTimeoutMs, 'ptc-runtime-node: timeoutMs'),
|
||||
sandboxPolicy,
|
||||
}
|
||||
}
|
||||
|
||||
async run(spec: AdmittedRun, controller: AbortController): Promise<PtcRunResult> {
|
||||
if (spec.sandboxPolicy === undefined) throw new Error('ptc-runtime-node: run requires a resolved sandbox policy')
|
||||
if (!isAbsolute(spec.cwd) || (spec.timeoutMs !== null && (!Number.isFinite(spec.timeoutMs) || spec.timeoutMs <= 0 || spec.timeoutMs > this.config.maxTimeoutMs))) throw new Error('ptc-runtime-node: run requires resolved cwd and timeout')
|
||||
const bindings = validateBindings(spec)
|
||||
return await this.execute(spec, spec.sandboxPolicy, bindings, controller)
|
||||
}
|
||||
|
||||
async release(): Promise<void> {}
|
||||
|
||||
private async execute(
|
||||
spec: AdmittedRun,
|
||||
policy: SandboxExecutionPolicy,
|
||||
bindings: Map<string, PtcBindingNamespace>,
|
||||
controller: AbortController,
|
||||
): Promise<PtcRunResult> {
|
||||
const replyBinding = AsyncLocalStorage.bind(replyToBinding)
|
||||
const output = new OutputLedger(this.config.maxOutputBytes)
|
||||
const logs: string[] = []
|
||||
const sandbox: PtcRunSandbox = { mode: policy.mode, denied: false }
|
||||
const result = Promise.withResolvers<PtcRunResult>()
|
||||
const signal = spec.signal
|
||||
let handle: SubprocessHandle | undefined
|
||||
let channel: JsonChannel | undefined
|
||||
let confined: ConfinedArgv | undefined
|
||||
let settled = false
|
||||
let timedOut = false
|
||||
let outputOverflow = false
|
||||
let overflowResult: PtcRunResult | undefined
|
||||
let stderr = ''
|
||||
let parsing = true
|
||||
const wallTimer = spec.timeoutMs === null ? undefined
|
||||
: setTimeout(() => { timedOut = true; controller.abort('execution deadline reached') }, spec.timeoutMs)
|
||||
let stopping: Promise<boolean> | undefined
|
||||
const stop = (): Promise<boolean> => {
|
||||
if (stopping !== undefined) return stopping
|
||||
const cleanup = handle === undefined ? Promise.resolve(true) : terminateProcess(handle, this.config.graceMs)
|
||||
stopping = cleanup.then((drained) => { unregister(); return drained }, (error: unknown) => {
|
||||
stopping = undefined
|
||||
throw error
|
||||
})
|
||||
return stopping
|
||||
}
|
||||
const finish = (failure?: PtcRunFailure, value?: PtcJsonValue): void => {
|
||||
if (settled) return
|
||||
settled = true
|
||||
clearTimeout(wallTimer)
|
||||
signal.removeEventListener('abort', onAbort)
|
||||
channel?.close()
|
||||
void (async () => {
|
||||
try {
|
||||
const drained = await stop()
|
||||
if (!drained && failure === undefined) {
|
||||
failure = { kind: 'worker-exit', message: 'Node process output did not close cleanly' }
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
failure = { kind: 'worker-exit', message: `managed process cleanup failed: ${messageOf(error)}` }
|
||||
}
|
||||
const outcome = outputOverflow ? overflowResult ?? output.limit(logs)
|
||||
: failure === undefined ? output.success(logs, value) : output.failure(logs, failure)
|
||||
result.resolve({ ...outcome, sandbox: { ...sandbox } })
|
||||
})()
|
||||
}
|
||||
const onAbort = (): void => {
|
||||
finish(timedOut
|
||||
? { kind: 'timeout', message: `execution deadline reached (${spec.timeoutMs}ms)` }
|
||||
: { kind: 'abort', message: messageOf(signal.reason) })
|
||||
}
|
||||
const unregister = this.processes.register(policy, async () => {
|
||||
controller.abort(new Error('sandbox mode changed'))
|
||||
await stop()
|
||||
})
|
||||
signal.addEventListener('abort', onAbort, { once: true })
|
||||
if (signal.aborted) onAbort()
|
||||
try {
|
||||
// Abort callbacks can settle execution before or during an awaited operation.
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition
|
||||
if (settled) return await result.promise
|
||||
const code = this.mode.stripTypes(spec.program)
|
||||
parsing = false
|
||||
const data: ProgramBootData = {
|
||||
code,
|
||||
programFormat: this.mode.format,
|
||||
namespaces: [...bindings.values()].map(binding => ({
|
||||
global: binding.global,
|
||||
names: Object.keys(binding.functions),
|
||||
...binding.errorClass ? { errorClass: binding.errorClass } : {},
|
||||
})),
|
||||
maxOutputBytes: this.config.maxOutputBytes,
|
||||
}
|
||||
const executable = await this.ctx.subprocess.resolveExecutable(this.config.launch.executable, undefined, signal)
|
||||
// Abort callbacks can settle execution before or during an awaited operation.
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition
|
||||
if (settled) return await result.promise
|
||||
const packaged = this.config.launch.kind === 'embedded'
|
||||
const heapFlag = `--max-old-space-size=${this.config.maxOldGenerationSizeMb}`
|
||||
const argv = [executable, ...packaged ? [] : [heapFlag],
|
||||
...bootstrapArgs(this.ctx.fs, this.config.launch, this.config.maxMessageBytes)]
|
||||
confined = policy.mode === 'danger-full-access' ? undefined : await this.ctx.sandbox.confine(argv, { ...policy, mode: policy.mode }, signal)
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition -- Cancellation can settle during awaited confinement.
|
||||
if (settled) return await result.promise
|
||||
if (confined !== undefined) sandbox.enforcement = confined.enforcement
|
||||
// Electron needs its Node-mode selector until bootstrap; the child then removes it with other ambient values.
|
||||
const env: NodeJS.ProcessEnv = Object.fromEntries(Object.keys(process.env)
|
||||
.filter(key => !STARTUP_ENVIRONMENT_NAMES.has(key.toUpperCase()) && key.toUpperCase() !== 'ELECTRON_RUN_AS_NODE')
|
||||
.map(key => [key, undefined]))
|
||||
if (packaged) {
|
||||
env.DSH_PTC_RUNTIME_NODE = '1'
|
||||
env.NODE_OPTIONS = heapFlag
|
||||
}
|
||||
handle = this.ctx.subprocess.spawn({ argv: confined?.argv ?? argv, cwd: spec.cwd, env, stdio: { stdin: 'ignore', stdout: 'pipe', stderr: 'pipe', control: 'pipe' }, graceMs: this.config.graceMs, signal })
|
||||
const launched = handle
|
||||
if (launched.control === undefined || launched.stdout === undefined || launched.stderr === undefined) {
|
||||
throw new Error('subprocess provider did not supply the requested control and output pipes')
|
||||
}
|
||||
const admit = (text: string): void => {
|
||||
if (outputOverflow) return
|
||||
if (!output.admit(text, logs)) {
|
||||
outputOverflow = true
|
||||
overflowResult = output.limit([...logs, text])
|
||||
finish({ kind: 'output-limit', message: `outer output exceeded ${this.config.maxOutputBytes} bytes` })
|
||||
}
|
||||
}
|
||||
const stdoutDecoder = new TextDecoder('utf-8', { ignoreBOM: true })
|
||||
const stderrDecoder = new TextDecoder('utf-8', { ignoreBOM: true })
|
||||
handle.stdout?.on('data', (chunk: Buffer) => {
|
||||
const text = stdoutDecoder.decode(chunk, { stream: true })
|
||||
if (text.length > 0) admit(text)
|
||||
})
|
||||
handle.stdout?.on('end', () => {
|
||||
const text = stdoutDecoder.decode()
|
||||
if (text.length > 0) admit(text)
|
||||
})
|
||||
handle.stdout?.on('error', (error: Error) => { finish({ kind: 'worker-exit', message: messageOf(error) }) })
|
||||
handle.stderr?.on('data', (chunk: Buffer) => {
|
||||
const text = stderrDecoder.decode(chunk, { stream: true })
|
||||
stderr = (stderr + text).slice(-this.config.maxOutputBytes)
|
||||
if (text.length > 0) admit(text)
|
||||
})
|
||||
handle.stderr?.on('end', () => {
|
||||
const text = stderrDecoder.decode()
|
||||
if (text.length > 0) admit(text)
|
||||
})
|
||||
handle.stderr?.on('error', (error: Error) => { finish({ kind: 'worker-exit', message: messageOf(error) }) })
|
||||
let ready = false
|
||||
let nextId = 1
|
||||
let pending = 0
|
||||
let pendingBytes = 0
|
||||
const protocolFailure = (message: string): void => { finish({ kind: 'protocol', message }) }
|
||||
const processFinished = (outcome: SubprocessOutcome): void => {
|
||||
if (settled) return
|
||||
finish((confined !== undefined && classifyRunnerFailure(outcome.exitCode, stderr, confined.runnerFailureRules) !== undefined)
|
||||
? { kind: 'sandbox-unavailable', message: stderr }
|
||||
: { kind: 'worker-exit', message: `Node process exited before completing (${String(outcome.exitCode)})${stderr ? `: ${stderr}` : ''}` })
|
||||
}
|
||||
const transport: JsonChannel = new JsonChannel(launched.control, this.config.maxMessageBytes, (raw, bytes) => {
|
||||
if (!record(raw)) { protocolFailure('invalid control frame'); return }
|
||||
if (!ready) {
|
||||
if (raw.type !== 'ready') { protocolFailure('program frame arrived before bootstrap readiness'); return }
|
||||
ready = true
|
||||
void transport.send({ type: 'boot', data }).catch((error: unknown) => { protocolFailure(messageOf(error)) })
|
||||
return
|
||||
}
|
||||
switch (raw.type) {
|
||||
case 'log':
|
||||
if (typeof raw.text !== 'string') { protocolFailure('invalid log frame'); return }
|
||||
admit(raw.text)
|
||||
return
|
||||
case 'output-limit': outputOverflow = true; finish(); return
|
||||
case 'done': {
|
||||
if (raw.error !== undefined) {
|
||||
if (!record(raw.error) || typeof raw.error.message !== 'string' || (raw.error.kind !== 'exception' && raw.error.kind !== 'invalid-output' && raw.error.kind !== 'output-limit')) { protocolFailure('invalid terminal error'); return }
|
||||
const failure = { kind: raw.error.kind, message: raw.error.message } as PtcRunFailure
|
||||
if (confined !== undefined) {
|
||||
sandbox.denied = confined.denialSignatures.some(signature =>
|
||||
failure.message.toLowerCase().includes(signature.toLowerCase()))
|
||||
}
|
||||
if (failure.kind === 'output-limit') outputOverflow = true
|
||||
finish(failure)
|
||||
return
|
||||
}
|
||||
const value = raw.value === undefined ? undefined : decodePtcJsonWire(raw.value)
|
||||
if (raw.value !== undefined && value === undefined) finish({ kind: 'invalid-output', message: 'program completion must be lossless JSON' })
|
||||
else finish(undefined, value)
|
||||
return
|
||||
}
|
||||
case 'call': {
|
||||
if (!Number.isSafeInteger(raw.id) || raw.id !== nextId || typeof raw.global !== 'string' || typeof raw.name !== 'string') { protocolFailure('invalid binding call identity'); return }
|
||||
nextId += 1
|
||||
const functions = bindings.get(raw.global)?.functions
|
||||
const fn = functions !== undefined && Object.hasOwn(functions, raw.name) ? functions[raw.name] : undefined
|
||||
if (typeof fn !== 'function') { protocolFailure('program requested an undeclared binding'); return }
|
||||
const args = decodePtcJsonWire(raw.args)
|
||||
if (args === undefined) { protocolFailure('binding arguments must be lossless JSON'); return }
|
||||
if (++pending > this.config.maxPendingCalls || (pendingBytes += bytes) > this.config.maxMessageBytes) { protocolFailure('pending binding calls exceed configured limits'); return }
|
||||
const id = raw.id
|
||||
void (async () => {
|
||||
const reply = await replyBinding(id, fn, args).finally(() => {
|
||||
pending -= 1
|
||||
pendingBytes -= bytes
|
||||
})
|
||||
if (!settled) await transport.send(reply)
|
||||
})().catch((error: unknown) => { protocolFailure(messageOf(error)) })
|
||||
return
|
||||
}
|
||||
default: protocolFailure('unknown control message')
|
||||
}
|
||||
}, (error, kind) => {
|
||||
if (kind === 'protocol') protocolFailure(messageOf(error))
|
||||
else if (ready) finish({ kind: 'worker-exit', message: messageOf(error) })
|
||||
else void launched.done.then(processFinished, (failure: unknown) => { finish({ kind: 'worker-exit', message: messageOf(failure) }) })
|
||||
})
|
||||
channel = transport
|
||||
void launched.done.then((outcome) => {
|
||||
// Allow queued control-frame callbacks to run before classifying a command exit.
|
||||
setImmediate(() => { processFinished(outcome) })
|
||||
}, (error: unknown) => { finish({ kind: confined !== undefined && isRunnerSpawnFailure(error, confined.argv[0], spec.cwd) ? 'sandbox-unavailable' : 'worker-exit', message: messageOf(error) }) })
|
||||
} catch (error: unknown) {
|
||||
finish({ kind: error instanceof SandboxUnavailableError ? 'sandbox-unavailable' : parsing ? 'exception' : 'worker-exit', message: messageOf(error) })
|
||||
}
|
||||
return await result.promise
|
||||
}
|
||||
}
|
||||
@@ -1,128 +1,24 @@
|
||||
/** Retained Node evaluations with per-evaluation bindings and output attribution. */
|
||||
import { AsyncLocalStorage } from 'node:async_hooks'
|
||||
/** Existing retained bootstrap entry delegates to the shared Node evaluation loop. */
|
||||
import type { Duplex } from 'node:stream'
|
||||
import { JsonChannel } from './channel.ts'
|
||||
import { captureStreamWrites, LogBuffer, makeConsoleShim, makeBindingErrorClass, makeNamespaces, prepareCompletion, prepareException, wireReplies } from './bootstrap.ts'
|
||||
import type { BindingErrorConstructor, PendingCall, PatchableStream } from './bootstrap.ts'
|
||||
import type { DoneMessage, ProgramBootData, ProgramToHost, ReplyMessage } from './protocol.ts'
|
||||
import { clearProgramEnvironment } from './environment.ts'
|
||||
import { createProgramEvaluator } from './evaluators/index.ts'
|
||||
import { runNodeMain } from './process.ts'
|
||||
import type { ProgramProcess } from './process.ts'
|
||||
import type { ProgramFormat } from './evaluators/types.ts'
|
||||
|
||||
/** Process state owned by the retained child; tests supply instance-local streams and environment. */
|
||||
export interface PersistentProcess {
|
||||
env: NodeJS.ProcessEnv
|
||||
stdout: PatchableStream
|
||||
stderr: PatchableStream
|
||||
/** Process state required by retained bootstraps that change cwd between evaluations. */
|
||||
export interface PersistentProcess extends ProgramProcess {
|
||||
chdir(directory: string): void
|
||||
}
|
||||
|
||||
interface Evaluation {
|
||||
id: number
|
||||
logs: LogBuffer
|
||||
maxOutputBytes: number
|
||||
active: boolean
|
||||
finish(result: Omit<DoneMessage, 'type'>): void
|
||||
}
|
||||
|
||||
/**
|
||||
* Serve sequential evaluations until the owning host closes the control channel.
|
||||
* Serve evaluations through the installed retained-bootstrap entry.
|
||||
* @param stream - Inherited private control endpoint.
|
||||
* @param maxMessageBytes - Maximum control frame and queued output bytes.
|
||||
* @param processState - Child environment, output and directory control, bound only in the program context.
|
||||
* @param format - Retained evaluation syntax; existing callers select REPL by default.
|
||||
* @returns After the channel closes and the evaluator releases its resources.
|
||||
* @param processState - Child environment, output and directory control.
|
||||
* @param format - Default syntax for hosts that omit it from program data.
|
||||
* @returns After control closure and evaluator cleanup.
|
||||
*/
|
||||
export async function runPersistentMain(stream: Duplex, maxMessageBytes: number, processState: PersistentProcess, format: ProgramFormat = 'repl'): Promise<void> {
|
||||
if (!Number.isSafeInteger(maxMessageBytes) || maxMessageBytes <= 0 || maxMessageBytes > 0xffff_ffff) throw new Error('invalid control message limit')
|
||||
clearProgramEnvironment(processState)
|
||||
const context = new AsyncLocalStorage<Evaluation>()
|
||||
const closed = Promise.withResolvers<void>()
|
||||
const pending = new Map<number, PendingCall>()
|
||||
const nextId = { value: 1 }
|
||||
const errors = new Map<string, { name: string; memberNameProperty: string; constructor: BindingErrorConstructor }>()
|
||||
const errorLookup = { get: (name: string) => errors.get(name)?.constructor }
|
||||
const replies: Array<(message: ReplyMessage) => void> = []
|
||||
const evaluator = createProgramEvaluator(format, (message) => {
|
||||
const current = context.getStore()
|
||||
if (current?.active && message) current.finish(prepareException(message, current.logs.remainingOutputBytes(), current.maxOutputBytes))
|
||||
})
|
||||
let current: Evaluation | undefined
|
||||
const send = (id: number, payload: ProgramToHost): void => {
|
||||
void channel.send({ id, payload }).catch(() => { channel.close(); closed.resolve() })
|
||||
}
|
||||
const port = {
|
||||
postMessage(payload: ProgramToHost): void {
|
||||
const evaluation = context.getStore()
|
||||
if (!evaluation?.active) throw new Error('PTC evaluation has ended; call this function from a new run_code evaluation')
|
||||
send(evaluation.id, payload)
|
||||
},
|
||||
on(_event: 'message', listener: (message: ReplyMessage) => void): void { replies.push(listener) },
|
||||
}
|
||||
wireReplies(port, pending)
|
||||
const logSink = { push(text: string): void { const evaluation = context.getStore(); if (evaluation?.active) evaluation.logs.push(text) } }
|
||||
evaluator.globals.console = makeConsoleShim(logSink)
|
||||
Object.defineProperty(evaluator.globals, 'process', { configurable: true, enumerable: true, writable: true, value: processState })
|
||||
captureStreamWrites(logSink, processState.stdout)
|
||||
captureStreamWrites(logSink, processState.stderr)
|
||||
const channel = new JsonChannel(stream, maxMessageBytes, (raw) => {
|
||||
// The host owns these frames. Model-written traffic is validated on the host.
|
||||
const message = raw as { type: 'run'; id: number; cwd: string; data: ProgramBootData } | ReplyMessage
|
||||
if (message.type === 'reply') {
|
||||
for (const listener of replies) listener(message)
|
||||
return
|
||||
}
|
||||
if (current?.active) { channel.close(); closed.resolve(); return }
|
||||
const { id, data } = message
|
||||
const logs = new LogBuffer(data.maxOutputBytes, (text) => { send(id, { type: 'log', text }) }, () => { send(id, { type: 'output-limit' }) })
|
||||
const evaluation: Evaluation = {
|
||||
id, logs, maxOutputBytes: data.maxOutputBytes, active: true,
|
||||
finish(result) {
|
||||
if (!evaluation.active) return
|
||||
evaluation.active = false
|
||||
for (const call of pending.values()) call.reject(new Error('PTC evaluation ended before the binding completed'))
|
||||
pending.clear()
|
||||
send(id, { type: 'done', ...result })
|
||||
},
|
||||
}
|
||||
current = evaluation
|
||||
context.run(evaluation, () => {
|
||||
try {
|
||||
processState.chdir(message.cwd)
|
||||
for (const namespace of data.namespaces) {
|
||||
const descriptor = namespace.errorClass
|
||||
if (!descriptor) { errors.delete(namespace.global); continue }
|
||||
const previous = errors.get(namespace.global)
|
||||
if (previous?.name === descriptor.name && previous.memberNameProperty === descriptor.memberNameProperty) continue
|
||||
const constructor = makeBindingErrorClass(descriptor)
|
||||
errors.set(namespace.global, { name: descriptor.name, memberNameProperty: descriptor.memberNameProperty, constructor })
|
||||
}
|
||||
const namespaces = makeNamespaces(data, port, pending, nextId, errorLookup)
|
||||
for (const [index, namespace] of data.namespaces.entries()) {
|
||||
Object.defineProperty(evaluator.globals, namespace.global, { configurable: true, value: namespaces[index] })
|
||||
if (namespace.errorClass) {
|
||||
Object.defineProperty(evaluator.globals, namespace.errorClass.name, {
|
||||
configurable: true, value: errorLookup.get(namespace.global),
|
||||
})
|
||||
}
|
||||
}
|
||||
evaluator.evaluate(data.code, (result) => {
|
||||
evaluation.finish(result.ok ? prepareCompletion(result.value, logs.remainingOutputBytes(), data.maxOutputBytes)
|
||||
: prepareException(result.error, logs.remainingOutputBytes(), data.maxOutputBytes))
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
evaluation.finish(prepareException(error, logs.remainingOutputBytes(), data.maxOutputBytes))
|
||||
}
|
||||
})
|
||||
}, () => { channel.close(); closed.resolve() })
|
||||
stream.once('close', () => { closed.resolve() })
|
||||
try {
|
||||
await channel.send({ type: 'ready' })
|
||||
await closed.promise
|
||||
} finally {
|
||||
if (current) current.active = false
|
||||
evaluator.close()
|
||||
channel.close()
|
||||
context.disable()
|
||||
}
|
||||
export function runPersistentMain(
|
||||
stream: Duplex, maxMessageBytes: number, processState: PersistentProcess, format: ProgramFormat = 'repl',
|
||||
): Promise<void> {
|
||||
return runNodeMain(stream, maxMessageBytes, processState, format)
|
||||
}
|
||||
@@ -1,86 +1,160 @@
|
||||
/** Node child execution over an inherited control channel; no Harness services run here. */
|
||||
/** Node child evaluation over the shared process protocol; the host owns process lifetime. */
|
||||
import { AsyncLocalStorage } from 'node:async_hooks'
|
||||
import type { Duplex } from 'node:stream'
|
||||
import { JsonChannel } from './channel.ts'
|
||||
import { runProgram } from './bootstrap.ts'
|
||||
import { captureStreamWrites, LogBuffer, makeConsoleShim, makeBindingErrorClass, makeNamespaces, prepareCompletion, prepareException, acceptReply } from './bootstrap.ts'
|
||||
import type { BindingErrorConstructor, PendingCall, PatchableStream } from './bootstrap.ts'
|
||||
import { clearProgramEnvironment } from './environment.ts'
|
||||
import type { PatchableStream } from './bootstrap.ts'
|
||||
import type { ReplyMessage, ProgramBootData, ProgramToHost } from './protocol.ts'
|
||||
import { createProgramEvaluator } from './evaluators/index.ts'
|
||||
import type { ProgramEvaluator, ProgramFormat } from './evaluators/types.ts'
|
||||
import type { DoneMessage, ProgramBootData, ProgramToHost, ReplyMessage, RunMessage } from './protocol.ts'
|
||||
|
||||
/** Process-owned environment and output streams consumed by the child bootstrap. */
|
||||
/** Process-owned environment, output and optional directory control consumed by the child. */
|
||||
export interface ProgramProcess {
|
||||
env: NodeJS.ProcessEnv
|
||||
stdout: PatchableStream
|
||||
stderr: PatchableStream
|
||||
exitCode: string | number | null | undefined
|
||||
exitCode?: string | number | null | undefined
|
||||
chdir?(directory: string): void
|
||||
}
|
||||
|
||||
interface Evaluation {
|
||||
/** The standalone boot entry has no evaluation envelope. */
|
||||
id: number | undefined
|
||||
logs: LogBuffer
|
||||
maxOutputBytes: number
|
||||
active: boolean
|
||||
finish(result: Omit<DoneMessage, 'type'>): void
|
||||
}
|
||||
|
||||
/**
|
||||
* Run one host-supplied program after the control handshake.
|
||||
* Serve host evaluations until the host closes the control channel.
|
||||
* The standalone boot frame remains accepted by packaged worker probes.
|
||||
* @param stream - Inherited, already-adopted control endpoint.
|
||||
* @param maxMessageBytes - Host-validated maximum frame and queued-write bytes.
|
||||
* @param processState - Environment, output streams and exit status of this Node child.
|
||||
* @returns After control output flushes and host shutdown is observed, or transport failure closes the channel.
|
||||
* @param maxMessageBytes - Maximum frame and queued-write bytes.
|
||||
* @param processState - Child environment, output streams, directory control and exit status.
|
||||
* @param defaultFormat - Syntax used by existing bootstraps when the host omits it.
|
||||
* @returns After control closure and evaluator cleanup; malformed startup rejects.
|
||||
*/
|
||||
export async function runNodeMain(stream: Duplex, maxMessageBytes: number, processState: ProgramProcess): Promise<void> {
|
||||
export async function runNodeMain(
|
||||
stream: Duplex, maxMessageBytes: number, processState: ProgramProcess, defaultFormat: ProgramFormat = 'function-body',
|
||||
): Promise<void> {
|
||||
if (!Number.isSafeInteger(maxMessageBytes) || maxMessageBytes <= 0 || maxMessageBytes > 0xffff_ffff) throw new Error('invalid control message limit')
|
||||
clearProgramEnvironment(processState)
|
||||
const boot = Promise.withResolvers<ProgramBootData>()
|
||||
const listeners: Array<(message: ReplyMessage) => void> = []
|
||||
let started = false
|
||||
let failed = false
|
||||
let terminalSent = false
|
||||
const hostClosed = Promise.withResolvers<void>()
|
||||
const onClose = (): void => { hostClosed.resolve() }
|
||||
stream.once('close', onClose)
|
||||
const context = new AsyncLocalStorage<Evaluation>()
|
||||
const closed = Promise.withResolvers<void>()
|
||||
const pending = new Map<number, PendingCall>()
|
||||
const nextId = { value: 1 }
|
||||
const errors = new Map<string, { name: string; memberNameProperty: string; constructor: BindingErrorConstructor }>()
|
||||
const errorLookup = { get: (name: string) => errors.get(name)?.constructor }
|
||||
let evaluator: ProgramEvaluator | undefined
|
||||
let current: Evaluation | undefined
|
||||
const onClose = (): void => { closed.resolve() }
|
||||
const fail = (): void => {
|
||||
processState.exitCode = 1
|
||||
channel.close()
|
||||
closed.resolve()
|
||||
}
|
||||
const send = (id: number | undefined, payload: ProgramToHost): void => {
|
||||
void channel.send(id === undefined ? payload : { id, payload }).catch(fail)
|
||||
}
|
||||
const sender = {
|
||||
postMessage(payload: ProgramToHost): void {
|
||||
const evaluation = context.getStore()
|
||||
if (!evaluation?.active) throw new Error('PTC evaluation has ended; call this function from a new run_code evaluation')
|
||||
send(evaluation.id, payload)
|
||||
},
|
||||
}
|
||||
const logSink = { push(text: string): void { const evaluation = context.getStore(); if (evaluation?.active) evaluation.logs.push(text) } }
|
||||
captureStreamWrites(logSink, processState.stdout)
|
||||
captureStreamWrites(logSink, processState.stderr)
|
||||
const channel = new JsonChannel(stream, maxMessageBytes, (raw) => {
|
||||
if (!started) {
|
||||
started = true
|
||||
if (typeof raw !== 'object' || raw === null || (raw as { type?: unknown }).type !== 'boot') {
|
||||
boot.reject(new Error('expected program boot frame'))
|
||||
return
|
||||
}
|
||||
boot.resolve((raw as { data: ProgramBootData }).data)
|
||||
// Host-owned run frames are trusted; model-written traffic is validated by the host.
|
||||
if (typeof raw !== 'object' || raw === null || !('type' in raw)
|
||||
|| (raw.type !== 'run' && raw.type !== 'boot' && raw.type !== 'reply')
|
||||
|| (raw.type === 'reply' && current === undefined)) {
|
||||
closed.reject(new Error('expected program boot or run frame'))
|
||||
channel.close()
|
||||
return
|
||||
}
|
||||
if (terminalSent) return
|
||||
for (const listener of listeners) listener(raw as ReplyMessage)
|
||||
}, (error, kind) => {
|
||||
if (!terminalSent || kind === 'protocol') {
|
||||
failed = true
|
||||
boot.reject(error)
|
||||
channel.close()
|
||||
processState.exitCode = 1
|
||||
const message = raw as RunMessage | { type: 'boot'; data: ProgramBootData } | ReplyMessage
|
||||
if (message.type === 'reply') {
|
||||
acceptReply(pending, message)
|
||||
return
|
||||
}
|
||||
hostClosed.resolve()
|
||||
if (current?.active) { fail(); return }
|
||||
// Standalone workers finish exactly one program and then await host shutdown.
|
||||
if (current !== undefined && current.id === undefined) return
|
||||
const id = message.type === 'run' ? message.id : undefined
|
||||
const { data } = message
|
||||
const logs = new LogBuffer(data.maxOutputBytes, (text) => { send(id, { type: 'log', text }) }, () => { send(id, { type: 'output-limit' }) })
|
||||
const evaluation: Evaluation = {
|
||||
id, logs, maxOutputBytes: data.maxOutputBytes, active: true,
|
||||
finish(result) {
|
||||
if (!evaluation.active) return
|
||||
evaluation.active = false
|
||||
if (id !== undefined) {
|
||||
for (const call of pending.values()) call.reject(new Error('PTC evaluation ended before the binding completed'))
|
||||
}
|
||||
pending.clear()
|
||||
send(id, { type: 'done', ...result })
|
||||
},
|
||||
}
|
||||
current = evaluation
|
||||
context.run(evaluation, () => {
|
||||
try {
|
||||
if (message.type === 'run') {
|
||||
if (processState.chdir === undefined) throw new Error('Node program process cannot change its working directory')
|
||||
processState.chdir(message.cwd)
|
||||
}
|
||||
if (evaluator === undefined) {
|
||||
evaluator = createProgramEvaluator(data.programFormat ?? defaultFormat, (error) => {
|
||||
const active = context.getStore()
|
||||
if (active?.active && error) active.finish(prepareException(error, active.logs.remainingOutputBytes(), active.maxOutputBytes))
|
||||
})
|
||||
evaluator.globals.console = makeConsoleShim(logSink)
|
||||
Object.defineProperty(evaluator.globals, 'process', { configurable: true, enumerable: true, writable: true, value: processState })
|
||||
}
|
||||
for (const namespace of data.namespaces) {
|
||||
const descriptor = namespace.errorClass
|
||||
if (!descriptor) { errors.delete(namespace.global); continue }
|
||||
const previous = errors.get(namespace.global)
|
||||
if (previous?.name === descriptor.name && previous.memberNameProperty === descriptor.memberNameProperty) continue
|
||||
const constructor = makeBindingErrorClass(descriptor)
|
||||
errors.set(namespace.global, { name: descriptor.name, memberNameProperty: descriptor.memberNameProperty, constructor })
|
||||
}
|
||||
const namespaces = makeNamespaces(data, sender, pending, nextId, errorLookup)
|
||||
for (const [index, namespace] of data.namespaces.entries()) {
|
||||
Object.defineProperty(evaluator.globals, namespace.global, { configurable: true, enumerable: true, value: namespaces[index] })
|
||||
if (namespace.errorClass) {
|
||||
Object.defineProperty(evaluator.globals, namespace.errorClass.name, {
|
||||
configurable: true, enumerable: true, value: errorLookup.get(namespace.global),
|
||||
})
|
||||
}
|
||||
}
|
||||
evaluator.evaluate(data.code, (result) => {
|
||||
evaluation.finish(result.ok ? prepareCompletion(result.value, logs.remainingOutputBytes(), data.maxOutputBytes)
|
||||
: prepareException(result.error, logs.remainingOutputBytes(), data.maxOutputBytes))
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
evaluation.finish(prepareException(error, logs.remainingOutputBytes(), data.maxOutputBytes))
|
||||
}
|
||||
})
|
||||
}, (error, kind) => {
|
||||
if (current === undefined) closed.reject(error)
|
||||
if (current?.active || kind === 'protocol' || current === undefined) fail()
|
||||
else closed.resolve()
|
||||
})
|
||||
const pending = new Set<Promise<void>>()
|
||||
const send = (message: ProgramToHost): void => {
|
||||
if (terminalSent) return
|
||||
if (message.type === 'done') terminalSent = true
|
||||
const task = channel.send(message).catch(() => {
|
||||
failed = true
|
||||
channel.close()
|
||||
processState.exitCode = 1
|
||||
hostClosed.resolve()
|
||||
}).finally(() => { pending.delete(task) })
|
||||
pending.add(task)
|
||||
}
|
||||
stream.once('close', onClose)
|
||||
try {
|
||||
await channel.send({ type: 'ready' })
|
||||
const data = await boot.promise
|
||||
await runProgram({
|
||||
postMessage: send,
|
||||
on: (_event, listener) => { listeners.push(listener) },
|
||||
}, data, { stdout: processState.stdout, stderr: processState.stderr })
|
||||
while (pending.size > 0) await Promise.all(pending)
|
||||
await closed.promise
|
||||
await channel.drain()
|
||||
// Host binding replies can race the terminal frame; the host owns channel shutdown.
|
||||
await hostClosed.promise
|
||||
} finally {
|
||||
stream.off('close', onClose)
|
||||
if (current) current.active = false
|
||||
evaluator?.close()
|
||||
channel.close()
|
||||
// Transport callbacks can set failed while the awaited program executes.
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition
|
||||
if (failed) processState.exitCode = 1
|
||||
context.disable()
|
||||
}
|
||||
}
|
||||
@@ -7,11 +7,11 @@
|
||||
import type { ProgramFormat } from './evaluators/types.ts'
|
||||
import type { PtcJsonWire } from './json-wire.ts'
|
||||
|
||||
/** What the host hands the program at spawn, after the readiness handshake. */
|
||||
/** One evaluation's source, binding declarations and output budget. */
|
||||
export interface ProgramBootData {
|
||||
/** The type-stripped (plain JS) program. */
|
||||
code: string
|
||||
/** One-shot evaluation syntax; omitted preserves async-function-body execution. */
|
||||
/** Evaluation syntax; omitted uses the installed bootstrap's default. */
|
||||
programFormat?: ProgramFormat
|
||||
/** Binding namespaces to materialize; functions themselves stay host-side. */
|
||||
namespaces: {
|
||||
@@ -23,6 +23,16 @@ export interface ProgramBootData {
|
||||
maxOutputBytes: number
|
||||
}
|
||||
|
||||
/** Host → program: one evaluation in a temporary or owner-retained process. */
|
||||
export interface RunMessage {
|
||||
type: 'run'
|
||||
/** Process-local evaluation identity, increasing for each use of a retained process. */
|
||||
id: number
|
||||
/** Directory applied before evaluation in the child process world. */
|
||||
cwd: string
|
||||
data: ProgramBootData
|
||||
}
|
||||
|
||||
/** Program → host: one bridged binding call. */
|
||||
interface CallMessage {
|
||||
type: 'call'
|
||||
|
||||
+108
-160
@@ -1,16 +1,14 @@
|
||||
/** Node PTC in-memory execution with retained processes scoped to caller-owned objects. */
|
||||
/** Managed Node process shared by temporary and owner-retained evaluations. */
|
||||
import { AsyncLocalStorage } from 'node:async_hooks'
|
||||
import { isAbsolute } from 'node:path'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import type { PtcRunRequest, PtcRunSpec, PtcRunResult, PtcRunSandbox, PtcRunFailure, PtcBindingNamespace, PtcJsonValue } from '@deepseek-ai/dsh-ptc-runtime'
|
||||
import { clampTimeout } from '@deepseek-ai/dsh-timeout'
|
||||
import type { PtcRunSpec, PtcRunResult, PtcRunSandbox, PtcRunFailure, PtcBindingNamespace, PtcJsonValue } from '@deepseek-ai/dsh-ptc-runtime'
|
||||
import { SandboxUnavailableError, classifyRunnerFailure, isRunnerSpawnFailure } from '@deepseek-ai/dsh-sandbox'
|
||||
import type { ConfinedArgv, SandboxExecutionPolicy } from '@deepseek-ai/dsh-sandbox'
|
||||
import type { SubprocessHandle, SubprocessOutcome } from '@deepseek-ai/dsh-subprocess'
|
||||
import type { PolicyProcesses } from './policy-processes.ts'
|
||||
import type { ResolvedConfig } from './config.ts'
|
||||
import type { ExecutionMode, ExecutionRunner } from './execution.ts'
|
||||
import { replyToBinding, validateBindings } from './bindings.ts'
|
||||
import type { ExecutionMode } from './execution.ts'
|
||||
import { replyToBinding } from './bindings.ts'
|
||||
import { JsonChannel } from './channel.ts'
|
||||
import { bootstrapArgs } from './launch.ts'
|
||||
import { STARTUP_ENVIRONMENT_NAMES } from './environment.ts'
|
||||
@@ -22,27 +20,12 @@ import type { ProgramBootData } from './protocol.ts'
|
||||
const messageOf = (error: unknown): string => error instanceof Error ? error.message : String(error)
|
||||
const record = (value: unknown): value is Record<string, unknown> => typeof value === 'object' && value !== null && !Array.isArray(value)
|
||||
|
||||
function persistentOutput(maxBytes: number): OutputLedger {
|
||||
return new OutputLedger(maxBytes, `retained variables were cleared; outer output exceeded ${maxBytes} bytes`)
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare requested extra grants without resolving paths in another execution world.
|
||||
* @param left - one policy's additional roots.
|
||||
* @param right - the other policy's additional roots.
|
||||
* @returns whether both contain the same paths, ignoring order and duplicates.
|
||||
*/
|
||||
function sameAdditionalRoots(left: readonly string[] | undefined, right: readonly string[] | undefined): boolean {
|
||||
const effective = (roots: readonly string[] | undefined): string[] => [...new Set(roots ?? [])].sort()
|
||||
const a = effective(left)
|
||||
const b = effective(right)
|
||||
return a.length === b.length && a.every((root, index) => root === b[index])
|
||||
}
|
||||
|
||||
interface Evaluation {
|
||||
id: number
|
||||
settled: boolean
|
||||
data: ProgramBootData
|
||||
spec: PtcRunSpec
|
||||
sandbox: PtcRunSandbox
|
||||
bindings: Map<string, PtcBindingNamespace>
|
||||
reply: typeof replyToBinding
|
||||
pending: number
|
||||
@@ -52,7 +35,7 @@ interface Evaluation {
|
||||
}
|
||||
|
||||
/** One managed process; all mutable program state belongs to this process alone. */
|
||||
class Kernel {
|
||||
export class ProgramWorker {
|
||||
private readonly unregister: (() => void) | undefined
|
||||
private confined: ConfinedArgv | undefined
|
||||
private handle?: SubprocessHandle
|
||||
@@ -71,6 +54,17 @@ class Kernel {
|
||||
private readonly controller = new AbortController()
|
||||
closed = false
|
||||
|
||||
/** Whether termination succeeded and this process no longer needs cleanup retries. */
|
||||
get cleanupComplete(): boolean { return this.closed && !this.cleanupPending }
|
||||
|
||||
/**
|
||||
* Own one process under a fixed file policy.
|
||||
* @param ctx - Provider execution services.
|
||||
* @param config - Validated resource limits and launch selection.
|
||||
* @param policy - Authority retained for the lifetime of this process.
|
||||
* @param processes - Session policy invalidation and cleanup coordination.
|
||||
* @param mode - Evaluator syntax and retained-state diagnostics.
|
||||
*/
|
||||
constructor(
|
||||
private readonly ctx: Context, private readonly config: ResolvedConfig,
|
||||
readonly policy: SandboxExecutionPolicy, processes: PolicyProcesses, private readonly mode: ExecutionMode,
|
||||
@@ -78,67 +72,97 @@ class Kernel {
|
||||
this.unregister = processes.register(policy, () => this.dispose())
|
||||
}
|
||||
|
||||
/** Reject reuse until the previous evaluation and required cleanup settle. */
|
||||
assertIdle(): void {
|
||||
if (this.running) throw new Error('persistent PTC does not accept concurrent runs for one state owner')
|
||||
if (this.cleanupPending) throw new Error('persistent PTC environment cleanup is incomplete; release the owner before running another program')
|
||||
}
|
||||
|
||||
async run(spec: PtcRunSpec, bindings: Map<string, PtcBindingNamespace>, code: string): Promise<PtcRunResult> {
|
||||
/**
|
||||
* Evaluate a program, retaining this process only for an enabled state owner.
|
||||
* @param spec - Resolved program inputs and cancellation.
|
||||
* @param bindings - Current host binding implementations.
|
||||
* @returns Output after any required process cleanup.
|
||||
*/
|
||||
async run(spec: PtcRunSpec, bindings: Map<string, PtcBindingNamespace>): Promise<PtcRunResult> {
|
||||
this.running = true
|
||||
const sandbox: PtcRunSandbox = { mode: this.policy.mode, denied: false }
|
||||
try {
|
||||
const result = await this.evaluate(spec, bindings, code)
|
||||
const result = await this.evaluate(spec, bindings, sandbox)
|
||||
if (this.confined !== undefined) {
|
||||
sandbox.enforcement = this.confined.enforcement
|
||||
sandbox.denied = result.error?.kind !== 'sandbox-unavailable' && this.confined.denialSignatures.some(signature =>
|
||||
sandbox.denied ||= result.error?.kind !== 'sandbox-unavailable' && this.confined.denialSignatures.some(signature =>
|
||||
result.error?.message.toLowerCase().includes(signature.toLowerCase()))
|
||||
}
|
||||
return { ...result, sandbox }
|
||||
} finally { this.running = false }
|
||||
}
|
||||
|
||||
private async evaluate(spec: PtcRunSpec, bindings: Map<string, PtcBindingNamespace>, code: string): Promise<PtcRunResult> {
|
||||
private output(): OutputLedger {
|
||||
return new OutputLedger(this.config.maxOutputBytes, this.mode.retainsState
|
||||
? `retained variables were cleared; outer output exceeded ${this.config.maxOutputBytes} bytes` : undefined)
|
||||
}
|
||||
|
||||
private message(message: string): string {
|
||||
return this.mode.retainsState ? `${message}; retained variables were cleared` : message
|
||||
}
|
||||
|
||||
private abortMessage(signal: AbortSignal | undefined): string {
|
||||
return this.mode.retainsState ? 'execution canceled; retained variables were cleared' : messageOf(signal?.reason)
|
||||
}
|
||||
|
||||
private async evaluate(spec: PtcRunSpec, bindings: Map<string, PtcBindingNamespace>, sandbox: PtcRunSandbox): Promise<PtcRunResult> {
|
||||
if (this.exitOutcome !== undefined) {
|
||||
const output = persistentOutput(this.config.maxOutputBytes)
|
||||
const output = this.output()
|
||||
const failure: PtcRunFailure = spec.signal?.aborted
|
||||
? { kind: 'abort', message: 'execution canceled; retained variables were cleared' }
|
||||
: { kind: 'worker-exit', message: `${this.exitMessage(this.exitOutcome)}; retained variables were cleared` }
|
||||
? { kind: 'abort', message: this.abortMessage(spec.signal) }
|
||||
: { kind: 'worker-exit', message: this.message(this.exitMessage(this.exitOutcome)) }
|
||||
try { await this.stop() } catch (error: unknown) {
|
||||
return output.failure([], { kind: 'worker-exit', message: `managed process cleanup failed: ${messageOf(error)}` })
|
||||
}
|
||||
return output.failure([], failure)
|
||||
}
|
||||
this.stderr = ''
|
||||
const output = persistentOutput(this.config.maxOutputBytes)
|
||||
const output = this.output()
|
||||
const logs: string[] = []
|
||||
let overflowed = false
|
||||
const result = Promise.withResolvers<PtcRunResult>()
|
||||
const timer = spec.timeoutMs === null ? undefined : setTimeout(() => {
|
||||
evaluation.finish({ kind: 'timeout', message: `execution deadline reached (${spec.timeoutMs}ms); retained variables were cleared` }, undefined, true)
|
||||
evaluation.finish({ kind: 'timeout', message: this.message(`execution deadline reached (${spec.timeoutMs}ms)`) }, undefined, true)
|
||||
}, spec.timeoutMs)
|
||||
const onAbort = (): void => { evaluation.finish({ kind: 'abort', message: 'execution canceled; retained variables were cleared' }, undefined, true) }
|
||||
const onAbort = (): void => { evaluation.finish({ kind: 'abort', message: this.abortMessage(spec.signal) }, undefined, true) }
|
||||
const evaluation: Evaluation = {
|
||||
id: ++this.sequence,
|
||||
data: { code, maxOutputBytes: this.config.maxOutputBytes, namespaces: [...bindings.values()].map(binding => ({
|
||||
global: binding.global, names: Object.keys(binding.functions), ...binding.errorClass ? { errorClass: binding.errorClass } : {},
|
||||
})) },
|
||||
spec, bindings, reply: AsyncLocalStorage.bind(replyToBinding), pending: 0, pendingBytes: 0,
|
||||
settled: false,
|
||||
data: {
|
||||
code: '', programFormat: this.mode.format, maxOutputBytes: this.config.maxOutputBytes,
|
||||
namespaces: [...bindings.values()].map(binding => ({
|
||||
global: binding.global, names: Object.keys(binding.functions), ...binding.errorClass ? { errorClass: binding.errorClass } : {},
|
||||
})),
|
||||
},
|
||||
spec, sandbox, bindings, reply: AsyncLocalStorage.bind(replyToBinding), pending: 0, pendingBytes: 0,
|
||||
admit: (text) => {
|
||||
if (overflowed) return
|
||||
if (!output.admit(text, logs)) {
|
||||
overflowed = true
|
||||
logs.push(text)
|
||||
evaluation.finish({ kind: 'output-limit', message: 'output limit exceeded' }, undefined, true)
|
||||
}
|
||||
},
|
||||
finish: (error, value, discard = false) => {
|
||||
if (this.active !== evaluation) return
|
||||
if (evaluation.settled) return
|
||||
evaluation.settled = true
|
||||
clearTimeout(timer)
|
||||
spec.signal?.removeEventListener('abort', onAbort)
|
||||
this.active = undefined
|
||||
const outcome = error?.kind === 'output-limit' ? output.limit(logs)
|
||||
const complete = (): PtcRunResult => overflowed || error?.kind === 'output-limit' ? output.limit(logs)
|
||||
: error ? output.failure(logs, error) : output.success(logs, value)
|
||||
const cleanup = discard || outcome.error?.kind === 'output-limit' || spec.stateOwner === undefined ? this.stop() : Promise.resolve(true)
|
||||
const outcome = complete()
|
||||
const cleanup = discard || outcome.error?.kind === 'output-limit' || !this.mode.retainsState || spec.stateOwner === undefined ? this.stop() : Promise.resolve(true)
|
||||
void cleanup.then((drained) => {
|
||||
result.resolve(drained ? outcome : output.failure(logs, { kind: 'worker-exit', message: 'managed process exited but its output did not drain cleanly' }))
|
||||
this.active = undefined
|
||||
result.resolve(drained ? complete() : output.failure(logs, { kind: 'worker-exit', message: 'managed process exited but its output did not drain cleanly' }))
|
||||
}, (failure: unknown) => {
|
||||
this.active = undefined
|
||||
result.resolve(output.failure(logs, { kind: 'worker-exit', message: `managed process cleanup failed: ${messageOf(failure)}` }))
|
||||
})
|
||||
},
|
||||
@@ -146,16 +170,23 @@ class Kernel {
|
||||
this.active = evaluation
|
||||
spec.signal?.addEventListener('abort', onAbort, { once: true })
|
||||
if (spec.signal?.aborted) onAbort()
|
||||
else if (this.ready) this.dispatch(evaluation)
|
||||
else {
|
||||
this.starting = this.start(spec.cwd)
|
||||
void this.starting.catch((error: unknown) => { this.fail(error instanceof SandboxUnavailableError || (this.confined !== undefined && isRunnerSpawnFailure(error, this.confined.argv[0], spec.cwd)) ? 'sandbox-unavailable' : 'worker-exit', messageOf(error)) })
|
||||
try { evaluation.data.code = this.mode.stripTypes(spec.program) } catch (error: unknown) {
|
||||
evaluation.finish({ kind: 'exception', message: messageOf(error) })
|
||||
return await result.promise
|
||||
}
|
||||
if (this.ready) this.dispatch(evaluation)
|
||||
else {
|
||||
this.starting = this.start(spec.cwd)
|
||||
void this.starting.catch((error: unknown) => { this.fail(error instanceof SandboxUnavailableError || (this.confined !== undefined && isRunnerSpawnFailure(error, this.confined.argv[0], spec.cwd)) ? 'sandbox-unavailable' : 'worker-exit', messageOf(error)) })
|
||||
}
|
||||
}
|
||||
return await result.promise
|
||||
}
|
||||
|
||||
/** Cancel active work and await managed process termination and output drainage. */
|
||||
async dispose(): Promise<void> {
|
||||
this.active?.finish({ kind: 'abort', message: 'PTC environment disposed; retained variables were cleared' }, undefined, true)
|
||||
this.active?.finish({ kind: 'abort', message: this.message('PTC environment disposed') }, undefined, true)
|
||||
await this.stop()
|
||||
}
|
||||
|
||||
@@ -166,20 +197,22 @@ class Kernel {
|
||||
this.controller.abort()
|
||||
this.channel?.close()
|
||||
this.stopped = (async () => {
|
||||
await this.starting?.catch(() => {})
|
||||
await this.starting?.catch((_error: unknown) => {
|
||||
// Evaluation reports startup failure; cleanup still owns any process already created.
|
||||
})
|
||||
const handle = this.handle
|
||||
return handle === undefined || await terminateProcess(handle, this.config.graceMs)
|
||||
})().then((drained) => {
|
||||
this.unregister?.()
|
||||
this.cleanupPending = false
|
||||
if (!drained) this.ctx.logger.warn('persistent Node process exited but its output did not drain cleanly')
|
||||
if (!drained) this.ctx.logger.warn('Node process exited but its output did not drain cleanly')
|
||||
return drained
|
||||
}, (error: unknown) => { this.stopped = undefined; throw error })
|
||||
return this.stopped
|
||||
}
|
||||
|
||||
private fail(kind: PtcRunFailure['kind'], message: string): void {
|
||||
if (this.active) this.active.finish({ kind, message: `${message}; retained variables were cleared` }, undefined, true)
|
||||
if (this.active) this.active.finish({ kind, message: this.message(message) }, undefined, true)
|
||||
else {
|
||||
if (!this.closed) this.ctx.logger.warn(message)
|
||||
void this.stop().catch((error: unknown) => { this.ctx.logger.warn(messageOf(error)) })
|
||||
@@ -198,17 +231,24 @@ class Kernel {
|
||||
private async start(cwd: string): Promise<void> {
|
||||
const executable = await this.ctx.subprocess.resolveExecutable(this.config.launch.executable, undefined, this.controller.signal)
|
||||
if (this.closed) return
|
||||
const argv = [executable, `--max-old-space-size=${this.config.maxOldGenerationSizeMb}`, ...bootstrapArgs(this.ctx.fs, this.config.launch, this.config.maxMessageBytes, true, this.mode.format)]
|
||||
const packaged = this.config.launch.kind === 'embedded'
|
||||
const heapFlag = `--max-old-space-size=${this.config.maxOldGenerationSizeMb}`
|
||||
const argv = [executable, ...packaged ? [] : [heapFlag],
|
||||
...bootstrapArgs(this.ctx.fs, this.config.launch, this.config.maxMessageBytes, this.mode.retainsState)]
|
||||
const policy = this.policy
|
||||
this.confined = policy.mode === 'danger-full-access' ? undefined : await this.ctx.sandbox.confine(argv, { ...policy, mode: policy.mode }, this.controller.signal)
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition -- Cancellation can close the kernel during confinement.
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition -- Cancellation can close the process during confinement.
|
||||
if (this.closed) return
|
||||
const env: NodeJS.ProcessEnv = Object.fromEntries(Object.keys(process.env)
|
||||
.filter(key => !STARTUP_ENVIRONMENT_NAMES.has(key.toUpperCase()) && key.toUpperCase() !== 'ELECTRON_RUN_AS_NODE')
|
||||
.map(key => [key, undefined]))
|
||||
if (packaged) {
|
||||
env.DSH_PTC_RUNTIME_NODE = '1'
|
||||
env.NODE_OPTIONS = heapFlag
|
||||
}
|
||||
const handle = this.ctx.subprocess.spawn({ argv: this.confined?.argv ?? argv, cwd, env, stdio: { stdin: 'ignore', stdout: 'pipe', stderr: 'pipe', control: 'pipe' }, graceMs: this.config.graceMs })
|
||||
this.handle = handle
|
||||
if (!handle.control || !handle.stdout || !handle.stderr) throw new Error('subprocess provider did not supply persistent PTC pipes')
|
||||
if (!handle.control || !handle.stdout || !handle.stderr) throw new Error('subprocess provider did not supply the requested control and output pipes')
|
||||
for (const stream of [handle.stdout, handle.stderr]) {
|
||||
stream.setEncoding('utf8')
|
||||
stream.on('data', (text: string) => {
|
||||
@@ -229,15 +269,19 @@ class Kernel {
|
||||
setImmediate(() => {
|
||||
if (!this.closed) this.fail(this.confined !== undefined && classifyRunnerFailure(outcome.exitCode, this.stderr, this.confined.runnerFailureRules) !== undefined ? 'sandbox-unavailable' : 'worker-exit', this.exitMessage(outcome))
|
||||
})
|
||||
}, (error: unknown) => { this.fail('worker-exit', messageOf(error)) })
|
||||
}, (error: unknown) => {
|
||||
this.fail(this.confined !== undefined && isRunnerSpawnFailure(error, this.confined.argv[0], cwd)
|
||||
? 'sandbox-unavailable' : 'worker-exit', messageOf(error))
|
||||
})
|
||||
}
|
||||
|
||||
private receive(raw: unknown, bytes: number): void {
|
||||
if (!record(raw)) { this.fail('protocol', 'invalid persistent PTC frame'); return }
|
||||
if (!record(raw)) { this.fail('protocol', 'invalid PTC frame'); return }
|
||||
const evaluation = this.active
|
||||
if (!evaluation) { this.fail('protocol', 'frame arrived without an active PTC evaluation'); return }
|
||||
if (evaluation.settled) return
|
||||
if (!this.ready) {
|
||||
if (raw.type !== 'ready') { this.fail('protocol', 'expected persistent PTC readiness'); return }
|
||||
if (raw.type !== 'ready') { this.fail('protocol', 'expected PTC readiness'); return }
|
||||
this.ready = true
|
||||
this.dispatch(evaluation)
|
||||
return
|
||||
@@ -247,16 +291,18 @@ class Kernel {
|
||||
switch (payload.type) {
|
||||
case 'log':
|
||||
if (typeof payload.text === 'string') evaluation.admit(payload.text)
|
||||
else this.fail('protocol', 'invalid persistent PTC log')
|
||||
else this.fail('protocol', 'invalid PTC log')
|
||||
return
|
||||
case 'output-limit': evaluation.finish({ kind: 'output-limit', message: 'output limit exceeded' }, undefined, true); return
|
||||
case 'done': {
|
||||
if (payload.error !== undefined) {
|
||||
const error = payload.error
|
||||
if (!record(error) || typeof error.message !== 'string' || (error.kind !== 'exception' && error.kind !== 'invalid-output' && error.kind !== 'output-limit')) {
|
||||
this.fail('protocol', 'invalid persistent PTC failure'); return
|
||||
this.fail('protocol', 'invalid PTC failure'); return
|
||||
}
|
||||
const failure = { kind: error.kind, message: error.message } as PtcRunFailure
|
||||
evaluation.sandbox.denied = this.confined?.denialSignatures.some(signature =>
|
||||
failure.message.toLowerCase().includes(signature.toLowerCase())) ?? false
|
||||
setImmediate(() => { evaluation.finish(failure, undefined, failure.kind === 'output-limit') })
|
||||
} else {
|
||||
const value = payload.value === undefined ? undefined : decodePtcJsonWire(payload.value)
|
||||
@@ -268,7 +314,7 @@ class Kernel {
|
||||
return
|
||||
}
|
||||
case 'call': {
|
||||
if (payload.id !== this.nextCall || typeof payload.global !== 'string' || typeof payload.name !== 'string') { this.fail('protocol', 'invalid persistent PTC binding identity'); return }
|
||||
if (!Number.isSafeInteger(payload.id) || payload.id !== this.nextCall || typeof payload.global !== 'string' || typeof payload.name !== 'string') { this.fail('protocol', 'invalid PTC binding identity'); return }
|
||||
this.nextCall += 1
|
||||
const args = decodePtcJsonWire(payload.args)
|
||||
if (args === undefined) { this.fail('protocol', 'binding arguments must be lossless JSON'); return }
|
||||
@@ -277,116 +323,18 @@ class Kernel {
|
||||
}
|
||||
const functions = evaluation.bindings.get(payload.global)?.functions
|
||||
const fn = functions && Object.hasOwn(functions, payload.name) ? functions[payload.name] : undefined
|
||||
if (!this.mode.retainsState && typeof fn !== 'function') { this.fail('protocol', 'program requested an undeclared binding'); return }
|
||||
const id = payload.id
|
||||
void (async () => {
|
||||
const reply = await evaluation.reply(id, fn, args).finally(() => {
|
||||
evaluation.pending -= 1
|
||||
evaluation.pendingBytes -= bytes
|
||||
})
|
||||
if (this.active === evaluation) await this.channel?.send(reply)
|
||||
if (this.active === evaluation && !evaluation.settled) await this.channel?.send(reply)
|
||||
})().catch((error: unknown) => { this.fail('protocol', messageOf(error)) })
|
||||
return
|
||||
}
|
||||
default: this.fail('protocol', 'unknown persistent PTC message')
|
||||
default: this.fail('protocol', 'unknown PTC message')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Sandboxed, best-effort retained state; restart, disposal, and interrupted execution clear it. */
|
||||
export class InMemoryExecution implements ExecutionRunner {
|
||||
private readonly owners = new WeakMap<object, Kernel>()
|
||||
private readonly kernels = new Set<Kernel>()
|
||||
private disposed = false
|
||||
|
||||
/**
|
||||
* Own retained execution resources under the existing Node provider.
|
||||
* @param ctx - Provider lifecycle and execution services.
|
||||
* @param config - Validated limits and selected executable.
|
||||
* @param processes - Provider-owned policy invalidation and cleanup.
|
||||
* @param mode - Program syntax and TypeScript erasure.
|
||||
*/
|
||||
constructor(
|
||||
private readonly ctx: Context, private readonly config: ResolvedConfig,
|
||||
private readonly processes: PolicyProcesses, private readonly mode: ExecutionMode,
|
||||
) {
|
||||
if (this.config.launch.kind === 'embedded') throw new Error('Persistent PTC requires a node-script launch')
|
||||
ctx.effect(() => async () => {
|
||||
this.disposed = true
|
||||
const results = await Promise.allSettled([...this.kernels].map(kernel => kernel.dispose()))
|
||||
this.kernels.clear()
|
||||
const failed = results.find(result => result.status === 'rejected')
|
||||
if (failed?.status === 'rejected') throw failed.reason
|
||||
}, 'Persistent PTC environments')
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the file policy, directory and deadline for a retained program context.
|
||||
* @param request - Program, bindings, owner, directory and elapsed budget.
|
||||
* @returns Resolved inputs with the selected file policy.
|
||||
*/
|
||||
resolve(request: PtcRunRequest): PtcRunSpec {
|
||||
if (this.disposed) throw new Error('persistent PTC: resolve after disposal')
|
||||
const sandboxPolicy = request.sandboxPolicy ?? this.ctx.sandboxPolicy.resolve()
|
||||
const cwd = request.cwd ?? sandboxPolicy.workspaceRoot
|
||||
if (!isAbsolute(cwd)) throw new Error('persistent PTC: cwd must be absolute')
|
||||
return { ...request, sandboxPolicy, cwd, timeoutMs: request.timeoutMs === null ? null : clampTimeout(request.timeoutMs, this.config.timeoutMs, this.config.maxTimeoutMs, 'persistent PTC timeoutMs') }
|
||||
}
|
||||
|
||||
/**
|
||||
* Evaluate statements using this owner's process, or a temporary process without an owner.
|
||||
* @param spec - Resolved request; concurrent calls for one owner are rejected.
|
||||
* @returns Logs, a final JSON expression, or a structured execution failure.
|
||||
*/
|
||||
async run(spec: PtcRunSpec): Promise<PtcRunResult> {
|
||||
if (this.disposed) throw new Error('persistent PTC: run after disposal')
|
||||
if (spec.sandboxPolicy === undefined || !isAbsolute(spec.cwd) || (spec.timeoutMs !== null && (!Number.isFinite(spec.timeoutMs) || spec.timeoutMs <= 0 || spec.timeoutMs > this.config.maxTimeoutMs))) throw new Error('persistent PTC: run requires resolved inputs with a sandbox policy')
|
||||
const bindings = validateBindings(spec)
|
||||
let kernel = spec.stateOwner === undefined ? undefined : this.owners.get(spec.stateOwner)
|
||||
kernel?.assertIdle()
|
||||
if (kernel && !kernel.closed && (kernel.policy.mode !== spec.sandboxPolicy.mode
|
||||
|| kernel.policy.workspaceRoot !== spec.sandboxPolicy.workspaceRoot
|
||||
|| kernel.policy.sessionId !== spec.sandboxPolicy.sessionId
|
||||
|| !sameAdditionalRoots(kernel.policy.additionalWritableRoots, spec.sandboxPolicy.additionalWritableRoots))) {
|
||||
throw new Error('persistent PTC: release the retained environment before changing its sandbox policy')
|
||||
}
|
||||
let code: string
|
||||
try { code = this.mode.stripTypes(spec.program) } catch (error: unknown) {
|
||||
const output = persistentOutput(this.config.maxOutputBytes)
|
||||
const outcome = output.failure([], { kind: 'exception', message: messageOf(error) })
|
||||
if (outcome.error?.kind === 'output-limit' && spec.stateOwner !== undefined) {
|
||||
try { await this.release(spec.stateOwner) } catch (failure: unknown) {
|
||||
return output.failure([], { kind: 'worker-exit', message: `managed process cleanup failed: ${messageOf(failure)}` })
|
||||
}
|
||||
}
|
||||
return outcome
|
||||
}
|
||||
if (!kernel || kernel.closed) {
|
||||
if (kernel) this.kernels.delete(kernel)
|
||||
// Direct callers can own mutable policy arrays; the retained process owns its grant snapshot.
|
||||
kernel = new Kernel(this.ctx, this.config, {
|
||||
...spec.sandboxPolicy,
|
||||
...spec.sandboxPolicy.additionalWritableRoots === undefined
|
||||
? {}
|
||||
: { additionalWritableRoots: [...spec.sandboxPolicy.additionalWritableRoots] },
|
||||
}, this.processes, this.mode)
|
||||
this.kernels.add(kernel)
|
||||
if (spec.stateOwner !== undefined) this.owners.set(spec.stateOwner, kernel)
|
||||
}
|
||||
try { return await kernel.run(spec, bindings, code) } finally {
|
||||
if (spec.stateOwner === undefined) this.kernels.delete(kernel)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Release retained state and await its process cleanup.
|
||||
* @param owner - Exact object supplied to run.
|
||||
* @returns Completion after the owned process and output streams stop.
|
||||
*/
|
||||
async release(owner: object): Promise<void> {
|
||||
const kernel = this.owners.get(owner)
|
||||
if (!kernel) return
|
||||
await kernel.dispose()
|
||||
if (this.owners.get(owner) === kernel) this.owners.delete(owner)
|
||||
this.kernels.delete(kernel)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user