diff --git a/bench/probes/probe-872-shell-host.mjs b/bench/probes/probe-872-shell-host.mjs new file mode 100644 index 00000000..6bc6ac43 --- /dev/null +++ b/bench/probes/probe-872-shell-host.mjs @@ -0,0 +1,224 @@ +/** Local candidate proof only: real host authority and kernel scheduling, no model calls or shipped pin change. */ +import fs from 'node:fs' +import path from 'node:path' +import os from 'node:os' +import assert from 'node:assert/strict' +import { createHash } from 'node:crypto' +import { setTimeout as delay } from 'node:timers/promises' + +const args = Object.fromEntries(Array.from({ length: (process.argv.length - 2) / 2 }, (_, i) => [process.argv[2 + i * 2].replace(/^--/, ''), process.argv[3 + i * 2]])) +for (const key of ['worker', 'bash', 'payload', 'sha256', 'report']) assert.ok(args[key], `--${key} required`) +assert.equal(process.platform, 'win32', 'C2 process ownership is Windows-only') +const hash = file => createHash('sha256').update(fs.readFileSync(file)).digest('hex') +assert.equal(hash(args.bash), args.sha256, 'candidate hash must match the explicit local proof input') +const home = fs.mkdtempSync(path.join(os.tmpdir(), 'ops-c2-host-home-')) +process.env.DSH_HOME = home +const workspace = fs.mkdtempSync(path.join(os.tmpdir(), 'ops-c2-host-work-')) +const modules = await Promise.all(['@deepseek-ai/cordis', '@deepseek-ai/dsh-agent', '@deepseek-ai/dsh-session', + '@deepseek-ai/dsh-session-projection', '@deepseek-ai/dsh-system-prompt', '@deepseek-ai/dsh-tools', + '@deepseek-ai/dsh-llm', '@deepseek-ai/dsh-agent-loop', '@deepseek-ai/dsh-sandbox-policy', + '@deepseek-ai/dsh-subprocess-local', '@deepseek-ai/dsh-sandbox-local', '@deepseek-ai/dsh-fs-local', + '@deepseek-ai/dsh-working-directory'].map(name => import(name))) +const [{ Context }, Agents, Sessions, Projections, Prompt, Tools, Llm, Loop, Policy, Subprocess, Sandbox, FileSystem, Directory] = modules +const { startResident } = await import('../../lib/execution/resident/client.js') +const { workerEnvelope } = await import('../../lib/execution/protocol.js') +const { residentShellAdmission } = await import('../../lib/execution/resident/shell-owner.js') +const { prepareNiubash } = await import('../../lib/niubash/runtime.js') +const ctx = new Context(); ctx.logger.level = 0 +const report = { kind: 'local-candidate-real-host-C2-proof', started: new Date().toISOString(), + worker: args.worker, workerSha256: hash(args.worker), bash: args.bash, bashSha256: args.sha256, + home, workspace, limitations: ['No model-facing publication, persistent Jobs, performance claim or shipped pin change.', + 'Actual Cordis/Agent/Session/policy/directory/subprocess/sandbox; approval and explicit shellEnv are instrumented public service shapes.'], + checks: [], events: [] } +let client, host, ownerFiber, kernelFiber, next = 0 +const handles = [], calls = new Map(), approvals = [] +let approve = async () => 'allowed-once' +const quote = value => `'${value.replaceAll("'", "'\\''")}'` +const node = quote(process.execPath.replaceAll('\\', '/')) +const sleep = ms => `${node} -e ${quote(`setTimeout(()=>{},${ms})`)}` +const event = (type, id) => report.events.find(value => value.type === type && value.requestId === id) +async function until(predicate, label, ms = 15_000) { + const deadline = Date.now() + ms + for (;;) { const value = predicate(); if (value) return value; if (Date.now() >= deadline) throw new Error(`${label} timed out`); await delay(10) } +} +async function within(work, label, ms = 15_000) { + let timer + try { return await Promise.race([work, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`${label} timed out`)), ms) })]) } + finally { clearTimeout(timer) } +} +async function check(name, run) { + const started = Date.now() + try { const detail = await run(); report.checks.push({ name, status: 'passed', ms: Date.now() - started, detail }); console.log(`ok ${name}`) } + catch (error) { report.checks.push({ name, status: 'failed', ms: Date.now() - started, error: String(error.stack ?? error) }); console.error(`FAIL ${name}: ${error.stack ?? error}`) } +} +async function agent(mode = 'danger-full-access') { + const handle = await ctx.get('agents').create({ sessionId: `c2-host-${++next}`, meta: { cwd: workspace }, agentOptions: { provider: 'fixture', model: 'never-called' } }) + handles.push(handle); Policy.setSandboxMode(handle.agent.session, mode) + return handle +} +function call(handle, command, args = {}, signal) { + const callId = `c2-call-${++next}`, previous = report.events.length + const result = host.execute({ command, timeoutMs: 10_000, ...args }, { agent: handle.agent, callId, signal }) + const queued = report.events.slice(previous).find(value => value.type === 'shell.execute') + assert.ok(queued, 'valid call must enqueue in native') + calls.set(queued.requestId, result) + return { id: queued.requestId, result, callId } +} +function okay(value) { assert.equal(value.status, 'completed', JSON.stringify(value)); assert.equal(value.cleanupConfirmed, true); return value } +try { + for (const plugin of [Agents, Sessions, Projections, Prompt, Tools, Llm]) await ctx.plugin(plugin.default, {}) + await ctx.plugin(Policy.default, { mode: 'danger-full-access', workspaceRoot: workspace }) + await ctx.plugin(Subprocess.default); await ctx.plugin(Sandbox.default); await ctx.plugin(FileSystem.default, { cwd: workspace }) + await ctx.plugin(Directory.default, { defaultDirectory: workspace }) + await ctx.plugin(inner => { + inner.provide('shellEnv', { collect: () => ({}) }) + inner.provide('approval', { async request(input) { approvals.push(input); return approve(input) } }) + inner.on('llm/stream', () => { throw new Error('No model request is permitted in this proof') }) + }) + await ctx.plugin(Loop.default, { agents: [] }) + kernelFiber = await ctx.plugin({ name: 'proof-kernel', inject: ['subprocess'], async apply(inner) { client = await startResident(inner, args.worker, workspace) } }) + report.kernelPid = client.pid + const transport = { + get available() { return client.shell.available }, get capabilities() { return client.shell.capabilities }, + send(message) { + report.events.push({ type: message.type, requestId: message.requestId, sessionKey: message.sessionKey, decision: message.decision, + closeAfter: message.closeAfter, at: Date.now() }) + return client.shell.send(message) + }, + subscribe(listener) { return client.shell.subscribe(message => { + report.events.push({ type: message.type, requestId: message.requestId, status: message.status, shellState: message.shellState, + cleanupConfirmed: message.cleanupConfirmed, hasShell: message.hasShell, at: Date.now(), error: message.error }) + listener(message) + }) }, + read: (...values) => client.shell.read(...values), abort: error => client.shell.abort(error), + } + ownerFiber = await ctx.plugin(residentShellAdmission({ transport, bash: { file: args.bash, source: 'explicit-local-backport-candidate' }, + async runtime(_bash, env, policy, signal) { + const runtime = await prepareNiubash(path.join(args.payload, 'niu.exe'), env.PATH ?? env.Path, signal, { workspaceRoot: policy.workspaceRoot }) + return { ...runtime, file: args.bash } + } })) + host = ctx.get('opsShellAdmission'); assert.ok(host, 'injected admission owner must be ready') + + await check('actual host preserves cwd/env/functions and explicit persistent workdir', async () => { + const handle = await agent(), child = path.join(workspace, 'child'), explicit = path.join(workspace, 'explicit') + fs.mkdirSync(child); fs.mkdirSync(explicit) + okay(await call(handle, "cd child\nexport KEEP='中文 value'\nkept() { printf '%s' \"$KEEP\"; }\nset -f\nprintf first").result) + const second = okay(await call(handle, "pwd; kept; printf '\\nflags:%s\\n' \"$-\"").result) + assert.match(second.stdout.text, /\/child\r?\n中文 value\r?\nflags:.*f/) + okay(await call(handle, 'printf changed', { workdir: explicit }).result) + const third = okay(await call(handle, 'pwd; kept').result) + assert.match(third.stdout.text, /\/explicit\r?\n中文 value/) + return { session: handle.agent.id, tail: third.stdout.text } + }) + await check('native FIFO serializes one session while a sibling overlaps', async () => { + const first = await agent(), sibling = await agent() + okay(await call(first, 'printf warm').result); okay(await call(sibling, 'printf warm').result) + const slow = call(first, `${sleep(650)}; printf slow`) + await until(() => event('shell.started', slow.id), 'slow start') + const queued = call(first, 'printf queued'), other = call(sibling, 'printf sibling') + const file = path.join(workspace, 'while-shell.txt'); fs.writeFileSync(file, 'same kernel file traffic') + const exchange = await client.invoke(workerEnvelope('file', { root: workspace, scope: 'c2-file', tool: 'inspect_local_file', read_only: true, + deny_roots: [], deny_credential_names: false, args: { path: file, view: 'text' } })) + assert.equal(exchange.value?.status, 'completed', JSON.stringify(exchange)) + okay(await other.result) + assert.equal(event('shell.started', queued.id), undefined) + assert.equal(event('shell.completed', slow.id), undefined) + okay(await slow.result); okay(await queued.result) + assert.ok(event('shell.started', queued.id).at >= event('shell.completed', slow.id).at) + }) + await check('queued native timeout and pending approval cancellation have no effects', async () => { + const handle = await agent('read-only') + okay(await call(handle, 'printf warm').result) + const slow = call(handle, `${sleep(600)}; printf slow`) + await until(() => event('shell.started', slow.id), 'blocking start') + const queued = call(handle, 'printf MUST_NOT_RUN', { timeoutMs: 100 }) + const expired = await queued.result + assert.equal(expired.effectState, 'none'); assert.notEqual(expired.status, 'completed') + assert.equal(event('shell.started', queued.id), undefined); assert.equal(event('shell.completed', slow.id), undefined) + okay(await slow.result) + const granted = Promise.withResolvers(); approve = () => granted.promise + const controller = new AbortController(), before = approvals.length + const pending = call(handle, 'printf MUST_NOT_RUN', { sandbox_permissions: 'danger-full-access', justification: 'Fixture one-call proof.' }, controller.signal) + await until(() => approvals.length > before, 'approval request') + controller.abort(); const cancelled = await pending.result + assert.equal(cancelled.effectState, 'none'); assert.equal(event('shell.started', pending.id), undefined) + granted.resolve('allowed-once'); approve = async () => 'allowed-once' + okay(await call(handle, 'printf after-cancel').result) + const late = Promise.withResolvers(); approve = () => late.promise + const approvalCount = approvals.length + const expires = call(handle, 'printf MUST_NOT_RUN', { timeoutMs: 500, sandbox_permissions: 'danger-full-access', justification: 'Approval deadline fixture.' }) + await until(() => approvals.length > approvalCount, 'deadline approval request') + const timedOut = await expires.result + assert.equal(timedOut.effectState, 'none'); assert.notEqual(timedOut.status, 'completed') + assert.equal(event('shell.started', expires.id), undefined) + late.resolve('allowed-once'); approve = async () => 'allowed-once' + okay(await call(handle, 'printf after-approval-timeout').result) + }) + await check('admission retirement keeps the granted head and closeAfter prevents reuse', async () => { + const handle = await agent('read-only') + okay(await call(handle, 'export STANDING=old; printf warm').result) + const before = approvals.length + const widened = call(handle, 'export WIDE=one; printf widened', { sandbox_permissions: 'danger-full-access', justification: 'One exact fixture command.' }) + const value = okay(await widened.result) + assert.equal(value.shellState, 'reset'); assert.equal(approvals.length, before + 1) + const decisions = report.events.filter(value => value.requestId === widened.id && value.type === 'shell.admission') + assert.deepEqual(decisions.map(value => value.decision), ['retire', 'launch']) + assert.equal(decisions.at(-1).closeAfter, true) + const next = okay(await call(handle, "printf '%s/%s' \"${STANDING-unset}\" \"${WIDE-unset}\"").result) + assert.equal(next.stdout.text, 'unset/unset') + }) + await check('policy control retirement cancels current head but re-admits queued follower', async () => { + const handle = await agent('danger-full-access') + okay(await call(handle, 'printf warm').result) + const active = call(handle, `${sleep(2000)}; printf MUST_NOT_FINISH`) + await until(() => event('shell.started', active.id), 'active head') + const queued = call(handle, 'printf reread-policy') + Policy.setSandboxMode(handle.agent.session, 'read-only') + const stopped = await active.result; assert.notEqual(stopped.status, 'completed'); assert.equal(stopped.shellState, 'reset') + assert.equal(okay(await queued.result).stdout.text, 'reread-policy') + assert.ok(report.events.some(value => value.type === 'shell.retire')) + }) + await check('bounded distinct output is copied through EOF then released', async () => { + const handle = await agent() + const source = "process.stdout.write('A'.repeat(200000)+'TAIL');process.stderr.write('ERR-END')" + const request = call(handle, `${node} -e ${quote(source)}`) + const value = okay(await request.result) + assert.equal(value.stdout.bytesTotal, 200004); assert.equal(value.stdout.lossy, true) + // C2's internal window explicitly retains a prefix, not the old tool's spill contract. + assert.equal(value.stdout.text, 'A'.repeat(131072)); assert.equal(value.stdout.text.includes('TAIL'), false) + assert.equal(value.stderr.text, 'ERR-END'); assert.equal(value.stderr.lossy, false) + assert.ok(event('shell.release', request.id)); assert.equal(client.available, true) + const tail = "process.stdout.write('NEXT-HEAD');setTimeout(()=>{process.stdout.write('-TAIL');process.stderr.write('ERR-TAIL')},100)" + const next = okay(await call(handle, `${node} -e ${quote(tail)}`).result) + assert.equal(next.stdout.text, 'NEXT-HEAD-TAIL'); assert.equal(next.stderr.text, 'ERR-TAIL') + assert.equal(next.stdout.lossy, false); assert.equal(next.stderr.lossy, false) + }) + await check('actual Agent disposal joins its active shell and leaves sibling usable', async () => { + const handle = await agent(), sibling = await agent() + okay(await call(sibling, 'printf warm').result) + const active = call(handle, sleep(2000)); await until(() => event('shell.started', active.id), 'disposed active head') + await handle.dispose() + const value = await active.result + assert.equal(value.cleanupConfirmed, true); assert.notEqual(value.status, 'completed') + assert.equal(okay(await call(sibling, 'printf survives').result).stdout.text, 'survives') + }) +} catch (error) { + report.fatal = String(error.stack ?? error) + console.error(report.fatal) +} finally { + try { + await ownerFiber?.dispose() + report.settledCalls = (await within(Promise.allSettled([...calls.values()]), 'all admitted calls settle')).length + for (const handle of handles) await handle.dispose() + await kernelFiber?.dispose() + await ctx.fiber.dispose() + if (client) assert.throws(() => process.kill(client.pid, 0), error => error?.code === 'ESRCH', 'kernel PID must be gone after owned teardown') + report.kernelGone = true + report.cleanupConfirmed = true + } catch (error) { report.cleanupConfirmed = false; report.cleanupError = String(error.stack ?? error) } + report.finished = new Date().toISOString() + report.status = !report.fatal && report.cleanupConfirmed && report.checks.every(value => value.status === 'passed') ? 'passed' : 'failed' + fs.writeFileSync(args.report, `${JSON.stringify(report, null, 2)}\n`) + console.log(JSON.stringify({ status: report.status, checks: report.checks.length, passed: report.checks.filter(value => value.status === 'passed').length, report: args.report })) + if (report.status !== 'passed') process.exitCode = 1 +} diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index e398c4e5..17e949a4 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -7,6 +7,9 @@ this package adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ## [Unreleased] +- Add an internal resident-shell host adapter that rechecks current authority at each native queue head, scopes one-call grants and awaits owner cleanup. Bounded transport and terminal output collection preserve explicit effect and state-loss facts; public session tools and recoverable spill remain pending (#872). +- 添加内部常驻 shell 宿主适配器：每次原生队首准入重查当前权限、限制单次批准范围并等待所有者回收。有界通道与终态输出收集保留明确效果和状态丢失事实；公开会话工具与可恢复 spill 仍待完成（#872）。 + - Add a Windows resident shell queue with per-request admission, deadlines, cancellation and explicit state-loss results. Bounded output replies preserve control-reader responsiveness, and a blocked start acknowledgement remains cancellable. Repository-only regressions exercise real public-host ownership; model-facing shell publication and DSH jobs remain pending (#872). - 添加 Windows 常驻 shell 队列，按请求准入、计时、取消并明确报告状态丢失。有界输出回复保持控制 reader 可响应，开始回执阻塞时仍能取消。仓库专用回归验证真实公开宿主所有权；模型可见 shell 发布与 DSH jobs 仍待完成（#872）。 diff --git a/docs/resident-shell-admission.md b/docs/resident-shell-admission.md new file mode 100644 index 00000000..b0cf4323 --- /dev/null +++ b/docs/resident-shell-admission.md @@ -0,0 +1,77 @@ +# Resident shell host admission + +中文：[常驻 shell 宿主准入](常驻-shell-宿主准入.md). + +This internal adapter connects current public host authority to the +[native shell queue](resident-shell-queues.md) for [#872](https://github.com/T-Auto/dsh-ops/issues/872). +It does not publish a tool or enable session execution for the shipped shell runtime. +The complete feature still requires eligible runtime distribution, shared recoverable output, +Native/PTC publication, kernel-owned fresh/background tasks and full acceptance evidence. + +## One owner, current authority + +The adapter attaches to the already-ready activation-owned kernel's `shell-v1` channel. It never +starts another kernel or schedules commands in the host. Native queue-head admission triggers +policy, approval, environment, directory, runtime and confinement preparation. + +Ownership uses the exact live Agent and Session objects. Each owner receives a private random +native key; a reused public session ID cannot acquire an old owner's shell. Agent disposal awaits +native close. Removing or replacing an injected authority provider disposes the admission service +and waits for owner cleanup. Kernel loss withdraws the transport, reports its confirmed cleanup +outcome and settles pending calls without replay or restart. + +Every asynchronous preparation boundary rechecks the original standing authority and caller +lifetime. A late approval cannot authorize a cancelled, timed-out, retired or disposed request. +An allowed-once grant belongs only to the same unstarted queue head, including retirement needed +to replace a narrower shell. The wider range uses `closeAfter`, so its state is gone before the +result is published. Queued followers receive independent current admission. + +Policy changes observed through session events retire the active shell while preserving queued +followers. Head-time rechecks remain authoritative when no change event is available. Host +environment changes also retire the live shell. The adapter snapshots its configured Bash +identity; configuration replacement must dispose that owning adapter instead of relabelling it. + +## Launch facts and working directory + +The environment comes from the running host's public subprocess scrub and current shell-env +service. Windows merges are case-insensitive and honor deleted entries. The outer private-control +marker is removed. Nonempty `NIU_ENV`, `BASH_ENV` and `ENV` entries are rejected by name without +printing their values, both before runtime preparation and after the final environment merge. + +The host resolves and validates the first directory. An omitted later `workdir` preserves the +shell's current directory. An explicit relative directory resolves against the current host +session directory and is sent as a persistent directory change for that command. Restricted +launches use the exact argv returned by the public sandbox provider; native code does not replace +the wrapper. Missing or invalid confinement fails before command bytes enter the shell. + +## Bounded correlation and honest output + +The transport tracks at most 32 retained requests, 32 owners and 64 controls, with one outstanding +read per stream per request. Cancellation is coalesced. Ordinary capacity refusal occurs locally +before submission. The shared outgoing frame window accommodates both file and shell controls; +each frame remains bounded and only one physical write is outstanding. + +Native owns queue-inclusive deadlines. The adapter's default is 120 seconds and its maximum is +10 minutes. It collects both terminal output streams before releasing the native record. Output +errors retain the other stream's evidence; contradictory offsets or completion receipts fail the +owned connection. Empty live output never implies EOF. The current protocol retains a 128 KiB +prefix per stream and reports observed totals and loss; it does not claim recoverable spill. + +Terminal state changes are applied synchronously before output collection awaits. A slow old +output tail cannot erase the next queue head's new shell facts. A start notification followed by +cancellation before command bytes may correctly finish with `effectState: none`. + +## Reproduction + +Focused suites cover admission, protocol correlation, kernel transport and host package scope: + +```console +node scripts/build.mjs +node test/run.mjs resident-shell-admission resident-shell-channel resident-kernel host-scope-source +``` + +`bench/probes/probe-872-shell-host.mjs` additionally uses actual Cordis, Agent, Session, policy, +directory, subprocess and sandbox services. Its approval and shell-env inputs are instrumented +public service shapes. The probe requires explicit worker, candidate, support payload, executable +hash and report paths; it neither changes the shipped runtime pin nor establishes model-facing +or complete application acceptance. See the PR validation record for exact hashes and results. diff --git "a/docs/\345\270\270\351\251\273-shell-\345\256\277\344\270\273\345\207\206\345\205\245.md" "b/docs/\345\270\270\351\251\273-shell-\345\256\277\344\270\273\345\207\206\345\205\245.md" new file mode 100644 index 00000000..694c9445 --- /dev/null +++ "b/docs/\345\270\270\351\251\273-shell-\345\256\277\344\270\273\345\207\206\345\205\245.md" @@ -0,0 +1,64 @@ +# 常驻 shell 宿主准入 + +English: [Resident shell host admission](resident-shell-admission.md). + +此内部适配器将当前公开宿主权限连接到 [原生 shell 队列](常驻-shell-队列协议.md)， +用于 [#872](https://github.com/T-Auto/dsh-ops/issues/872)。它不发布工具，也不为已发布的 +shell runtime 启用会话执行。完整功能仍需要合格 runtime 发行、共享可恢复输出、Native/PTC +发布、内核所有的 fresh/后台任务以及完整验收证据。 + +## 单一所有者与当前权限 + +适配器连接已经 ready、由本次激活所有的内核 `shell-v1` 通道，不另起内核，也不在宿主调度 +命令。原生队首准入触发策略、批准、环境、目录、runtime 与沙箱约束准备。 + +所有权绑定精确的存活 Agent 和 Session 对象。每个所有者使用随机私有原生 key，复用公开 +session ID 不能取得旧所有者的 shell。Agent 释放会等待原生 close。注入的权限提供者被移除 +或替换时，准入服务释放并等待所有 shell 回收。内核丢失会撤销通道、报告已确认的回收结果， +结束待处理调用，不重放、不重启。 + +每次异步准备返回后都重查原有常驻权限与调用方生命周期。迟到批准不能授权已取消、超时、 +退役或释放的请求。allowed-once 只属于同一个尚未开始的队首，包括为替换较窄 shell 而退役 +后重新准入。较宽权限范围使用 `closeAfter`，在发布结果前结束其状态；后续请求单独准入。 + +session 事件观察到策略变化时，退役活动 shell 并保留排队请求。没有变化事件时，队首重查 +仍具有最终决定权。宿主环境变化也会退役活动 shell。配置中的 Bash 身份在适配器内固定； +配置替换必须释放原适配器，不能给存活 shell 换标签。 + +## 启动事实与工作目录 + +环境来自当前宿主公开 subprocess 清理接口与 shell-env 服务。Windows 合并忽略大小写并 +处理删除项，移除外层私有控制标记。非空 `NIU_ENV`、`BASH_ENV`、`ENV` 按变量名拒绝， +不打印值；runtime 准备前与最终环境合并后都检查。 + +初始目录由宿主解析、验证。后续省略 `workdir` 保留 shell 当前目录；显式相对路径按当前 +宿主会话目录解析，作为该命令之前的持久目录变更发送。受限启动使用公开 sandbox 返回的 +原样 argv，原生代码不替换 wrapper。约束缺失或无效时，在命令字节发送前失败。 + +## 有界关联与真实输出 + +通道最多保留 32 请求、32 所有者和 64 控制操作，每个请求每个流最多一个未完成读取。 +重复取消合并。普通容量不足在本地提交前拒绝。共享发送窗口容纳文件与 shell 控制， +每帧仍有字节上限，实际写入同时仅有一次。 + +包括排队时间的截止由原生内核负责；适配器默认 120 秒，上限 10 分钟。适配器收集两个 +终态输出流后才释放原生记录。输出错误保留另一个流的证据；矛盾的偏移或完成收据使连接 +失败。活动流为空不代表 EOF。当前每流保留 128 KiB 前缀，报告观测总量和丢失，不声称 +具有可恢复 spill。 + +终态状态变化在异步收集输出之前同步应用，因此旧请求缓慢收集尾部不能清除下一队首的新 +启动事实。开始通知之后、命令字节之前取消，最终 `effectState: none` 仍是真实结果。 + +## 复现 + +定向测试覆盖准入、协议关联、内核通道和宿主包作用域： + +```console +node scripts/build.mjs +node test/run.mjs resident-shell-admission resident-shell-channel resident-kernel host-scope-source +``` + +`bench/probes/probe-872-shell-host.mjs` 另使用真实 Cordis、Agent、Session、策略、目录、 +subprocess 与 sandbox。批准与 shell-env 输入使用公开服务形状的受控实现。探针要求显式 +worker、候选 shell、支持目录、可执行文件 hash 和报告路径，不改变发行 pin，也不构成 +模型可见发布或完整应用验收。精确 hash 与结果见 PR 验证记录。 diff --git a/src/adapter/dsh/host.ts b/src/adapter/dsh/host.ts index d5782a51..a99ffe4e 100644 --- a/src/adapter/dsh/host.ts +++ b/src/adapter/dsh/host.ts @@ -14,6 +14,17 @@ export async function loadUserMessage(): Promise> { + const anchor = hostEntryAnchor() + let source: string + try { source = anchor ? pathToFileURL(createRequire(anchor).resolve('@deepseek-ai/dsh-subprocess')).href : import.meta.resolve('@deepseek-ai/dsh-subprocess') } + catch { throw new Error('EXECUTOR_UNAVAILABLE: the host public subprocess environment API is unavailable.') } + const host = await import(source) as typeof import('@deepseek-ai/dsh-subprocess') + if (typeof host.scrubbedParentEnv !== 'function') throw new Error('EXECUTOR_UNAVAILABLE: the host public environment scrub is unavailable.') + return host.scrubbedParentEnv() +} + export function service(ctx: Pick, name: string): T | undefined { try { const value = ctx.get(name as never) diff --git a/src/execution/resident/client.ts b/src/execution/resident/client.ts index fc566e3d..6eaae450 100644 --- a/src/execution/resident/client.ts +++ b/src/execution/resident/client.ts @@ -5,6 +5,8 @@ import { record } from '../boundary.js' import type { workerEnvelope } from '../protocol.js' import type { WorkerExchange } from '../transport.js' import { ENVELOPE_BYTES, ResidentFrames, encodeFrame } from './frames.js' +import { ResidentShellChannel } from './shell-channel.js' +import type { ShellTransport } from './shell-protocol.js' type Envelope = ReturnType interface Pending { @@ -21,6 +23,7 @@ export interface ResidentClient { readonly file: string readonly pid: number readonly available: boolean + readonly shell: ShellTransport invoke(envelope: Envelope, signal?: AbortSignal, started?: () => void): Promise stop(): Promise } @@ -40,6 +43,7 @@ function exchange(value: unknown): value is WorkerExchange { class Client implements ResidentClient { pid = 0 + readonly shell: ResidentShellChannel private readonly pending = new Map() private readonly frames: ResidentFrames private readonly hello: Promise @@ -55,7 +59,12 @@ class Client implements ResidentClient { // The startup waiter is installed immediately; all paths also await it below. void this.hello.catch(() => {}) if (!handle.control) throw new Error('The host did not provide a resident control pipe') - this.frames = new ResidentFrames(handle.control, value => this.receive(value), error => this.fail(error)) + this.shell = new ResidentShellChannel(() => this.available, (value, sending) => this.frames.send(encodeFrame(value), sending), error => this.fail(error)) + this.frames = new ResidentFrames(handle.control, value => this.receive(value), error => this.fail(error), + // 32 file + 32 shell executions, their 64 cancels, 32 admissions, + // 64 retire/close controls, 64 stream reads, 32 releases and shutdown. + // Each frame remains byte-bounded; one outstanding write preserves backpressure. + () => this.shell.capabilities.includes('shell-v1') ? 321 : 65) this.timer = setTimeout(() => this.fail(new Error('Resident kernel handshake exceeded 10000ms')), 10_000) handle.done.then(() => this.fail(new Error('Resident kernel exited')), error => this.fail(new Error(`Resident kernel failed: ${String(error)}`))) } @@ -68,12 +77,16 @@ class Client implements ResidentClient { if (!record(value)) throw new Error('Resident message is not an object') if (value.type === 'ready') { if (this.pid || value.version !== 1 || !Number.isSafeInteger(value.pid) || Number(value.pid) < 1) throw new Error('Invalid resident handshake') + if (value.capabilities !== undefined && (!Array.isArray(value.capabilities) || value.capabilities.length > 16 + || value.capabilities.some(capability => typeof capability !== 'string' || capability.length > 64))) throw new Error('Invalid resident capabilities') + this.shell.capabilities = Object.freeze([...(value.capabilities as string[] | undefined) ?? []]) this.pid = Number(value.pid) clearTimeout(this.timer) this.ready() return } if (!this.pid || typeof value.requestId !== 'string') throw new Error('Resident message precedes handshake or lacks requestId') + if (this.shell.receive(value)) return const pending = this.pending.get(value.requestId) if (!pending) throw new Error('Resident message has an unknown requestId') if (value.type === 'started') { @@ -112,6 +125,7 @@ class Client implements ResidentClient { try { diagnostic = this.handle.collected.stderr?.readFrom(0).text ?? '' } catch { /* provider failure */ } for (const request of pending) if (request.sent) request.resolve({ exitCode: null, interrupted: true, readers: true, lossy: false, stderr: [error.message.slice(0, 1000), cleanupFailure.slice(0, 1000), diagnostic.slice(-2000)].join('\n') }) + this.shell.unavailable(error.message.slice(0, 1000), cleanupFailure === '') })() } invoke(envelope: Envelope, signal?: AbortSignal, started?: () => void): Promise { diff --git a/src/execution/resident/frames.ts b/src/execution/resident/frames.ts index cf038cd9..f37f794e 100644 --- a/src/execution/resident/frames.ts +++ b/src/execution/resident/frames.ts @@ -3,7 +3,7 @@ import type { Duplex } from 'node:stream' export const ENVELOPE_BYTES = 2 * 1024 * 1024 export const FRAME_BYTES = ENVELOPE_BYTES + 256 -const QUEUED_FRAMES = 65 // 32 executions, one cancellation each, one shutdown. +const QUEUED_FRAMES = 65 // File-only: 32 executions, one cancellation each, one shutdown. interface Queued { bytes: Buffer; sending(): void } export function encodeFrame(value: unknown): Buffer { @@ -17,7 +17,8 @@ export class ResidentFrames { private readonly queue: Queued[] = [] private writing = false private stopped = false - constructor(private readonly pipe: Duplex, receive: (value: unknown) => void, private readonly failed: (error: Error) => void) { + constructor(private readonly pipe: Duplex, receive: (value: unknown) => void, private readonly failed: (error: Error) => void, + private readonly capacity: () => number = () => QUEUED_FRAMES) { pipe.on('error', error => failed(error)) pipe.on('end', () => failed(new Error('Resident control channel ended'))) pipe.on('close', () => failed(new Error('Resident control channel closed'))) @@ -43,7 +44,7 @@ export class ResidentFrames { /** Return a remover for a frame that has not reached the writable stream yet. */ send(bytes: Buffer, sending: () => void = () => {}): () => boolean { if (this.stopped) throw new Error('Resident control channel is closed') - if (this.queue.length >= QUEUED_FRAMES) throw new Error('Resident control queue is full') + if (this.queue.length >= this.capacity()) throw new Error('Resident control queue is full') const entry = { bytes, sending } this.queue.push(entry) this.pump() diff --git a/src/execution/resident/shell-admission.ts b/src/execution/resident/shell-admission.ts new file mode 100644 index 00000000..f38d4a81 --- /dev/null +++ b/src/execution/resident/shell-admission.ts @@ -0,0 +1,297 @@ +/** Host authority at the native queue head. This module neither schedules commands nor publishes tools. */ +import path from 'node:path' +import fs from 'node:fs/promises' +import { randomUUID } from 'node:crypto' +import type { Agent, Context, Session } from '../../adapter/dsh/types.js' +import type { ShellCall, ShellPolicy } from '../../shells/types.js' +import { errorMessage } from '../../shells/types.js' +import { approveBashPolicy } from '../../shells/approval.js' +import { DEFAULT_BASH_TIMEOUT_MS, MAX_BASH_TIMEOUT_MS } from '../../shells/constants.js' +import { sameAuthority } from '../../authority-paths.js' +import { resolveSessionDirectory } from '../../adapter/dsh/working-directory.js' +import { sessionWorkdir } from '../../session-shells.js' +import { prepareShellLaunch, shellEnvironment, environmentIdentity, type ShellLaunchOptions } from './shell-launch.js' +import type { ShellAdmission, ShellCompleted, ShellFailure, ShellKernelMessage, ShellTransport } from './shell-protocol.js' +import { collectShellOutput, type ShellOutputPair } from './shell-output.js' + +type Args = { command: string; workdir?: string; timeoutMs: number; sandbox_permissions?: string; justification?: string } +interface Facts { standing: ShellPolicy; effective: ShellPolicy; environment: string } +interface Owner { + agent: Agent; session: Session; key: string; facts?: Facts; closed: boolean + closing?: Promise; retiring?: Promise +} +interface Pending { + owner: Owner; args: Args; call: ShellCall; signal: AbortSignal; abort: AbortController + sent: boolean; started: boolean; admitting: boolean; atHead?: boolean; terminal?: boolean; removeAbort(): void + grant?: { standing: ShellPolicy; effective: ShellPolicy } + resolve(value: ShellAdmissionResult): void +} +interface Control { owner: Owner; resolve(): void; reject(error: Error): void } +export interface ShellAdmissionOptions extends ShellLaunchOptions { + transport: ShellTransport + /** The owning publication may revoke access synchronously before its async reconciliation. */ + active?(agent: Agent): boolean +} +export type ShellAdmissionResult = ShellCompleted & ShellOutputPair +const MAX_PENDING = 32 +const MAX_COMMAND_BYTES = 1024 * 1024 +const OUTPUT_BYTES = 128 * 1024 +const snapshot = (policy: ShellPolicy): ShellPolicy => ({ ...policy, additionalWritableRoots: [...policy.additionalWritableRoots ?? []] }) +const same = (left: ShellPolicy, right: ShellPolicy) => sameAuthority(left, right) && left.sessionId === right.sessionId +function failure(error: unknown): ShellFailure { + const message = errorMessage(error) + const code = error instanceof Error && error.name === 'AbortError' ? 'CANCELLED' + : error instanceof Error && error.name === 'BashArgumentError' ? 'INVALID_ARGUMENT' + : /^([A-Z][A-Z_]+):/.exec(message)?.[1] ?? 'EXECUTOR_UNAVAILABLE' + return { code, message } +} +function emptyOutput(lossy = false): ShellOutputPair { + return { stdout: { text: '', bytesTotal: 0, lossy }, stderr: { text: '', bytesTotal: 0, lossy } } +} +function refused(requestId: string, error: ShellFailure, sent = false, cleanupConfirmed = true): ShellAdmissionResult { + return { type: 'shell.completed', requestId, status: sent ? 'failed' : 'rejected', exitCode: null, + effectState: sent ? 'unknown' : 'none', shellState: 'reset', cleanupConfirmed, stdoutTotal: 0, stderrTotal: 0, error, ...emptyOutput(sent) } +} +function argumentsFor(input: Record): Args { + if (typeof input.command !== 'string' || !input.command.trim() || input.command.includes('\0') + || Buffer.byteLength(input.command) > MAX_COMMAND_BYTES) throw new Error('INVALID_ARGUMENT: command must be nonempty and at most 1 MiB, without NUL.') + if (input.script_path !== undefined || input.run_in_background !== undefined) throw new Error('INVALID_ARGUMENT: this internal foreground protocol accepts command only.') + if (input.workdir !== undefined && (typeof input.workdir !== 'string' || input.workdir.includes('\0'))) throw new Error('INVALID_ARGUMENT: workdir must be a string without NUL.') + const timeoutMs = input.timeoutMs ?? DEFAULT_BASH_TIMEOUT_MS + if (typeof timeoutMs !== 'number' || !Number.isInteger(timeoutMs) || timeoutMs < 1 || timeoutMs > MAX_BASH_TIMEOUT_MS) throw new Error(`INVALID_ARGUMENT: timeoutMs must be an integer from 1 to ${MAX_BASH_TIMEOUT_MS}.`) + // Escalation validation belongs at admission, against the then-current standing policy. + return { ...input, command: input.command, workdir: typeof input.workdir === 'string' && input.workdir.trim() ? input.workdir : undefined, timeoutMs } as Args +} + +export function createShellAdmissionHost(ctx: Context, options: ShellAdmissionOptions) { + const { transport } = options + // Config replacement must dispose this owning context. Mutable caller config cannot relabel a live shell. + options = { ...options, bash: { ...options.bash } } + if (!transport.available || !transport.capabilities.includes('shell-v1')) throw new Error('EXECUTOR_UNAVAILABLE: the resident kernel does not support shell-v1.') + const owners = new Map(), identities = new Map() + const disposed = new WeakSet(), pending = new Map(), controls = new Map() + let stopped = false, disposal: Promise | undefined + let unavailable: Extract | undefined + + function current(owner: Owner): ShellPolicy { + if (stopped || owner.closed || disposed.has(owner.agent) || owner.agent.session !== owner.session + || ctx.get('agents')?.get(owner.agent.id) !== owner.agent || options.active?.(owner.agent) === false) { + throw new Error('OWNER_DISPOSED: the exact calling Agent/Session is no longer active.') + } + const policy = ctx.get('sandboxPolicy')?.resolve({ session: owner.session }) + if (!policy) throw new Error('SANDBOX_UNAVAILABLE: no current session policy; no command executed.') + return snapshot(policy) + } + function assertCurrent(item: Pending, standing: ShellPolicy) { + item.signal.throwIfAborted() + if (!same(current(item.owner), standing)) throw new Error('AUTHORITY_CHANGED: session policy changed before admission; no command executed.') + } + function finish(requestId: string, value: ShellAdmissionResult) { + const item = pending.get(requestId) + if (!item) return + pending.delete(requestId) + item.removeAbort() + item.abort.abort() + item.resolve(value) + } + function control(owner: Owner, type: 'shell.retire' | 'shell.close', reason: string): Promise { + const requestId = randomUUID() + return new Promise((resolve, reject) => { + controls.set(requestId, { owner, resolve, reject }) + try { transport.send({ type, requestId, sessionKey: owner.key, reason }) } + catch (error) { transport.abort(new Error(errorMessage(error))) } + }) + } + function retire(owner: Owner): Promise { + if (owner.closed) return owner.closing ?? Promise.resolve() + if (!owner.retiring) { + owner.facts = undefined + // Retire affects the current head only. Queued requests keep their own + // admission opportunity under the next policy; late head callbacks stop. + for (const item of pending.values()) if (item.owner === owner && item.atHead && !item.terminal) { + item.removeAbort() + item.abort.abort(new Error('AUTHORITY_CHANGED: the active shell is being retired.')) + } + owner.retiring = control(owner, 'shell.retire', 'Session authority changed') + .catch(error => { void close(owner).catch(() => {}); throw error }) + .finally(() => { owner.retiring = undefined }) + } + return owner.retiring + } + function close(owner: Owner): Promise { + if (owner.closing) return owner.closing + owner.closed = true + disposed.add(owner.agent) + for (const item of pending.values()) if (item.owner === owner) item.abort.abort(new Error('OWNER_DISPOSED: session owner is closing.')) + const closed = unavailable ? unavailable.cleanupConfirmed ? Promise.resolve() : Promise.reject(new Error(unavailable.error.message)) + : control(owner, 'shell.close', 'Session owner disposed') + owner.closing = closed.then(() => { + owners.delete(owner.session) + if (identities.get(owner.agent.id) === owner) identities.delete(owner.agent.id) + }) + return owner.closing + } + function ownerFor(agent: Agent | undefined): Owner { + if (!agent?.session || disposed.has(agent)) throw new Error('OWNER_DISPOSED: an active authoritative Agent/Session is required.') + const existing = owners.get(agent.session) + if (existing && existing.agent !== agent) throw new Error('OWNER_CONFLICT: this Session belongs to another Agent.') + if (existing) { current(existing); return existing } + if (identities.has(agent.id)) throw new Error('OWNER_CLOSING: the previous owner with this session id has not finished cleanup.') + if (owners.size >= 32) throw new Error('EXECUTOR_BUSY: resident shell owner capacity is full; dispose an owner before opening another.') + const owner: Owner = { agent, session: agent.session, key: randomUUID(), closed: false } + current(owner) + owners.set(owner.session, owner); identities.set(agent.id, owner) + agent.ctx.effect(() => () => close(owner)) + return owner + } + async function directory(item: Pending, standing: ShellPolicy): Promise { + const basis = await resolveSessionDirectory(ctx, item.owner.agent, item.signal) + assertCurrent(item, standing) + const cwd = sessionWorkdir(item.args, { header: { cwd: basis.cwd } }) + if (typeof cwd !== 'string' || !path.isAbsolute(cwd) || cwd.includes('\0')) throw new Error('INVALID_ARGUMENT: workdir must resolve to an absolute path.') + const info = await fs.stat(cwd) + assertCurrent(item, standing) + if (!info.isDirectory()) throw new Error('INVALID_ARGUMENT: workdir must be an existing directory.') + return cwd + } + async function admit(event: Extract) { + const item = pending.get(event.requestId) + if (!item) return + if (item.admitting || item.started || item.terminal) { void close(item.owner).catch(() => {}); return } + item.admitting = true + item.atHead = true + let reply: ShellAdmission + try { + const standing = current(item.owner) + assertCurrent(item, standing) + const env = await shellEnvironment(ctx, { ...item.call, signal: item.signal }) + assertCurrent(item, standing) + if (item.owner.retiring) { item.admitting = false; return } + const environment = environmentIdentity(env) + const facts = item.owner.facts + if (event.hasShell && (!facts || !same(facts.standing, standing) || facts.environment !== environment)) { + reply = { type: 'shell.admission', requestId: event.requestId, decision: 'retire' } + } else { + if (!item.grant) item.grant = { standing, effective: snapshot(await approveBashPolicy(ctx, item.args, { ...item.call, signal: item.signal }, standing)) } + assertCurrent(item, item.grant.standing) + const effective = item.grant.effective + if (event.hasShell && !same(facts!.effective, effective)) { + reply = { type: 'shell.admission', requestId: event.requestId, decision: 'retire' } + } else { + const cwd = !event.hasShell || item.args.workdir !== undefined ? await directory(item, standing) : undefined + assertCurrent(item, standing) + if (event.hasShell) reply = { type: 'shell.admission', requestId: event.requestId, decision: 'reuse', ...(cwd ? { commandCwd: cwd } : {}) } + else { + const endpoints = event.bootstrap + if (!endpoints || Object.values(endpoints).some(value => typeof value !== 'string' || !value || value.includes('\0') || value.length > 512)) throw new Error('PROTOCOL_ERROR: missing trusted bootstrap endpoints.') + const prepared = await prepareShellLaunch(ctx, options, endpoints, cwd!, env, effective, item.signal, () => assertCurrent(item, standing)) + assertCurrent(item, standing) + item.owner.facts = { standing, effective, environment } + reply = { type: 'shell.admission', requestId: event.requestId, decision: 'launch', launch: prepared.launch, + closeAfter: !same(standing, effective), ...(item.args.workdir ? { commandCwd: cwd } : {}) } + } + } + } + } catch (error) { reply = { type: 'shell.admission', requestId: event.requestId, decision: 'reject', error: failure(error) } } + item.admitting = false + if (pending.get(event.requestId) !== item || item.terminal || item.signal.aborted || item.owner.closed || item.owner.retiring || stopped) return + // Every async boundary above rechecks authority; there is no await between the last check and enqueue. + try { transport.send(reply) } + catch (error) { transport.abort(new Error(errorMessage(error))) } + } + function receive(event: ShellKernelMessage) { + if (event.type === 'shell.output') return // read() owns offset-correlated output receipts. + if (event.type === 'shell.admit') { void admit(event); return } + if (event.type === 'shell.started') { const item = pending.get(event.requestId); if (item) item.started = true; return } + if (event.type === 'shell.completed') { + const item = pending.get(event.requestId) + if (!item || item.terminal) return + item.terminal = true + // Native may admit the next queued request while these terminal streams + // are still being copied. Apply this receipt's state change now, never + // after awaits that could overwrite the next shell's launch facts. + if (event.shellState === 'reset') item.owner.facts = undefined + item.removeAbort() + item.abort.abort(new Error('The kernel has settled this request.')) + if (!transport.available) { finish(event.requestId, { ...event, ...emptyOutput(event.effectState !== 'none') }); return } + void collectShellOutput(transport, event).then(output => { + if (!pending.has(event.requestId)) return + transport.send({ type: 'shell.release', requestId: event.requestId }) + finish(event.requestId, { ...event, ...output, status: output.error ? 'failed' : event.status, + ...(event.error ?? output.error ? { error: event.error ?? output.error } : {}) }) + }).catch(error => { transport.abort(new Error(errorMessage(error))) }) + return + } + if (event.type === 'shell.closed') { + const value = controls.get(event.requestId) + if (!value) return + controls.delete(event.requestId) + value.owner.facts = undefined + if (event.cleanupConfirmed) value.resolve() + else value.reject(new Error(event.error?.message ?? 'Shell cleanup was not confirmed.')) + return + } + stopped = true + unavailable = event + for (const owner of owners.values()) owner.facts = undefined + for (const [id, item] of pending) finish(id, refused(id, event.error, item.sent, event.cleanupConfirmed)) + for (const value of controls.values()) { + if (event.cleanupConfirmed) value.resolve() + else value.reject(new Error(event.error.message)) + } + controls.clear() + } + const unsubscribe = transport.subscribe(receive) + const removeEvent = ctx.on('session/event', (session) => { + const owner = owners.get(session) + if (!owner || owner.closed || !owner.facts) return + try { if (!same(current(owner), owner.facts.standing)) void retire(owner).catch(() => {}) } + catch { void close(owner).catch(() => {}) } + }) + const removeDisposed = ctx.on('session/disposed', session => { const owner = owners.get(session); if (owner) void close(owner).catch(() => {}) }) + + function execute(input: Record, call: ShellCall): Promise { + const requestId = randomUUID() + let args: Args, owner: Owner + try { + call.signal?.throwIfAborted() + if (stopped || !transport.available) throw new Error('EXECUTOR_UNAVAILABLE: resident shell transport is unavailable.') + if (pending.size >= MAX_PENDING) throw new Error('EXECUTOR_BUSY: resident shell correlation capacity is full.') + args = argumentsFor(input); owner = ownerFor(call.agent) + } catch (error) { return Promise.resolve(refused(requestId, failure(error))) } + return new Promise(resolve => { + const abort = new AbortController() + const signal = call.signal ? AbortSignal.any([call.signal, abort.signal]) : abort.signal + let cancelled = false + const cancel = () => { + if (!item.sent || cancelled || !pending.has(requestId)) return + cancelled = true + try { transport.send({ type: 'cancel', requestId }) } catch (error) { transport.abort(new Error(errorMessage(error))) } + } + const item: Pending = { owner, args, call: { ...call, agent: owner.agent }, signal, abort, sent: false, started: false, admitting: false, + removeAbort: () => signal.removeEventListener('abort', cancel), resolve } + pending.set(requestId, item) + signal.addEventListener('abort', cancel, { once: true }) + try { + signal.throwIfAborted() + item.sent = true + transport.send({ type: 'shell.execute', requestId, sessionKey: owner.key, command: args.command, + timeoutMs: args.timeoutMs, stdoutBytes: OUTPUT_BYTES, stderrBytes: OUTPUT_BYTES }) + } catch (error) { finish(requestId, refused(requestId, failure(error))) } + }) + } + function dispose(): Promise { + if (!disposal) { + stopped = true + removeEvent(); removeDisposed() + disposal = Promise.allSettled([...owners.values()].map(close)).then(results => { + unsubscribe() + const failed = results.find(result => result.status === 'rejected') + if (failed?.status === 'rejected') throw failed.reason + }) + } + return disposal + } + ctx.effect(() => dispose) + return { execute, dispose } +} diff --git a/src/execution/resident/shell-channel.ts b/src/execution/resident/shell-channel.ts new file mode 100644 index 00000000..b26f65c6 --- /dev/null +++ b/src/execution/resident/shell-channel.ts @@ -0,0 +1,160 @@ +/** Bounded shell correlation on the existing resident control pipe; scheduling stays native. */ +import { record } from '../boundary.js' +import type { ShellCompleted, ShellHostMessage, ShellKernelMessage, ShellOutput, ShellStream, ShellTransport } from './shell-protocol.js' + +interface Request { sent: boolean; started: boolean; completed?: boolean; local?: boolean; cancelled?: boolean; awaitingAdmission?: boolean; remove?: () => boolean } +interface Read { offset: number; maxBytes: number; resolve(value: ShellOutput): void; reject(error: Error): void } +const identity = (value: unknown): value is string => typeof value === 'string' && /^[A-Za-z0-9_-]{1,128}$/.test(value) +const count = (value: unknown): boolean => Number.isSafeInteger(value) && Number(value) >= 0 +const error = (value: unknown): boolean => record(value) && typeof value.code === 'string' && typeof value.message === 'string' + +export class ResidentShellChannel implements ShellTransport { + capabilities: readonly string[] = [] + private readonly requests = new Map() + private readonly controls = new Map() + private readonly sessions = new Set() + private readonly reads = new Map() + private listener?: (message: ShellKernelMessage) => void + private terminal?: Extract + constructor(private readonly isAvailable: () => boolean, + private readonly write: (value: ShellHostMessage, sending?: () => void) => () => boolean, + readonly abort: (error: Error) => void) {} + get available(): boolean { return !this.terminal && this.isAvailable() } + subscribe(listener: (message: ShellKernelMessage) => void): () => void { + if (this.listener) throw new Error('Resident shell already has an admission owner') + this.listener = listener + if (this.terminal) queueMicrotask(() => { if (this.listener === listener) listener(this.terminal!) }) + return () => { if (this.listener === listener) this.listener = undefined } + } + send(message: ShellHostMessage): void { + if (!this.available || !this.capabilities.includes('shell-v1')) throw new Error('Resident shell capability is unavailable') + if (!identity(message.requestId)) throw new Error('Invalid shell request identity') + if (message.type === 'shell.execute') { + if (this.requests.size >= 32 || this.requests.has(message.requestId) || this.controls.has(message.requestId)) throw new Error('Resident shell admission is full or request identity is duplicated') + if (!identity(message.sessionKey) || (!this.sessions.has(message.sessionKey) && this.sessions.size >= 32)) throw new Error('Resident shell owner capacity is full or invalid') + const newSession = !this.sessions.has(message.sessionKey) + this.sessions.add(message.sessionKey) + const request: Request = { sent: false, started: false } + this.requests.set(message.requestId, request) + try { request.remove = this.write(message, () => { request.sent = true }) } + catch (error) { this.requests.delete(message.requestId); if (newSession) this.sessions.delete(message.sessionKey); throw error } + return + } + if (message.type === 'shell.retire' || message.type === 'shell.close') { + if (this.controls.size >= 64 || this.controls.has(message.requestId) || this.requests.has(message.requestId)) throw new Error('Resident shell control capacity is full or identity is duplicated') + if ([...this.controls.values()].some(control => control.sessionKey === message.sessionKey && control.close === (message.type === 'shell.close'))) throw new Error('Duplicate pending shell owner control') + this.controls.set(message.requestId, { sessionKey: message.sessionKey, close: message.type === 'shell.close' }) + try { this.write(message) } catch (error) { this.controls.delete(message.requestId); throw error } + return + } + const request = this.requests.get(message.requestId) + if (!request) throw new Error('Unknown resident shell request') + if (message.type === 'shell.release') { + if (!request.completed || ['stdout', 'stderr'].some(stream => this.reads.has(`${message.requestId}:${stream}`))) throw new Error('Shell output is still active') + if (!request.local) this.write(message) + this.requests.delete(message.requestId) + return + } + if (message.type === 'shell.read') { + throw new Error('Shell reads require the offset-correlated read() method') + } + if (request.completed) throw new Error('Shell execution already completed') + if (message.type === 'shell.admission') { + if (!request.awaitingAdmission) throw new Error('Shell admission reply was not requested') + this.write(message) + request.awaitingAdmission = false + return + } + if (message.type === 'cancel' && request.cancelled) return + if (message.type === 'cancel') request.cancelled = true + if (message.type === 'cancel' && !request.sent && request.remove?.()) { + request.completed = true + request.local = true + this.listener?.({ type: 'shell.completed', requestId: message.requestId, status: 'cancelled', exitCode: null, + effectState: 'none', shellState: 'preserved', cleanupConfirmed: true, stdoutTotal: 0, stderrTotal: 0, + error: { code: 'CANCELLED', message: 'Cancelled before the execution frame was written.' } }) + return + } + this.write(message) + } + read(requestId: string, stream: ShellStream, offset: number, maxBytes: number): Promise { + const key = `${requestId}:${stream}` + if (!this.available || !this.capabilities.includes('shell-v1')) return Promise.reject(new Error('Resident shell capability is unavailable')) + if (!this.requests.has(requestId)) return Promise.reject(new Error('Unknown resident shell request')) + if (!['stdout', 'stderr'].includes(stream) || !count(offset) || !Number.isInteger(maxBytes) || maxBytes < 1 || maxBytes > 65536 + || this.reads.has(key)) return Promise.reject(new Error('Invalid or concurrent shell stream read')) + if (this.requests.get(requestId)?.local) return Promise.resolve({ type: 'shell.output', requestId, stream, + offset, nextOffset: offset, bytesBase64: '', lossy: offset !== 0, eof: true }) + return new Promise((resolve, reject) => { + this.reads.set(key, { offset, maxBytes, resolve, reject }) + try { this.write({ type: 'shell.read', requestId, stream, offset, maxBytes }) } + catch (error) { this.reads.delete(key); reject(error) } + }) + } + receive(value: Record): boolean { + if (typeof value.type !== 'string' || !value.type.startsWith('shell.')) return false + if (!this.capabilities.includes('shell-v1') || !identity(value.requestId)) throw new Error('Unnegotiated or invalid shell response') + if (value.type === 'shell.output') { + const key = `${value.requestId}:${value.stream}`, read = this.reads.get(key) + if (!read || value.offset !== read.offset || !count(value.nextOffset) || Number(value.nextOffset) < read.offset + || typeof value.lossy !== 'boolean' || typeof value.eof !== 'boolean' || typeof value.bytesBase64 !== 'string' + || (value.error !== undefined && !error(value.error))) throw new Error('Invalid shell output receipt') + const bytes = Buffer.from(value.bytesBase64, 'base64'), delta = Number(value.nextOffset) - read.offset + if (bytes.length > read.maxBytes || bytes.toString('base64') !== value.bytesBase64 || delta < bytes.length + || (!value.lossy && delta !== bytes.length)) throw new Error('Invalid shell output bytes or offsets') + this.reads.delete(key) + read.resolve(value as unknown as ShellOutput) + return true + } + if (value.type === 'shell.closed') { + const control = this.controls.get(value.requestId) + if (!control || typeof value.cleanupConfirmed !== 'boolean' + || (!value.cleanupConfirmed && !error(value.error))) throw new Error('Invalid shell cleanup receipt') + this.controls.delete(value.requestId) + if (control.close && value.cleanupConfirmed) this.sessions.delete(control.sessionKey) + } else { + const request = this.requests.get(value.requestId) + if (!request?.sent || request.completed) throw new Error('Unknown, completed or unwritten shell response identity') + if (value.type === 'shell.admit') { + if (request.started || request.awaitingAdmission || typeof value.hasShell !== 'boolean' || (!value.hasShell && (!record(value.bootstrap) + || ![value.bootstrap.commandPipe, value.bootstrap.receiptPipe].every(endpoint => typeof endpoint === 'string' && endpoint.length > 0 && endpoint.length <= 512 && !endpoint.includes('\0'))))) { + throw new Error('Invalid shell admission request') + } + request.awaitingAdmission = true + } else if (value.type === 'shell.started') { + if (request.started) throw new Error('Resident shell command started twice') + request.started = true + } else if (value.type === 'shell.completed') { + if (!['completed', 'failed', 'cancelled', 'rejected'].includes(String(value.status)) + || !(value.exitCode === null || Number.isSafeInteger(value.exitCode)) + || !['none', 'unknown'].includes(String(value.effectState)) || !['preserved', 'reset'].includes(String(value.shellState)) + || typeof value.cleanupConfirmed !== 'boolean' || !count(value.stdoutTotal) || !count(value.stderrTotal) + || (value.status !== 'completed' && !error(value.error)) + || (value.status === 'completed' && (!request.started || !value.cleanupConfirmed)) + || (value.status === 'rejected' && (request.started || value.effectState !== 'none'))) throw new Error('Invalid shell completion receipt') + request.completed = true + } else throw new Error('Unknown resident shell response') + } + if (!this.listener) throw new Error('Resident shell response has no admission owner') + this.listener(value as unknown as ShellKernelMessage) + return true + } + /** Called only once outer managed-range cleanup has finished, including its failure result. */ + unavailable(message: string, cleanupConfirmed: boolean): void { + if (this.terminal) return + const failure = { code: 'EXECUTOR_UNAVAILABLE', message } + this.terminal = { type: 'shell.unavailable', cleanupConfirmed, error: failure } + const requests = [...this.requests] + this.requests.clear(); this.controls.clear(); this.sessions.clear() + for (const read of this.reads.values()) read.reject(new Error(message)) + this.reads.clear() + for (const [requestId, request] of requests) { + if (request.completed) continue + const result: ShellCompleted = { type: 'shell.completed', requestId, status: request.sent ? 'failed' : 'rejected', exitCode: null, + effectState: request.sent ? 'unknown' : 'none', shellState: 'reset', cleanupConfirmed, + stdoutTotal: 0, stderrTotal: 0, error: failure } + this.listener?.(result) + } + this.listener?.(this.terminal) + } +} diff --git a/src/execution/resident/shell-launch.ts b/src/execution/resident/shell-launch.ts new file mode 100644 index 00000000..a0a4d310 --- /dev/null +++ b/src/execution/resident/shell-launch.ts @@ -0,0 +1,82 @@ +/** Prepare one exact public-confined launch; never replace its wrapper in native code. */ +import path from 'node:path' +import { createHash } from 'node:crypto' +import type { ConfinedArgv } from '../../adapter/dsh/types.js' +import { loadScrubbedParentEnv } from '../../adapter/dsh/host.js' +import type { ShellCall, ShellContext, ShellPolicy } from '../../shells/types.js' +import { shellEnvOverrides } from '../../shells/constants.js' +import { confinedBash } from '../../bash-ipc.js' +import { isNiubashExecutable, prepareNiubash, type NativeShellRuntime } from '../../niubash/runtime.js' +import { prepareWorkspace } from '../../workspace-preparation.js' +import type { ShellBootstrap, ShellLaunch } from './shell-protocol.js' +import { residentShellBootstrap } from './shell-bootstrap.js' + +export interface ShellLaunchOptions { + bash: { file: string; source: string } + /** Owned fixed driver arguments only; command text never enters this callback. */ + bootstrap?(endpoints: ShellBootstrap): string[] + runtime?: typeof prepareShellRuntime +} +/** Full explicit environment, including Windows case-insensitive tombstones. */ +export function mergeShellEnvironment(...layers: NodeJS.ProcessEnv[]): Record { + const output: Record = {} + for (const layer of layers) for (const [name, value] of Object.entries(layer)) { + if (process.platform === 'win32') for (const key of Object.keys(output)) if (key.toUpperCase() === name.toUpperCase()) delete output[key] + if (value === undefined) delete output[name] + else output[name] = value + } + // This native child must never inherit the outer host's private control descriptor. + for (const key of Object.keys(output)) if (key.toUpperCase() === 'DSH_SUBPROCESS_CONTROL') delete output[key] + return output +} +export async function shellEnvironment(ctx: ShellContext, call: ShellCall): Promise> { + const parent = await loadScrubbedParentEnv() + const env = mergeShellEnvironment(parent, shellEnvOverrides(parent, ctx.get('shellEnv')?.collect(call) ?? {})) + assertBootstrapEnvironment(env) + return env +} +/** Startup scripts can replace the reserved driver before its first receipt. */ +export function assertBootstrapEnvironment(env: Record): void { + const configured = Object.keys(env).filter(name => ['NIU_ENV', 'BASH_ENV', 'ENV'].includes(name.toUpperCase()) && env[name] !== '') + if (configured.length) throw new Error(`UNSUPPORTED_STARTUP_ENV: session execution cannot run ${configured.join(', ')} startup scripts. Remove those environment entries or use fresh execution.`) +} +/** Hash is kept private and never logged; changing host facts resets instead of overwriting exports. */ +export function environmentIdentity(env: Record): string { + return createHash('sha256').update(JSON.stringify(Object.entries(env).sort(([a], [b]) => a.localeCompare(b)))).digest('hex') +} +export async function prepareShellRuntime(bash: ShellLaunchOptions['bash'], env: NodeJS.ProcessEnv, + policy: ShellPolicy, signal: AbortSignal): Promise { + if (!path.isAbsolute(bash.file)) throw new Error('EXECUTOR_UNAVAILABLE: Bash requires an absolute executable.') + if (process.platform === 'win32' && isNiubashExecutable(bash.file)) { + return prepareNiubash(bash.file, env.PATH ?? env.Path, signal, { workspaceRoot: policy.workspaceRoot }) + } + const runtime = policy.mode === 'danger-full-access' ? { file: bash.file } : confinedBash(bash.file, { source: bash.source }) + return { file: runtime.file, env: runtime.env ?? {} } +} +export async function prepareShellLaunch(ctx: ShellContext, options: ShellLaunchOptions, endpoints: ShellBootstrap, + cwd: string, env: Record, policy: ShellPolicy, signal: AbortSignal, assertCurrent: () => void, +): Promise<{ launch: ShellLaunch; confinement?: ConfinedArgv }> { + assertCurrent() + const runtime = await (options.runtime ?? prepareShellRuntime)(options.bash, env, policy, signal) + assertCurrent() + const completeEnv = mergeShellEnvironment(env, runtime.env) + assertBootstrapEnvironment(completeEnv) + await prepareWorkspace(policy) + assertCurrent() + const argv = [runtime.file, ...(options.bootstrap ?? residentShellBootstrap)(endpoints)] + if (!path.isAbsolute(argv[0]) || argv.some(value => typeof value !== 'string' || value.includes('\0'))) { + throw new Error('EXECUTOR_UNAVAILABLE: invalid trusted shell bootstrap argv.') + } + let confinement: ConfinedArgv | undefined + if (policy.mode !== 'danger-full-access') { + const sandbox = ctx.get('sandbox') + if (!sandbox?.confine) throw new Error('SANDBOX_UNAVAILABLE: no confinement provider; no command executed.') + confinement = await sandbox.confine(argv, { ...policy, mode: policy.mode }, signal) + assertCurrent() + if (!confinement?.argv?.length || !['full', 'partial'].includes(confinement.enforcement) + || !path.isAbsolute(confinement.argv[0]) || confinement.argv.some(value => typeof value !== 'string' || value.includes('\0'))) { + throw new Error('SANDBOX_UNAVAILABLE: invalid confinement result; no command executed.') + } + } + return { launch: { argv: confinement?.argv ?? argv, cwd, env: completeEnv }, confinement } +} diff --git a/src/execution/resident/shell-output.ts b/src/execution/resident/shell-output.ts new file mode 100644 index 00000000..2d52bcd9 --- /dev/null +++ b/src/execution/resident/shell-output.ts @@ -0,0 +1,37 @@ +/** Copy terminal output before releasing the native record. Empty live tails are not EOF. */ +import type { ShellCompleted, ShellFailure, ShellStream, ShellTransport } from './shell-protocol.js' +export interface ShellCapture { text: string; bytesTotal: number; lossy: boolean; error?: ShellFailure } +export interface ShellOutputPair { stdout: ShellCapture; stderr: ShellCapture; error?: ShellFailure } +const RETAINED_BYTES = 128 * 1024 + +export async function collectShellOutput(transport: ShellTransport, completed: ShellCompleted): Promise { + async function collect(stream: ShellStream, total: number): Promise { + let offset = 0, bytes = Buffer.alloc(0), lossy = false + // Native retention is bounded; this also bounds malformed tiny-progress receipts. + for (let reads = 0; reads < 256; reads++) { + const value = await transport.read(completed.requestId, stream, offset, 65536) + const block = Buffer.from(value.bytesBase64, 'base64') + if (value.nextOffset > total || (!value.error && ((value.nextOffset === offset && !value.eof) + || (value.eof && value.nextOffset !== total)))) throw new Error('Shell output disagrees with its terminal receipt') + bytes = Buffer.concat([bytes, block]) + lossy ||= value.lossy || bytes.length > RETAINED_BYTES + if (bytes.length > RETAINED_BYTES) bytes = bytes.subarray(bytes.length - RETAINED_BYTES) + offset = value.nextOffset + if (value.error) return { text: bytes.toString('utf8'), bytesTotal: total, lossy: true, error: value.error } + if (value.eof) return { text: bytes.toString('utf8'), bytesTotal: total, lossy } + } + throw new Error('Shell output exceeded its bounded terminal read count') + } + const guarded = (stream: ShellStream, total: number) => collect(stream, total).catch(error => { + // A contradictory terminal receipt is a protocol failure. Stop the owned + // connection before joining the sibling reader, which might still be pending. + transport.abort(error instanceof Error ? error : new Error(String(error))) + throw error + }) + const results = await Promise.allSettled([guarded('stdout', completed.stdoutTotal), guarded('stderr', completed.stderrTotal)]) + const failed = results.find(result => result.status === 'rejected') + if (failed?.status === 'rejected') throw failed.reason + const stdout = (results[0] as PromiseFulfilledResult).value + const stderr = (results[1] as PromiseFulfilledResult).value + return { stdout, stderr, ...(stdout.error ?? stderr.error ? { error: stdout.error ?? stderr.error } : {}) } +} diff --git a/src/execution/resident/shell-owner.ts b/src/execution/resident/shell-owner.ts new file mode 100644 index 00000000..9ed3fc30 --- /dev/null +++ b/src/execution/resident/shell-owner.ts @@ -0,0 +1,23 @@ +/** Service-scoped activation for the internal shell adapter, before any tool publication. */ +import type { Context } from '../../adapter/dsh/types.js' +import { createShellAdmissionHost, type ShellAdmissionOptions } from './shell-admission.js' + +declare module '@deepseek-ai/cordis' { + interface Context { opsShellAdmission: ReturnType } +} + +/** + * Withdrawal/replacement of any launch authority provider disposes this owner + * and awaits every shell close before a replacement adapter can be exposed. + * The transport itself belongs to the already-ready, activation-owned kernel. + */ +export function residentShellAdmission(options: ShellAdmissionOptions) { + return { + name: 'dsh-ops-shell-admission', + inject: ['agents', 'sandboxPolicy', 'sandbox', 'shellEnv', 'workingDirectory'], + apply(ctx: Context): void { + if (ctx.get('opsShellAdmission', false)) throw new Error('A resident shell admission owner is already mounted in this scope.') + ctx.provide('opsShellAdmission', createShellAdmissionHost(ctx, options)) + }, + } +} diff --git a/src/execution/resident/shell-protocol.ts b/src/execution/resident/shell-protocol.ts new file mode 100644 index 00000000..31c1148b --- /dev/null +++ b/src/execution/resident/shell-protocol.ts @@ -0,0 +1,44 @@ +/** Internal shell wire contract. No tool is published until the native driver supports it. */ +export interface ShellBootstrap { commandPipe: string; receiptPipe: string } +export interface ShellLaunch { argv: string[]; cwd: string; env: Record } +export interface ShellFailure { code: string; message: string } +export type ShellStream = 'stdout' | 'stderr' +export interface ShellOutput { + type: 'shell.output'; requestId: string; stream: ShellStream + offset: number; nextOffset: number; bytesBase64: string; lossy: boolean; eof: boolean + error?: ShellFailure +} +export type ShellAdmission = { type: 'shell.admission'; requestId: string } & ( + | { decision: 'launch'; launch: ShellLaunch; closeAfter: boolean; commandCwd?: string } + | { decision: 'reuse'; commandCwd?: string } + | { decision: 'retire' } + | { decision: 'reject'; error: ShellFailure }) +export interface ShellCompleted { + type: 'shell.completed'; requestId: string + status: 'completed' | 'failed' | 'cancelled' | 'rejected' + exitCode: number | null; effectState: 'none' | 'unknown' + shellState: 'preserved' | 'reset'; cleanupConfirmed: boolean + stdoutTotal: number; stderrTotal: number; error?: ShellFailure +} +export type ShellHostMessage = ShellAdmission + | { type: 'shell.execute'; requestId: string; sessionKey: string; command: string; timeoutMs: number; stdoutBytes: number; stderrBytes: number } + | { type: 'shell.retire' | 'shell.close'; requestId: string; sessionKey: string; reason: string } + | { type: 'cancel'; requestId: string } + | { type: 'shell.read'; requestId: string; stream: ShellStream; offset: number; maxBytes: number } + | { type: 'shell.release'; requestId: string } +export type ShellKernelMessage = ShellCompleted | ShellOutput + | { type: 'shell.admit'; requestId: string; hasShell: boolean; bootstrap?: ShellBootstrap } + | { type: 'shell.started'; requestId: string } + | { type: 'shell.closed'; requestId: string; cleanupConfirmed: boolean; error?: ShellFailure } + // Transport emits this only after its owned kernel range has settled (or cleanup failed). + | { type: 'shell.unavailable'; cleanupConfirmed: boolean; error: ShellFailure } +export interface ShellTransport { + readonly available: boolean + readonly capabilities: readonly string[] + /** Synchronous bounded enqueue; a throw proves the frame was not accepted. */ + send(message: ShellHostMessage): void + read(requestId: string, stream: ShellStream, offset: number, maxBytes: number): Promise + /** Fail the shared connection; unavailable is delivered only after its owned range is joined. */ + abort(error: Error): void + subscribe(listener: (message: ShellKernelMessage) => void): () => void +} diff --git a/test/resident-shell-admission.test.mjs b/test/resident-shell-admission.test.mjs new file mode 100644 index 00000000..745a482c --- /dev/null +++ b/test/resident-shell-admission.test.mjs @@ -0,0 +1,928 @@ +/** Real host ownership and policy over a scripted kernel-owned shell queue. */ +import fs from 'node:fs' +import path from 'node:path' +import { inspect } from 'node:util' +import Agents from '@deepseek-ai/dsh-agent' +import Sessions from '@deepseek-ai/dsh-session' +import Projections from '@deepseek-ai/dsh-session-projection' +import Llm from '@deepseek-ai/dsh-llm' +import Loop from '@deepseek-ai/dsh-agent-loop' +import Policy, { setSandboxMode } from '@deepseek-ai/dsh-sandbox-policy' +import { createShellAdmissionHost } from '../lib/execution/resident/shell-admission.js' +import { residentShellAdmission } from '../lib/execution/resident/shell-owner.js' +import { assert, test, report, withTempDir } from './lib/harness.mjs' +import { bootHost } from './lib/host.mjs' + +let identity = 0 +const turn = () => new Promise(resolve => setImmediate(resolve)) +async function bounded(promise, label) { + let timer + try { + return await Promise.race([promise, new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${label} did not settle`)), 15_000) + })]) + } finally { clearTimeout(timer) } +} + +async function fixture(run, { mode = 'danger-full-access', providerOwned = false } = {}) { + await withTempDir('resident-shell-admission', async directory => { + const { ctx } = await bootHost({ withWorkingDirectory: false }) + const frames = [], listeners = new Set(), watchers = new Set(), handles = [] + const approvals = [], confinements = [], preparations = [], bootstraps = [], calls = [], logs = [], aborts = [], reads = [], environmentCalls = [] + const closed = new Set(), pending = new Map(), retained = new Map() + const state = { ensureCalls: 0, directoryError: undefined, approval: async () => 'allowed-once', + environment: {}, runtimeEnvironment: {}, cleanup: false, expectedCleanupFailure: false, sendFailure: undefined, onAbort: undefined, + outputRead: undefined, autoRetire: true } + ctx.logger.exporter({ levels: { default: 3 }, export: message => logs.push(inspect(message.args, { depth: null })) }) + let host, owner, unavailable = false + const emit = frame => { + if (frame.type === 'shell.unavailable') unavailable = true + for (const listener of [...listeners]) listener(frame) + } + const complete = (request, overrides = {}) => { + pending.delete(request.requestId) + const receipt = { type: 'shell.completed', requestId: request.requestId, status: 'completed', exitCode: 0, + effectState: 'unknown', shellState: 'preserved', cleanupConfirmed: true, + stdoutTotal: 0, stderrTotal: 0, ...overrides } + retained.set(request.requestId, receipt) + emit(receipt) + } + const acknowledge = frame => { + if (closed.has(frame.requestId)) return + closed.add(frame.requestId) + emit({ type: 'shell.closed', requestId: frame.requestId, cleanupConfirmed: true }) + if (frame.type === 'shell.close') for (const request of [...pending.values()]) { + if (request.sessionKey === frame.sessionKey) complete(request, { + status: 'cancelled', exitCode: null, effectState: 'none', shellState: 'reset', + }) + } + } + const transport = { + available: true, capabilities: ['shell-v1'], + send(frame) { + if (!transport.available) throw new Error('Fixture transport is unavailable') + if (state.sendFailure?.(frame)) throw new Error('Fixture control write rejected') + frames.push(structuredClone(frame)) + if (frame.type === 'shell.execute') pending.set(frame.requestId, frame) + if (frame.type === 'shell.release') assert.equal(retained.delete(frame.requestId), true, 'Only retained terminal output can be released') + for (const watcher of [...watchers]) watcher() + // A retirement closes old shell state but leaves this kernel's queue intact. + if (state.autoRetire && frame.type === 'shell.retire' || state.cleanup && frame.type === 'shell.close') { + queueMicrotask(() => acknowledge(frame)) + } + }, + async read(requestId, stream, offset, maxBytes) { + reads.push({ requestId, stream, offset, maxBytes }) + assert.equal(transport.available, true) + const receipt = retained.get(requestId) + assert.ok(receipt, 'Output reads follow a terminal receipt in this fixture') + if (state.outputRead) return state.outputRead({ requestId, stream, offset, maxBytes }, receipt) + assert.equal(receipt[stream === 'stdout' ? 'stdoutTotal' : 'stderrTotal'], 0) + return { type: 'shell.output', requestId, stream, offset, nextOffset: offset, bytesBase64: '', lossy: false, eof: true } + }, + abort(error) { aborts.push(error); transport.available = false; state.onAbort?.(error) }, + subscribe(listener) { listeners.add(listener); return () => listeners.delete(listener) }, + } + const message = async (type, after = 0, predicate = () => true) => { + const find = () => frames.slice(after).find(frame => frame.type === type && predicate(frame)) + if (find()) return find() + let check + try { + return await bounded(new Promise(resolve => { + check = () => { const frame = find(); if (frame) resolve(frame) } + watchers.add(check) + check() + }), `kernel message ${type}`) + } finally { watchers.delete(check) } + } + try { + for (const plugin of [Agents, Sessions, Projections, Llm]) await ctx.plugin(plugin, {}) + await ctx.plugin(Policy, { mode, workspaceRoot: directory }) + await ctx.plugin(inner => { + inner.provide('workingDirectory', { + get: session => session.header.cwd, + async ensure(agent) { + state.ensureCalls++ + if (state.directoryError) throw state.directoryError + return agent.session.header.cwd + }, + }) + inner.provide('approval', { async request(request) { + approvals.push(request) + return state.approval(request) + } }) + inner.provide('shellEnv', { collect: exec => { environmentCalls.push(exec); return { ...state.environment } } }) + inner.on('llm/stream', () => { throw new Error('This admission fixture must never request a model') }) + }) + const sandboxProvider = await ctx.plugin(inner => { + inner.provide('sandbox', { async confine(argv, policy, signal) { + signal?.throwIfAborted() + const wrapped = [process.execPath, 'fixture-public-sandbox', ...argv] + confinements.push({ argv: [...argv], wrapped, policy: structuredClone(policy) }) + return { argv: wrapped, enforcement: 'full' } + } }) + }) + await ctx.plugin(Loop, { agents: [] }) + const options = { + transport, bash: { file: process.execPath, source: 'fixture' }, + async runtime(...args) { preparations.push(args); return { file: process.execPath, env: { ...state.runtimeEnvironment } } }, + bootstrap(endpoints) { + bootstraps.push(structuredClone(endpoints)) + return ['--noprofile', '--norc', '-c', 'fixture-driver', endpoints.commandPipe, endpoints.receiptPipe] + }, + } + if (providerOwned) { + owner = await ctx.plugin(residentShellAdmission(options)) + host = ctx.get('opsShellAdmission') + assert.ok(host, 'The public owner helper publishes its ready admission service') + } else owner = await ctx.plugin({ name: 'fixture-shell-admission-owner', apply(inner) { + host = createShellAdmissionHost(inner, options) + } }) + const agent = async (cwd = directory) => { + const handle = await ctx.get('agents').create({ sessionId: `shell-admission-${++identity}`, + meta: { cwd }, agentOptions: { provider: 'fixture', model: 'unused' } }) + handles.push(handle) + return handle + } + const submit = async (handle, args = {}, signal = new AbortController().signal) => { + const start = frames.length + const result = host.execute({ command: 'printf fixture', ...args }, { + agent: handle.agent, callId: `shell-call-${++identity}`, signal, + }) + calls.push(result) + void result.catch(() => {}) + const request = await message('shell.execute', start) + return { request, result } + } + const admit = async (call, hasShell = false) => { + const start = frames.length + emit({ type: 'shell.admit', requestId: call.request.requestId, hasShell, + ...hasShell ? {} : { bootstrap: { + commandPipe: `\\\\.\\pipe\\fixture-command-${++identity}`, + receiptPipe: `\\\\.\\pipe\\fixture-receipt-${identity}`, + } } }) + return message('shell.admission', start, frame => frame.requestId === call.request.requestId) + } + const finish = async (call, overrides = {}) => { + emit({ type: 'shell.started', requestId: call.request.requestId }) + complete(call.request, overrides) + return bounded(call.result, 'shell result') + } + await run({ ctx, host, owner, directory, agent, submit, admit, finish, complete, emit, acknowledge, + message, frames, approvals, confinements, preparations, bootstraps, state, transport, logs, aborts, reads, sandboxProvider, environmentCalls }) + } finally { + state.cleanup = true + if (!transport.available && !unavailable) emit({ type: 'shell.unavailable', cleanupConfirmed: true, + error: { code: 'EXECUTOR_UNAVAILABLE', message: 'Fixture teardown joined the scripted range.' } }) + for (const frame of frames) if (frame.type === 'shell.close' || frame.type === 'shell.retire') acknowledge(frame) + const failures = [] + const cleanup = async (work, label) => { + try { await bounded(work, label) } catch (error) { failures.push(error) } + } + await cleanup(host?.dispose() ?? Promise.resolve(), 'host cleanup') + for (const handle of handles) await cleanup(handle.dispose(), 'Agent cleanup') + await cleanup(ctx.fiber.dispose(), 'Cordis cleanup') + await bounded(Promise.allSettled(calls), 'pending request cleanup') + // The unconfirmed-loss case asserts the cleanup rejection before reaching this teardown. + if (!state.expectedCleanupFailure && failures.length) throw new AggregateError(failures, 'Fixture cleanup failed') + } + }) +} + +await test('the kernel owns FIFO admission and reads the current policy when a queued head arrives', async () => { + await fixture(async f => { + const owner = await f.agent() + const first = await f.submit(owner, { command: 'printf first' }) + const second = await f.submit(owner, { command: 'printf second', timeoutMs: 2_147_483_647 }) + assert.equal(first.request.timeoutMs, 120_000) + assert.equal(second.request.timeoutMs, 2_147_483_647) + const beforeInvalid = f.frames.length + const invalid = await f.host.execute({ command: 'printf invalid-timeout', timeoutMs: 2_147_483_648 }, + { agent: owner.agent, callId: 'over-maximum-timeout' }) + assert.equal(invalid.status, 'rejected') + assert.equal(invalid.error.code, 'INVALID_ARGUMENT') + assert.equal(invalid.effectState, 'none') + assert.equal(f.frames.length, beforeInvalid) + assert.equal(first.request.sessionKey, second.request.sessionKey) + assert.notEqual(first.request.requestId, second.request.requestId) + assert.equal(f.frames.filter(frame => frame.type === 'shell.execute').length, 2) + assert.equal(f.frames.filter(frame => frame.type === 'shell.admission').length, 0) + assert.equal(f.preparations.length, 0) + assert.equal(f.confinements.length, 0) + setSandboxMode(owner.agent.session, 'workspace-write') + const admitted = await f.admit(first) + assert.equal(admitted.decision, 'launch') + assert.equal(f.confinements.at(-1).policy.mode, 'workspace-write') + assert.equal(admitted.launch.cwd, f.directory) + assert.equal(admitted.launch.env.NO_COLOR, '1') + assert.equal(admitted.launch.env.TERM, 'dumb') + assert.equal(admitted.launch.env.PAGER, 'cat') + assert.equal(admitted.launch.env.LC_MESSAGES, 'C') + assert.equal(admitted.launch.argv[0], process.execPath) + assert.equal(admitted.launch.argv.includes(first.request.command), false) + assert.deepEqual(admitted.launch.argv, f.confinements.at(-1).wrapped) + assert.deepEqual(f.confinements.at(-1).argv.slice(1), ['--noprofile', '--norc', '-c', 'fixture-driver', + f.bootstraps[0].commandPipe, f.bootstraps[0].receiptPipe]) + const result = await f.finish(first) + assert.equal(result.status, 'completed') + assert.deepEqual(result.stdout, { text: '', bytesTotal: 0, lossy: false }) + assert.deepEqual(result.stderr, { text: '', bytesTotal: 0, lossy: false }) + assert.equal(f.frames.some(frame => frame.type === 'shell.release' && frame.requestId === first.request.requestId), true) + assert.equal((await f.admit(second, true)).decision, 'reuse') + await f.finish(second) + assert.equal(f.preparations.length, 1) + }) +}) + +await test('a committed mode change retires old shell state while preserving the kernel queue', async () => { + await fixture(async f => { + const owner = await f.agent() + const first = await f.submit(owner) + assert.equal((await f.admit(first)).decision, 'launch') + await f.finish(first) + const asked = Promise.withResolvers(), grant = Promise.withResolvers() + f.state.approval = request => { asked.resolve(request); return grant.promise } + const head = await f.submit(owner, { sandbox_permissions: 'danger-full-access', justification: 'Pending head approval.' }) + const queued = await f.submit(owner) + f.emit({ type: 'shell.admit', requestId: head.request.requestId, hasShell: true }) + await bounded(asked.promise, 'approval at the active queue head') + const start = f.frames.length + setSandboxMode(owner.agent.session, 'read-only') + const retired = await f.message('shell.retire', start) + assert.equal(retired.sessionKey, queued.request.sessionKey) + grant.resolve('allowed-once') + await turn() + assert.equal(f.frames.slice(start).some(frame => frame.type === 'shell.admission' && frame.requestId === head.request.requestId), false) + assert.equal(f.frames.slice(start).some(frame => frame.type === 'cancel' && frame.requestId === queued.request.requestId), false) + assert.equal(f.aborts.length, 0) + f.complete(head.request, { status: 'cancelled', exitCode: null, effectState: 'none', shellState: 'reset', + error: { code: 'CANCELLED', message: 'Native retired the approval-pending head.' } }) + assert.equal((await head.result).effectState, 'none') + assert.equal(f.frames.filter(frame => frame.type === 'shell.execute').length, 3) + const admitted = await f.admit(queued) + assert.equal(admitted.decision, 'launch') + assert.equal(f.confinements.at(-1).policy.mode, 'read-only') + assert.equal((await f.finish(queued)).status, 'completed') + }, { mode: 'workspace-write' }) +}) + +await test('policy changes during delayed approval reject the queued command without effects', async () => { + await fixture(async f => { + const owner = await f.agent(), grant = Promise.withResolvers(), asked = Promise.withResolvers() + f.state.approval = request => { asked.resolve(request); return grant.promise } + const call = await f.submit(owner, { sandbox_permissions: 'danger-full-access', justification: 'Fixture escalation.' }) + assert.equal(f.approvals.length, 0, 'approval belongs to the kernel-admitted head') + const admission = f.admit(call) + const request = await bounded(asked.promise, 'approval request') + assert.equal(request.agent, owner.agent) + setSandboxMode(owner.agent.session, 'workspace-write') + grant.resolve('allowed-once') + const rejected = await admission + assert.equal(rejected.decision, 'reject') + assert.equal(rejected.error.code, 'AUTHORITY_CHANGED') + assert.equal(f.preparations.length, 0) + assert.equal(f.confinements.length, 0) + f.complete(call.request, { status: 'rejected', exitCode: null, effectState: 'none', shellState: 'reset', error: rejected.error }) + const value = await bounded(call.result, 'rejected command') + assert.equal(value.status, 'rejected') + assert.equal(value.effectState, 'none') + }, { mode: 'read-only' }) +}) + +await test('allowed-once widening closes its shell and is never inherited by the next call', async () => { + await fixture(async f => { + const owner = await f.agent() + const widened = await f.submit(owner, { sandbox_permissions: 'danger-full-access', justification: 'One command only.' }) + const launch = await f.admit(widened) + assert.equal(launch.decision, 'launch') + assert.equal(launch.closeAfter, true) + assert.equal(f.approvals.length, 1) + assert.equal(f.confinements.length, 0) + await f.finish(widened, { shellState: 'reset' }) + assert.equal(f.ctx.get('sandboxPolicy').resolve({ session: owner.agent.session }).mode, 'workspace-write') + const ordinary = await f.submit(owner) + assert.equal((await f.admit(ordinary)).decision, 'launch') + assert.equal(f.confinements.at(-1).policy.mode, 'workspace-write') + await f.finish(ordinary) + const nextGrant = await f.submit(owner, { sandbox_permissions: 'danger-full-access', justification: 'A separate command.' }) + const retired = await f.admit(nextGrant, true) + assert.equal(retired.decision, 'retire') + const separate = await f.admit(nextGrant) + assert.equal(separate.decision, 'launch') + assert.equal(separate.closeAfter, true) + assert.equal(f.approvals.length, 2) + await f.finish(nextGrant, { shellState: 'reset' }) + }, { mode: 'workspace-write' }) +}) + +await test('reuse omits cwd recovery and passes explicit cwd to the existing shell', async () => { + await fixture(async f => { + const owner = await f.agent(), child = path.join(f.directory, 'child') + fs.mkdirSync(child) + const first = await f.submit(owner) + assert.equal((await f.admit(first)).decision, 'launch') + await f.finish(first) + const ensures = f.state.ensureCalls + f.state.directoryError = new Error('Reuse must preserve the live shell directory without recovery') + const reused = await f.submit(owner) + const reuse = await f.admit(reused, true) + assert.equal(reuse.decision, 'reuse') + assert.equal(reuse.commandCwd, undefined) + assert.equal(f.state.ensureCalls, ensures) + await f.finish(reused) + f.state.directoryError = undefined + const explicit = await f.submit(owner, { workdir: child }) + const changed = await f.admit(explicit, true) + assert.equal(changed.decision, 'reuse') + assert.equal(changed.commandCwd, child) + await f.finish(explicit) + const following = await f.submit(owner) + const followingAdmission = await f.admit(following, true) + assert.equal(followingAdmission.decision, 'reuse') + assert.equal(followingAdmission.commandCwd, undefined) + await f.finish(following) + assert.equal(f.preparations.length, 1) + }) +}) + +await test('Agent disposal awaits its close receipt while a sibling continues on the same host', async () => { + await fixture(async f => { + const first = await f.agent(), sibling = await f.agent() + const firstCall = await f.submit(first), siblingCall = await f.submit(sibling) + assert.notEqual(firstCall.request.sessionKey, siblingCall.request.sessionKey) + assert.equal((await f.admit(firstCall)).decision, 'launch') + assert.equal((await f.admit(siblingCall)).decision, 'launch') + await f.finish(firstCall); await f.finish(siblingCall) + const start = f.frames.length + let disposed = false + const disposal = first.dispose().then(() => { disposed = true }) + const close = await f.message('shell.close', start, frame => frame.sessionKey === firstCall.request.sessionKey) + await turn() + assert.equal(disposed, false) + const another = await f.submit(sibling) + assert.equal((await f.admit(another, true)).decision, 'reuse') + assert.equal((await f.finish(another)).status, 'completed') + assert.equal(f.frames.slice(start).some(frame => frame.type === 'shell.close' && frame.sessionKey === siblingCall.request.sessionKey), false) + f.acknowledge(close) + await bounded(disposal, 'first Agent disposal') + assert.equal(disposed, true) + }) +}) + +await test('the owning Cordis plugin awaits kernel cleanup before disposal completes', async () => { + await fixture(async f => { + const owner = await f.agent(), call = await f.submit(owner) + assert.equal((await f.admit(call)).decision, 'launch') + await f.finish(call) + let disposed = false + const start = f.frames.length + const disposal = f.owner.dispose().then(() => { disposed = true }) + const close = await f.message('shell.close', start) + await turn() + assert.equal(disposed, false) + f.acknowledge(close) + await bounded(disposal, 'admission owner disposal') + assert.equal(disposed, true) + }) +}) + +await test('a pre-aborted call never enters the kernel queue', async () => { + await fixture(async f => { + const owner = await f.agent(), cancellation = new AbortController() + cancellation.abort(new DOMException('Fixture cancelled before enqueue', 'AbortError')) + const result = await f.host.execute({ command: 'printf forbidden' }, { + agent: owner.agent, signal: cancellation.signal, callId: 'pre-aborted', + }) + assert.equal(result.status, 'rejected') + assert.equal(result.effectState, 'none') + assert.equal(result.error.code, 'CANCELLED') + assert.equal(f.frames.length, 0) + assert.equal(f.preparations.length, 0) + }) +}) + +await test('a queued abort sends cancel and waits for the kernel completion receipt', async () => { + await fixture(async f => { + const owner = await f.agent(), cancellation = new AbortController() + const call = await f.submit(owner, {}, cancellation.signal) + let settled = false + void call.result.then(() => { settled = true }) + const start = f.frames.length + cancellation.abort(new DOMException('Fixture queued cancellation', 'AbortError')) + const cancelled = await f.message('cancel', start) + assert.equal(cancelled.requestId, call.request.requestId) + await turn() + assert.equal(settled, false) + assert.deepEqual(f.frames.slice(start), [{ type: 'cancel', requestId: call.request.requestId }]) + assert.equal(f.preparations.length, 0) + f.complete(call.request, { status: 'cancelled', exitCode: null, effectState: 'none', shellState: 'reset' }) + const result = await bounded(call.result, 'kernel cancellation receipt') + assert.equal(result.status, 'cancelled') + assert.equal(result.effectState, 'none') + assert.equal(result.cleanupConfirmed, true) + }) +}) + +await test('cancellation during approval ignores a grant arriving after kernel completion', async () => { + await fixture(async f => { + const owner = await f.agent(), cancellation = new AbortController() + const asked = Promise.withResolvers(), grant = Promise.withResolvers() + f.state.approval = request => { asked.resolve(request); return grant.promise } + const call = await f.submit(owner, { sandbox_permissions: 'danger-full-access', justification: 'Cancelled approval.' }, cancellation.signal) + f.emit({ type: 'shell.admit', requestId: call.request.requestId, hasShell: false, + bootstrap: { commandPipe: 'fixture-command', receiptPipe: 'fixture-receipt' } }) + await bounded(asked.promise, 'pending approval') + const start = f.frames.length + cancellation.abort(new DOMException('Fixture cancelled during approval', 'AbortError')) + assert.equal((await f.message('cancel', start)).requestId, call.request.requestId) + f.complete(call.request, { status: 'cancelled', exitCode: null, effectState: 'none', shellState: 'reset' }) + assert.equal((await bounded(call.result, 'cancelled approval result')).effectState, 'none') + grant.resolve('allowed-once') + await turn() + assert.equal(f.frames.some(frame => frame.type === 'shell.admission' && frame.requestId === call.request.requestId), false) + assert.equal(f.preparations.length, 0) + assert.equal(f.confinements.length, 0) + }, { mode: 'read-only' }) +}) + +await test('an environment baseline change retires old state without exposing its values in diagnostics', async () => { + await fixture(async f => { + const owner = await f.agent(), before = 'fixture-env-before-private', after = 'fixture-env-after-private' + f.state.environment = { RESIDENT_ADMISSION_FIXTURE_VALUE: before } + const first = await f.submit(owner) + const initial = await f.admit(first) + assert.equal(initial.launch.env.RESIDENT_ADMISSION_FIXTURE_VALUE, before) + await f.finish(first) + const next = await f.submit(owner) + f.state.environment = { RESIDENT_ADMISSION_FIXTURE_VALUE: after } + const retired = await f.admit(next, true) + assert.deepEqual(retired, { type: 'shell.admission', requestId: next.request.requestId, decision: 'retire' }) + const fresh = await f.admit(next) + assert.equal(fresh.decision, 'launch') + assert.equal(fresh.launch.env.RESIDENT_ADMISSION_FIXTURE_VALUE, after) + await f.finish(next) + assert.equal(f.logs.some(log => log.includes(before) || log.includes(after)), false, 'Environment values must not appear in logs') + }) +}) + +for (const cleanupConfirmed of [true, false]) { + await test(`transport loss preserves ${cleanupConfirmed ? 'confirmed' : 'unconfirmed'} cleanup and never sends to the dead kernel`, async () => { + await fixture(async f => { + const owner = await f.agent(), call = await f.submit(owner) + assert.equal((await f.admit(call)).decision, 'launch') + f.emit({ type: 'shell.started', requestId: call.request.requestId }) + f.transport.available = false + const start = f.frames.length + f.emit({ type: 'shell.unavailable', cleanupConfirmed, + error: { code: 'EXECUTOR_UNAVAILABLE', message: 'Fixture kernel transport lost; cleanup observation retained.' } }) + const result = await bounded(call.result, 'lost transport result') + assert.equal(result.status, 'failed') + assert.equal(result.error.code, 'EXECUTOR_UNAVAILABLE') + assert.equal(result.effectState, 'unknown') + assert.equal(result.cleanupConfirmed, cleanupConfirmed) + if (cleanupConfirmed) await bounded(f.host.dispose(), 'confirmed transport-loss cleanup') + else { + f.state.expectedCleanupFailure = true + await assert.rejects(bounded(f.host.dispose(), 'unconfirmed transport-loss cleanup'), /Fixture kernel transport lost/) + } + assert.deepEqual(f.frames.slice(start), []) + const refused = await f.host.execute({ command: 'printf forbidden' }, { agent: owner.agent, callId: 'after-loss' }) + assert.equal(refused.effectState, 'none') + assert.equal(refused.status, 'rejected') + assert.deepEqual(f.frames.slice(start), []) + }) + }) +} + +await test('a failed admission reply waits for transport cleanup instead of abandoning the kernel head', async () => { + await fixture(async f => { + const owner = await f.agent(), aborted = Promise.withResolvers() + const call = await f.submit(owner) + let settled = false + void call.result.then(() => { settled = true }) + f.state.sendFailure = frame => frame.type === 'shell.admission' + f.state.onAbort = error => aborted.resolve(error) + f.emit({ type: 'shell.admit', requestId: call.request.requestId, hasShell: false, + bootstrap: { commandPipe: 'fixture-command', receiptPipe: 'fixture-receipt' } }) + assert.match((await bounded(aborted.promise, 'admission transport abort')).message, /Fixture control write rejected/) + await turn() + assert.equal(settled, false) + assert.equal(f.aborts.length, 1) + assert.equal(f.frames.some(frame => frame.type === 'shell.admission'), false) + f.emit({ type: 'shell.unavailable', cleanupConfirmed: true, + error: { code: 'EXECUTOR_UNAVAILABLE', message: 'Fixture transport failure joined its range.' } }) + const result = await bounded(call.result, 'failed admission cleanup receipt') + assert.equal(result.status, 'failed') + assert.equal(result.effectState, 'unknown') + assert.equal(result.cleanupConfirmed, true) + }) +}) + +await test('disposal joins a delayed sibling close even when another owner cleanup fails', async () => { + await fixture(async f => { + const first = await f.agent(), sibling = await f.agent() + const a = await f.submit(first), b = await f.submit(sibling) + await f.admit(a); await f.finish(a) + await f.admit(b); await f.finish(b) + const start = f.frames.length + let settled = false + const disposal = f.host.dispose() + void disposal.then(() => { settled = true }, () => { settled = true }) + const firstClose = await f.message('shell.close', start, frame => frame.sessionKey === a.request.sessionKey) + const siblingClose = await f.message('shell.close', start, frame => frame.sessionKey === b.request.sessionKey) + f.state.expectedCleanupFailure = true + f.emit({ type: 'shell.closed', requestId: firstClose.requestId, cleanupConfirmed: false, + error: { code: 'EXECUTOR_UNAVAILABLE', message: 'Fixture first owner cleanup failed.' } }) + await turn() + assert.equal(settled, false) + f.acknowledge(siblingClose) + await assert.rejects(bounded(disposal, 'all owner cleanup'), /Fixture first owner cleanup failed/) + assert.equal(settled, true) + }) +}) + +await test('32 idle owners consume the owner budget and disposal joins every close', async () => { + await fixture(async f => { + for (let index = 0; index < 32; index++) { + const owner = await f.agent(), call = await f.submit(owner) + assert.equal((await f.admit(call)).decision, 'launch') + await f.finish(call) + } + const extra = await f.agent(), start = f.frames.length + const refused = await f.host.execute({ command: 'printf overflow' }, { agent: extra.agent, callId: 'idle-owner-overflow' }) + assert.equal(refused.status, 'rejected') + assert.equal(refused.error.code, 'EXECUTOR_BUSY') + assert.equal(refused.effectState, 'none') + assert.equal(f.frames.length, start, 'The 33rd owner never enters the kernel') + let disposed = false + const disposal = f.host.dispose().then(() => { disposed = true }) + await turn() + const closes = f.frames.slice(start).filter(frame => frame.type === 'shell.close') + assert.equal(closes.length, 32) + assert.equal(new Set(closes.map(frame => frame.sessionKey)).size, 32) + for (const close of closes.slice(0, -1)) f.acknowledge(close) + await turn() + assert.equal(disposed, false) + f.acknowledge(closes.at(-1)) + await bounded(disposal, 'all 32 idle owners') + assert.equal(disposed, true) + }) +}) + +await test('terminal collection copies both streams before release and before the result settles', async () => { + await fixture(async f => { + const owner = await f.agent(), call = await f.submit(owner) + await f.admit(call) + const stdout = Buffer.from('out snowman ☃\n'), stderr = Buffer.from('stderr: 不同\n') + const split = stdout.indexOf(Buffer.from('☃')) + 1 + const outGate = Promise.withResolvers(), errGate = Promise.withResolvers() + const outAsked = Promise.withResolvers(), errAsked = Promise.withResolvers() + const reply = (request, bytes, eof) => ({ type: 'shell.output', requestId: request.requestId, stream: request.stream, + offset: request.offset, nextOffset: request.offset + bytes.length, bytesBase64: bytes.toString('base64'), lossy: false, eof }) + f.state.outputRead = async request => { + if (request.stream === 'stderr') { errAsked.resolve(); await errGate.promise; return reply(request, stderr, true) } + if (request.offset === 0) return reply(request, stdout.subarray(0, split), false) + assert.equal(request.offset, split) + outAsked.resolve(); await outGate.promise + return reply(request, stdout.subarray(split), true) + } + let settled = false + const completion = f.finish(call, { stdoutTotal: stdout.length, stderrTotal: stderr.length }) + void completion.then(() => { settled = true }) + await bounded(Promise.all([outAsked.promise, errAsked.promise]), 'both output streams') + assert.equal(settled, false) + assert.equal(f.frames.some(frame => frame.type === 'shell.release'), false) + outGate.resolve(); await turn() + assert.equal(settled, false, 'The stderr copy still owns the terminal record') + assert.equal(f.frames.some(frame => frame.type === 'shell.release'), false) + errGate.resolve() + const result = await completion + assert.deepEqual(result.stdout, { text: stdout.toString('utf8'), bytesTotal: stdout.length, lossy: false }) + assert.deepEqual(result.stderr, { text: stderr.toString('utf8'), bytesTotal: stderr.length, lossy: false }) + assert.equal(f.frames.filter(frame => frame.type === 'shell.release' && frame.requestId === call.request.requestId).length, 1) + assert.equal(f.reads.filter(read => read.stream === 'stdout').length, 2) + assert.equal(f.reads.filter(read => read.stream === 'stderr').length, 1) + }) +}) + +for (const [label, total, bytes] of [['premature EOF', 3, Buffer.from('a')], ['cursor beyond total', 1, Buffer.from('ab')]]) { + await test(`contradictory terminal output ${label} aborts and waits for joined cleanup`, async () => { + await fixture(async f => { + const owner = await f.agent(), call = await f.submit(owner), aborted = Promise.withResolvers() + await f.admit(call) + f.state.onAbort = error => aborted.resolve(error) + f.state.outputRead = request => { + const data = request.stream === 'stdout' ? bytes : Buffer.alloc(0) + return { type: 'shell.output', requestId: request.requestId, stream: request.stream, offset: request.offset, + nextOffset: request.offset + data.length, bytesBase64: data.toString('base64'), lossy: false, eof: true } + } + let settled = false + const completion = f.finish(call, { stdoutTotal: total }) + void completion.then(() => { settled = true }) + assert.match((await bounded(aborted.promise, 'contradictory terminal abort')).message, /terminal receipt/) + await turn() + assert.equal(settled, false) + assert.equal(f.frames.some(frame => frame.type === 'shell.release'), false) + f.emit({ type: 'shell.unavailable', cleanupConfirmed: true, + error: { code: 'EXECUTOR_UNAVAILABLE', message: 'Contradictory output transport joined its range.' } }) + const result = await completion + assert.equal(result.status, 'failed') + assert.equal(result.effectState, 'unknown') + assert.equal(result.cleanupConfirmed, true) + assert.equal(result.stdout.lossy, true) + assert.equal(f.frames.some(frame => frame.type === 'shell.release'), false) + }) + }) +} + +await test('withdrawing the real Cordis sandbox provider awaits dependent shell cleanup', async () => { + await fixture(async f => { + const owner = await f.agent(), call = await f.submit(owner) + assert.equal((await f.admit(call)).decision, 'launch') + await f.finish(call) + const start = f.frames.length + let disposed = false + const withdrawal = f.sandboxProvider.dispose().then(() => { disposed = true }) + const close = await f.message('shell.close', start, frame => frame.sessionKey === call.request.sessionKey) + await turn() + assert.equal(disposed, false, 'Authority provider disposal must join the dependent shell owner') + assert.equal(f.ctx.get('opsShellAdmission'), undefined, 'The withdrawing owner cannot remain publicly available') + const retained = await f.host.execute({ command: 'printf forbidden' }, { agent: owner.agent, callId: 'withdrawn-provider' }) + assert.equal(retained.status, 'rejected') + assert.equal(retained.effectState, 'none') + assert.equal(f.frames.slice(start).some(frame => frame.type === 'shell.execute'), false) + f.acknowledge(close) + await bounded(withdrawal, 'sandbox provider withdrawal') + assert.equal(disposed, true) + assert.equal(f.ctx.get('sandbox'), undefined) + assert.equal(f.ctx.get('opsShellAdmission'), undefined) + }, { providerOwned: true }) +}) + +await test('a capture error retains bytes and joins the other stream without aborting sibling sessions', async () => { + for (const nativeError of [undefined, { code: 'COMMAND_FAILED', message: 'The command failed before capture ended.' }]) { + await fixture(async f => { + const owner = await f.agent(), sibling = await f.agent(), call = await f.submit(owner) + await f.admit(call) + const captureError = { code: 'OUTPUT_FAILED', message: 'Fixture output capture became unavailable.' } + const stderr = Buffer.from('distinct stderr\n'), stderrGate = Promise.withResolvers(), stderrAsked = Promise.withResolvers() + f.state.outputRead = async request => { + let bytes = Buffer.alloc(0), error, eof = true + if (request.requestId === call.request.requestId) { + if (request.stream === 'stdout') { bytes = Buffer.from('part'); error = captureError; eof = false } + else { stderrAsked.resolve(); await stderrGate.promise; bytes = stderr } + } + return { type: 'shell.output', requestId: request.requestId, stream: request.stream, offset: request.offset, + nextOffset: request.offset + bytes.length, bytesBase64: bytes.toString('base64'), lossy: false, eof, ...(error ? { error } : {}) } + } + let settled = false + const completion = f.finish(call, { stdoutTotal: 10, stderrTotal: stderr.length, + ...(nativeError ? { status: 'failed', exitCode: 42, error: nativeError } : {}) }) + void completion.then(() => { settled = true }) + await bounded(stderrAsked.promise, 'capture sibling stream') + await turn() + assert.equal(settled, false) + assert.equal(f.frames.some(frame => frame.type === 'shell.release' && frame.requestId === call.request.requestId), false) + assert.equal(f.aborts.length, 0) + const other = await f.submit(sibling) + assert.equal((await f.admit(other)).decision, 'launch') + assert.equal((await f.finish(other)).status, 'completed') + stderrGate.resolve() + const result = await completion + assert.equal(result.status, 'failed') + assert.deepEqual(result.error, nativeError ?? captureError) + assert.deepEqual(result.stdout, { text: 'part', bytesTotal: 10, lossy: true, error: captureError }) + assert.deepEqual(result.stderr, { text: stderr.toString(), bytesTotal: stderr.length, lossy: false }) + assert.equal(f.frames.filter(frame => frame.type === 'shell.release' && frame.requestId === call.request.requestId).length, 1) + assert.equal(f.aborts.length, 0) + assert.equal(f.transport.available, true) + }) + } +}) + +await test('an older reset completion cannot erase the next shell while its output drain is pending', async () => { + await fixture(async f => { + const owner = await f.agent(), first = await f.submit(owner) + assert.equal((await f.admit(first)).decision, 'launch') + const tails = Promise.withResolvers(), readersStarted = Promise.withResolvers() + let readers = 0, firstSettled = false + f.state.outputRead = async request => { + if (request.requestId === first.request.requestId) { + if (++readers === 2) readersStarted.resolve() + await tails.promise + } + return { type: 'shell.output', requestId: request.requestId, stream: request.stream, + offset: request.offset, nextOffset: request.offset, bytesBase64: '', lossy: false, eof: true } + } + const firstCompletion = f.finish(first, { shellState: 'reset' }) + void firstCompletion.then(() => { firstSettled = true }) + await bounded(readersStarted.promise, 'older reset output readers') + assert.equal(firstSettled, false) + const second = await f.submit(owner) + assert.equal((await f.admit(second)).decision, 'launch') + assert.equal((await f.finish(second)).shellState, 'preserved') + assert.equal(firstSettled, false) + tails.resolve() + assert.equal((await firstCompletion).shellState, 'reset') + const third = await f.submit(owner) + assert.equal((await f.admit(third, true)).decision, 'reuse', 'The old output drain must not erase the newer launch facts') + await f.finish(third) + assert.equal(f.preparations.length, 2) + }) +}) + +await test('a locally refused enqueue preserves the existing shell launch facts', async () => { + await fixture(async f => { + const owner = await f.agent(), first = await f.submit(owner) + assert.equal((await f.admit(first)).decision, 'launch') + await f.finish(first) + const start = f.frames.length + f.state.sendFailure = frame => frame.type === 'shell.execute' + const refused = await f.host.execute({ command: 'printf not-enqueued' }, { agent: owner.agent, callId: 'local-enqueue-refusal' }) + assert.equal(refused.status, 'rejected') + assert.equal(refused.effectState, 'none') + assert.equal(f.frames.length, start) + assert.equal(f.aborts.length, 0) + f.state.sendFailure = undefined + const next = await f.submit(owner) + assert.equal((await f.admit(next, true)).decision, 'reuse') + await f.finish(next) + assert.equal(f.preparations.length, 1) + }) +}) + +await test('native cancellation suppresses a late approval while terminal output still drains', async () => { + await fixture(async f => { + const owner = await f.agent(), asked = Promise.withResolvers(), grant = Promise.withResolvers() + const tails = Promise.withResolvers(), readersStarted = Promise.withResolvers() + let readers = 0, settled = false + f.state.approval = request => { asked.resolve(request); return grant.promise } + const call = await f.submit(owner, { sandbox_permissions: 'danger-full-access', justification: 'Native-cancelled approval.' }) + void call.result.then(() => { settled = true }) + f.emit({ type: 'shell.admit', requestId: call.request.requestId, hasShell: false, + bootstrap: { commandPipe: 'fixture-command', receiptPipe: 'fixture-receipt' } }) + await bounded(asked.promise, 'approval before native cancellation') + f.state.outputRead = async request => { + if (++readers === 2) readersStarted.resolve() + await tails.promise + return { type: 'shell.output', requestId: request.requestId, stream: request.stream, + offset: request.offset, nextOffset: request.offset, bytesBase64: '', lossy: false, eof: true } + } + const start = f.frames.length + f.complete(call.request, { status: 'cancelled', exitCode: null, effectState: 'none', shellState: 'reset', + error: { code: 'CANCELLED', message: 'Native queue retired the head before execution.' } }) + await bounded(readersStarted.promise, 'native-cancelled output readers') + grant.resolve('allowed-once') + await turn() + assert.equal(settled, false) + assert.equal(f.frames.slice(start).some(frame => frame.type === 'shell.admission' || frame.type === 'cancel'), false) + assert.equal(f.aborts.length, 0) + assert.equal(f.preparations.length, 0) + assert.equal(f.confinements.length, 0) + tails.resolve() + const result = await bounded(call.result, 'native-cancelled final output') + assert.equal(result.status, 'cancelled') + assert.equal(result.effectState, 'none') + assert.equal(f.frames.filter(frame => frame.type === 'shell.release' && frame.requestId === call.request.requestId).length, 1) + assert.equal(f.aborts.length, 0) + }, { mode: 'read-only' }) +}) + +await test('queued admission snapshots the original Agent and call identity before caller mutation', async () => { + await fixture(async f => { + const siblingDirectory = path.join(f.directory, 'other-session') + fs.mkdirSync(siblingDirectory) + const original = await f.agent(), sibling = await f.agent(siblingDirectory) + const shared = { agent: original.agent, callId: 'original-call-id', signal: new AbortController().signal } + const start = f.frames.length + const result = f.host.execute({ command: 'printf original-owner', sandbox_permissions: 'danger-full-access', + justification: 'Approve this original session only.' }, shared) + const request = await f.message('shell.execute', start) + shared.agent = sibling.agent + shared.callId = 'mutated-call-id' + const admitted = await f.admit({ request, result }) + assert.equal(admitted.decision, 'launch') + assert.equal(admitted.closeAfter, true) + assert.equal(f.approvals.length, 1) + assert.equal(f.approvals[0].agent, original.agent) + assert.equal(f.approvals[0].callId, 'original-call-id') + assert.equal(f.environmentCalls.length, 1) + assert.equal(f.environmentCalls[0].agent, original.agent) + assert.equal(f.environmentCalls[0].callId, 'original-call-id') + assert.equal(admitted.launch.cwd, original.agent.session.header.cwd) + assert.notEqual(admitted.launch.cwd, sibling.agent.session.header.cwd) + assert.equal((await f.finish({ request, result }, { shellState: 'reset' })).status, 'completed') + }, { mode: 'read-only' }) +}) + +await test('caller cancellation suppresses approval replies before the native terminal receipt', async () => { + await fixture(async f => { + const owner = await f.agent(), cancellation = new AbortController() + const asked = Promise.withResolvers(), grant = Promise.withResolvers() + f.state.approval = request => { asked.resolve(request); return grant.promise } + const call = await f.submit(owner, { sandbox_permissions: 'danger-full-access', justification: 'Cancelled queue head.' }, cancellation.signal) + f.emit({ type: 'shell.admit', requestId: call.request.requestId, hasShell: false, + bootstrap: { commandPipe: 'fixture-command', receiptPipe: 'fixture-receipt' } }) + await bounded(asked.promise, 'approval before caller cancellation') + let settled = false + void call.result.then(() => { settled = true }) + const start = f.frames.length + cancellation.abort(new DOMException('Fixture caller cancellation', 'AbortError')) + assert.equal((await f.message('cancel', start)).requestId, call.request.requestId) + grant.resolve('allowed-once') + await turn() + assert.equal(settled, false) + assert.equal(f.frames.slice(start).some(frame => frame.type === 'shell.admission'), false) + assert.equal(f.aborts.length, 0) + assert.equal(f.preparations.length, 0) + f.complete(call.request, { status: 'cancelled', exitCode: null, effectState: 'none', shellState: 'reset', + error: { code: 'CANCELLED', message: 'Native joined the cancelled head.' } }) + const result = await bounded(call.result, 'caller cancellation terminal') + assert.equal(result.status, 'cancelled') + assert.equal(result.effectState, 'none') + assert.equal(f.aborts.length, 0) + }, { mode: 'read-only' }) +}) + +await test('failed retirement closes only that owner and refuses reuse until cleanup is joined', async () => { + await fixture(async f => { + const owner = await f.agent(), sibling = await f.agent() + const first = await f.submit(owner), other = await f.submit(sibling) + await f.admit(first); await f.finish(first) + await f.admit(other); await f.finish(other) + f.state.autoRetire = false + const start = f.frames.length + setSandboxMode(owner.agent.session, 'read-only') + const retire = await f.message('shell.retire', start) + f.emit({ type: 'shell.closed', requestId: retire.requestId, cleanupConfirmed: false, + error: { code: 'EXECUTOR_UNAVAILABLE', message: 'Fixture retirement cleanup failed.' } }) + const close = await f.message('shell.close', start, frame => frame.sessionKey === first.request.sessionKey) + const beforeRefusal = f.frames.length + const refused = await f.host.execute({ command: 'printf forbidden-reuse' }, { agent: owner.agent, callId: 'failed-retirement' }) + assert.equal(refused.status, 'rejected') + assert.equal(refused.effectState, 'none') + assert.equal(f.frames.length, beforeRefusal) + const next = await f.submit(sibling) + assert.equal((await f.admit(next, true)).decision, 'reuse') + assert.equal((await f.finish(next)).status, 'completed') + assert.equal(f.frames.slice(start).some(frame => frame.type === 'shell.close' && frame.sessionKey === other.request.sessionKey), false) + assert.equal(f.aborts.length, 0) + f.acknowledge(close) + await bounded(owner.dispose(), 'failed retirement owner close') + }) +}) + +await test('startup script variables reject before approval or runtime without disclosing their values', async () => { + await fixture(async f => { + const owner = await f.agent(), value = 'fixture-private-startup-script' + for (const name of ['NIU_ENV', 'nIu_eNv', 'BASH_ENV', 'bAsH_eNv', 'ENV', 'eNv']) { + f.state.environment = { [name]: value } + const call = await f.submit(owner, { sandbox_permissions: 'danger-full-access', justification: 'Must never request approval.' }) + const admission = await f.admit(call) + assert.equal(admission.decision, 'reject') + assert.equal(admission.error.code, 'UNSUPPORTED_STARTUP_ENV') + assert.equal(admission.error.message.includes(name), true) + assert.equal(admission.error.message.includes(value), false) + assert.equal(f.approvals.length, 0) + assert.equal(f.preparations.length, 0) + assert.equal(f.confinements.length, 0) + assert.equal(f.bootstraps.length, 0) + f.complete(call.request, { status: 'rejected', exitCode: null, effectState: 'none', shellState: 'reset', error: admission.error }) + const result = await bounded(call.result, 'startup environment refusal') + assert.equal(result.status, 'rejected') + assert.equal(result.effectState, 'none') + assert.equal(result.error.code, 'UNSUPPORTED_STARTUP_ENV') + } + assert.equal(JSON.stringify(f.frames).includes(value), false) + assert.equal(f.logs.join('\n').includes(value), false) + f.state.environment = { NIU_ENV: '', BASH_ENV: '', ENV: '' } + const ordinary = await f.submit(owner) + assert.equal((await f.admit(ordinary)).decision, 'launch') + assert.equal((await f.finish(ordinary)).status, 'completed') + }, { mode: 'read-only' }) +}) + +await test('runtime environment merge cannot add a startup script before confinement or bootstrap', async () => { + await fixture(async f => { + const owner = await f.agent(), value = 'fixture-private-runtime-startup' + f.state.environment = { NIU_ENV: '', BASH_ENV: '', ENV: '' } + f.state.runtimeEnvironment = { nIu_eNv: value, bAsH_eNv: value, eNv: value } + const call = await f.submit(owner) + const admission = await f.admit(call) + assert.equal(admission.decision, 'reject') + assert.equal(admission.error.code, 'UNSUPPORTED_STARTUP_ENV') + for (const name of Object.keys(f.state.runtimeEnvironment)) assert.equal(admission.error.message.includes(name), true) + assert.equal(admission.error.message.includes(value), false) + assert.equal(f.preparations.length, 1) + assert.equal(f.confinements.length, 0) + assert.equal(f.bootstraps.length, 0) + f.complete(call.request, { status: 'rejected', exitCode: null, effectState: 'none', shellState: 'reset', error: admission.error }) + const result = await bounded(call.result, 'runtime startup environment refusal') + assert.equal(result.effectState, 'none') + assert.equal(result.error.code, 'UNSUPPORTED_STARTUP_ENV') + assert.equal(JSON.stringify(f.frames).includes(value), false) + assert.equal(f.logs.join('\n').includes(value), false) + }, { mode: 'read-only' }) +}) + +report('resident-shell-admission') diff --git a/test/resident-shell-channel.test.mjs b/test/resident-shell-channel.test.mjs new file mode 100644 index 00000000..a5283a5e --- /dev/null +++ b/test/resident-shell-channel.test.mjs @@ -0,0 +1,472 @@ +/** Real resident client multiplexing over a scripted public fd7 duplex handle. */ +import { Duplex, PassThrough, Writable } from 'node:stream' +import { Context } from '@deepseek-ai/cordis' +import { startResident } from '../lib/execution/resident/client.js' +import { workerEnvelope } from '../lib/execution/protocol.js' +import { assert, report, test } from './lib/harness.mjs' + +const turn = () => new Promise(resolve => setImmediate(resolve)) +const exchange = { exitCode: 0, interrupted: false, readers: true, lossy: false, stderr: '', value: { status: 'completed' } } +const file = requestId => ({ ...workerEnvelope('file', { tool: 'stat', args: {}, read_only: true }), requestId }) +const execute = requestId => ({ type: 'shell.execute', requestId, sessionKey: 'fixture-session', command: 'printf fixture', + timeoutMs: 1000, stdoutBytes: 1024, stderrBytes: 1024 }) +const completed = (requestId, overrides = {}) => ({ type: 'shell.completed', requestId, status: 'completed', exitCode: 0, + effectState: 'unknown', shellState: 'preserved', cleanupConfirmed: true, stdoutTotal: 0, stderrTotal: 0, ...overrides }) +const output = (requestId, stream, offset, bytes, overrides = {}) => ({ type: 'shell.output', requestId, stream, + offset, nextOffset: offset + bytes.length, bytesBase64: bytes.toString('base64'), lossy: false, eof: false, ...overrides }) +async function bounded(work, label) { + let timer + try { + return await Promise.race([work, new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${label} did not settle`)), 5000) + })]) + } finally { clearTimeout(timer) } +} + +async function fixture({ capabilities = ['shell-v1'], omitCapabilities = false, ready = true, holdWrites = false, holdExit = false } = {}) { + const ctx = new Context(); ctx.logger.level = 0 + const input = new PassThrough(), frames = [], writes = [], events = [], watchers = new Set() + const ended = Promise.withResolvers(), empty = Promise.withResolvers() + let client, owner, spec, spawns = 0, terminated = 0, waits = 0, finished = false, holdAll = false + const send = frame => input.write(`${JSON.stringify(frame)}\n`) + const finish = () => { + if (!finished) { finished = true; ended.resolve({ exitCode: 0, signal: null }); control.destroy() } + if (!holdExit) empty.resolve(true) + } + const writable = new Writable({ write(bytes, _encoding, callback) { + const frame = JSON.parse(bytes.toString('utf8')) + frames.push(frame) + if (holdAll || holdWrites && (frame.type === 'execute' || frame.type === 'shell.execute')) writes.push(callback) + else callback() + if (frame.type === 'shutdown') queueMicrotask(finish) + } }) + const control = Duplex.from({ readable: input, writable }) + const handle = { + control, done: ended.promise, + collected: { stdout: { readFrom: () => ({ text: '', lossy: false }) }, stderr: { readFrom: () => ({ text: '', lossy: false }) } }, + terminate() { terminated++; finish() }, + waitForExit(signal) { + waits++ + if (!signal) return empty.promise + return Promise.race([empty.promise, new Promise(resolve => { + if (signal.aborted) resolve(false) + else signal.addEventListener('abort', () => resolve(false), { once: true }) + })]) + }, + } + await ctx.plugin(inner => inner.provide('subprocess', { spawn(value) { + spawns++; spec = value + if (ready) queueMicrotask(() => send({ type: 'ready', version: 1, pid: 123, + ...omitCapabilities ? {} : { capabilities } })) + return handle + } })) + const starting = ctx.plugin({ name: 'fixture-resident-shell-channel', inject: ['subprocess'], async apply(inner) { + client = await startResident(inner, '/fixture-worker', '/fixture-workspace') + client.shell.subscribe(event => { events.push(event); for (const watcher of [...watchers]) watcher() }) + } }).then(value => { owner = value; return client }) + void starting.catch(() => {}) + const event = async (type, requestId) => { + const find = () => events.find(value => value.type === type && (requestId === undefined || value.requestId === requestId)) + if (find()) return find() + let check + try { + return await bounded(new Promise(resolve => { + check = () => { const found = find(); if (found) resolve(found) } + watchers.add(check); check() + }), `shell event ${type} ${requestId ?? ''}`) + } finally { watchers.delete(check) } + } + return { starting, send, input, frames, writes, events, event, handle, finish, + holdAllWrites: () => { holdAll = true }, + releaseWrites: () => { holdAll = false; holdWrites = false; while (writes.length) writes.shift()() }, + releaseExit: value => empty.resolve(value ?? true), + get spec() { return spec }, get spawns() { return spawns }, get terminated() { return terminated }, get waits() { return waits }, + async dispose() { + empty.resolve(true) + holdAll = false; holdWrites = false + while (writes.length) writes.shift()() + if (owner) await bounded(owner.dispose(), 'resident owner disposal') + else { finish(); await starting.catch(() => {}) } + await bounded(ctx.fiber.dispose(), 'Cordis disposal') + }, + } +} + +await test('optional shell negotiation preserves file-only kernels and uses the existing control pipe', async () => { + for (const options of [{ omitCapabilities: true }, { capabilities: [] }, { capabilities: ['future-capability'] }]) { + const host = await fixture(options), client = await host.starting + try { + assert.equal(client.available, true) + assert.throws(() => client.shell.send(execute('disabled-shell')), /capability is unavailable/) + await assert.rejects(client.shell.read('disabled-shell', 'stdout', 0, 8), /unavailable/) + const call = client.invoke(file('file-only')) + host.send({ type: 'completed', requestId: 'file-only', exchange }) + assert.deepEqual(await call, exchange) + assert.deepEqual(host.spec.argv, ['/fixture-worker', '--serve']) + assert.equal(host.spec.stdio.control, 'pipe') + assert.equal(host.spawns, 1) + assert.equal(host.terminated, 0) + } finally { await host.dispose() } + } +}) + +await test('malformed optional capabilities fail the handshake through owned cleanup', async () => { + for (const capabilities of [null, 'shell-v1', Array(17).fill('shell-v1'), ['x'.repeat(65)]]) { + const host = await fixture({ capabilities }) + try { + await assert.rejects(bounded(host.starting, 'invalid capabilities'), /capabilities/) + assert.ok(host.terminated > 0) + assert.ok(host.waits > 0) + } finally { await host.dispose() } + } +}) + +await test('file and shell responses interleave without crossing their correlations', async () => { + const host = await fixture(), client = await host.starting + try { + let fileStarted = 0 + const call = client.invoke(file('file-call'), undefined, () => { fileStarted++ }) + client.shell.send(execute('shell-first')); client.shell.send(execute('shell-second')) + await turn() + host.send({ type: 'started', requestId: 'file-call' }) + host.send({ type: 'shell.admit', requestId: 'shell-second', hasShell: true }) + await host.event('shell.admit', 'shell-second') + client.shell.send({ type: 'shell.admission', requestId: 'shell-second', decision: 'reuse' }) + host.send({ type: 'shell.started', requestId: 'shell-second' }) + host.send({ type: 'completed', requestId: 'file-call', exchange }) + host.send({ type: 'shell.admit', requestId: 'shell-first', hasShell: false, + bootstrap: { commandPipe: 'fixture-command', receiptPipe: 'fixture-receipt' } }) + await host.event('shell.admit', 'shell-first') + client.shell.send({ type: 'shell.admission', requestId: 'shell-first', decision: 'launch', closeAfter: false, + launch: { argv: ['/fixture-shell', 'fixed-bootstrap'], cwd: '/fixture-workspace', env: {} } }) + host.send({ type: 'shell.started', requestId: 'shell-first' }) + host.send(completed('shell-second')); host.send(completed('shell-first')) + assert.deepEqual(await call, exchange) + await host.event('shell.completed', 'shell-first') + assert.equal(fileStarted, 1) + assert.deepEqual(host.events.map(value => `${value.type}:${value.requestId}`), [ + 'shell.admit:shell-second', 'shell.started:shell-second', 'shell.admit:shell-first', + 'shell.started:shell-first', 'shell.completed:shell-second', 'shell.completed:shell-first', + ]) + client.shell.send({ type: 'shell.release', requestId: 'shell-first' }) + client.shell.send({ type: 'shell.release', requestId: 'shell-second' }) + assert.equal(client.available, true) + assert.equal(host.spawns, 1) + assert.equal(host.terminated, 0) + } finally { await host.dispose() } +}) + +await test('shell ownership and the retained correlation window stay bounded until release', async () => { + const host = await fixture(), client = await host.starting + try { + assert.throws(() => client.shell.subscribe(() => {}), /admission owner/) + for (let index = 0; index < 32; index++) { + const id = `retained-${index}` + client.shell.send(execute(id)); await turn() + host.send({ type: 'shell.started', requestId: id }); host.send(completed(id)) + } + await host.event('shell.completed', 'retained-31') + assert.throws(() => client.shell.send(execute('overflow')), /full/) + assert.throws(() => client.shell.send(execute('retained-0')), /duplicated|full/) + const independent = client.invoke(file('independent-file')) + host.send({ type: 'completed', requestId: 'independent-file', exchange }) + assert.deepEqual(await independent, exchange) + client.shell.send({ type: 'shell.release', requestId: 'retained-0' }) + client.shell.send(execute('released-capacity')) + await turn() + assert.equal(host.frames.some(frame => frame.type === 'shell.execute' && frame.requestId === 'released-capacity'), true) + assert.equal(client.available, true) + } finally { await host.dispose() } +}) + +await test('unsent cancellation is local, has no effects, and retains only empty output until release', async () => { + const host = await fixture({ holdWrites: true }), client = await host.starting + try { + client.shell.send(execute('blocking-write')) + client.shell.send(execute('unsent')) + client.shell.send({ type: 'cancel', requestId: 'unsent' }) + const cancelled = await host.event('shell.completed', 'unsent') + assert.equal(cancelled.status, 'cancelled') + assert.equal(cancelled.effectState, 'none') + assert.equal(cancelled.cleanupConfirmed, true) + assert.equal(host.frames.length, 1) + const empty = await client.shell.read('unsent', 'stdout', 0, 64) + assert.deepEqual(empty, output('unsent', 'stdout', 0, Buffer.alloc(0), { eof: true })) + client.shell.send({ type: 'shell.release', requestId: 'unsent' }) + host.writes.shift()(); await turn() + assert.equal(host.frames.some(frame => frame.requestId === 'unsent'), false) + host.send({ type: 'shell.started', requestId: 'blocking-write' }); host.send(completed('blocking-write')) + await host.event('shell.completed', 'blocking-write') + assert.equal(client.available, true) + assert.equal(host.terminated, 0) + } finally { await host.dispose() } +}) + +await test('sent failures remain unknown and unsent failures remain none until the outer range joins', async () => { + const host = await fixture({ holdWrites: true, holdExit: true }), client = await host.starting + try { + client.shell.send(execute('sent')); client.shell.send(execute('unsent')) + host.finish(); await turn() + assert.equal(client.available, false) + assert.equal(client.shell.available, false) + assert.equal(host.events.length, 0) + assert.ok(host.terminated > 0) + assert.ok(host.waits > 0) + assert.throws(() => client.shell.send(execute('after-loss')), /unavailable/) + host.releaseExit() + const sent = await host.event('shell.completed', 'sent') + const unsent = await host.event('shell.completed', 'unsent') + assert.equal(sent.status, 'failed'); assert.equal(sent.effectState, 'unknown') + assert.equal(unsent.status, 'rejected'); assert.equal(unsent.effectState, 'none') + assert.equal((await host.event('shell.unavailable')).cleanupConfirmed, true) + assert.equal(host.spawns, 1) + } finally { await host.dispose() } +}) + +await test('malformed or duplicate shell terminals fail the outer client and await joined cleanup', async () => { + const cases = [ + ['unknown identity', host => host.send(completed('unknown'))], + ['success without started', host => host.send(completed('shell-pending'))], + ['negative output count', host => { host.send({ type: 'shell.started', requestId: 'shell-pending' }); host.send(completed('shell-pending', { stdoutTotal: -1 })) }], + ['duplicate started', host => { host.send({ type: 'shell.started', requestId: 'shell-pending' }); host.send({ type: 'shell.started', requestId: 'shell-pending' }) }], + ['duplicate terminal', host => { host.send({ type: 'shell.started', requestId: 'shell-pending' }); host.send(completed('shell-pending')); host.send(completed('shell-pending')) }], + ['uncorrelated cleanup', host => host.send({ type: 'shell.closed', requestId: 'missing-close', cleanupConfirmed: true })], + ] + for (const [label, corrupt] of cases) { + const host = await fixture({ holdExit: true }), client = await host.starting + try { + let settled = false + const fileCall = client.invoke(file('file-pending')) + void fileCall.then(() => { settled = true }) + client.shell.send(execute('shell-pending')); await turn() + corrupt(host); await turn() + assert.equal(client.available, false, label) + assert.equal(settled, false, label) + assert.equal(host.events.some(event => event.type === 'shell.unavailable'), false, label) + assert.ok(host.terminated > 0, label) + host.releaseExit() + assert.equal((await fileCall).interrupted, true, label) + assert.equal((await host.event('shell.unavailable')).cleanupConfirmed, true, label) + assert.equal(host.events.filter(event => event.type === 'shell.completed' && event.requestId === 'shell-pending').length, 1, label) + assert.equal(host.spawns, 1, label) + } finally { await host.dispose() } + } +}) + +await test('bounded stream reads preserve binary bytes, stream separation and live empty output', async () => { + const host = await fixture(), client = await host.starting + try { + client.shell.send(execute('reading')); await turn() + const stdout = client.shell.read('reading', 'stdout', 0, 65536) + const stderr = client.shell.read('reading', 'stderr', 0, 8) + await assert.rejects(client.shell.read('reading', 'stdout', 0, 1), /concurrent/) + host.send(output('reading', 'stderr', 0, Buffer.from('warn'))) + host.send(output('reading', 'stdout', 0, Buffer.alloc(0))) + assert.deepEqual(await stdout, output('reading', 'stdout', 0, Buffer.alloc(0))) + assert.deepEqual(await stderr, output('reading', 'stderr', 0, Buffer.from('warn'))) + const bytes = Buffer.alloc(65536, 255); bytes[0] = 0 + const maximum = client.shell.read('reading', 'stdout', 0, 65536) + host.send(output('reading', 'stdout', 0, bytes)) + assert.deepEqual(Buffer.from((await maximum).bytesBase64, 'base64'), bytes) + await turn() + const written = host.frames.length + for (const args of [['stdout', -1, 1], ['stdout', 0.5, 1], ['stdout', 0, 0], ['stdout', 0, 65537], ['stdin', 0, 1]]) { + await assert.rejects(client.shell.read('reading', ...args), /Invalid/) + } + await turn() + assert.equal(host.frames.length, written) + assert.equal(client.available, true) + } finally { await host.dispose() } +}) + +await test('completed output supports explicit cursor reads and cannot release an active read', async () => { + const host = await fixture(), client = await host.starting + try { + client.shell.send(execute('completed-output')); await turn() + host.send({ type: 'shell.started', requestId: 'completed-output' }) + host.send(completed('completed-output', { stdoutTotal: 20 })) + await host.event('shell.completed', 'completed-output') + const read = client.shell.read('completed-output', 'stdout', 5, 8) + assert.throws(() => client.shell.send({ type: 'shell.release', requestId: 'completed-output' }), /active/) + const gap = output('completed-output', 'stdout', 5, Buffer.from('ok'), { nextOffset: 20, lossy: true, eof: true }) + host.send(gap) + assert.deepEqual(await read, gap) + const end = client.shell.read('completed-output', 'stdout', 20, 8) + host.send(output('completed-output', 'stdout', 20, Buffer.alloc(0), { eof: true })) + assert.equal((await end).eof, true) + client.shell.send({ type: 'shell.release', requestId: 'completed-output' }); await turn() + assert.equal(host.frames.filter(frame => frame.type === 'shell.release' && frame.requestId === 'completed-output').length, 1) + await assert.rejects(client.shell.read('completed-output', 'stdout', 20, 8), /Unknown/) + assert.throws(() => client.shell.send({ type: 'shell.release', requestId: 'completed-output' }), /Unknown/) + assert.equal(client.available, true) + } finally { await host.dispose() } +}) + +await test('invalid output receipts reject pending reads only after outer cleanup', async () => { + const cases = [ + ['wrong request', { requestId: 'other' }], ['wrong stream', { stream: 'stderr' }], + ['wrong requested offset', { offset: 3 }], ['backward cursor', { nextOffset: 3 }], + ['noncanonical base64', { bytesBase64: 'YQ' }], ['invalid base64', { bytesBase64: '!!!!' }], + ['oversized bytes', { bytesBase64: Buffer.from('abcd').toString('base64'), nextOffset: 8 }], + ['lossless gap', { nextOffset: 9 }], ['lossy cursor before bytes', { lossy: true, nextOffset: 4 }], + ['invalid loss marker', { lossy: 'false' }], ['invalid eof marker', { eof: 1 }], + ['invalid capture error code', { error: { code: 1, message: 'bad' } }], + ['missing capture error message', { error: { code: 'OUTPUT_FAILED' } }], + ] + for (const [label, overrides] of cases) { + const host = await fixture({ holdExit: true }), client = await host.starting + try { + client.shell.send(execute('bad-output')); await turn() + let settled = false + const read = client.shell.read('bad-output', 'stdout', 4, 3) + void read.then(() => { settled = true }, () => { settled = true }) + const rejection = assert.rejects(read, /shell output/) + host.send(output('bad-output', 'stdout', 4, Buffer.from('a'), overrides)); await turn() + assert.equal(client.available, false, label) + assert.equal(settled, false, label) + assert.equal(host.events.some(event => event.type === 'shell.unavailable'), false, label) + host.releaseExit(); await rejection + assert.equal((await host.event('shell.unavailable')).cleanupConfirmed, true, label) + assert.ok(host.terminated > 0, label) + } finally { await host.dispose() } + } +}) + +await test('unconfirmed outer cleanup remains visible in shell failure receipts', async () => { + const host = await fixture({ holdExit: true }), client = await host.starting + try { + client.shell.send(execute('uncertain')); await turn() + host.finish(); await turn() + assert.equal(host.events.length, 0) + host.releaseExit(false) + const failure = await host.event('shell.completed', 'uncertain') + assert.equal(failure.effectState, 'unknown') + assert.equal(failure.cleanupConfirmed, false) + assert.equal((await host.event('shell.unavailable')).cleanupConfirmed, false) + // Let fixture disposal observe a now-empty synthetic range without altering the recorded failure. + host.handle.waitForExit = async () => true + } finally { await host.dispose() } +}) + +await test('32 idle sessions keep their budget until a confirmed close frees one slot', async () => { + const host = await fixture(), client = await host.starting + try { + for (let index = 0; index < 32; index++) { + const id = `idle-${index}` + client.shell.send({ ...execute(id), sessionKey: `session-${index}` }); await turn() + host.send({ type: 'shell.started', requestId: id }); host.send(completed(id)) + await host.event('shell.completed', id) + client.shell.send({ type: 'shell.release', requestId: id }) + } + await turn() + const extra = { ...execute('idle-overflow'), sessionKey: 'session-overflow' }, count = host.frames.length + assert.throws(() => client.shell.send(extra), /owner capacity/) + assert.equal(host.frames.length, count) + client.shell.send({ type: 'shell.close', requestId: 'free-idle-owner', sessionKey: 'session-0', reason: 'Fixture release' }) + assert.throws(() => client.shell.send(extra), /owner capacity/) + await turn() + host.send({ type: 'shell.closed', requestId: 'free-idle-owner', cleanupConfirmed: true }) + await host.event('shell.closed', 'free-idle-owner') + client.shell.send(extra); await turn() + assert.equal(host.frames.filter(frame => frame.type === 'shell.execute' && frame.requestId === 'idle-overflow').length, 1) + assert.equal(client.available, true) + assert.equal(host.terminated, 0) + } finally { await host.dispose() } +}) + +await test('accepted file and shell traffic retains its reserved control capacity under backpressure', async () => { + const host = await fixture({ holdWrites: true }), client = await host.starting + const files = [], controllers = [], reads = [] + try { + for (let index = 0; index < 32; index++) client.shell.send({ ...execute(`pressure-shell-${index}`), sessionKey: `pressure-owner-${index}` }) + for (let index = 0; index < 32; index++) { + const controller = new AbortController(); controllers.push(controller) + const call = client.invoke(file(`pressure-file-${index}`), controller.signal) + void call.catch(() => {}); files.push(call) + } + // Hand every accepted execution to the writable boundary, retaining the last callback. + for (let index = 0; index < 63; index++) { assert.equal(host.writes.length, 1); host.writes.shift()(); await turn() } + assert.equal(host.frames.length, 64) + assert.equal(host.writes.length, 1) + host.holdAllWrites() + for (let index = 0; index < 32; index++) host.send({ type: 'shell.admit', requestId: `pressure-shell-${index}`, hasShell: true }) + await turn() + for (let index = 0; index < 32; index++) client.shell.send({ type: 'shell.admission', requestId: `pressure-shell-${index}`, decision: 'reuse' }) + for (let index = 0; index < 32; index++) for (const stream of ['stdout', 'stderr']) { + const read = client.shell.read(`pressure-shell-${index}`, stream, 0, 64) + void read.catch(() => {}); reads.push(read) + } + for (let index = 0; index < 32; index++) { client.shell.send({ type: 'cancel', requestId: `pressure-shell-${index}` }); controllers[index].abort() } + for (let index = 0; index < 32; index++) for (const type of ['shell.retire', 'shell.close']) { + client.shell.send({ type, requestId: `${type === 'shell.close' ? 'close' : 'retire'}-${index}`, + sessionKey: `pressure-owner-${index}`, reason: 'Fixture cleanup budget' }) + } + assert.equal(host.frames.length, 64, 'No control callback may bypass the blocked execution write') + assert.equal(client.available, true) + assert.equal(host.terminated, 0) + // Advance admissions and all 64 reads. Each output answers a physically written read. + let answered = 0 + while (answered < 64) { + assert.equal(host.writes.length, 1) + host.writes.shift()(); await turn() + const frame = host.frames.at(-1) + if (frame.type === 'shell.read') { + host.send(output(frame.requestId, frame.stream, frame.offset, Buffer.alloc(0))) + answered++ + } + } + await Promise.all(reads) + for (let index = 0; index < 32; index++) { + host.send(completed(`pressure-shell-${index}`, { status: 'cancelled', exitCode: null, effectState: 'none', + error: { code: 'CANCELLED', message: 'Kernel cancelled before command start.' } })) + host.send({ type: 'completed', requestId: `pressure-file-${index}`, exchange }) + } + await Promise.all(files) + await host.event('shell.completed', 'pressure-shell-31') + for (let index = 0; index < 32; index++) client.shell.send({ type: 'shell.release', requestId: `pressure-shell-${index}` }) + assert.equal(client.available, true) + assert.equal(host.terminated, 0) + host.releaseWrites(); await turn() + const counts = Object.fromEntries(['execute', 'shell.execute', 'shell.admission', 'shell.read', 'cancel', 'shell.retire', 'shell.close', 'shell.release'] + .map(type => [type, host.frames.filter(frame => frame.type === type).length])) + assert.deepEqual(counts, { execute: 32, 'shell.execute': 32, 'shell.admission': 32, 'shell.read': 64, + cancel: 64, 'shell.retire': 32, 'shell.close': 32, 'shell.release': 32 }) + for (const frame of host.frames.filter(frame => frame.type === 'shell.close' || frame.type === 'shell.retire')) { + host.send({ type: 'shell.closed', requestId: frame.requestId, cleanupConfirmed: true }) + } + await host.event('shell.closed', 'close-31') + assert.equal(client.available, true) + assert.equal(host.spawns, 1) + assert.equal(host.terminated, 0) + } finally { await host.dispose(); await Promise.allSettled([...files, ...reads]) } +}) + +await test('a typed per-stream capture error preserves the channel and raw reads cannot bypass correlation', async () => { + const host = await fixture(), client = await host.starting + try { + client.shell.send(execute('capture-failure')); client.shell.send(execute('healthy-sibling')); await turn() + host.send({ type: 'shell.started', requestId: 'capture-failure' }) + host.send({ type: 'shell.started', requestId: 'healthy-sibling' }) + host.send(completed('capture-failure', { stdoutTotal: 10 })) + await host.event('shell.completed', 'capture-failure') + const raw = { type: 'shell.read', requestId: 'capture-failure', stream: 'stdout', offset: 0, maxBytes: 64 } + assert.throws(() => client.shell.send(raw), /read\(\) method/) + const read = client.shell.read('capture-failure', 'stdout', 0, 64) + assert.throws(() => client.shell.send(raw), /read\(\) method/) + const receipt = output('capture-failure', 'stdout', 0, Buffer.from('part'), { + eof: false, error: { code: 'OUTPUT_FAILED', message: 'Capture failed after partial output.' }, + }) + host.send(receipt) + assert.deepEqual(await read, receipt) + client.shell.send({ type: 'shell.release', requestId: 'capture-failure' }) + host.send(completed('healthy-sibling')) + assert.equal((await host.event('shell.completed', 'healthy-sibling')).status, 'completed') + client.shell.send({ type: 'shell.release', requestId: 'healthy-sibling' }); await turn() + assert.equal(host.frames.filter(frame => frame.type === 'shell.read').length, 1) + assert.equal(host.events.some(event => event.type === 'shell.unavailable'), false) + assert.equal(client.available, true) + assert.equal(host.terminated, 0) + } finally { await host.dispose() } +}) + +report('resident-shell-channel')