feat(resident): recheck shell authority at queue heads

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）。
 

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.

docs/常驻-shell-宿主准入.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 验证记录。

src/adapter/dsh/host.ts

@@ -14,6 +14,17 @@ export async function loadUserMessage(): Promise<typeof import('@deepseek-ai/dsh
   return llm.createUserMessage
 }
 
+/** Use the running host's public scrub, including its active proxy environment. */
+export async function loadScrubbedParentEnv(): Promise<Record<string, string>> {
+  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<T>(ctx: Pick<Context, 'get'>, name: string): T | undefined {
   try {
     const value = ctx.get(name as never)

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<typeof workerEnvelope>
 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<WorkerExchange>
   stop(): Promise<void>
 }
@@ -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<string, Pending>()
   private readonly frames: ResidentFrames
   private readonly hello: Promise<void>
@@ -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<WorkerExchange> {

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()