diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 83ecf19e..531d4ef7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -131,6 +131,10 @@ jobs:
         if: runner.os == 'Windows'
         shell: bash
         run: node bench/probes/probe-872-channel-spill.mjs native/target/debug/examples/shell_driver.exe runtime/win32-x64/niubash/niu.exe runtime/win32-x64/niubash/niu.exe --stock-foreground
+      - name: Windows session spill with stock foreground commands
+        if: runner.os == 'Windows'
+        shell: bash
+        run: node bench/probes/probe-872-session-spill.mjs native/target/debug/dsh-ops-worker.exe runtime/win32-x64/niubash/niu.exe runtime/win32-x64/niubash/niu.exe session-spill-report.json --stock-foreground
       - name: Assert the Linux worker is statically linked
         if: runner.os == 'Linux'
         shell: bash
diff --git a/bench/probes/probe-872-session-spill.mjs b/bench/probes/probe-872-session-spill.mjs
new file mode 100644
index 00000000..b9a1d936
--- /dev/null
+++ b/bench/probes/probe-872-session-spill.mjs
@@ -0,0 +1,166 @@
+/** Actual internal task/session kernel, public confinement, and the existing recovery ledger. */
+import assert from 'node:assert/strict'
+import fs from 'node:fs'
+import os from 'node:os'
+import path from 'node:path'
+import { createHash } from 'node:crypto'
+import { setTimeout as delay } from 'node:timers/promises'
+import Subprocess from '@deepseek-ai/dsh-subprocess-local'
+import Sandbox from '@deepseek-ai/dsh-sandbox-local'
+import Policy, { setSandboxMode } from '@deepseek-ai/dsh-sandbox-policy'
+import FileSystem from '@deepseek-ai/dsh-fs-local'
+import { residentKernel } from '../../lib/execution/resident/provider.js'
+import { residentShellAdmission } from '../../lib/execution/resident/shell-owner.js'
+import { createTaskExecutionHost } from '../../lib/execution/resident/task-owner.js'
+import { prepareNiubash } from '../../lib/niubash/runtime.js'
+import { loadScrubbedParentEnv } from '../../lib/adapter/dsh/host.js'
+import { mergeShellEnvironment } from '../../lib/execution/resident/shell-launch.js'
+import { getSpillLedger, SpillLedger } from '../../lib/shells/spill-ledger.js'
+import { sweepSpills } from '../../lib/shells/spill-retention.js'
+import { invoke as fileCall } from '../../lib/execution/files.js'
+
+assert.equal(process.platform, 'win32', 'session process ownership is Windows-only')
+const [worker, shell, support, output] = process.argv.slice(2, 6).map(value => path.resolve(value))
+const stock = process.argv.includes('--stock-foreground')
+const hash = file => createHash('sha256').update(fs.readFileSync(file)).digest('hex')
+if (stock) assert.equal(hash(shell), hash(support))
+const workspace = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), 'ops-session-spill-')))
+process.env.DSH_HOME = fs.mkdtempSync(path.join(os.tmpdir(), 'ops-session-spill-home-'))
+fs.mkdirSync(path.join(workspace, 'child'))
+const { createResidentTaskJobs } = await import('../../test/lib/resident-task-jobs.mjs')
+const report = { kind: stock ? 'stock-foreground-session-spill-only' : 'candidate-session-spill',
+  worker, workerSha256: hash(worker), shell, shellSha256: hash(shell), workspace, checks: [],
+  limitations: ['Internal adapters; no model publication or shipped runtime change.',
+    stock ? 'Stock foreground proof does not qualify stock asynchronous IO for persistent sessions.' : 'Explicit local backport candidate, not a published runtime.'] }
+let host, tasks, admission, client, lastRequest
+const sealed = new Set(), observed = new Set(), calls = []
+const quote = value => `'${value.replaceAll("'", "'\\''")}'`
+const node = quote(process.execPath.replaceAll('\\', '/'))
+const external = code => `${node} -e ${quote(code)}`
+const alive = pid => { try { process.kill(pid, 0); return true } catch (error) { if (error.code === 'ESRCH') return false; throw error } }
+async function until(predicate, label, ms = 15000) {
+  const deadline = Date.now() + ms
+  for (;;) { const value = await predicate(); if (value) return value; if (Date.now() > deadline) throw Error(`${label} timed out`); await delay(10) }
+}
+async function within(promise, label, ms = 25000) {
+  let timer
+  try { return await Promise.race([promise, new Promise((_, reject) => { timer = setTimeout(() => reject(Error(`${label} timed out`)), ms) })]) }
+  finally { clearTimeout(timer) }
+}
+async function check(name, run) {
+  const detail = await run(); report.checks.push({ name, status: 'passed', detail }); console.log(`ok ${name}`)
+}
+function okay(value) { assert.equal(value.status, 'completed', JSON.stringify(value)); assert.equal(value.cleanupConfirmed, true); return value }
+try {
+  host = await createResidentTaskJobs({ cwd: workspace, initialAgents: 3 })
+  const { ctx, agents } = host
+  await ctx.plugin(Policy, { mode: 'danger-full-access', workspaceRoot: workspace })
+  await ctx.plugin(Subprocess); await ctx.plugin(Sandbox, {}); await ctx.plugin(FileSystem, { cwd: workspace })
+  await ctx.plugin(inner => inner.provide('shellEnv', { collect: () => ({}) }))
+  await ctx.plugin(residentKernel({ binaryPath: worker }))
+  client = ctx.get('opsKernel'); observed.add(client.pid); report.kernelPid = client.pid
+  assert.ok(client.shell.capabilities.includes('shell-spill-v1'))
+  assert.ok(client.task.capabilities.includes('task-spill-v1'))
+  const transport = {
+    get available() { return client.shell.available }, get capabilities() { return client.shell.capabilities },
+    send(message) { if (message.type === 'shell.execute') lastRequest = message.requestId; return client.shell.send(message) },
+    read: (...args) => client.shell.read(...args), subscribe: listener => client.shell.subscribe(listener), abort: error => client.shell.abort(error),
+  }
+  await ctx.plugin(residentShellAdmission({ transport, bash: { file: shell, source: report.kind },
+    async runtime(_bash, env, policy, signal) {
+      const prepared = await prepareNiubash(support, env.PATH ?? env.Path, signal, { workspaceRoot: policy.workspaceRoot })
+      return { ...prepared, file: shell }
+    } }))
+  admission = ctx.get('opsShellAdmission')
+  await ctx.plugin({ inject: ['agents'], apply(inner) { tasks = createTaskExecutionHost(inner, client.task) } })
+  const ledger = getSpillLedger(), env = mergeShellEnvironment(await loadScrubbedParentEnv())
+  function call(agent, command, signal) {
+    const result = admission.execute({ command, timeoutMs: 20000 }, { agent, signal })
+    calls.push(result); return within(result, 'session call')
+  }
+  function recover(stream, expected, agent) {
+    assert.equal(stream.bytesTotal, expected.length); assert.equal(stream.lossy, true)
+    assert.equal(stream.spillOrigin, 'ops-native-v1'); assert.equal(stream.spillPid, client.pid); assert.equal(stream.spillSealed, true)
+    assert.deepEqual(fs.readFileSync(stream.spillPath), expected)
+    const entry = ledger.entries().find(entry => entry.path === stream.spillPath)
+    assert.ok(entry); assert.equal(entry.sessionId, agent.session.id); assert.equal(entry.bytes, expected.length)
+    assert.equal(entry.pid, client.pid); assert.equal(entry.origin, 'ops-native-v1'); sealed.add(stream.spillPath)
+  }
+  await check('exact memory boundary remains memory only', async () => {
+    const value = okay(await call(agents[0], external("require('fs').writeSync(1,Buffer.alloc(131072,120))")))
+    assert.equal(value.stdout.bytesTotal, 131072); assert.equal(value.stdout.spillPath, undefined)
+    assert.equal(value.stdout.lossy, false); assert.equal(ledger.entries().length, 0)
+  })
+  const modes = ['danger-full-access', 'read-only', 'workspace-write']
+  for (const [index, mode] of modes.entries()) await check(`${mode}: retained state and exact binary recovery after release`, async () => {
+    const agent = agents[index]; setSandboxMode(agent.session, mode)
+    const state = okay(await call(agent, "cd child; export KEEP_SESSION=state; kept_fn(){ printf function; }; set -f; printf '%s' \"$$\""))
+    observed.add(Number(state.stdout.text))
+    const length = 1024 * 1024, expected = Buffer.alloc(length)
+    for (let i = 0; i < length; i++) expected[i] = i % 251
+    const code = `const fs=require('fs'),b=Buffer.alloc(${length});for(let i=0;i<b.length;i++)b[i]=i%251;fs.writeSync(1,b);fs.writeSync(2,b)`
+    const result = okay(await call(agent, external(code) + (stock ? '' : ' & wait')))
+    recover(result.stdout, expected, agent); recover(result.stderr, expected, agent)
+    assert.equal(result.stdout.text, expected.subarray(0, 131072).toString('utf8'))
+    const preserved = okay(await call(agent, 'printf "%s:%s:" "${PWD##*/}" "$KEEP_SESSION"; kept_fn; printf ":%s" "$-"'))
+    assert.match(preserved.stdout.text, /^child:state:function:.*f/)
+    return { stdout: result.stdout.spillPath, stderr: result.stderr.spillPath, sessionId: agent.session.id }
+  })
+  await check('fresh task and all session channels use one kernel store and ledger', async () => {
+    const execution = tasks.start({ agent: agents[0] }, { lifetime: 'foreground', source: 'shared-store-fixture',
+      spec: { command: 'fixed fixture', workdir: workspace, timeoutMs: 20000, graceMs: 1, includeDiagnostics: true },
+      async admit() { return { argv: [process.execPath, '-e', "require('fs').writeSync(1,'TASK_HEADER'+'t'.repeat(70000))"], cwd: workspace, env } } })
+    const value = await within(execution.result(), 'shared task')
+    assert.equal(value.error, undefined); sealed.add(value.stdout.spillPath)
+    assert.equal(fs.readFileSync(value.stdout.spillPath, 'utf8'), 'TASK_HEADER' + 't'.repeat(70000))
+    assert.equal(new Set([...sealed].map(file => path.dirname(file))).size, 1)
+    const reloaded = new SpillLedger(ledger.file)
+    assert.equal(reloaded.entries().filter(entry => sealed.has(entry.path)).length, sealed.size)
+    return { directory: path.dirname(value.stdout.spillPath), files: sealed.size }
+  })
+  await check('existing file authority recovers a session spill and rejects writes outside the workspace', async () => {
+    const agent = agents[1], result = okay(await call(agent, external("require('fs').writeSync(1,'SESSION_RECOVERY_HEADER\\n'+'r'.repeat(150000))")))
+    const file = result.stdout.spillPath; sealed.add(file)
+    const fileContext = { agent, signal: AbortSignal.timeout(15000) }
+    const read = await fileCall(ctx, worker, 'inspect_local_file', { path: file }, fileContext, fileContext.signal)
+    assert.equal(read.status, 'completed', JSON.stringify(read)); assert.ok(JSON.stringify(read).includes('SESSION_RECOVERY_HEADER'))
+    const write = await fileCall(ctx, worker, 'write_file', { path: file, content: 'forbidden' }, fileContext, fileContext.signal)
+    assert.equal(write.status, 'rejected'); assert.ok(fs.readFileSync(file, 'utf8').startsWith('SESSION_RECOVERY_HEADER'))
+  })
+  await check('complete-file cap withdraws recovery while bounded memory still drains', async () => {
+    const value = okay(await call(agents[0], external("const fs=require('fs'),b=Buffer.alloc(65536,120);for(let i=0;i<1024;i++)fs.writeSync(1,b);fs.writeSync(1,'!')")))
+    assert.equal(value.stdout.bytesTotal, 64 * 1024 * 1024 + 1); assert.equal(value.stdout.text, 'x'.repeat(131072))
+    assert.equal(value.stdout.spillPath, undefined); assert.match(value.stdout.spillError, /cap exceeded/)
+  })
+  await check('retention removes only sealed files recorded by both adapters', async () => {
+    const swept = sweepSpills({ ledger, now: Date.now() + 40 * 86400000 })
+    assert.deepEqual(new Set(swept.deleted), sealed)
+    for (const file of sealed) assert.equal(fs.existsSync(file), false)
+    assert.equal(okay(await call(agents[2], 'printf sibling')).stdout.text, 'sibling')
+    return { deleted: swept.deleted.length }
+  })
+  await check('kernel loss withdraws unsealed live output and never records a recovery claim', async () => {
+    const result = call(agents[0], external("const fs=require('fs');fs.writeSync(1,'LIVE_PID='+process.pid+'\\n'+'z'.repeat(150000));setInterval(()=>{},1000)"))
+    const id = lastRequest
+    const live = await until(async () => {
+      const value = await client.shell.read(id, 'stdout', 0, 65536)
+      return value.spillPath ? value : undefined
+    }, 'live session spill')
+    assert.equal(live.spillSealed, false); assert.equal(live.eof, false)
+    assert.equal(ledger.entries().some(entry => entry.path === live.spillPath), false)
+    const pid = /^LIVE_PID=(\d+)/.exec(fs.readFileSync(live.spillPath, 'utf8')); assert.ok(pid); observed.add(Number(pid[1]))
+    process.kill(client.pid, 'SIGKILL')
+    const lost = await result
+    assert.equal(lost.effectState, 'unknown'); assert.equal(lost.cleanupConfirmed, true)
+    assert.equal(lost.stdout.spillPath, undefined); assert.equal(fs.existsSync(live.spillPath), false)
+    assert.equal(ledger.entries().some(entry => entry.path === live.spillPath), false)
+    assert.deepEqual([...observed].filter(alive), [])
+  })
+  report.status = 'passed'
+} catch (error) { report.status = 'failed'; report.error = String(error.stack ?? error); process.exitCode = 1 }
+finally {
+  try { await tasks?.dispose(); await admission?.dispose(); await host?.dispose(); await Promise.allSettled(calls); report.cleanupConfirmed = true }
+  catch (error) { report.cleanupConfirmed = false; report.cleanupError = String(error.stack ?? error); process.exitCode = 1 }
+  report.survivorsAfterCleanup = [...observed].filter(alive)
+  fs.writeFileSync(output, JSON.stringify(report, null, 2) + '\n'); console.log(JSON.stringify(report))
+}
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index f0f70ff3..d26a45fc 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]
 
+- Connect persistent session streams to the kernel's shared task spill store and existing retention ledger, with `shell-spill-v1`, actual-kernel-PID validation, sealed terminal recovery and exact Session ownership. Public activation still requires a qualified runtime; stock foreground CI does not qualify asynchronous output (#872).
+- 将持久会话输出接入内核与任务共享的溢写存储及现有保留账本，增加 `shell-spill-v1`、实际内核 PID 校验、封存终态恢复和准确 Session 归属。公共启用仍需已验证的运行时；现有运行时的前台 CI 不代表异步输出已合格（#872）。
+
 - Connect resident task output to the shared native spill store and existing settled-file retention. Complete native streams retain exact kernel/session ownership, while unknown cleanup remains poisoned. The new `task-spill-v1` capability gates later production fresh migration; persistent channels and public fresh activation remain pending (#872).
 - 将常驻任务输出接入共享原生 spill store 和既有终态文件保留流程。完整原生流保留精确内核/会话所有权，回收未知则保持中毒状态。新增 `task-spill-v1` capability 作为后续生产 fresh 迁移门槛；持久通道和公开 fresh 启用仍待完成（#872）。
 
diff --git a/docs/session-spill.md b/docs/session-spill.md
new file mode 100644
index 00000000..7cb7d083
--- /dev/null
+++ b/docs/session-spill.md
@@ -0,0 +1,65 @@
+# Session output recovery
+
+中文：[会话输出溢写](会话输出溢写.md)
+
+This internal #872 slice connects persistent session output to the same native
+spill store and retention ledger as fresh tasks. It does not publish session
+tools, change the default executor, or qualify the stock Windows runtime for
+persistent asynchronous commands.
+
+## Ownership and completion
+
+The resident kernel allocates one `SpillStore`, clones it into the task and shell
+managers before `READY`, and then advertises `shell-spill-v1` alongside
+`shell-v1`. Store configuration performs no filesystem I/O. The first stream
+overflow creates the private directory lazily in its owned capture reader.
+Creation failure degrades to bounded memory and `spillError`; it does not prevent
+kernel startup or file-tool operation.
+
+Session stdout and stderr retain their existing 128 KiB prefixes. Crossing that
+boundary starts a complete raw file, including bytes already captured in memory,
+up to 64 MiB per stream. Control receipts, framing and original process
+diagnostics never enter these files. Exact-limit streams remain memory-only.
+Exceeding the file cap, or failing a write or seal, withdraws the complete path
+while bounded capture continues. The underlying private origin, no-follow checks,
+owned I/O cancellation and irreversible poison semantics are described in
+[private-channel spill](private-channel-spill.md) and [native raw spill](native-raw-spill.md).
+
+Each correlated `shell.output` receipt may include `spillPath`,
+`spillOrigin: "ops-native-v1"`, `spillPid`, `spillSealed`, and separate `spillError`
+degradation. The host requires negotiated spill support, all origin fields
+together, and the PID from the actual resident handshake. Terminal collection
+joins both stream reads before releasing the native record. Unsealed terminal
+metadata or metadata that changes between terminal reads fails the protocol.
+Raw capture errors retain partial bytes without inventing EOF or a recovery path.
+
+Only sealed, error-free stream captures with confirmed cleanup enter the existing
+ledger, attributed to the exact calling Session. An ordinary command failure or
+cancellation may still leave a fully captured, sealed stream; a raw capture error
+or unknown cleanup cannot. There is no second ledger or host spill writer.
+Releasing a native record preserves its sealed files for the existing bounded
+retention pass. Recovery uses existing absolute-path file-read authority; write
+permissions are unchanged.
+
+## Validation and activation boundary
+
+The real public-host probe is:
+
+```powershell
+node bench/probes/probe-872-session-spill.mjs <worker.exe> <candidate-niu.exe> <stock-payload-niu.exe> <report.json>
+```
+
+Its nine checks cover the exact memory boundary; complete binary stdout/stderr
+recovery with retained cwd, exports, functions and options in all three sandbox
+modes; one store shared with a fresh task; ledger reload and exact Session
+attribution; file-read recovery and write refusal; the 64 MiB cap; retention; and
+kernel death with an unsealed stream. The local backport candidate additionally
+runs external background commands followed by `wait`, and passed all nine checks
+with confirmed cleanup and no survivors.
+
+CI passes `--stock-foreground` with the pinned stock executable. That mode proves
+the same storage/lifecycle path with foreground external commands only. It does
+not establish stock asynchronous output fidelity. Public session activation must
+separately require `shell-v1`, `shell-spill-v1`, and a qualified runtime. Internal
+bounded `shell-v1` fixtures remain compatible. POSIX shell ownership remains
+unsupported; no process-group fallback or silent scope change is introduced.
diff --git "a/docs/\344\274\232\350\257\235\350\276\223\345\207\272\346\272\242\345\206\231.md" "b/docs/\344\274\232\350\257\235\350\276\223\345\207\272\346\272\242\345\206\231.md"
new file mode 100644
index 00000000..f90a97c0
--- /dev/null
+++ "b/docs/\344\274\232\350\257\235\350\276\223\345\207\272\346\272\242\345\206\231.md"
@@ -0,0 +1,48 @@
+# 会话输出溢写
+
+English: [Session output recovery](session-spill.md)
+
+本 #872 内部切片把持久会话输出接入新鲜任务已使用的同一个原生溢写存储和保留账本。
+它不发布会话工具、不改变默认执行器，也不将现有 Windows 固定运行时认定为支持持久异步命令。
+
+## 所有权与完成条件
+
+常驻内核只创建一个 `SpillStore`，在 `READY` 前将其克隆交给任务与 Shell 管理器，
+随后在 `shell-v1` 之外公布 `shell-spill-v1`。配置存储不执行文件系统 I/O；第一次输出
+溢出由拥有该捕获流的读取线程延迟创建私有目录。创建失败时保留有界内存输出并报告
+`spillError`，不会阻止内核启动或文件工具运行。
+
+会话 stdout、stderr 仍分别保留 128 KiB 前缀。超过此边界才创建完整原始文件，包含此前
+已在内存捕获的字节，每个流上限为 64 MiB。控制回执、协议前缀和原进程诊断不进入文件。
+恰好达到内存上限时不创建文件；超过文件上限，或写入、封存失败时，撤回完整文件路径，
+同时继续有界捕获。私有来源、禁止跟随重解析点、I/O 取消所有权及不可撤销的污染语义见
+[私有通道溢写](私有Shell通道共享溢写.md)与[原生原始输出溢写](原生原始输出溢写基础.md)。
+
+每条关联的 `shell.output` 回执可带 `spillPath`、`spillOrigin: "ops-native-v1"`、
+`spillPid`、`spillSealed`，以及独立的退化说明 `spillError`。宿主要求已协商溢写能力、
+来源字段齐全，并且 PID 必须匹配实际常驻握手。终态收集等待两个流均完成后才释放原生
+记录。终态路径未封存，或多次终态读取间元数据变化，都会判为协议错误。原始捕获错误
+保留已有字节，但不会虚构 EOF 或恢复路径。
+
+只有清理已确认、已封存且自身无错误的捕获流才进入现有账本，归属于准确的调用 Session。
+普通命令失败或取消仍可能留下完整捕获并封存的流；原始捕获错误或清理未知不能如此记录。
+没有第二份账本或宿主溢写器。释放原生记录后，封存文件仍由现有有界保留机制管理。
+恢复读取沿用现有绝对路径文件权限，写入权限不变。
+
+## 验证与启用边界
+
+实际公共宿主探针命令为：
+
+```powershell
+node bench/probes/probe-872-session-spill.mjs <worker.exe> <candidate-niu.exe> <stock-payload-niu.exe> <report.json>
+```
+
+九项检查覆盖：恰好达到内存上限；三种沙箱模式下完整二进制 stdout/stderr 恢复及工作目录、
+导出变量、函数、选项保留；与新鲜任务共享同一存储；账本重载及准确 Session 归属；
+文件读取恢复与写入拒绝；64 MiB 上限；保留清理；存在未封存流时内核死亡。
+本地回移修复候选还执行外部后台命令及 `wait`，九项全部通过，清理已确认且无存活进程。
+
+CI 以固定运行时及 `--stock-foreground` 验证同一路径中的前台外部命令。
+这不证明现有运行时的异步输出完整性。公共会话启用仍须分别要求 `shell-v1`、
+`shell-spill-v1` 和已验证的运行时。内部有界 `shell-v1` 测试夹具保持兼容。
+POSIX Shell 所有权仍不支持；不引入进程组降级或静默改变清理范围。
diff --git a/native/examples/shell_started_blocked.rs b/native/examples/shell_started_blocked.rs
index ee9400de..ade33831 100644
--- a/native/examples/shell_started_blocked.rs
+++ b/native/examples/shell_started_blocked.rs
@@ -62,6 +62,10 @@ mod probe {
                 let _ = shutdown.shutdown();
             },
         )?;
+        manager.enable_spill(ops_process::shell::SpillStore::new(
+            &std::env::temp_dir(),
+            ops_process::shell::MAX_SPILL_BYTES,
+        )?)?;
         (manager.output())(
             json!({"type":"ready","pid":std::process::id(),"capabilities":shell::capabilities()}),
         )?;
diff --git a/native/src/resident.rs b/native/src/resident.rs
index 508cb2c0..43dbcf7f 100644
--- a/native/src/resident.rs
+++ b/native/src/resident.rs
@@ -139,11 +139,17 @@ pub fn serve() -> io::Result<()> {
             &std::env::temp_dir(),
             ops_process::shell::MAX_SPILL_BYTES,
         )?;
-        task_manager.enable_spill(spill_store)?;
+        task_manager.enable_spill(spill_store.clone())?;
+        shells.enable_spill(spill_store)?;
     }
     let active: Active = Arc::default();
     let mut tasks: Vec<thread::JoinHandle<()>> = Vec::new();
     let mut capabilities = crate::shell::capabilities();
+    #[cfg(windows)]
+    capabilities
+        .as_array_mut()
+        .unwrap()
+        .push(json!("shell-spill-v1"));
     capabilities
         .as_array_mut()
         .unwrap()
diff --git a/native/src/shell/manager.rs b/native/src/shell/manager.rs
index 2eda99d2..a9435e87 100644
--- a/native/src/shell/manager.rs
+++ b/native/src/shell/manager.rs
@@ -1,11 +1,11 @@
 use super::{protocol, session};
-use ops_process::shell::ChannelView;
+use ops_process::shell::{ChannelView, SpillStore};
 use serde_json::{Value, json};
 use std::{
     collections::{HashMap, HashSet, VecDeque},
     io,
     sync::{
-        Arc, Condvar, Mutex,
+        Arc, Condvar, Mutex, OnceLock,
         atomic::{AtomicBool, Ordering},
         mpsc,
     },
@@ -18,6 +18,7 @@ type Output = Arc<dyn Fn(Value) -> io::Result<()> + Send + Sync>;
 type Started = Arc<dyn Fn(Value) -> io::Result<mpsc::Receiver<io::Result<()>>> + Send + Sync>;
 type Response = (Value, Option<mpsc::SyncSender<io::Result<()>>>);
 pub(super) struct State {
+    pub spill: OnceLock<SpillStore>,
     pub output: Output,
     pub started: Started,
     pub fatal: Arc<dyn Fn() + Send + Sync>,
@@ -110,6 +111,7 @@ impl Manager {
         let cleanup_fatal = fatal.clone();
         Ok(Self {
             state: Arc::new(State {
+                spill: OnceLock::new(),
                 output: Arc::new(move |value| {
                     sender.try_send((value, None)).map_err(|error| {
                         // A deadline/session completion can fail after the last
@@ -144,6 +146,18 @@ impl Manager {
     pub(super) fn test_state(&self) -> Arc<State> {
         self.state.clone()
     }
+    /// Install the same allocation-only store used by tasks, before any shell admission.
+    pub fn enable_spill(&mut self, store: SpillStore) -> io::Result<()> {
+        if !self.state.requests.lock().unwrap().is_empty()
+            || !self.state.sessions.lock().unwrap().is_empty()
+        {
+            return Err(protocol::invalid("shell spill must precede admission"));
+        }
+        self.state
+            .spill
+            .set(store)
+            .map_err(|_| protocol::invalid("shell spill already configured"))
+    }
     pub fn contains(&self, id: &str) -> bool {
         self.state.requests.lock().unwrap().contains_key(id)
             || self.state.controls.lock().unwrap().contains(id)
@@ -395,6 +409,15 @@ impl Manager {
             if let Some(error) = error {
                 response["error"] = json!({"code":"OUTPUT_INCOMPLETE","message":error});
             }
+            if let Some(spill) = snapshot.spill {
+                response["spillPath"] = json!(spill.path);
+                response["spillOrigin"] = json!("ops-native-v1");
+                response["spillPid"] = json!(spill.pid);
+                response["spillSealed"] = json!(spill.sealed);
+            }
+            if let Some(error) = snapshot.spill_error {
+                response["spillError"] = json!(error);
+            }
         } else {
             response["eof"] = json!(completed && offset == 0);
             if offset != 0 {
diff --git a/native/src/shell/session.rs b/native/src/shell/session.rs
index 7d9cec74..d938d37b 100644
--- a/native/src/shell/session.rs
+++ b/native/src/shell/session.rs
@@ -236,7 +236,11 @@ impl Worker {
                     return self.failure(request, None, error, "SHELL_RESET");
                 }
             } else if self.driver.is_none() {
-                match Driver::prepare() {
+                let prepared = match self.state.spill.get() {
+                    Some(store) => Driver::prepare_with_spill(Some(store.clone())),
+                    None => Driver::prepare(),
+                };
+                match prepared {
                     Ok(driver) => self.driver = Some(driver),
                     Err(error) => {
                         return self.failure(request, None, error, "EXECUTOR_UNAVAILABLE");
diff --git a/native/src/shell/tests.rs b/native/src/shell/tests.rs
index 412fa4d7..bac704c3 100644
--- a/native/src/shell/tests.rs
+++ b/native/src/shell/tests.rs
@@ -38,6 +38,28 @@ fn reject(manager: &mut Manager, id: &str) {
         .unwrap();
 }
 
+#[test]
+fn shared_spill_configuration_is_allocation_only_and_precedes_admission() {
+    let root =
+        std::env::temp_dir().join(format!("ops-uncreated-shell-store-{}", std::process::id()));
+    assert!(!root.exists());
+    let store = ops_process::shell::SpillStore::new(&root, 1024).unwrap();
+    let (mut manager, output) = fixture();
+    manager.enable_spill(store.clone()).unwrap();
+    assert!(
+        !root.exists(),
+        "configuration must not perform coordinator filesystem I/O"
+    );
+    assert!(manager.enable_spill(store.clone()).is_err());
+    manager.shutdown().unwrap();
+    let (mut manager, late_output) = fixture();
+    execute(&mut manager, "late", "owner", 30_000);
+    assert!(manager.enable_spill(store).is_err());
+    manager.shutdown().unwrap();
+    drop(output);
+    drop(late_output);
+}
+
 #[test]
 fn queued_expiry_and_cancellation_do_not_wait_for_the_active_approval() {
     let (mut manager, output) = fixture();
diff --git a/src/execution/resident/client.ts b/src/execution/resident/client.ts
index f7eb1354..d73b3658 100644
--- a/src/execution/resident/client.ts
+++ b/src/execution/resident/client.ts
@@ -63,7 +63,7 @@ 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.shell = new ResidentShellChannel(() => this.available, (value, sending) => this.frames.send(encodeFrame(value), sending), error => this.fail(error))
+    this.shell = new ResidentShellChannel(() => this.available, (value, sending) => this.frames.send(encodeFrame(value), sending), error => this.fail(error), () => this.pid)
     this.task = new ResidentTaskChannel(() => this.available, (value, sending) => this.frames.send(encodeFrame(value), sending), error => this.fail(error), () => this.pid)
     this.frames = new ResidentFrames(handle.control, value => this.receive(value), error => this.fail(error),
       // 32 file + 32 shell executions, their 64 cancels, 32 admissions,
diff --git a/src/execution/resident/shell-admission.ts b/src/execution/resident/shell-admission.ts
index f38d4a81..11ec535d 100644
--- a/src/execution/resident/shell-admission.ts
+++ b/src/execution/resident/shell-admission.ts
@@ -13,6 +13,7 @@ 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'
+import { recordSettledSpills } from '../../shells/spill-settlement.js'
 
 type Args = { command: string; workdir?: string; timeoutMs: number; sandbox_permissions?: string; justification?: string }
 interface Facts { standing: ShellPolicy; effective: ShellPolicy; environment: string }
@@ -217,6 +218,12 @@ export function createShellAdmissionHost(ctx: Context, options: ShellAdmissionOp
       void collectShellOutput(transport, event).then(output => {
         if (!pending.has(event.requestId)) return
         transport.send({ type: 'shell.release', requestId: event.requestId })
+        recordSettledSpills([output.stdout, output.stderr].flatMap(stream =>
+          event.cleanupConfirmed && !stream.error && !stream.spillError && stream.spillPath
+            && stream.spillSealed === true && stream.spillOrigin === 'ops-native-v1' && stream.spillPid !== undefined
+            ? [{ path: stream.spillPath, bytes: stream.bytesTotal, origin: stream.spillOrigin,
+              pid: stream.spillPid, sessionId: item.owner.session.id }]
+            : []))
         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))) })
diff --git a/src/execution/resident/shell-channel.ts b/src/execution/resident/shell-channel.ts
index b26f65c6..16216998 100644
--- a/src/execution/resident/shell-channel.ts
+++ b/src/execution/resident/shell-channel.ts
@@ -18,7 +18,7 @@ export class ResidentShellChannel implements ShellTransport {
   private terminal?: Extract<ShellKernelMessage, { type: 'shell.unavailable' }>
   constructor(private readonly isAvailable: () => boolean,
     private readonly write: (value: ShellHostMessage, sending?: () => void) => () => boolean,
-    readonly abort: (error: Error) => void) {}
+    readonly abort: (error: Error) => void, private readonly kernelPid: () => number) {}
   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')
@@ -98,7 +98,14 @@ export class ResidentShellChannel implements ShellTransport {
       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')
+        || (value.error !== undefined && !error(value.error))
+        || (value.spillError !== undefined && typeof value.spillError !== 'string')
+        || (value.spillPath === undefined
+          ? [value.spillOrigin, value.spillPid, value.spillSealed].some(item => item !== undefined)
+          : !this.capabilities.includes('shell-spill-v1') || value.spillOrigin !== 'ops-native-v1'
+            || value.spillPid !== this.kernelPid() || typeof value.spillSealed !== 'boolean'
+            || typeof value.spillPath !== 'string' || !value.spillPath || value.spillPath.length > 4096
+            || value.spillPath.includes('\0'))) 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')
diff --git a/src/execution/resident/shell-output.ts b/src/execution/resident/shell-output.ts
index 2d52bcd9..201366d2 100644
--- a/src/execution/resident/shell-output.ts
+++ b/src/execution/resident/shell-output.ts
@@ -1,15 +1,22 @@
 /** 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 }
+import type { ShellCompleted, ShellFailure, ShellOutput, ShellStream, ShellTransport } from './shell-protocol.js'
+export interface ShellCapture extends Pick<ShellOutput, 'spillPath' | 'spillOrigin' | 'spillPid' | 'spillSealed' | 'spillError'> {
+  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<ShellOutputPair> {
   async function collect(stream: ShellStream, total: number): Promise<ShellCapture> {
     let offset = 0, bytes = Buffer.alloc(0), lossy = false
+    let spillIdentity: string | undefined
     // 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 identity = JSON.stringify([value.spillPath, value.spillOrigin, value.spillPid, value.spillSealed, value.spillError])
+      if (spillIdentity !== undefined && identity !== spillIdentity) throw new Error('Shell terminal spill metadata changed during collection')
+      spillIdentity = identity
+      if (!value.error && value.spillPath && value.spillSealed !== true) throw new Error('Shell terminal spill was not sealed')
       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')
@@ -17,8 +24,12 @@ export async function collectShellOutput(transport: ShellTransport, completed: S
       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 }
+      const spillError = value.spillError === undefined ? {} : { spillError: value.spillError }
+      if (value.error) return { text: bytes.toString('utf8'), bytesTotal: total, lossy: true, error: value.error, ...spillError }
+      if (value.eof) return { text: bytes.toString('utf8'), bytesTotal: total, lossy, ...spillError,
+        ...(completed.cleanupConfirmed && value.spillPath && !value.spillError ? {
+          spillPath: value.spillPath, spillOrigin: value.spillOrigin, spillPid: value.spillPid, spillSealed: value.spillSealed,
+        } : {}) }
     }
     throw new Error('Shell output exceeded its bounded terminal read count')
   }
diff --git a/src/execution/resident/shell-protocol.ts b/src/execution/resident/shell-protocol.ts
index 31c1148b..520d70f2 100644
--- a/src/execution/resident/shell-protocol.ts
+++ b/src/execution/resident/shell-protocol.ts
@@ -7,6 +7,7 @@ export interface ShellOutput {
   type: 'shell.output'; requestId: string; stream: ShellStream
   offset: number; nextOffset: number; bytesBase64: string; lossy: boolean; eof: boolean
   error?: ShellFailure
+  spillPath?: string; spillOrigin?: 'ops-native-v1'; spillPid?: number; spillSealed?: boolean; spillError?: string
 }
 export type ShellAdmission = { type: 'shell.admission'; requestId: string } & (
   | { decision: 'launch'; launch: ShellLaunch; closeAfter: boolean; commandCwd?: string }
diff --git a/test/resident-shell-admission.test.mjs b/test/resident-shell-admission.test.mjs
index 745a482c..8fb333cb 100644
--- a/test/resident-shell-admission.test.mjs
+++ b/test/resident-shell-admission.test.mjs
@@ -12,6 +12,8 @@ import { createShellAdmissionHost } from '../lib/execution/resident/shell-admiss
 import { residentShellAdmission } from '../lib/execution/resident/shell-owner.js'
 import { assert, test, report, withTempDir } from './lib/harness.mjs'
 import { bootHost } from './lib/host.mjs'
+import { randomUUID } from 'node:crypto'
+import { getSpillLedger } from '../lib/shells/spill-ledger.js'
 
 let identity = 0
 const turn = () => new Promise(resolve => setImmediate(resolve))
@@ -925,4 +927,69 @@ await test('runtime environment merge cannot add a startup script before confine
   }, { mode: 'read-only' })
 })
 
+await test('sealed session spill is retained only after both copies and attributed to the exact Session', async () => {
+  await fixture(async f => {
+    const owner = await f.agent(), sibling = await f.agent(), call = await f.submit(owner)
+    await f.admit(call)
+    const file = path.join(f.directory, `dsh-ops-native-123-${randomUUID().replaceAll('-', '')}`, `stdout-${randomUUID().replaceAll('-', '')}.log`)
+    const gate = Promise.withResolvers(), asked = Promise.withResolvers(), ledger = getSpillLedger()
+    f.state.outputRead = async request => {
+      if (request.stream === 'stderr') { asked.resolve(); await gate.promise }
+      return { type: 'shell.output', ...request, nextOffset: request.stream === 'stdout' ? 200000 : 0,
+        bytesBase64: request.stream === 'stdout' ? Buffer.from('prefix').toString('base64') : '',
+        lossy: request.stream === 'stdout', eof: true, ...(request.stream === 'stdout' ? {
+          spillPath: file, spillOrigin: 'ops-native-v1', spillPid: 123, spillSealed: true,
+        } : {}) }
+    }
+    const finishing = f.finish(call, { stdoutTotal: 200000 })
+    await bounded(asked.promise, 'sibling stream'); await turn()
+    assert.equal(ledger.entries().some(entry => entry.path === file), false)
+    gate.resolve()
+    const result = await finishing
+    assert.equal(result.stdout.spillPath, file); assert.equal(result.stdout.text, 'prefix')
+    assert.equal(result.stdout.bytesTotal, 200000); assert.equal(result.stdout.lossy, true)
+    const entry = ledger.entries().find(entry => entry.path === file)
+    assert.equal(entry.sessionId, owner.agent.session.id); assert.notEqual(entry.sessionId, sibling.agent.session.id)
+    assert.equal(entry.pid, 123); assert.equal(entry.origin, 'ops-native-v1'); assert.equal(entry.bytes, 200000)
+    assert.equal(f.frames.some(frame => frame.type === 'shell.release'), true)
+  })
+})
+
+for (const mode of ['raw error', 'spill degradation', 'unknown cleanup']) {
+  await test(`${mode} cannot publish or settle a session recovery path`, async () => {
+    await fixture(async f => {
+      const owner = await f.agent(), call = await f.submit(owner)
+      await f.admit(call)
+      const file = path.join(f.directory, `dsh-ops-native-123-${randomUUID().replaceAll('-', '')}`, `stdout-${randomUUID().replaceAll('-', '')}.log`)
+      f.state.outputRead = request => ({ type: 'shell.output', ...request, nextOffset: request.stream === 'stdout' ? 9 : 0,
+        bytesBase64: '', lossy: request.stream === 'stdout', eof: mode !== 'raw error' || request.stream !== 'stdout',
+        ...(request.stream === 'stdout' ? { spillPath: file, spillOrigin: 'ops-native-v1', spillPid: 123, spillSealed: true,
+          ...(mode === 'raw error' ? { error: { code: 'OUTPUT_INCOMPLETE', message: 'fixture capture failed' } } : {}),
+          ...(mode === 'spill degradation' ? { spillError: 'fixture complete-file close failed' } : {}),
+        } : {}) })
+      const result = await f.finish(call, { stdoutTotal: 9, cleanupConfirmed: mode !== 'unknown cleanup' })
+      assert.equal(result.stdout.spillPath, undefined)
+      assert.equal(getSpillLedger().entries().some(entry => entry.path === file), false)
+      assert.equal(f.aborts.length, 0)
+      if (mode === 'raw error') assert.equal(result.status, 'failed')
+      if (mode === 'spill degradation') assert.equal(result.stdout.spillError, 'fixture complete-file close failed')
+    })
+  })
+}
+
+await test('an unsealed terminal spill fails the protocol before ledger handoff', 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 => ({ type: 'shell.output', ...request, nextOffset: 0, bytesBase64: '', lossy: false, eof: true,
+      spillPath: '/unsealed', spillOrigin: 'ops-native-v1', spillPid: 123, spillSealed: false })
+    const finishing = f.finish(call)
+    assert.match((await bounded(aborted.promise, 'unsealed spill abort')).message, /not sealed/)
+    assert.equal(f.frames.some(frame => frame.type === 'shell.release'), false)
+    f.emit({ type: 'shell.unavailable', cleanupConfirmed: true, error: { code: 'EXECUTOR_UNAVAILABLE', message: 'fixture joined kernel' } })
+    assert.equal((await finishing).stdout.spillPath, undefined)
+  })
+})
+
 report('resident-shell-admission')
diff --git a/test/resident-shell-channel.test.mjs b/test/resident-shell-channel.test.mjs
index a5283a5e..caea7c56 100644
--- a/test/resident-shell-channel.test.mjs
+++ b/test/resident-shell-channel.test.mjs
@@ -469,4 +469,32 @@ await test('a typed per-stream capture error preserves the channel and raw reads
   } finally { await host.dispose() }
 })
 
+await test('session spill receipts bind to negotiated support and the ready kernel PID', async () => {
+  const spill = { spillPath: '/fixture/dsh-ops-native-123-0123456789abcdef0123456789abcdef/stdout-0123456789abcdef0123456789abcdef.log',
+    spillOrigin: 'ops-native-v1', spillPid: 123, spillSealed: true }
+  const cases = [
+    ['valid', spill, true, true],
+    ['unnegotiated', spill, false, false],
+    ['wrong kernel PID', { ...spill, spillPid: 124 }, true, false],
+    ['wrong origin', { ...spill, spillOrigin: 'host' }, true, false],
+    ['missing seal', { ...spill, spillSealed: undefined }, true, false],
+    ['orphan metadata', { ...spill, spillPath: undefined }, true, false],
+    ['invalid path', { ...spill, spillPath: 'a\0b' }, true, false],
+    ['nontext degradation', { spillError: {} }, true, false],
+    ['spill degradation', { spillError: 'file cap exceeded' }, true, true],
+  ]
+  for (const [label, fields, supported, valid] of cases) {
+    const host = await fixture({ capabilities: ['shell-v1', ...supported ? ['shell-spill-v1'] : []] }), client = await host.starting
+    try {
+      client.shell.send(execute('spill')); await turn()
+      const read = client.shell.read('spill', 'stdout', 0, 64)
+      const settled = valid ? read : assert.rejects(read, /Invalid shell output/)
+      host.send(output('spill', 'stdout', 0, Buffer.from('prefix'), { eof: true, ...fields }))
+      await settled
+      assert.equal(client.available, valid, label)
+      if (!valid) assert.equal((await host.event('shell.unavailable')).cleanupConfirmed, true, label)
+    } finally { await host.dispose() }
+  }
+})
+
 report('resident-shell-channel')
diff --git a/test/resident-shell-output.test.mjs b/test/resident-shell-output.test.mjs
new file mode 100644
index 00000000..e5cf6370
--- /dev/null
+++ b/test/resident-shell-output.test.mjs
@@ -0,0 +1,32 @@
+/** Terminal storage identity cannot change while bounded output is copied. */
+import { collectShellOutput } from '../lib/execution/resident/shell-output.js'
+import { assert, report, test } from './lib/harness.mjs'
+
+await test('changed terminal spill metadata aborts and joins the pending sibling reader', async () => {
+  const gate = Promise.withResolvers(), failures = []
+  let stderrSettled = false, reads = 0
+  const transport = {
+    available: true, capabilities: ['shell-v1', 'shell-spill-v1'],
+    abort(error) { failures.push(error); gate.resolve() },
+    async read(requestId, stream, offset) {
+      reads++
+      if (stream === 'stderr') {
+        await gate.promise; stderrSettled = true
+        return { type: 'shell.output', requestId, stream, offset, nextOffset: 0, bytesBase64: '', lossy: false, eof: true }
+      }
+      return { type: 'shell.output', requestId, stream, offset, nextOffset: offset + 1,
+        bytesBase64: Buffer.from('x').toString('base64'), lossy: false, eof: offset === 1,
+        spillPath: `/owned/${offset === 0 ? 'first' : 'replacement'}`, spillOrigin: 'ops-native-v1', spillPid: 123, spillSealed: true }
+    },
+  }
+  // The fixture's watchdog releases its own reader even if the consistency
+  // guard is reverted; a broken test must fail instead of hanging.
+  const watchdog = setTimeout(() => gate.resolve(), 1000)
+  try {
+    await assert.rejects(collectShellOutput(transport, { requestId: 'terminal', stdoutTotal: 2, stderrTotal: 0, cleanupConfirmed: true }), /metadata changed/)
+    assert.equal(stderrSettled, true)
+    assert.equal(failures.length, 1); assert.equal(reads, 3)
+  } finally { clearTimeout(watchdog); gate.resolve() }
+})
+
+report('resident-shell-output')