feat(core): add local managed journal snapshot recovery (#15626)

This commit is contained in:
jinye authored and GitHub committed 2026-10-26 22:47:40 +08:00
1 parent e11294b728
commit ca9d5f7ffa
34 files changed
+3941 -765

No files matched your search

@@ -154,6 +154,9 @@ PR3 [#15599](https://github.com/QwenLM/qwen-code/pull/15599) is merged into the
commit proof. Renaming a transcript directly from its journal adapter violates
the current lease's identity proof. Publish the native snapshot durably before
advancing a new native replay floor, separate from context-summary coverage.
[PR4's local design](2026-10-26-managed-local-journal-snapshots.md) implements
a one-time sealed-resume conversion with the complete prefix retained; it
does not implement rotation, GC, compact state or bounded memory/storage.
5. **Authority window:** add durable historical lookup/read-through and migrate
synchronous sequence-1 readers before eviction. Bound `transactions`, event
IDs, receipt/checkpoint/domain/extension indexes, Actions, and extension
@@ -123,6 +123,8 @@ PR3 [#15599](https://github.com/QwenLM/qwen-code/pull/15599) 已合入 PR2 分
固定 inode、元数据、rolling hash 与 sealed commit proof。在 journal 适配器中
直接 rename transcript 会违反当前 lease 的身份校验。先持久化原生快照,再推进
新的原生重放下界;该下界独立于上下文摘要覆盖范围。
[PR4 本地设计](2026-10-26-managed-local-journal-snapshots.zh-CN.md) 实现仅一次的
sealed 恢复转换,保留完整前缀;不实现轮换、GC、紧凑状态或内存/存储有界。
5. **Authority 窗口:** 淘汰前增加持久历史查询或 read-through,并迁移从 sequence 1
开始的同步消费者。除 `events` 外,还要约束 `transactions`、event ID、receipt、
checkpoint、domain、extension 索引、Action 和 extension 唯一性证据。
@@ -0,0 +1,135 @@
# Local Managed journal snapshot recovery (PR4)
[English](2026-10-26-managed-local-journal-snapshots.md) | [简体中文](2026-10-26-managed-local-journal-snapshots.zh-CN.md)
## Problem and boundaries
Issue #13184 requires a native recovery boundary before authority eviction. The
local journal currently reads the complete JSONL file, while its writer lease
authenticates both physical transcript bytes and a logical commit proof. A
journal adapter cannot independently rename that file without invalidating the
writer's proof.
This slice moves the complete committed prefix into immutable Session resources
and replaces the active file with its original ownership/header records, one
snapshot base record, and a contiguous future tail. Conversion happens once,
when a normally sealed local Session next resumes, before the final restore projection,
activation installation, or recorder binding. New Sessions, live Sessions and
read-only requests do not convert. A torn tail follows existing recovery rather
than being incorporated into a snapshot.
There is no automatic snapshot rotation, garbage collection, compact authority
state or memory eviction. Full history and authority indexes still grow. PR2 and
PR3 define remote snapshot recovery; this local envelope preserves the same
source facts without inventing server revisions or writer generations.
## Representation and interfaces
The local manifest binds its format, Session key, ordered immutable page refs,
complete-prefix byte length and SHA-256, committed sequence, commit digest, last
record UUID, resource closure and latest title/source refs. Pages contain exact
JSONL bytes and end at transaction boundaries, with a 16 MiB page maximum.
Manifest and pages use dedicated resource kinds and content-derived IDs in the
existing Session resource root. Publication is idempotent, durable and refuses
conflicting existing bytes. Existing resource size limits also apply.
The physical file retains its original first ownership record and Managed header,
then a single `managed_session_snapshot_base_v1` record and future transactions.
The base has its own UUID, is a known non-conversation control record, and is
excluded from logical history. Its manifest ref authenticates the complete
prefix. Event and logical header format versions remain 1; only the physical
header requires `managed-session/2`. Reader capability is separate from the
minimum-reader stamp used for new ordinary Sessions.
A common local writer acquisition/preparation entry serves the adapter,
authority and Config. Config first validates restore admission and ownership,
then converts and reacquires the final projection. Invalid restore requests
must not rewrite the main file. Static read-only access never converts. `SessionWriterLease.replaceManagedTranscript` performs
ownership-checked replacement; the existing legacy replacement method continues
to reject Managed leases. Local reads explicitly receive the runtime base,
Session key and optional frozen physical byte bound. The shared HTTP journal
interface gains no local-only option or hook.
The implementation is split between the local writer preparation, the native
snapshot replay wrapper and a filesystem/display codec. The shared
`managed-session-resource-references` module enumerates declared schema paths;
channel inputs include attachment refs while their opaque reply context and
ordinary user JSON remain data. The existing HTTP collector is unchanged.
## Conversion and durability
1. Authenticate the sealed predecessor's physical proof and logical commit
proof. Only a nonempty, fully committed v1 source without a base is eligible.
2. Install a schema-3 writer with local `format_version=2`, then durably sync its
lock directory before installing the new layout. New ordinary Sessions remain
format 1; converted Session acquisition, seal and reclaim never downgrade.
3. Validate schema-declared header/event references and nested resources through
their resource-kind parsers. Publish pages before the manifest. Arbitrary JSON
shaped like a ref is not a dependency; inconsistent refs for one ID fail.
4. Write and sync the temporary main file, recheck owner and source hash, rename,
sync the parent directory and repin inode, metadata and rolling hash.
5. Re-read the installed snapshot plus tail and verify the original sequence,
marker digest, last UUID and command-replay facts before admitting work.
A pre-rename failure leaves the old file and reports failure. An uncertain
replacement or failed post-install verification stops mutations and preserves
the writer fence. Abort/release cannot remove a format-2 barrier. A verified
failed open may restore its original logical seal. Missing pages, manifests or
dependencies are corruption, never an empty Session or a reason to fall back to
another head. Lock claim/primary transitions for format 2 are directory-synced
before claims protecting them are removed. Process-crash testing does not prove
power-loss durability or qualify an untested filesystem.
This integrity gate also applies to pre-existing sealed v1 Sessions selected for
conversion. Missing, truncated or digest-mismatched declared dependencies stop
resume before activation, even if an older reader tolerated that damage. The
original transcript remains unchanged and the format-2 barrier remains sealed;
there is no compatibility bypass or automatic pruning. Restoring the originally
referenced resource bytes allows a subsequent resume to retry the same history.
Static inspection of an unconverted v1 transcript keeps its previous read
semantics and never upgrades it. A journal handle's first read can prepare only
because that handle already owns the writer; it is a resume operation. Aborting
that writer seals its proof instead of deleting the upgraded barrier.
## Readers and compatibility
The logical scan includes all events, transactions and recovery facts.
Snapshot-aware reads verify the committed tail's complete resource closure as
well as the manifest census; uncommitted and frozen-boundary-excluded references
are not admitted or read.
`maxBytes`, `committedBytes`, `headerBytes` and `uncommittedBytes` continue to
describe the current physical file so tail repair truncates the correct bytes.
Read one verified file generation; frozen transcript cursors retain their
inode/length contract and return the existing 409 after replacement.
Authority, restore/history projections and failed-open cleanup use the same
snapshot-aware reader. Synchronous title/source reads scan the full committed
physical tail before using manifest anchors; this intentionally costs O(tail).
Read the controlled complete header/base rather than assuming they fit in the
64 KiB display window. Keep first-record project authorization and existing
Legacy-only refusal. Resources remain in place through archive/unarchive and are
removed through whole-Session resource cleanup.
The version barrier prevents old writer takeover and execution. Existing old
maintenance readers do not enforce that version ceiling; archive/delete must
remain safe rather than claiming every old operation is blocked. Read-only
compatibility failures must not be interpreted as Legacy or a new Session.
## Verification and acceptance
- Exact-byte, multipage recovery; original receipts and conflict rejection;
checkpoints, activation and nested resources; repeated resumes publish once.
- Frozen byte bounds, UTF-8, empty tail, torn-tail physical truncation and append
continuity; restored logical and physical seal proofs.
- Failures during barrier install, publication, write/fsync, rename and proof
update; independent-process crashes and old/new writer compatibility.
- Pagination during replacement, large headers, title/source updates buried
beyond display windows, archive/unarchive/delete and missing/corrupt/cross-scope
or symlinked dependencies.
- Focused Core regressions, root build/typecheck/bundle, formatting and lint,
two clean self-audit passes and independent review.
The global CLI capability check is only a baseline discovery step. If it exposes
no native snapshot operation, direct adapter/process tests exercise the applicable
contract. Executed results belong in the E2E report at the tested revision. This
design does not establish Linux/Windows, power-loss or production acceptance.
@@ -0,0 +1,100 @@
# 本地 Managed journal 快照恢复(PR4)
[English](2026-10-26-managed-local-journal-snapshots.md) | [简体中文](2026-10-26-managed-local-journal-snapshots.zh-CN.md)
## 问题与边界
Issue #13184 需要先建立原生恢复边界,才能淘汰 authority 状态。本地 journal 当前
读取完整 JSONL 文件,而 writer lease 同时认证物理 transcript 字节和逻辑提交证明。
适配器不能独自 rename 文件,否则会使 writer 的证明失效。
本片把完整已提交前缀移入不可变 Session 资源,将活跃文件替换为原 ownership/header、
一条快照 base 记录及连续的后续尾部。仅在正常 sealed 的本地 Session 下一次恢复时
转换一次,发生在最终恢复投影、activation 安装和 recorder 绑定之前。新 Session、运行中
Session 和只读请求不转换。未提交尾部沿用现有恢复流程,不纳入快照。
不包含自动轮换、垃圾回收、紧凑 authority 状态或内存淘汰。完整历史和 authority 索引
仍会增长。PR2/PR3 定义远端快照恢复;本地信封保存相同源事实,不伪造服务端 revision
或 writer generation。
## 表示与接口
本地 manifest 绑定格式、Session key、有序不可变 page refs、完整前缀长度与 SHA-256、
提交 sequence、commit digest、最后 record UUID、资源闭包及最新 title/source refs。
页面保存准确 JSONL 字节,以事务边界结束,每页最多 16 MiB。Manifest 与页面使用专用
resource kind 和内容派生 ID,位于既有 Session resource root。发布幂等且耐久,拒绝
已存在的冲突字节;既有资源大小限制仍适用。
物理文件保留原 ownership 首记录和 Managed header,随后是唯一的
`managed_session_snapshot_base_v1` 与后续事务。Base 使用独立 UUID,是已知的
non-conversation 控制记录,不进入逻辑历史;其 manifest ref 认证完整前缀。事件与
逻辑 header 格式仍为 1,仅物理 header 要求 `managed-session/2`。读取能力与新建
普通 Session 使用的最低读取版本盖章分离。
统一本地 writer 获取/准备入口服务适配器、authority 和 Config。Config 先验证已有恢复投影的准入和归属,再转换并重新获取最终投影;无效恢复请求不得改写主日志。只读静态读取不会执行转换。
`SessionWriterLease.replaceManagedTranscript` 执行带 owner 校验的替换,既有 Legacy
替换方法继续拒绝 Managed。本地读取显式接收 runtime base、Session key 和可选的冻结
物理字节边界。公共 HTTP journal 接口不增加本地专用选项或 hook。
实现分为本地 writer 准备、原生快照重放封装与文件系统/显示编解码器。
公共 `managed-session-resource-references` 模块枚举 schema 明确声明的路径;channel
输入包含附件 ref,其 opaque reply context 与普通用户 JSON 保持为数据。
既有 HTTP 收集器保持原有行为。
## 转换与耐久性
1. 认证 sealed 前任的物理证明及逻辑提交证明。仅非空、完整已提交且没有 base 的 v1
源日志具备转换资格。
2. 安装 schema 3、本地 `format_version=2` 的 writer,并在安装新布局前同步锁目录。
新建普通 Session 保持格式 1;转换后的获取、seal 和 reclaim 不得降级。
3. 验证 schema 声明的 header/event 引用,通过 resource kind 解析器展开深层依赖。
先发布页面,再发布 manifest。任意 JSON 外形不是依赖,同一 ID 的冲突引用应拒绝。
4. 写入并同步临时主文件,复核 owner 与源哈希,rename、同步父目录,重新固定 inode、
元数据和 rolling hash。
5. 重读安装后的快照与尾部,验证原 sequence、marker digest、最后 UUID 和命令重放
事实,再准入新工作。
Rename 前失败保留旧文件并返回错误;结果不确定或安装后验证失败时停止写入并保留围栏。
Abort/release 不得删除格式 2 barrier;经过验证的失败打开可以恢复原逻辑 seal。缺失的
页面、manifest 或依赖属于损坏,不能当空 Session,也不能回退另一个 head。格式 2 的
锁 claim/primary 转换必须在移除保护 claim 之前同步目录。进程崩溃测试不证明掉电耐久性,
也不代表未测试文件系统获得资格。
此完整性门禁同样适用于被选择转换的既有 sealed v1 Session。声明依赖缺失、截断或摘要
不一致时,即使旧读取器曾容忍这些损坏,也必须在 activation 前阻断恢复。原 transcript
保持不变,格式 2 barrier 保持 sealed;不提供兼容绕过或自动裁剪。恢复引用所指向的原始
资源字节后,下次恢复可以重试同一份历史。未转换 v1 transcript 的静态查看保留原读取语义,
且不触发升级。Journal handle 首次 read 可以准备转换,是因为该 handle 已持有 writer,
属于恢复操作;abort 会封存其证明,而不是删除已升级的 barrier。
## 读取与兼容性
逻辑 scan 包含完整事件、事务和恢复事实。快照读取同时校验 manifest 资源清单和已提交尾部
的完整资源闭包;未提交及冻结边界外的引用不会被准入或读取。`maxBytes`、`committedBytes`、`headerBytes`
和 `uncommittedBytes` 继续描述当前物理文件,确保尾部修复截断准确字节。一次读取只接受
同一已验证文件 generation;冻结游标保留 inode/长度契约,替换后返回既有 409。
Authority、恢复/历史投影和失败打开收尾共享快照读取器。同步 title/source 先扫描完整
已提交物理尾部,再使用 manifest 锚点,明确接受 O(tail) 成本。受控地读取完整 header/base,
不能假定它们位于 64 KiB 显示窗口内。保留首记录项目授权和既有 Legacy 专用入口拒绝。
Archive/unarchive 不移动资源,永久删除沿用整棵 Session 资源目录清理。
版本 barrier 防止旧 writer 接管与执行。既有旧维护读取器没有版本上限,因此应保证
archive/delete 安全,不能声称所有旧操作都被阻断。只读兼容失败不得被识别为 Legacy
或新 Session。
## 验证与验收
- 准确字节和多页恢复、原 receipt 与冲突拒绝、checkpoint/activation/嵌套资源、重复
恢复仅发布一次。
- 冻结字节边界、UTF-8、空尾部、未提交尾部物理截断、追加连续性,以及逻辑/物理 seal
证明。
- Barrier 安装、发布、write/fsync、rename、证明更新故障,以及独立进程崩溃和新旧 writer
兼容。
- 替换时分页、大 header、埋在显示窗口外的 title/source 更新、archive/unarchive/delete,
及缺失、损坏、跨 scope、symlink 依赖。
- 聚焦 Core 回归、根 build/typecheck/bundle、格式与 lint、两轮干净自审及独立审查。
全局 CLI 能力检查只用于发现基线。没有原生快照操作入口时,通过适配器/进程测试覆盖
实际契约。已执行结果以测试 revision 记入 E2E 报告。本设计不代表 Linux/Windows、
掉电或生产验收。
@@ -10,10 +10,17 @@ import { tmpdir } from 'node:os';
import { createHash } from 'node:crypto';
import { createServer } from 'node:http';
import path from 'node:path';
import { LocalJsonlManagedSessionJournalStore } from '@qwen-code/qwen-code-core/managed-runtime/local-jsonl-managed-session-journal-store.js';
import {
LocalJsonlManagedSessionJournalHandle,
LocalJsonlManagedSessionJournalStore,
} from '@qwen-code/qwen-code-core/managed-runtime/local-jsonl-managed-session-journal-store.js';
import { SessionWriterLease } from '@qwen-code/qwen-code-core/services/session-writer-lease.js';
import { LocalManagedSessionResourceStore } from '@qwen-code/qwen-code-core/managed-runtime/managed-session-resources.js';
import type { ManagedSessionDurableRef } from '@qwen-code/qwen-code-core/managed-runtime/managed-session-records.js';
import type { ManagedSessionJournalHandle } from '@qwen-code/qwen-code-core/managed-runtime/managed-session-storage.js';
import type {
ManagedSessionJournalHandle,
ManagedSessionJournalStore,
} from '@qwen-code/qwen-code-core/managed-runtime/managed-session-storage.js';
import { HOSTED_STORE_PREFIX } from '../scratch-dir.js';
// An HTTP transport fixture, not a substitute for the Spring/MySQL slice.
@@ -25,11 +32,21 @@ export async function startHostedSessionStore(sessionId: string) {
sessionId,
};
const transcriptPath = path.join(root, `${sessionId}.jsonl`);
const journal = new LocalJsonlManagedSessionJournalStore({
runtimeBaseDir: root,
sessionId,
transcriptPath,
});
// This transport fixture retains the HTTP journal's native transaction shape.
const journal: ManagedSessionJournalStore = {
open: async (request) =>
new LocalJsonlManagedSessionJournalHandle(
await SessionWriterLease.acquire({
runtimeBaseDir: root,
sessionId,
transcriptPath,
takeoverPolicy: 'certified',
lockSchema: { schemaVersion: 3, formatVersion: 1 },
}),
request.sessionKey,
true,
),
};
const resources = LocalManagedSessionResourceStore.create({
runtimeBaseDir: root,
sessionKey,
@@ -44,7 +61,11 @@ export async function startHostedSessionStore(sessionId: string) {
let writerGeneration = 0;
let seals = 0;
const scan = () =>
LocalJsonlManagedSessionJournalStore.read(transcriptPath, sessionKey);
LocalJsonlManagedSessionJournalStore.read({
transcriptPath,
sessionKey,
runtimeBaseDir: root,
});
const server = createServer((req, res) => {
res.setHeader('Cache-Control', 'no-store');
const request = (async () => {
File diff suppressed because it is too large. Load diff
@@ -328,10 +328,11 @@ it('records a durable Shell receipt when cancellation recovery settles a produce
expect(f.cancelled()).toBe(true);
expect(requests.filter((r) => r.endsWith(':cancel'))).toHaveLength(1);
expect(requests.some((r) => /:start|:prepare/.test(r))).toBe(false);
const journal = await LocalJsonlManagedSessionJournalStore.read(
const journal = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath,
key,
);
runtimeBaseDir: root!,
sessionKey: key,
});
const receipts = journal.events.filter(
(event) =>
event.kind === 'tool.receipt' &&
@@ -401,10 +402,11 @@ it.each([
response: { executionStatus: 'cancelled' },
},
});
const journal = await LocalJsonlManagedSessionJournalStore.read(
f.transcriptPath,
f.key,
);
const journal = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath: f.transcriptPath,
runtimeBaseDir: root!,
sessionKey: f.key,
});
expect(
journal.events.filter((event) => event.kind === 'tool.receipt'),
).toHaveLength(0);
@@ -414,10 +416,11 @@ it.each([
if (repeated.status === 'runnable') {
expect(repeated.checkpoint.tools!.items[0].outcomeRef).toEqual(ref);
}
const repeatedJournal = await LocalJsonlManagedSessionJournalStore.read(
f.transcriptPath,
f.key,
);
const repeatedJournal = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath: f.transcriptPath,
runtimeBaseDir: root!,
sessionKey: f.key,
});
expect(
repeatedJournal.events.filter((event) => event.kind === 'tool.receipt'),
).toHaveLength(0);
@@ -451,10 +454,11 @@ it.each([
'in_progress',
);
}
const journal = await LocalJsonlManagedSessionJournalStore.read(
f.transcriptPath,
f.key,
);
const journal = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath: f.transcriptPath,
runtimeBaseDir: root!,
sessionKey: f.key,
});
expect(
journal.events.filter((event) => event.kind === 'tool.receipt'),
).toHaveLength(0);
+32 -4
View File
@@ -362,6 +362,10 @@ import {
ManagedSessionUncommittedTailError,
} from '../managed-runtime/managed-session-authority.js';
import { LocalJsonlManagedSessionJournalStore } from '../managed-runtime/local-jsonl-managed-session-journal-store.js';
import {
acquireLocalManagedSessionWriter,
prepareLocalManagedSessionWriter,
} from '../managed-runtime/local-managed-session-writer.js';
import { LocalManagedSessionResourceStore } from '../managed-runtime/managed-session-resources.js';
import {
LocalManagedRuntimeOutcomes,
@@ -4998,10 +5002,14 @@ export class Config {
ReturnType<typeof LocalJsonlManagedSessionJournalStore.read>
>;
try {
scan = await LocalJsonlManagedSessionJournalStore.read(
scan = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath,
localManagedSessionKey(this.storage.getProjectRoot(), this.sessionId),
);
runtimeBaseDir: this.sessionRuntimeBaseDir,
sessionKey: localManagedSessionKey(
this.storage.getProjectRoot(),
this.sessionId,
),
});
} catch {
return;
}
@@ -5079,7 +5087,10 @@ export class Config {
if (!recorder) throw new SessionWriterUnavailableError();
let lease: SessionWriterLease | undefined;
try {
lease = await SessionWriterLease.acquire({
const acquire = managed
? acquireLocalManagedSessionWriter
: SessionWriterLease.acquire;
lease = await acquire({
runtimeBaseDir: this.sessionRuntimeBaseDir,
sessionId: this.sessionId,
transcriptPath: this.getTranscriptPath(),
@@ -5139,6 +5150,23 @@ export class Config {
'managed',
);
}
if (
managed &&
(await prepareLocalManagedSessionWriter(
lease,
localManagedSessionKey(
this.storage.getProjectRoot(),
this.sessionId,
),
))
) {
projection = await this.sessionRestoreProjectionSource();
assertSessionExecutionEngine(
projection?.executionEngine,
this.sessionId,
'managed',
);
}
this.setSessionRestoreProjection(projection);
} else if (
managed &&
@@ -231,10 +231,11 @@ describe('Managed Session log recording', () => {
MANAGED_SESSION_COMMIT_SUBTYPE,
]).toContain(record['subtype']);
}
const scan = await LocalJsonlManagedSessionJournalStore.read(
sessionService().getSessionTranscriptPath(SESSION_ID),
localManagedSessionKey(projectDir, SESSION_ID),
);
const scan = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath: sessionService().getSessionTranscriptPath(SESSION_ID),
runtimeBaseDir: runtimeDir,
sessionKey: localManagedSessionKey(projectDir, SESSION_ID),
});
expect(scan.uncommitted).toBe(0);
// The session is the activation's worker, with a five-minute horizon.
expect(scan.activation?.workerId).toBe(SESSION_ID);
@@ -290,10 +291,11 @@ describe('Managed Session log recording', () => {
const sealed = await lockRecord();
expect(sealed).toMatchObject({ state: 'sealed', schema_version: 3 });
// The log records that the session stopped advancing before the seal.
const scan = await LocalJsonlManagedSessionJournalStore.read(
sessionService().getSessionTranscriptPath(SESSION_ID),
localManagedSessionKey(projectDir, SESSION_ID),
);
const scan = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath: sessionService().getSessionTranscriptPath(SESSION_ID),
runtimeBaseDir: runtimeDir,
sessionKey: localManagedSessionKey(projectDir, SESSION_ID),
});
expect(scan.activation?.phase).toBe('released');
expect(scan.uncommitted).toBe(0);
const bytesAtClose = (
@@ -643,11 +645,18 @@ describe('Managed Session log recording', () => {
false,
);
await recorder.flush();
const raw = await readFile(
sessionService().getSessionTranscriptPath(SESSION_ID),
'utf8',
);
expect(raw.split('"domain":"session_source"')).toHaveLength(2);
const journal = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath: sessionService().getSessionTranscriptPath(SESSION_ID),
runtimeBaseDir: runtimeDir,
sessionKey: localManagedSessionKey(projectDir, SESSION_ID),
});
expect(
journal.events.filter(
(event) =>
event.kind === 'domain.committed' &&
event.payload['domain'] === 'session_source',
),
).toHaveLength(1);
await restored.closeSessionWriter();
});
@@ -801,10 +810,11 @@ describe('Managed Session log recording', () => {
});
// The failed anchor does not skip the stop: the log records that the
// session stopped advancing before the seal.
const scan = await LocalJsonlManagedSessionJournalStore.read(
sessionService().getSessionTranscriptPath(SESSION_ID),
localManagedSessionKey(projectDir, SESSION_ID),
);
const scan = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath: sessionService().getSessionTranscriptPath(SESSION_ID),
runtimeBaseDir: runtimeDir,
sessionKey: localManagedSessionKey(projectDir, SESSION_ID),
});
expect(scan.activation?.phase).toBe('released');
const restored = await start(restoringConfig());
expect(await activeChatTexts(restored)).toEqual(['committed prompt']);
@@ -1404,7 +1414,7 @@ describe('Managed Session log recording', () => {
await restored.closeSessionWriter();
});
it('restores a session whose title body is missing', async () => {
it('refuses snapshot conversion when the title dependency is missing', async () => {
const first = await start(managedConfig());
await first
.getChatRecordingService()!
@@ -1425,12 +1435,11 @@ describe('Managed Session log recording', () => {
replay: { kind: 'none' },
}),
).resolves.toBeDefined();
const restored = await start(restoringConfig());
expect(
restored.getChatRecordingService()!.getCurrentCustomTitle(),
).toBeUndefined();
expect(await activeChatTexts(restored)).toEqual(['first prompt']);
await restored.closeSessionWriter();
await expect(start(restoringConfig())).rejects.toThrow(/not present/);
expect(await lockRecord()).toMatchObject({
state: 'sealed',
format_version: 2,
});
});
it('gives back the seal it took over when the log does not match it', async () => {
@@ -1584,13 +1593,14 @@ describe('Managed Session log recording', () => {
}
expect(zeroed).toBeGreaterThan(0);
// The restore repair leaves the unreadable checkpoint to the gate: the
// session opens blocked rather than failing to open at all.
const restored = await start(restoringConfig());
const block = restored.getManagedSessionBlock();
expect(block).toBeInstanceOf(ManagedRuntimeOutcomeUnknownError);
expect(block?.message).toContain('cannot be read');
await restored.closeSessionWriter();
// Conversion refuses an incomplete resource closure before admitting work.
await expect(start(restoringConfig())).rejects.toThrow(
/on disk where|digest|not present/,
);
expect(await lockRecord()).toMatchObject({
state: 'sealed',
format_version: 2,
});
});
it('restores a log whose Runtime calls settled, without blocking', async () => {
@@ -5,14 +5,14 @@
*/
import { readFile, rename, unlink, writeFile } from 'node:fs/promises';
import { SessionWriterLease } from '../services/session-writer-lease.js';
import type { SessionWriterLease } from '../services/session-writer-lease.js';
import {
ManagedSessionRecordError,
MANAGED_SESSION_FORMAT_VERSION,
} from './managed-session-records.js';
acquireLocalManagedSessionWriter,
prepareLocalManagedSessionWriter,
} from './local-managed-session-writer.js';
import { readLocalManagedJournal } from './local-managed-journal-snapshot.js';
import { ManagedSessionRecordError } from './managed-session-records.js';
import {
emptyManagedSessionJournalScan,
scanManagedSessionJournal,
type ManagedSessionCommitProof,
type ManagedSessionJournalHandle,
type ManagedSessionJournalScan,
@@ -25,6 +25,7 @@ interface LocalJsonlManagedSessionJournalStoreOptions {
readonly sessionId: string;
readonly transcriptPath: string;
readonly lease?: SessionWriterLease;
readonly requireNew?: boolean;
}
/** Local-file implementation retained for standalone CLI and development. */
@@ -41,23 +42,16 @@ export class LocalJsonlManagedSessionJournalStore
const adopted = this.options.lease !== undefined;
const lease =
this.options.lease ??
(await SessionWriterLease.acquire({
(await acquireLocalManagedSessionWriter({
runtimeBaseDir: this.options.runtimeBaseDir,
sessionId: this.options.sessionId,
transcriptPath: this.options.transcriptPath,
takeoverPolicy: 'certified',
// A Managed writer pins its log format into the lock: baseline
// binaries refuse the unknown schema, which is the write barrier
// that keeps this log off legacy writers.
lockSchema: {
schemaVersion: 3,
formatVersion: MANAGED_SESSION_FORMAT_VERSION,
},
}));
return new LocalJsonlManagedSessionJournalHandle(
lease,
request.sessionKey,
!adopted,
this.options.requireNew !== true,
);
}
@@ -68,20 +62,13 @@ export class LocalJsonlManagedSessionJournalStore
return new LocalJsonlManagedSessionJournalHandle(lease, sessionKey, false);
}
static async read(
transcriptPath: string,
sessionKey: ManagedSessionKey,
maxBytes?: number,
): Promise<ManagedSessionJournalScan> {
try {
const bytes = await readFile(transcriptPath);
return scanManagedSessionJournal(bytes, sessionKey, maxBytes);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
return emptyManagedSessionJournalScan();
}
throw error;
}
static async read(options: {
readonly transcriptPath: string;
readonly runtimeBaseDir: string;
readonly sessionKey: ManagedSessionKey;
readonly maxBytes?: number;
}): Promise<ManagedSessionJournalScan> {
return readLocalManagedJournal(options);
}
}
@@ -92,18 +79,30 @@ export class LocalJsonlManagedSessionJournalHandle
private readonly lease: SessionWriterLease,
readonly sessionKey: ManagedSessionKey,
private readonly ownsLease: boolean,
private readonly prepareOnRead = true,
) {}
read(
private preparation: Promise<boolean> | undefined;
async read(
options: {
readonly maxBytes?: number;
} = {},
): Promise<ManagedSessionJournalScan> {
return LocalJsonlManagedSessionJournalStore.read(
this.lease.transcriptPath,
this.sessionKey,
options.maxBytes,
);
if (this.prepareOnRead) {
// This handle already owns the writer; static inspection never prepares.
this.preparation ??= prepareLocalManagedSessionWriter(
this.lease,
this.sessionKey,
);
await this.preparation;
}
return LocalJsonlManagedSessionJournalStore.read({
transcriptPath: this.lease.transcriptPath,
runtimeBaseDir: this.lease.runtimeBaseDir,
sessionKey: this.sessionKey,
...options,
});
}
async appendTransaction(records: readonly unknown[]): Promise<void> {
@@ -133,9 +132,14 @@ export class LocalJsonlManagedSessionJournalHandle
}
async abort(): Promise<void> {
if (this.ownsLease) {
await this.lease.release();
if (!this.ownsLease) return;
if ((this.lease.managedJournalFormatVersion ?? 0) >= 2) {
// Retain the format barrier so an older writer cannot acquire the Session.
const proof = this.lease.takeoverCommitProof;
if (proof) await this.lease.sealForHandoff(proof);
return;
}
await this.lease.release();
}
async recoverUncommittedTail(): Promise<{
@@ -0,0 +1,887 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { createHash, randomUUID } from 'node:crypto';
import * as fs from 'node:fs/promises';
import fsSync from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { afterEach, describe, expect, it, vi } from 'vitest';
import {
getSessionWriterLockPath,
SessionWriterLease,
} from '../services/session-writer-lease.js';
import { Storage } from '../config/storage.js';
import { SessionService } from '../services/sessionService.js';
import {
localManagedSessionKey,
isManagedSessionTranscriptSync,
isManagedExecutionTranscriptSync,
readManagedSessionTitleInfoSync,
} from '../utils/sessionStorageUtils.js';
import { LocalJsonlManagedSessionJournalStore } from './local-jsonl-managed-session-journal-store.js';
import { LocalManagedSessionAuthority } from './managed-session-authority.js';
import { openManagedSession } from './managed-session-assembly.js';
import { LocalManagedSessionResourceStore } from './managed-session-resources.js';
import { encodeChannelInputEnvelope } from './managed-channel-operations.js';
import {
LOCAL_MANAGED_SNAPSHOT_BASE,
LOCAL_MANAGED_SNAPSHOT_KIND,
LOCAL_MANAGED_SNAPSHOT_PAGE_KIND,
parseLocalJournalBase,
readLocalJournalBytes,
readLocalJournalManifest,
} from './local-managed-journal-snapshot.js';
vi.mock('node:fs/promises', async (importOriginal) => ({
...(await importOriginal<typeof import('node:fs/promises')>()),
}));
const roots: string[] = [];
afterEach(async () => {
vi.restoreAllMocks();
await Promise.all(
roots
.splice(0)
.map((root) => fs.rm(root, { recursive: true, force: true })),
);
});
const digest = (text: string) =>
createHash('sha256').update(text).digest('hex');
async function fixture(ownershipPadding = 0) {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-local-snapshot-'));
roots.push(root);
const runtimeBaseDir = path.join(root, 'runtime');
const project = path.join(root, 'project');
await fs.mkdir(project, { recursive: true });
const sessionId = randomUUID();
const sessionKey = localManagedSessionKey(project, sessionId);
const transcriptPath = path.join(
new Storage(project, runtimeBaseDir).getProjectDir(),
'chats',
`${sessionId}.jsonl`,
);
await fs.mkdir(path.dirname(transcriptPath), { recursive: true });
if (ownershipPadding) {
await fs.writeFile(
transcriptPath,
`${JSON.stringify({
version: 'x'.repeat(ownershipPadding),
uuid: randomUUID(),
parentUuid: null,
sessionId,
type: 'system',
subtype: 'session_execution_engine',
cwd: project,
systemPayload: { version: 1, engine: 'managed' },
})}\n`,
);
}
const options = { runtimeBaseDir, transcriptPath, sessionId };
const resources = LocalManagedSessionResourceStore.create({
runtimeBaseDir,
sessionKey,
});
const create = {
definitionRef: await resources.publish(
'managed-session-definition',
Buffer.from('{}'),
),
rootSnapshotRef: await resources.publish(
'managed-session-root-snapshot',
Buffer.from('{}'),
),
createdBy: 'snapshot-test',
};
const input = {
inputId: 'input-1',
turnId: 'turn-1',
source: 'web_shell',
contentRef: await resources.publish(
'managed-test',
Buffer.from('中文 input'),
),
admissionRef: await resources.publish(
'managed-test',
Buffer.from('admission'),
),
deadline: null,
wakeReason: 'input',
};
const command = (id: string) => ({
sessionKey,
commandId: id,
operation: 'submitInput',
contentDigest: digest(id),
});
const open = async (isNew = false) => {
const journal = await new LocalJsonlManagedSessionJournalStore(
options,
).open({ sessionKey });
const authority = await LocalManagedSessionAuthority.open({
journal,
sessionKey,
resources,
cwd: project,
version: 'test',
...(isNew ? { create } : {}),
...(journal.takeoverCommitProof
? { expectedCommitProof: journal.takeoverCommitProof }
: {}),
});
return { journal, authority };
};
const first = await open(true);
const receipt = await first.authority.submitInput(command('input'), input);
return {
...options,
root,
project,
sessionKey,
resources,
first,
open,
input,
command,
receipt,
};
}
async function snapshot(f: Awaited<ReturnType<typeof fixture>>) {
const bytes = await fs.readFile(f.transcriptPath);
const base = parseLocalJournalBase(bytes, f.sessionKey)!;
expect(base).toBeDefined();
return {
bytes,
base,
manifest: readLocalJournalManifest(
f.runtimeBaseDir,
f.sessionKey,
base.manifest,
),
};
}
async function countSnapshots(f: Awaited<ReturnType<typeof fixture>>) {
return (
await fs.readdir(
path.join(f.resources.sessionRoot, LOCAL_MANAGED_SNAPSHOT_KIND),
)
).length;
}
describe('local native journal snapshots', () => {
it('converts only sealed resume, preserves original bytes and command replay, then never rotates', async () => {
const f = await fixture();
const original = await fs.readFile(f.transcriptPath);
expect(original.includes(LOCAL_MANAGED_SNAPSHOT_BASE)).toBe(false);
await f.first.authority.close();
await LocalJsonlManagedSessionJournalStore.read({
...f,
sessionKey: f.sessionKey,
});
expect(await fs.readFile(f.transcriptPath)).toEqual(original);
const resumed = await f.open();
const { bytes, manifest } = await snapshot(f);
expect(bytes.includes('managed-session/2')).toBe(true);
expect(bytes.equals(original)).toBe(false);
expect(
Buffer.concat(
await Promise.all(manifest.pages.map((ref) => f.resources.read(ref))),
),
).toEqual(original);
expect(
(await resumed.authority.submitInput(f.command('input'), f.input))
.transactionId,
).toBe(f.receipt.transactionId);
await expect(
resumed.authority.submitInput(
{ ...f.command('input'), contentDigest: digest('conflict') },
f.input,
),
).rejects.toThrow(/different content/);
await resumed.authority.submitInput(f.command('tail'), {
...f.input,
inputId: 'tail',
turnId: 'tail',
});
const proof = resumed.authority.commitProof;
await resumed.authority.close();
const second = await f.open();
expect(second.authority.commitProof).toEqual(proof);
expect(await countSnapshots(f)).toBe(1);
await second.authority.close();
});
it('keeps frozen offsets physical and truncates only an uncommitted tail', async () => {
const f = await fixture();
await f.first.authority.close();
const resumed = await f.open();
const frozen = (await fs.stat(f.transcriptPath)).size;
await resumed.authority.submitInput(f.command('tail'), {
...f.input,
inputId: 'tail',
turnId: 'tail',
});
const historical = await LocalJsonlManagedSessionJournalStore.read({
...f,
maxBytes: frozen,
});
expect(historical.committed).toBe(f.receipt.committedSequence);
expect(historical.committedBytes).toBe(frozen);
const before = await fs.readFile(f.transcriptPath);
const committed = await resumed.journal.read();
const event = {
...committed.events[0],
sequence: committed.committed + 1,
eventId: 'uncommitted',
payload: {
...committed.events[0].payload,
contentRef: {
...f.input.contentRef,
resourceId: 'uncommitted-missing',
},
},
};
await resumed.journal.appendTransaction([
{
uuid: randomUUID(),
subtype: 'managed_session_event_v1',
managedSession: event,
},
]);
expect((await resumed.journal.read()).uncommitted).toBe(1);
await resumed.journal.recoverUncommittedTail();
expect(await fs.readFile(f.transcriptPath)).toEqual(before);
await resumed.authority.close();
});
it('keeps a frozen prefix readable when a writer appends during the physical read', async () => {
const f = await fixture();
await f.first.authority.close();
const original = await fs.readFile(f.transcriptPath);
const readSync = fsSync.readSync;
const spy = vi
.spyOn(fsSync, 'readSync')
.mockImplementationOnce((...args) => {
const count = Reflect.apply(readSync, fsSync, args) as number;
fsSync.appendFileSync(
f.transcriptPath,
'a concurrent uncommitted tail',
);
return count;
});
try {
const frozen = await LocalJsonlManagedSessionJournalStore.read({
...f,
maxBytes: original.length,
});
expect(frozen.committed).toBe(f.receipt.committedSequence);
expect(frozen.uncommitted).toBe(0);
} finally {
spy.mockRestore();
}
});
it('refuses a frozen prefix rewrite even when the file also grows', async () => {
const f = await fixture();
await f.first.authority.close();
const original = await fs.readFile(f.transcriptPath);
const readSync = fsSync.readSync;
const spy = vi
.spyOn(fsSync, 'readSync')
.mockImplementationOnce((...args) => {
const count = Reflect.apply(readSync, fsSync, args) as number;
const modified = Buffer.from(original);
modified[0] ^= 1;
fsSync.writeFileSync(
f.transcriptPath,
Buffer.concat([modified, Buffer.from('a concurrent tail')]),
);
return count;
});
try {
expect(() =>
readLocalJournalBytes(f.transcriptPath, original.length),
).toThrow(/file changed while reading/);
} finally {
spy.mockRestore();
}
});
it.each(['manifest', 'page', 'dependency'] as const)(
'refuses a missing %s without treating the Session as empty',
async (kind) => {
const f = await fixture();
await f.first.authority.close();
const resumed = await f.open();
const { base, manifest } = await snapshot(f);
const ref =
kind === 'manifest'
? base.manifest
: kind === 'page'
? manifest.pages[0]
: manifest.resources[0];
await fs.unlink(
path.join(f.resources.sessionRoot, ref.kind, ref.resourceId),
);
await expect(resumed.journal.read()).rejects.toThrow();
await resumed.authority.close();
const lock = JSON.parse(
await fs.readFile(
getSessionWriterLockPath(f.runtimeBaseDir, f.sessionId),
'utf8',
),
);
expect(lock).toMatchObject({ state: 'sealed', format_version: 2 });
},
);
it.each(
(['definition', 'root', 'input', 'nested'] as const).flatMap((dependency) =>
(['missing', 'corrupt'] as const).map((damage) => ({
dependency,
damage,
})),
),
)(
'blocks an existing sealed v1 Session with a $damage $dependency before activation',
async ({ dependency, damage }) => {
const f = await fixture();
let nestedRef = f.input.contentRef;
if (dependency === 'nested') {
const attachment = await f.resources.publish(
'managed-channel-attachment',
Buffer.from('nested attachment'),
);
nestedRef = attachment;
const contentRef = await f.resources.publish(
'managed-input',
encodeChannelInputEnvelope({
channelInstanceId: 'channel',
accountId: 'account',
accountGeneration: 1,
platformEventId: 'event',
semanticRevision: 1,
routeId: 'route',
routeRevision: 1,
senderId: 'sender',
chatId: null,
threadId: null,
subject: null,
text: 'hello',
attachments: [
{
fileName: 'input.txt',
mimeType: 'text/plain',
byteLength: attachment.byteLength,
digest: attachment.digest,
ref: attachment,
},
],
replyContext: {},
}),
);
await f.first.authority.submitInput(f.command('channel'), {
...f.input,
inputId: 'channel',
turnId: 'channel',
source: 'channel',
contentRef,
});
}
const ref =
dependency === 'definition'
? f.first.authority.sessionHeader.definitionRef
: dependency === 'root'
? f.first.authority.sessionHeader.rootSnapshotRef
: dependency === 'nested'
? nestedRef
: f.input.contentRef;
await f.first.authority.close();
const before = await fs.readFile(f.transcriptPath);
expect(before.includes('managed-session/1')).toBe(true);
expect(before.includes(LOCAL_MANAGED_SNAPSHOT_BASE)).toBe(false);
const resourcePath = path.join(
f.resources.sessionRoot,
ref.kind,
ref.resourceId,
);
const original = await fs.readFile(resourcePath);
if (damage === 'missing') await fs.unlink(resourcePath);
else await fs.writeFile(resourcePath, Buffer.alloc(0));
await LocalJsonlManagedSessionJournalStore.read(f);
expect(await fs.readFile(f.transcriptPath)).toEqual(before);
const options = {
...f,
cwd: f.project,
version: 'test',
workerId: 'test',
activationLeaseDurationMs: 60_000,
};
await expect(openManagedSession(options)).rejects.toThrow(/resource/);
expect(await fs.readFile(f.transcriptPath)).toEqual(before);
expect(
(await LocalJsonlManagedSessionJournalStore.read(f)).events.some(
(event) => event.kind === 'activation.changed',
),
).toBe(false);
expect(
JSON.parse(
await fs.readFile(
getSessionWriterLockPath(f.runtimeBaseDir, f.sessionId),
'utf8',
),
),
).toMatchObject({ state: 'sealed', format_version: 2 });
for (const kind of [
LOCAL_MANAGED_SNAPSHOT_KIND,
LOCAL_MANAGED_SNAPSHOT_PAGE_KIND,
]) {
await expect(
fs.access(path.join(f.resources.sessionRoot, kind)),
).rejects.toMatchObject({ code: 'ENOENT' });
}
await fs.writeFile(resourcePath, original);
const resumed = await openManagedSession(options);
expect(
(await resumed.authority.submitInput(f.command('input'), f.input))
.transactionId,
).toBe(f.receipt.transactionId);
await resumed.close();
},
);
it('refuses cross-scope reads and a symlinked page', async () => {
const f = await fixture();
await f.first.authority.close();
const resumed = await f.open();
await expect(
LocalJsonlManagedSessionJournalStore.read({
...f,
sessionKey: { ...f.sessionKey, workspaceId: 'other' },
}),
).rejects.toThrow(/scope|different session/);
const { manifest } = await snapshot(f);
const page = manifest.pages[0];
const file = path.join(f.resources.sessionRoot, page.kind, page.resourceId);
const outside = path.join(f.root, 'outside');
await fs.copyFile(file, outside);
await fs.unlink(file);
await fs.symlink(outside, file);
await expect(resumed.journal.read()).rejects.toThrow();
await resumed.authority.close();
});
it('refuses a missing committed-tail dependency before a converted Session resumes', async () => {
const f = await fixture();
await f.first.authority.close();
const resumed = await f.open();
const frozen = (await fs.stat(f.transcriptPath)).size;
const contentRef = await f.resources.publish(
'managed-input',
Buffer.from(JSON.stringify({ text: 'tail input' })),
);
await resumed.authority.submitInput(f.command('tail-resource'), {
...f.input,
inputId: 'tail-resource',
turnId: 'tail-resource',
contentRef,
});
const { manifest } = await snapshot(f);
expect(manifest.resources).not.toContainEqual(contentRef);
await resumed.authority.close();
const before = await fs.readFile(f.transcriptPath);
await fs.unlink(
path.join(
f.resources.sessionRoot,
contentRef.kind,
contentRef.resourceId,
),
);
expect(
(
await LocalJsonlManagedSessionJournalStore.read({
...f,
maxBytes: frozen,
})
).committed,
).toBe(f.receipt.committedSequence);
await expect(
LocalJsonlManagedSessionJournalStore.read(f),
).rejects.toThrow();
await expect(
openManagedSession({
...f,
cwd: f.project,
version: 'test',
workerId: 'test',
activationLeaseDurationMs: 60_000,
}),
).rejects.toThrow();
expect(await fs.readFile(f.transcriptPath)).toEqual(before);
expect(
JSON.parse(
await fs.readFile(
getSessionWriterLockPath(f.runtimeBaseDir, f.sessionId),
'utf8',
),
),
).toMatchObject({ state: 'sealed', format_version: 2 });
});
it('refuses a different writer scope before changing the physical layout', async () => {
const f = await fixture();
await f.first.authority.close();
const before = await fs.readFile(f.transcriptPath);
const journal = await new LocalJsonlManagedSessionJournalStore(f).open({
sessionKey: { ...f.sessionKey, workspaceId: 'other' },
});
await expect(journal.read()).rejects.toThrow(/different Session/);
expect(await fs.readFile(f.transcriptPath)).toEqual(before);
await journal.abort();
});
it('never downgrades a converted file while reclaiming a lower-format stale lock', async () => {
const f = await fixture();
await f.first.authority.close();
const converted = await f.open();
await converted.authority.close();
const lockPath = getSessionWriterLockPath(f.runtimeBaseDir, f.sessionId);
const sealed = JSON.parse(await fs.readFile(lockPath, 'utf8')) as Record<
string,
unknown
>;
const active = Object.fromEntries(
Object.entries(sealed).filter(
([key]) =>
![
'transcript',
'sealed_at',
'last_commit_sequence',
'committed_prefix_hash',
].includes(key),
),
);
await fs.writeFile(
lockPath,
JSON.stringify({
...active,
state: 'active',
format_version: 1,
pid: 999_999,
}),
);
const resumed = await f.open();
expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({
state: 'active',
format_version: 2,
});
await resumed.authority.close();
expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({
state: 'sealed',
format_version: 2,
});
await expect(
SessionWriterLease.acquire({
...f,
takeoverPolicy: 'certified',
lockSchema: { schemaVersion: 3, formatVersion: 1 },
}),
).rejects.toThrow();
});
it('preserves the old file and a sealed barrier when publication fails', async () => {
const f = await fixture();
await f.first.authority.close();
const before = await fs.readFile(f.transcriptPath);
const publish = vi
.spyOn(
LocalManagedSessionResourceStore.prototype,
'publishJournalSnapshot',
)
.mockRejectedValueOnce(new Error('publish failed'));
await expect(f.open()).rejects.toThrow('publish failed');
expect(await fs.readFile(f.transcriptPath)).toEqual(before);
expect(
JSON.parse(
await fs.readFile(
getSessionWriterLockPath(f.runtimeBaseDir, f.sessionId),
'utf8',
),
),
).toMatchObject({ state: 'sealed', format_version: 2 });
publish.mockRestore();
const resumed = await f.open();
await resumed.authority.close();
});
it('preserves a fence after an uncertain rename and rejects old writer takeover', async () => {
const f = await fixture();
await f.first.authority.close();
const rename = fs.rename;
vi.spyOn(fs, 'rename').mockImplementation(async (from, to) => {
await rename(from, to);
if (to === f.transcriptPath) throw new Error('rename outcome unknown');
});
await expect(f.open()).rejects.toThrow();
expect(
(await fs.readFile(f.transcriptPath)).includes(
LOCAL_MANAGED_SNAPSHOT_BASE,
),
).toBe(true);
const lock = JSON.parse(
await fs.readFile(
getSessionWriterLockPath(f.runtimeBaseDir, f.sessionId),
'utf8',
),
);
expect(lock).toMatchObject({ state: 'active', format_version: 2 });
await expect(
SessionWriterLease.acquire({
...f,
takeoverPolicy: 'certified',
lockSchema: { schemaVersion: 3, formatVersion: 1 },
}),
).rejects.toThrow();
});
it('returns the latest title even when its tail update is beyond the display window', async () => {
const f = await fixture();
const title = (authority: LocalManagedSessionAuthority, id: string) =>
authority.commitDomainRecord(
f.command(id),
{
domain: 'session_metadata',
content: { title: id, titleSource: 'manual' },
},
{ class: 'trusted_entry' },
);
await title(f.first.authority, 'before');
await f.first.authority.close();
const resumed = await f.open();
expect(
readManagedSessionTitleInfoSync(f.transcriptPath, f.runtimeBaseDir),
).toMatchObject({ title: 'before' });
await title(resumed.authority, 'after');
for (let index = 0; index < 3; index++) {
await resumed.authority.appendExecutionEvent(
f.command(`padding-${index}`),
(sequence) => ({
v: 1,
sequence,
eventId: `padding-${index}`,
sessionKey: f.sessionKey,
kind: 'cancel.requested',
occurredAt: 1,
payload: {
requestId: `padding-${index}`,
target: { padding: 'x'.repeat(32 * 1024) },
reason: 'test',
requestedBy: 'test',
},
}),
{ class: 'trusted_entry' },
);
}
expect(
readManagedSessionTitleInfoSync(f.transcriptPath, f.runtimeBaseDir),
).toMatchObject({ title: 'after' });
await resumed.authority.close();
});
it('classifies and reads a snapshot whose first control subtype is beyond the head window', async () => {
const f = await fixture(100_000);
await f.first.authority.commitDomainRecord(
f.command('title'),
{
domain: 'session_metadata',
content: { title: 'large owner', titleSource: 'manual' },
},
{ class: 'trusted_entry' },
);
await f.first.authority.close();
const resumed = await f.open();
const { bytes } = await snapshot(f);
expect(bytes.subarray(0, 65536).includes('session_execution_engine')).toBe(
false,
);
expect(isManagedSessionTranscriptSync(f.transcriptPath)).toBe(true);
expect(isManagedExecutionTranscriptSync(f.transcriptPath)).toBe(true);
const service = new SessionService(f.project, {
runtimeBaseDir: f.runtimeBaseDir,
});
expect(() => service.assertLegacySessionExecution(f.sessionId)).toThrow(
/cannot execute with legacy/,
);
await expect(
service.renameSession(f.sessionId, 'legacy rename'),
).rejects.toThrow(/rename must go through its session authority/);
expect(
readManagedSessionTitleInfoSync(f.transcriptPath, f.runtimeBaseDir),
).toMatchObject({ title: 'large owner' });
await resumed.authority.close();
await fs.writeFile(
f.transcriptPath,
bytes.subarray(0, bytes.indexOf(10) + 1),
);
expect(isManagedExecutionTranscriptSync(f.transcriptPath)).toBe(true);
const owner = bytes.subarray(0, bytes.indexOf(10));
await fs.writeFile(f.transcriptPath, owner);
expect(isManagedExecutionTranscriptSync(f.transcriptPath)).toBe(true);
await fs.writeFile(
f.transcriptPath,
Buffer.concat([
owner,
Buffer.from(
`${JSON.stringify({ uuid: 'glued', type: 'system', subtype: 'turn_result' })}\n`,
),
]),
);
expect(isManagedExecutionTranscriptSync(f.transcriptPath)).toBe(true);
expect(() => service.assertLegacySessionExecution(f.sessionId)).toThrow(
/cannot execute with legacy/,
);
});
it('pages large committed prefixes without guessing refs in arbitrary cancel JSON', async () => {
const f = await fixture();
for (let index = 0; index < 25; index++) {
await f.first.authority.appendExecutionEvent(
f.command(`large-${index}`),
(sequence) => ({
v: 1,
sequence,
eventId: `large-${index}`,
sessionKey: f.sessionKey,
kind: 'cancel.requested',
occurredAt: 1,
payload: {
requestId: `large-${index}`,
target: {
padding: 'x'.repeat(700_000),
resourceId: 'not-a-ref',
kind: 'opaque',
schemaVersion: 1,
byteLength: 4,
digest: 'a'.repeat(64),
},
reason: 'test',
requestedBy: 'test',
},
}),
{ class: 'trusted_entry' },
);
}
const proof = f.first.authority.commitProof;
await f.first.authority.close();
const resumed = await f.open();
const { manifest } = await snapshot(f);
expect(manifest.pages.length).toBeGreaterThan(1);
expect(
manifest.pages.every(
(ref) =>
ref.kind === LOCAL_MANAGED_SNAPSHOT_PAGE_KIND &&
ref.byteLength <= 16 * 1024 * 1024,
),
).toBe(true);
expect(
manifest.resources.some((ref) => ref.resourceId === 'not-a-ref'),
).toBe(false);
expect(resumed.authority.commitProof).toEqual(proof);
await resumed.authority.close();
}, 60_000);
it('preserves opaque input JSON without treating a ref-shaped object as a dependency', async () => {
const f = await fixture();
const contentRef = await f.resources.publish(
'managed-input',
Buffer.from(
JSON.stringify({
opaque: { ...f.input.contentRef, resourceId: 'user-json-example' },
}),
),
);
await f.first.authority.submitInput(f.command('opaque'), {
...f.input,
inputId: 'opaque',
turnId: 'opaque',
contentRef,
});
await f.first.authority.close();
const resumed = await f.open();
const { manifest } = await snapshot(f);
expect(manifest.resources).toContainEqual(contentRef);
expect(
manifest.resources.some((ref) => ref.resourceId === 'user-json-example'),
).toBe(false);
await resumed.authority.close();
});
it('retains channel attachments while leaving opaque reply context alone', async () => {
const f = await fixture();
const attachment = await f.resources.publish(
'managed-channel-attachment',
Buffer.from('attachment'),
);
const contentRef = await f.resources.publish(
'managed-input',
encodeChannelInputEnvelope({
channelInstanceId: 'channel',
accountId: 'account',
accountGeneration: 1,
platformEventId: 'event',
semanticRevision: 1,
routeId: 'route',
routeRevision: 1,
senderId: 'sender',
chatId: null,
threadId: null,
subject: null,
text: 'hello',
attachments: [
{
fileName: 'test.txt',
mimeType: 'text/plain',
byteLength: attachment.byteLength,
digest: attachment.digest,
ref: attachment,
},
],
replyContext: {
...f.input.contentRef,
resourceId: 'opaque-reply-context',
},
}),
);
await f.first.authority.submitInput(f.command('channel'), {
...f.input,
inputId: 'channel',
turnId: 'channel',
source: 'channel',
contentRef,
});
await f.first.authority.close();
const resumed = await f.open();
const { manifest } = await snapshot(f);
expect(manifest.resources).toContainEqual(attachment);
expect(
manifest.resources.some(
(ref) => ref.resourceId === 'opaque-reply-context',
),
).toBe(false);
await fs.unlink(
path.join(
f.resources.sessionRoot,
attachment.kind,
attachment.resourceId,
),
);
await expect(resumed.journal.read()).rejects.toThrow();
await resumed.authority.close();
});
});
@@ -0,0 +1,124 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import {
ManagedSessionRecordError,
managedSessionEventRefs,
type ManagedSessionKey,
} from './managed-session-records.js';
import { managedSessionNestedResourceRefs } from './managed-session-resource-references.js';
import {
scanManagedSessionJournal,
type ManagedSessionJournalScan,
} from './managed-session-storage.js';
import {
readLocalJournalBytes,
parseLocalJournalBase,
readLocalJournalManifest,
readLocalJournalResource,
localJournalDigest,
localJournalPhysicalPrefix,
} from '../utils/local-managed-journal-snapshot.js';
export * from '../utils/local-managed-journal-snapshot.js';
function requireSnapshot(
condition: boolean,
detail: string,
): asserts condition {
if (!condition)
throw new ManagedSessionRecordError(`Local journal snapshot ${detail}.`);
}
export function readLocalManagedJournal(options: {
readonly transcriptPath: string;
readonly runtimeBaseDir: string;
readonly sessionKey: ManagedSessionKey;
readonly maxBytes?: number;
}): ManagedSessionJournalScan {
let bytes: Buffer;
try {
bytes = readLocalJournalBytes(options.transcriptPath, options.maxBytes);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT')
return scanManagedSessionJournal(Buffer.alloc(0), options.sessionKey);
throw error;
}
const base = parseLocalJournalBase(bytes, options.sessionKey);
if (!base) return scanManagedSessionJournal(bytes, options.sessionKey);
const manifest = readLocalJournalManifest(
options.runtimeBaseDir,
options.sessionKey,
base.manifest,
);
const chunks = manifest.pages.map((ref) =>
readLocalJournalResource(options.runtimeBaseDir, options.sessionKey, ref),
);
const prefix = Buffer.concat(chunks);
requireSnapshot(
prefix.length === manifest.byteLength &&
localJournalDigest(prefix) === manifest.digest,
'prefix bytes differ',
);
const original = scanManagedSessionJournal(prefix, options.sessionKey);
requireSnapshot(
original.uncommitted === 0 &&
original.committedBytes === prefix.length &&
original.committed === manifest.committedSequence &&
original.lastMarkerDigest === manifest.lastCommitDigest &&
original.lastRecordUuid === manifest.lastRecordUuid,
'prefix commit proof differs',
);
requireSnapshot(
localJournalPhysicalPrefix(prefix, original.header!).equals(
bytes.subarray(0, base.headerEnd),
),
'ownership prefix differs from its source',
);
for (const ref of manifest.resources)
readLocalJournalResource(options.runtimeBaseDir, options.sessionKey, ref);
const logical = scanManagedSessionJournal(
Buffer.concat([prefix, bytes.subarray(base.baseEnd)]),
options.sessionKey,
);
const verified = new Map(
manifest.resources.map((ref) => [ref.resourceId, ref]),
);
const pending = logical.events
.filter((event) => event.sequence > manifest.committedSequence)
.flatMap(managedSessionEventRefs);
while (pending.length) {
const ref = pending.pop()!;
const previous = verified.get(ref.resourceId);
if (previous) {
requireSnapshot(
previous.kind === ref.kind &&
previous.schemaVersion === ref.schemaVersion &&
previous.byteLength === ref.byteLength &&
previous.digest === ref.digest,
'tail resource references conflict',
);
continue;
}
verified.set(ref.resourceId, ref);
pending.push(
...managedSessionNestedResourceRefs(
ref,
readLocalJournalResource(
options.runtimeBaseDir,
options.sessionKey,
ref,
),
),
);
}
return {
...logical,
headerBytes: base.headerEnd,
committedBytes: base.baseEnd + logical.committedBytes - prefix.length,
uncommittedBytes:
bytes.length - (base.baseEnd + logical.committedBytes - prefix.length),
};
}
@@ -0,0 +1,256 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { randomUUID } from 'node:crypto';
import {
SessionWriterLease,
type AcquireSessionWriterLeaseOptions,
} from '../services/session-writer-lease.js';
import {
managedSessionEventRefs,
managedSessionKeysEqual,
type ManagedSessionKey,
ManagedSessionRecordError,
parseManagedSessionHeader,
type ManagedSessionDurableRef,
} from './managed-session-records.js';
import { LocalManagedSessionResourceStore } from './managed-session-resources.js';
import { managedSessionNestedResourceRefs } from './managed-session-resource-references.js';
import { scanManagedSessionJournal } from './managed-session-storage.js';
import {
LOCAL_MANAGED_JOURNAL_FORMAT,
LOCAL_MANAGED_SNAPSHOT_BASE,
LOCAL_MANAGED_SNAPSHOT_FORMAT,
LOCAL_MANAGED_SNAPSHOT_KIND,
LOCAL_MANAGED_SNAPSHOT_PAGE_BYTES,
LOCAL_MANAGED_SNAPSHOT_PAGE_KIND,
localJournalDigest,
localJournalObject,
localJournalPhysicalPrefix,
parseLocalJournalBase,
readLocalJournalBytes,
readLocalManagedJournal,
type LocalManagedJournalManifest,
} from './local-managed-journal-snapshot.js';
function sourceHeader(bytes: Buffer) {
for (const line of bytes.toString('utf8').split('\n')) {
if (!line) continue;
const record = localJournalObject(JSON.parse(line));
if (record['subtype'] === 'managed_session_header_v1') {
return parseManagedSessionHeader(record['managedSession']);
}
}
return undefined;
}
/** Acquire writer capability; preparation waits for restore admission. */
export async function acquireLocalManagedSessionWriter(
options: AcquireSessionWriterLeaseOptions,
): Promise<SessionWriterLease> {
let header;
try {
header = sourceHeader(readLocalJournalBytes(options.transcriptPath));
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
}
const lease = await SessionWriterLease.acquire({
...options,
takeoverPolicy: 'certified',
lockSchema: {
schemaVersion: 3,
formatVersion:
header?.minimumReader === 'managed-session/2'
? LOCAL_MANAGED_JOURNAL_FORMAT
: 1,
resumeFormatVersion: LOCAL_MANAGED_JOURNAL_FORMAT,
},
});
return lease;
}
export async function prepareLocalManagedSessionWriter(
lease: SessionWriterLease,
expectedSessionKey: ManagedSessionKey,
): Promise<boolean> {
const options = {
runtimeBaseDir: lease.runtimeBaseDir,
transcriptPath: lease.transcriptPath,
sessionId: lease.sessionId,
};
const proof = lease.takeoverCommitProof;
if (
!proof ||
lease.managedJournalFormatVersion !== LOCAL_MANAGED_JOURNAL_FORMAT
)
return false;
let header;
try {
const bytes = Buffer.from(await lease.readTranscript());
header = sourceHeader(bytes);
if (!header) return false;
if (
header.sessionKey.sessionId !== options.sessionId ||
!managedSessionKeysEqual(header.sessionKey, expectedSessionKey)
) {
throw new ManagedSessionRecordError(
'Local snapshot source has a different Session.',
);
}
if (parseLocalJournalBase(bytes, header.sessionKey)) return false;
const scan = scanManagedSessionJournal(bytes, header.sessionKey);
if (
scan.committed !== proof.last_commit_sequence ||
(scan.lastMarkerDigest ?? '0'.repeat(64)) !== proof.committed_prefix_hash
) {
const { ManagedSessionConflictError } = await import(
'./managed-session-authority.js'
);
throw new ManagedSessionConflictError(
'Local snapshot source does not match the sealed writer proof.',
);
}
if (scan.uncommitted > 0 || scan.committed === 0) return false;
const resources = LocalManagedSessionResourceStore.create({
runtimeBaseDir: options.runtimeBaseDir,
sessionKey: header.sessionKey,
});
const pending = [
header.definitionRef,
header.rootSnapshotRef,
...(header.baseTranscriptProof ? [header.baseTranscriptProof] : []),
...scan.events.flatMap(managedSessionEventRefs),
];
const refs = new Map<string, ManagedSessionDurableRef>();
while (pending.length) {
const ref = pending.pop()!;
const previous = refs.get(ref.resourceId);
if (previous) {
if (
previous.kind !== ref.kind ||
previous.schemaVersion !== ref.schemaVersion ||
previous.byteLength !== ref.byteLength ||
previous.digest !== ref.digest
) {
throw new ManagedSessionRecordError(
'Local snapshot resource references conflict.',
);
}
continue;
}
refs.set(ref.resourceId, ref);
pending.push(
...managedSessionNestedResourceRefs(ref, await resources.read(ref)),
);
}
const pages: ManagedSessionDurableRef[] = [];
let pageStart = 0;
let transactionEnd = 0;
let offset = 0;
while (offset < bytes.length) {
const end = bytes.indexOf(10, offset);
if (end < 0)
throw new ManagedSessionRecordError(
'Local snapshot source has a torn line.',
);
const record = localJournalObject(
JSON.parse(bytes.subarray(offset, end).toString('utf8')),
);
if (record['subtype'] === 'managed_session_commit_v1') {
if (end + 1 - pageStart > LOCAL_MANAGED_SNAPSHOT_PAGE_BYTES) {
if (transactionEnd === pageStart)
throw new ManagedSessionRecordError(
'Local snapshot transaction exceeds its page budget.',
);
pages.push(
await resources.publishJournalSnapshot(
LOCAL_MANAGED_SNAPSHOT_PAGE_KIND,
bytes.subarray(pageStart, transactionEnd),
),
);
pageStart = transactionEnd;
}
transactionEnd = end + 1;
if (transactionEnd - pageStart > LOCAL_MANAGED_SNAPSHOT_PAGE_BYTES)
throw new ManagedSessionRecordError(
'Local snapshot transaction exceeds its page budget.',
);
}
offset = end + 1;
}
pages.push(
await resources.publishJournalSnapshot(
LOCAL_MANAGED_SNAPSHOT_PAGE_KIND,
bytes.subarray(pageStart),
),
);
const latestRef = (domain: string) =>
scan.events.findLast(
(event) =>
event.kind === 'domain.committed' &&
event.payload['domain'] === domain,
)?.payload['recordRef'] as ManagedSessionDurableRef | undefined;
const manifest: LocalManagedJournalManifest = {
format: LOCAL_MANAGED_SNAPSHOT_FORMAT,
sessionKey: header.sessionKey,
pages,
byteLength: bytes.length,
digest: localJournalDigest(bytes),
committedSequence: scan.committed,
lastCommitDigest: scan.lastMarkerDigest!,
lastRecordUuid: scan.lastRecordUuid!,
resources: [...refs.values()].sort((a, b) =>
a.resourceId.localeCompare(b.resourceId),
),
titleRef: latestRef('session_metadata') ?? null,
sourceRef: latestRef('session_source') ?? null,
};
const manifestRef = await resources.publishJournalSnapshot(
LOCAL_MANAGED_SNAPSHOT_KIND,
Buffer.from(JSON.stringify(manifest)),
);
const physicalPrefix = localJournalPhysicalPrefix(bytes, header);
const headerRecord = localJournalObject(
JSON.parse(physicalPrefix.toString('utf8').trimEnd().split('\n').at(-1)!),
);
const base = {
uuid: randomUUID(),
parentUuid: headerRecord['uuid'],
type: 'system',
subtype: LOCAL_MANAGED_SNAPSHOT_BASE,
sessionId: options.sessionId,
managedSessionSnapshot: {
format: LOCAL_MANAGED_SNAPSHOT_FORMAT,
sessionKey: header.sessionKey,
manifest: manifestRef,
},
};
const sessionKey = header.sessionKey;
await lease.replaceManagedTranscript(
`${physicalPrefix.toString('utf8')}${JSON.stringify(base)}\n`,
manifest.digest,
() => {
const restored = readLocalManagedJournal({ ...options, sessionKey });
if (
restored.committed !== scan.committed ||
restored.lastMarkerDigest !== scan.lastMarkerDigest ||
restored.lastRecordUuid !== scan.lastRecordUuid ||
JSON.stringify([...restored.transactions]) !==
JSON.stringify([...scan.transactions])
) {
throw new ManagedSessionRecordError(
'Local snapshot replacement changed the logical commit proof.',
);
}
},
);
return true;
} catch (error) {
// This preserves the barrier; an uncertain physical replacement refuses seal.
await lease.sealForHandoff(proof).catch(() => undefined);
throw error;
}
}
@@ -340,10 +340,11 @@ it('recovers a null-manifest cancellation under a new activation without relaxin
},
},
});
const journal = await LocalJsonlManagedSessionJournalStore.read(
f.transcriptPath,
f.key,
);
const journal = await LocalJsonlManagedSessionJournalStore.read({
transcriptPath: f.transcriptPath,
runtimeBaseDir: f.lease.runtimeBaseDir,
sessionKey: f.key,
});
expect(journal.events.filter((e) => e.kind === 'tool.receipt')).toHaveLength(
1,
);
@@ -35,6 +35,7 @@ import {
type ManagedSessionCommand,
} from './managed-session-authority.js';
import type { ManagedSessionDurableRef } from './managed-session-records.js';
import { LocalManagedSessionResourceStore } from './managed-session-resources.js';
import {
managedRuntimeDispatchGate,
resetManagedRuntimeDispatchGatesForTest,
@@ -85,7 +86,11 @@ async function createWorkspace(): Promise<Workspace> {
return { runtimeBaseDir, projectRoot, transcriptPath };
}
function open(workspace: Workspace): Promise<ManagedSession> {
async function open(workspace: Workspace): Promise<ManagedSession> {
const resources = LocalManagedSessionResourceStore.create({
runtimeBaseDir: workspace.runtimeBaseDir,
sessionKey,
});
return openManagedSession({
runtimeBaseDir: workspace.runtimeBaseDir,
sessionId,
@@ -96,8 +101,14 @@ function open(workspace: Workspace): Promise<ManagedSession> {
workerId: 'worker-1',
activationLeaseDurationMs: 60_000,
create: {
definitionRef: ref('managed-definition'),
rootSnapshotRef: ref('managed-root'),
definitionRef: await resources.publish(
'managed-definition',
Buffer.from('{}'),
),
rootSnapshotRef: await resources.publish(
'managed-root',
Buffer.from('{}'),
),
createdBy: 'daemon',
},
});
@@ -24,6 +24,7 @@ import { TurnBudget } from '../core/turn-budget.js';
import { uiTelemetryService, type UiEvent } from '../telemetry/uiTelemetry.js';
import { ApiResponseEvent } from '../telemetry/types.js';
import { EVENT_API_RESPONSE } from '../telemetry/constants.js';
import { LocalManagedSessionResourceStore } from './managed-session-resources.js';
const operation = {
operationId: 'operation-1',
@@ -55,13 +56,6 @@ afterEach(async () => {
async function fixture() {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'hook-model-'));
dirs.push(root);
const ref = {
resourceId: 'resource',
kind: 'managed-test',
schemaVersion: 1,
byteLength: 2,
digest: '9'.repeat(64),
};
const options = {
runtimeBaseDir: root,
sessionId: '550e8400-e29b-41d4-a716-446655440000',
@@ -76,6 +70,8 @@ async function fixture() {
workerId: 'worker-1',
activationLeaseDurationMs: 60_000,
};
const resources = LocalManagedSessionResourceStore.create(options);
const ref = await resources.publish('managed-test', Buffer.from('{}'));
const session = await openManagedSession({
...options,
create: { definitionRef: ref, rootSnapshotRef: ref, createdBy: 'test' },
@@ -15,6 +15,7 @@ import {
type ManagedSession,
} from './managed-session-assembly.js';
import { LocalManagedSessionAuthority } from './managed-session-authority.js';
import { LocalManagedSessionResourceStore } from './managed-session-resources.js';
import {
LocalManagedRuntimeOutcomes,
unresolvedRuntimeWorkReason,
@@ -56,17 +57,6 @@ afterEach(async () => {
// survives across attempts of the paged-Turn test.
let pagedOutcomesAttempt = 0;
async function resource(value: string) {
const { createHash } = await import('node:crypto');
return {
resourceId: `ref-${value}`,
kind: value,
schemaVersion: 1,
byteLength: Buffer.byteLength(value),
digest: createHash('sha256').update(value).digest('hex'),
};
}
interface OpenedSession {
readonly session: ManagedSession;
/** Finishes the adopted lease as a handoff seal, so a reopen takes it over. */
@@ -78,6 +68,10 @@ async function openSession(root: string, id: string): Promise<OpenedSession> {
const transcriptPath = path.join(runtimeBaseDir, 'chats', `${id}.jsonl`);
await fs.mkdir(path.dirname(transcriptPath), { recursive: true });
const sessionKey = sessionKeyOf(id);
const resources = LocalManagedSessionResourceStore.create({
runtimeBaseDir,
sessionKey,
});
const lease = await LocalManagedSessionAuthority.acquireWriter({
runtimeBaseDir,
sessionId: id,
@@ -94,8 +88,8 @@ async function openSession(root: string, id: string): Promise<OpenedSession> {
activationLeaseDurationMs: 60_000,
lease,
create: {
definitionRef: await resource('definition'),
rootSnapshotRef: await resource('root'),
definitionRef: await resources.publish('definition', Buffer.from('{}')),
rootSnapshotRef: await resources.publish('root', Buffer.from('{}')),
createdBy: 'test',
},
});
@@ -960,75 +954,92 @@ describe('restored runtime block', () => {
}
});
it('re-records a settled result whose outcome body no longer reads, as an error', async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-outcomes-'));
roots.add(root);
const sessionId = 'session-lost-outcome';
{
const { session, seal } = await openSession(root, sessionId);
const outcomes = new LocalManagedRuntimeOutcomes(session);
await outcomes.admit(admission('call-a'));
await outcomes.settle({
functionCallId: 'call-a',
executionStatus: 'success',
payload: {
it.each(['live', 'sealed'] as const)(
'handles a corrupted outcome in a %s Session',
async (state) => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-outcomes-'));
roots.add(root);
const sessionId = 'session-lost-outcome';
const { session: restored, seal } = await openSession(root, sessionId);
{
const session = restored;
const outcomes = new LocalManagedRuntimeOutcomes(session);
await outcomes.admit(admission('call-a'));
await outcomes.settle({
functionCallId: 'call-a',
executionStatus: 'success',
responseParts: [{ type: 'text', text: 'the answer' }],
},
});
await seal();
// The outcome blob is lost after the close — truncated, rotated away.
const outcomesDir = path.join(
root,
'runtime',
'resources',
sessionId,
'managed-tool-outcome',
);
let blanked = 0;
for (const file of await fs.readdir(outcomesDir)) {
if (file.startsWith('.')) continue;
await fs.writeFile(path.join(outcomesDir, file), Buffer.alloc(0));
blanked += 1;
payload: {
executionStatus: 'success',
responseParts: [{ type: 'text', text: 'the answer' }],
},
});
if (state === 'sealed') await seal();
// The recorded outcome is corrupted, either during execution or after seal.
const outcomesDir = path.join(
root,
'runtime',
'resources',
sessionId,
'managed-tool-outcome',
);
let blanked = 0;
for (const file of await fs.readdir(outcomesDir)) {
if (file.startsWith('.')) continue;
await fs.writeFile(path.join(outcomesDir, file), Buffer.alloc(0));
blanked += 1;
}
expect(blanked).toBeGreaterThan(0);
}
expect(blanked).toBeGreaterThan(0);
}
const { session: restored } = await openSession(root, sessionId);
try {
// The record the model's history needs exists either way: without it
// the assistant's functionCall pairs with nothing, and the provider
// rejects the next request as malformed with no gate left to say why.
await expect(
new LocalManagedRuntimeOutcomes(restored).recoverCommittedReceipts(),
).resolves.toBeUndefined();
const resultsRecorded = events(restored, 'message.committed').filter(
(event) => event.payload['role'] === 'tool_result',
);
expect(resultsRecorded).toHaveLength(1);
const body = await restored.resources.read(
resultsRecorded[0]!.payload['contentRef'] as never,
);
const record = JSON.parse(body.toString()) as {
message?: { parts?: Array<Record<string, unknown>> };
toolCallResult?: {
callId?: string;
status?: string;
error?: unknown;
if (state === 'sealed') {
const transcript = path.join(
root,
'runtime',
'chats',
`${sessionId}.jsonl`,
);
const before = await fs.readFile(transcript);
await expect(openSession(root, sessionId)).rejects.toThrow(
/bytes on disk/,
);
expect(await fs.readFile(transcript)).toEqual(before);
return;
}
try {
// The record the model's history needs exists either way: without it
// the assistant's functionCall pairs with nothing, and the provider
// rejects the next request as malformed with no gate left to say why.
await expect(
new LocalManagedRuntimeOutcomes(restored).recoverCommittedReceipts(),
).resolves.toBeUndefined();
const resultsRecorded = events(restored, 'message.committed').filter(
(event) => event.payload['role'] === 'tool_result',
);
expect(resultsRecorded).toHaveLength(1);
const body = await restored.resources.read(
resultsRecorded[0]!.payload['contentRef'] as never,
);
const record = JSON.parse(body.toString()) as {
message?: { parts?: Array<Record<string, unknown>> };
toolCallResult?: {
callId?: string;
status?: string;
error?: unknown;
};
};
};
expect(record.message?.parts?.[0]?.['functionResponse']).toMatchObject({
id: 'call-a',
name: 'read_file',
response: { error: expect.stringContaining('cannot be read') },
});
expect(record.toolCallResult?.callId).toBe('call-a');
expect(record.toolCallResult?.status).toBe('error');
expect(record.toolCallResult?.error).toBeDefined();
} finally {
await restored.close();
}
});
expect(record.message?.parts?.[0]?.['functionResponse']).toMatchObject({
id: 'call-a',
name: 'read_file',
response: { error: expect.stringContaining('cannot be read') },
});
expect(record.toolCallResult?.callId).toBe('call-a');
expect(record.toolCallResult?.status).toBe('error');
expect(record.toolCallResult?.error).toBeDefined();
} finally {
await restored.close();
}
},
);
it('recovers a cancelled outcome with the cancellation the live path reports', async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-outcomes-'));
@@ -1470,51 +1481,76 @@ describe('restored runtime block', () => {
}
});
it('answers a reason for a checkpoint that cannot be read', async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-outcomes-'));
roots.add(root);
{
const { session, seal } = await openSession(root, 'session-gone-state');
const outcomes = new LocalManagedRuntimeOutcomes(session);
await outcomes.admit(admission('call-a'));
await seal();
// A checkpoint whose state no longer exists on disk.
const stateRoot = path.join(
it.each(['live', 'sealed'] as const)(
'handles an unreadable checkpoint in a %s Session',
async (state) => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-outcomes-'));
roots.add(root);
const { session: restored, seal } = await openSession(
root,
'runtime',
'resources',
'session-gone-state',
);
let removed = 0;
for (const kind of await fs.readdir(stateRoot)) {
const dir = path.join(stateRoot, kind);
for (const file of await fs.readdir(dir)) {
if (file.startsWith('.')) continue;
const candidate = path.join(dir, file);
const content = await fs.readFile(candidate);
removed += content.includes('ckpt-') ? 1 : 0;
if (content.includes('ckpt-')) {
await fs.writeFile(candidate, Buffer.alloc(0));
{
const session = restored;
const outcomes = new LocalManagedRuntimeOutcomes(session);
await outcomes.admit(admission('call-a'));
if (state === 'sealed') await seal();
// A checkpoint whose state no longer exists on disk.
const stateRoot = path.join(
root,
'runtime',
'resources',
'session-gone-state',
);
let removed = 0;
for (const kind of await fs.readdir(stateRoot)) {
const dir = path.join(stateRoot, kind);
for (const file of await fs.readdir(dir)) {
if (file.startsWith('.')) continue;
const candidate = path.join(dir, file);
const content = await fs.readFile(candidate);
removed += content.includes('ckpt-') ? 1 : 0;
if (content.includes('ckpt-')) {
await fs.writeFile(candidate, Buffer.alloc(0));
}
}
}
expect(removed).toBeGreaterThan(0);
}
expect(removed).toBeGreaterThan(0);
}
const { session: restored } = await openSession(root, 'session-gone-state');
try {
// The restore repair leaves an unreadable checkpoint to the gate: it
// no-ops rather than failing the open the gate is about to block.
await expect(
new LocalManagedRuntimeOutcomes(restored).recoverCommittedReceipts(),
).resolves.toBeUndefined();
await expect(
unresolvedRuntimeWorkReason(restored.authority),
).resolves.toContain('cannot be read');
} finally {
await restored.close();
}
});
if (state === 'sealed') {
const transcript = path.join(
root,
'runtime',
'chats',
'session-gone-state.jsonl',
);
const before = await fs.readFile(transcript);
await expect(openSession(root, 'session-gone-state')).rejects.toThrow(
/bytes on disk/,
);
expect(await fs.readFile(transcript)).toEqual(before);
return;
}
try {
await expect(restored.authority.readCheckpointState()).rejects.toThrow(
/bytes on disk/,
);
// A live authority still has its parsed checkpoint cached.
await expect(
restored.authority.readParsedCheckpointState(),
).resolves.toMatchObject({ ok: true });
await expect(
new LocalManagedRuntimeOutcomes(restored).recoverCommittedReceipts(),
).resolves.toBeUndefined();
await expect(
unresolvedRuntimeWorkReason(restored.authority),
).resolves.toContain('never settled');
} finally {
await restored.close();
}
},
);
it('answers a reason for a checkpoint it cannot parse', async () => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'qwen-outcomes-'));
@@ -23,7 +23,6 @@ import {
} from './managed-session-assembly.js';
import { LocalJsonlManagedSessionJournalStore } from './local-jsonl-managed-session-journal-store.js';
import { readManagedSessionTitleInfoSync } from '../utils/sessionStorageUtils.js';
import type { ManagedSessionDurableRef } from './managed-session-records.js';
import { ManagedSessionAppendUnsettledError } from './managed-session-records.js';
import { LocalManagedSessionResourceStore } from './managed-session-resources.js';
import {
@@ -32,7 +31,6 @@ import {
type ManagedSessionResourceStore,
} from './managed-session-storage.js';
const DIGEST = '9'.repeat(64);
const sessionId = '550e8400-e29b-41d4-a716-446655440000';
const sessionKey = { tenantId: 't1', workspaceId: 'w1', sessionId };
const temporaryDirectories = new Set<string>();
@@ -46,16 +44,6 @@ afterEach(async () => {
temporaryDirectories.clear();
});
function ref(kind = 'managed-test'): ManagedSessionDurableRef {
return {
resourceId: 'res-1',
kind,
schemaVersion: 1,
byteLength: 4,
digest: DIGEST,
};
}
interface Workspace {
runtimeBaseDir: string;
projectRoot: string;
@@ -78,7 +66,7 @@ async function createWorkspace(): Promise<Workspace> {
return { runtimeBaseDir, projectRoot, transcriptPath };
}
function open(
async function open(
workspace: Workspace,
options: {
create?: boolean;
@@ -88,6 +76,10 @@ function open(
requireNew?: boolean;
} = {},
): Promise<ManagedSession> {
const roots = LocalManagedSessionResourceStore.create({
runtimeBaseDir: workspace.runtimeBaseDir,
sessionKey,
});
return openManagedSession({
runtimeBaseDir: workspace.runtimeBaseDir,
sessionId,
@@ -109,8 +101,14 @@ function open(
? {}
: {
create: {
definitionRef: ref('managed-definition'),
rootSnapshotRef: ref('managed-root'),
definitionRef: await roots.publish(
'managed-session-definition',
Buffer.from('{}'),
),
rootSnapshotRef: await roots.publish(
'managed-session-root-snapshot',
Buffer.from('{}'),
),
createdBy: 'daemon',
},
}),
@@ -261,6 +261,7 @@ export async function openManagedSession(
runtimeBaseDir: options.runtimeBaseDir,
sessionId: options.sessionId,
transcriptPath: options.transcriptPath,
requireNew: options.requireNew,
...(options.lease === undefined ? {} : { lease: options.lease }),
});
const journal = await journalStore.open({ sessionKey: options.sessionKey });
@@ -2132,6 +2132,10 @@ describe('managed session write barrier at rest', () => {
sessionId: fixture.sessionId,
transcriptPath: fixture.transcriptPath,
});
const resources = LocalManagedSessionResourceStore.create({
runtimeBaseDir: fixture.runtimeBaseDir,
sessionKey: sessionKeyFor(fixture),
});
const authority = await LocalManagedSessionAuthority.open({
journal: LocalJsonlManagedSessionJournalStore.fromLease(
lease,
@@ -2141,12 +2145,25 @@ describe('managed session write barrier at rest', () => {
cwd: '/workspace',
version: 'test',
create: {
definitionRef: ref('managed-definition'),
rootSnapshotRef: ref('managed-root'),
definitionRef: await resources.publish(
'managed-session-definition',
Buffer.from('{}'),
),
rootSnapshotRef: await resources.publish(
'managed-session-root-snapshot',
Buffer.from('{}'),
),
createdBy: 'daemon',
},
});
await authority.submitInput(inputCommand(fixture), inputRequest);
await authority.submitInput(inputCommand(fixture), {
...inputRequest,
contentRef: await resources.publish('managed-test', Buffer.from('input')),
admissionRef: await resources.publish(
'managed-test',
Buffer.from('admission'),
),
});
await authority.close();
}
@@ -2187,9 +2204,24 @@ describe('managed session write barrier at rest', () => {
version: 'test',
});
expect(authority.committedSequence).toBe(2);
const resources = LocalManagedSessionResourceStore.create({
runtimeBaseDir: fixture.runtimeBaseDir,
sessionKey: sessionKeyFor(fixture),
});
const receipt = await authority.submitInput(
inputCommand(fixture, { commandId: 'cmd-after-seal' }),
{ ...inputRequest, inputId: 'in-after-seal' },
{
...inputRequest,
inputId: 'in-after-seal',
contentRef: await resources.publish(
'managed-test',
Buffer.from('after-seal input'),
),
admissionRef: await resources.publish(
'managed-test',
Buffer.from('after-seal admission'),
),
},
);
expect(receipt.firstSequence).toBe(3);
await authority.close();
@@ -2277,8 +2309,14 @@ describe('managed session checkpoints', () => {
? {}
: {
create: {
definitionRef: ref('managed-definition'),
rootSnapshotRef: ref('managed-root'),
definitionRef: await store.publish(
'managed-session-definition',
Buffer.from('{}'),
),
rootSnapshotRef: await store.publish(
'managed-session-root-snapshot',
Buffer.from('{}'),
),
createdBy: 'daemon',
},
}),
@@ -2321,7 +2359,10 @@ describe('managed session checkpoints', () => {
phase: 'active',
leaseDurationMs: 60_000,
expiresAt: 2,
installRef: ref(),
installRef: await harness.store.publish(
'managed-test',
Buffer.from('install'),
),
boundaryRef: null,
},
},
@@ -2429,7 +2470,17 @@ describe('managed session checkpoints', () => {
it('chains each checkpoint and recovers the newest after a cold reopen', async () => {
const fixture = await createFixture();
const first = await openWithResources(fixture);
await first.authority.submitInput(inputCommand(fixture), inputRequest);
await first.authority.submitInput(inputCommand(fixture), {
...inputRequest,
contentRef: await first.store.publish(
'managed-test',
Buffer.from('input'),
),
admissionRef: await first.store.publish(
'managed-test',
Buffer.from('admission'),
),
});
await activate(first, 3);
const one = await first.authority.commitCheckpoint(
inputCommand(fixture, {
@@ -14,9 +14,10 @@ import {
parseManagedConfigInstallRecord,
type ManagedConfigBinding,
} from './managed-config-install.js';
import { SessionWriterLease } from '../services/session-writer-lease.js';
import type { SessionWriterLease } from '../services/session-writer-lease.js';
import { managedToolDigest } from '../tools/managed-tool-protocol.js';
import { LocalJsonlManagedSessionJournalStore } from './local-jsonl-managed-session-journal-store.js';
import { acquireLocalManagedSessionWriter } from './local-managed-session-writer.js';
import {
isDefinitionPinConsistent,
isTerminalRunState,
@@ -790,16 +791,7 @@ export class LocalManagedSessionAuthority {
sessionId: string;
transcriptPath: string;
}): Promise<SessionWriterLease> {
return SessionWriterLease.acquire({
runtimeBaseDir: options.runtimeBaseDir,
sessionId: options.sessionId,
transcriptPath: options.transcriptPath,
takeoverPolicy: 'certified',
lockSchema: {
schemaVersion: 3,
formatVersion: MANAGED_SESSION_FORMAT_VERSION,
},
});
return acquireLocalManagedSessionWriter(options);
}
/** The commit position the current log stands at, for sealing or proofs. */
@@ -4434,10 +4426,11 @@ function actionDecisionsMatch(
* a byte length has to keep answering from it, or a page it serves would mix in
* records appended after the cursor was issued.
*/
export async function readManagedSessionLog(
path: string,
sessionKey: ManagedSessionKey,
maxBytes?: number,
): Promise<ManagedSessionJournalScan> {
return LocalJsonlManagedSessionJournalStore.read(path, sessionKey, maxBytes);
export async function readManagedSessionLog(options: {
readonly transcriptPath: string;
readonly runtimeBaseDir: string;
readonly sessionKey: ManagedSessionKey;
readonly maxBytes?: number;
}): Promise<ManagedSessionJournalScan> {
return LocalJsonlManagedSessionJournalStore.read(options);
}
@@ -48,16 +48,6 @@ afterEach(async () => {
temporaryDirectories.clear();
});
function ref(kind = 'managed-test'): ManagedSessionDurableRef {
return {
resourceId: 'res-1',
kind,
schemaVersion: 1,
byteLength: 4,
digest: DIGEST,
};
}
function command(operation: string, commandId: string) {
return { operation, commandId, sessionKey, contentDigest: DIGEST };
}
@@ -101,8 +91,14 @@ async function createHarness(): Promise<Harness> {
version: 'test',
resources: store,
create: {
definitionRef: ref('managed-definition'),
rootSnapshotRef: ref('managed-root'),
definitionRef: await store.publish(
'managed-session-definition',
Buffer.from('{}'),
),
rootSnapshotRef: await store.publish(
'managed-session-root-snapshot',
Buffer.from('{}'),
),
createdBy: 'daemon',
},
});
@@ -129,7 +125,7 @@ async function createHarness(): Promise<Harness> {
phase: 'active',
leaseDurationMs: 60_000,
expiresAt: 2,
installRef: ref(),
installRef: await store.publish('managed-test', Buffer.from('test')),
boundaryRef: null,
},
},
@@ -429,10 +425,11 @@ describe('managed session message projection', () => {
'utf8',
),
);
const scan = await readManagedSessionLog(
harness.transcriptPath,
const scan = await readManagedSessionLog({
transcriptPath: harness.transcriptPath,
runtimeBaseDir: harness.runtimeBaseDir,
sessionKey,
);
});
const event = parseManagedSessionEvent({
v: 1,
sequence: scan.committed + 1,
@@ -500,10 +497,11 @@ describe('managed session message projection', () => {
'managed-file_history',
Buffer.from('null', 'utf8'),
);
const scan = await readManagedSessionLog(
harness.transcriptPath,
const scan = await readManagedSessionLog({
transcriptPath: harness.transcriptPath,
runtimeBaseDir: harness.runtimeBaseDir,
sessionKey,
);
});
const event = parseManagedSessionEvent({
v: 1,
sequence: scan.committed + 1,
@@ -630,10 +628,11 @@ describe('managed session message projection', () => {
await harness.close();
// The writer refuses this event, so it is spliced into the committed log
// the way a writer from before that fence could have left it.
const scan = await readManagedSessionLog(
harness.transcriptPath,
const scan = await readManagedSessionLog({
transcriptPath: harness.transcriptPath,
runtimeBaseDir: harness.runtimeBaseDir,
sessionKey,
);
});
const event = parseManagedSessionEvent({
v: 1,
sequence: scan.committed + 1,
@@ -712,10 +711,11 @@ describe('managed session message projection', () => {
} finally {
await harness.close();
}
const scan = await readManagedSessionLog(
harness.transcriptPath,
const scan = await readManagedSessionLog({
transcriptPath: harness.transcriptPath,
runtimeBaseDir: harness.runtimeBaseDir,
sessionKey,
);
});
await expect(
projectManagedSessionTitleInfo({ scan, resources: harness.store }),
@@ -873,14 +873,18 @@ describe('managed session message projection', () => {
expect(await reopened.readCheckpointState()).toEqual(
withState ? state : undefined,
);
expect(await fs.readFile(harness.transcriptPath)).toEqual(before);
const afterOpen = await fs.readFile(harness.transcriptPath);
expect(afterOpen.includes('managed_session_snapshot_base_v1')).toBe(
true,
);
expect(afterOpen.equals(before)).toBe(false);
const sink = new ManagedSessionRecordSink(
reopened,
harness.store,
() => HOLDS,
);
await sink.write(branch);
expect(await fs.readFile(harness.transcriptPath)).toEqual(before);
expect(await fs.readFile(harness.transcriptPath)).toEqual(afterOpen);
await expect(
sink.write({ ...branch, parentUuid: 'changed' }),
).rejects.toThrow(/different content/);
@@ -1131,10 +1135,11 @@ describe('managed session message projection', () => {
} finally {
await harness.close();
}
const scan = await readManagedSessionLog(
harness.transcriptPath,
const scan = await readManagedSessionLog({
transcriptPath: harness.transcriptPath,
runtimeBaseDir: harness.runtimeBaseDir,
sessionKey,
);
});
const marker = {
transactionId: 'historical-batch',
commandId: 'historical-batch',
@@ -208,11 +208,7 @@ export async function readManagedSessionRecords(options: {
/** Bounds the projection to a frozen snapshot's byte length. */
readonly maxBytes?: number;
}): Promise<ChatRecord[]> {
const scan = await readManagedSessionLog(
options.transcriptPath,
options.sessionKey,
options.maxBytes,
);
const scan = await readManagedSessionLog(options);
const resources = LocalManagedSessionResourceStore.create({
runtimeBaseDir: options.runtimeBaseDir,
sessionKey: options.sessionKey,
@@ -237,11 +233,7 @@ export async function readManagedSessionRecordsAndTitle(options: {
records: ChatRecord[];
titleInfo: { title?: string; source?: 'auto' | 'manual' };
}> {
const scan = await readManagedSessionLog(
options.transcriptPath,
options.sessionKey,
options.maxBytes,
);
const scan = await readManagedSessionLog(options);
const resources = LocalManagedSessionResourceStore.create({
runtimeBaseDir: options.runtimeBaseDir,
sessionKey: options.sessionKey,
@@ -995,17 +995,13 @@ describe('managed session header', () => {
it('refuses a newer format or reader requirement', () => {
headerError({ formatVersion: 2 }, /is not supported by this reader/);
headerError(
{ minimumReader: 'managed-session/2' },
/is not supported by this reader/,
);
headerError(
{ minimumReader: 'managed-session/3' },
/is not supported by this reader/,
);
});
it('accepts an older minimum-reader requirement and returns its token', () => {
it('accepts supported minimum-reader requirements and returns their tokens', () => {
expect(
parseManagedSessionHeader(header({ minimumReader: 'managed-session/0' }))
.minimumReader,
@@ -1014,6 +1010,10 @@ describe('managed session header', () => {
parseManagedSessionHeader(header({ minimumReader: 'managed-session/1' }))
.minimumReader,
).toBe('managed-session/1');
expect(
parseManagedSessionHeader(header({ minimumReader: 'managed-session/2' }))
.minimumReader,
).toBe('managed-session/2');
});
it('refuses a malformed minimum-reader token', () => {
@@ -4,6 +4,15 @@
* SPDX-License-Identifier: Apache-2.0
*/
import {
MANAGED_SESSION_LIMITS,
ManagedSessionRecordError,
} from '../utils/managed-session-format.js';
export {
MANAGED_SESSION_LIMITS,
ManagedSessionRecordError,
} from '../utils/managed-session-format.js';
import { createHash } from 'node:crypto';
import { parseBranchCheckpointPayload } from '../services/branch-points.js';
import type { ChatRecord } from '../services/chatRecordingService.js';
@@ -27,6 +36,7 @@ export const MANAGED_SESSION_FORMAT_VERSION = 1;
// created in between (H3 round-5 verification matrix), so the stamp
// rises only when a change genuinely breaks an older reader mid-scan.
export const MANAGED_SESSION_MINIMUM_READER = 'managed-session/1';
export const MANAGED_SESSION_READER_CAPABILITY = 'managed-session/2';
const MANAGED_SESSION_DOMAIN_RECORD_VERSION = 1;
const MAX_ERROR_VALUE_LENGTH = 4096;
@@ -35,22 +45,6 @@ export const MANAGED_SESSION_HEADER_SUBTYPE = 'managed_session_header_v1';
export const MANAGED_SESSION_EVENT_SUBTYPE = 'managed_session_event_v1';
export const MANAGED_SESSION_COMMIT_SUBTYPE = 'managed_session_commit_v1';
export const MANAGED_SESSION_LIMITS = {
maxIdBytes: 512,
// Not frozen by the storage spec: a bound for free-form enum-adjacent fields
// such as `source` or `stopReason`, borrowed from its error-message cap.
maxTextBytes: 4096,
maxTimeMs: 8_640_000_000_000_000,
maxJsonDepth: 64,
maxHeaderBytes: 64 * 1024,
maxEventBytes: 1024 * 1024,
maxCommitMarkerBytes: 64 * 1024,
maxTransactionEvents: 256,
maxTransactionBytes: 8 * 1024 * 1024,
defaultReadEvents: 100,
maxReadEvents: 256,
} as const;
export const MANAGED_SESSION_EVENT_KINDS = [
'input.accepted',
'wake.requested',
@@ -422,15 +416,6 @@ export interface ManagedSessionCommitMarker {
readonly previousCommitDigest: string | null;
}
export class ManagedSessionRecordError extends Error {
readonly code: string = 'managed_session_invalid_record';
constructor(message: string) {
super(message);
this.name = 'ManagedSessionRecordError';
}
}
/**
* The writer died on an earlier append: nothing new will ever commit, so
* callers must not read this as a retriable or a shape conflict.
@@ -1795,7 +1780,7 @@ export function parseManagedSessionHeader(
const minimumReader = record['minimumReader'];
const requiredReaderVersion = managedSessionReaderVersion(minimumReader);
const currentReaderVersion = managedSessionReaderVersion(
MANAGED_SESSION_MINIMUM_READER,
MANAGED_SESSION_READER_CAPABILITY,
);
if (
requiredReaderVersion === null ||
@@ -0,0 +1,174 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { expect, it } from 'vitest';
import { managedSessionNestedResourceRefs } from './managed-session-resource-references.js';
import {
createInitialHarnessCheckpoint,
encodeHarnessCheckpointV1,
} from './managed-harness-checkpoint.js';
import type { ManagedSessionDurableRef } from './managed-session-records.js';
import { encodeChildLaunchEnvelope } from './managed-child-operations.js';
const ref = (kind: string, id = kind): ManagedSessionDurableRef => ({
resourceId: id,
kind,
schemaVersion: 1,
byteLength: 2,
digest: 'a'.repeat(64),
});
const json = (value: unknown) => Buffer.from(JSON.stringify(value));
it('takes only declared definition edges, ignoring arbitrary nested data', () => {
const instructionsRef = ref('managed-agent-instructions');
const skillsRef = ref('managed-agent-skills');
const bundleRef = ref('managed-extension-config-bundle');
const contributionsRef = ref('managed-extension-contributions');
expect(
managedSessionNestedResourceRefs(
ref('managed-definition'),
json({
agentDefinition: { instructionsRef, skillsRef },
sessionSkillsRef: skillsRef,
publicChildExtensionAgent: {
effectiveConfigRef: bundleRef,
extensionAgent: { bundleRef, contributionsRef },
},
arbitrary: { userData: ref('not-a-dependency') },
}),
),
).toEqual([
skillsRef,
instructionsRef,
skillsRef,
bundleRef,
bundleRef,
contributionsRef,
]);
});
it('retains public child launch Skills and extension dependencies', () => {
const skillsRef = ref('managed-agent-skills');
const bundleRef = ref('managed-extension-config-bundle');
const contributionsRef = ref('managed-extension-contributions');
const bytes = encodeChildLaunchEnvelope({
description: 'child',
prompt: 'work',
definition: {
definitionId: 'agent',
definitionRevision: 1,
definitionDigest: 'a'.repeat(64),
},
publicLaunch: {
version: 2,
sourcePromptId: 'prompt',
approvalMode: 'yolo',
approvalRevision: 0,
modelId: 'model',
modelRevision: 0,
modelRoute: {
authType: 'openai',
endpointDigest: 'a'.repeat(64),
providerPin: 'b'.repeat(64),
},
skillsRef,
extensionAgent: {
configRevision: 1,
bundleRef,
contributionsRef,
pin: {
extensionId: 'extension',
packageRevision: 1,
packageDigest: 'a'.repeat(64),
},
agentName: 'agent',
toolNames: ['read_file'],
},
},
});
expect(managedSessionNestedResourceRefs(ref('managed-input'), bytes)).toEqual(
[skillsRef, bundleRef, contributionsRef],
);
});
it('includes extension contribution pages and explicit CSI preparation inputs', () => {
const contributions = ref('managed-extension-contributions');
expect(
managedSessionNestedResourceRefs(
ref('managed-extension-config-bundle'),
json({
packages: [{ contributionsRef: contributions }],
}),
),
).toEqual([contributions]);
const part = ref('managed-extension-contributions-part');
expect(
managedSessionNestedResourceRefs(contributions, json({ parts: [part] })),
).toEqual([part]);
const intent = ref('managed-file_history', 'intent');
const input = ref('managed-tool-input');
const definition = ref('managed-tool-definition');
expect(
managedSessionNestedResourceRefs(
ref('managed-file_history'),
json({
schemaVersion: 2,
profile: 'csi-files-retirement/1',
preparation: {
intentRef: intent,
invocations: [{ inputRef: input, toolDefinitionRef: definition }],
},
}),
),
).toEqual([intent, input, definition]);
});
it('preserves inline and paged checkpoint references without walking message content', () => {
const checkpoint = createInitialHarnessCheckpoint({
sessionKey: {
tenantId: 'tenant',
workspaceId: 'workspace',
sessionId: 'session',
},
checkpointId: 'checkpoint',
coveredSequence: 1,
activationId: 'activation',
turnId: 'turn',
promptId: null,
definitionRevision: 'definition',
configRevision: 'config',
inputDigest: 'a'.repeat(64),
previousCheckpointId: null,
});
const history = ref('managed-api-history');
const state = {
...checkpoint,
resume: { ...checkpoint.resume, apiHistoryRef: history },
};
expect(
managedSessionNestedResourceRefs(
ref('managed-checkpoint'),
encodeHarnessCheckpointV1(state),
),
).toEqual([history]);
const page = ref('managed-checkpoint-history');
expect(
managedSessionNestedResourceRefs(
ref('managed-checkpoint'),
json({
...state,
storageProfile: 'managed-checkpoint-paged/1',
history: { ref: page, toolCount: 1, bindingCount: 0 },
}),
),
).toEqual([history, page]);
expect(
managedSessionNestedResourceRefs(
history,
json({ message: ref('not-a-dependency') }),
),
).toEqual([]);
});
@@ -0,0 +1,233 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import {
MANAGED_SESSION_DOMAINS,
assertManagedSessionDurableRef,
parseManagedSessionRecordJson,
type ManagedSessionDurableRef,
type ManagedSessionJsonValue,
} from './managed-session-records.js';
import { decodeChannelInputEnvelope } from './managed-channel-operations.js';
import { decodeChildLaunchEnvelope } from './managed-child-operations.js';
import { tryParseHarnessCheckpointV1 } from './managed-harness-checkpoint.js';
import {
CHECKPOINT_HISTORY_KIND,
checkpointHistoryPage,
checkpointStorageBody,
isPagedHarnessCheckpoint,
} from './managed-checkpoint-history.js';
import {
HOSTED_RECOVERY_RESOURCE_KINDS,
hostedRecoveryReferences,
parseHostedRecoveryResource,
} from './hosted-recovery-records.js';
import {
MANAGED_MESSAGE_CHUNKS_KIND,
managedMessageChunkParts,
} from './managed-message-chunks.js';
import {
MANAGED_TOOL_RESULT_KINDS,
parseToolResultManifestBytes,
} from './managed-tool-result.js';
// Paths are the ref positions declared by each resource schema. Opaque input,
// tool arguments/results, replyContext and message Parts are never traversed.
const RESOURCE_REF_PATHS: Readonly<Record<string, readonly string[]>> = {
'managed-tool_stage': ['inputRef', 'permissionRef', 'resultRef'],
'managed-history_maintenance': ['requestRef', 'candidateRef'],
'managed-mcp_configuration': ['catalogRef'],
'managed-mcp_operation': ['argsRef', 'resultRef'],
'managed-hook_registration': ['catalogRef'],
'managed-hook_execution': ['planRef', 'inputRef', 'resultRef'],
'managed-monitor_run': [
'commandRef',
'startReceiptRef',
'outputRef',
'lastObservationRef',
],
'managed-child_run': [
'commandRef',
'startReceiptRef',
'outputRef',
'inputRef',
'resultRef',
'terminalReceiptRef',
],
'managed-child_acceptance': ['contentRef', 'terminalReceiptRef'],
'managed-channel_route': ['policyRef'],
'managed-channel_delivery': [
'contentRef',
'segments.*.contentRef',
'segments.*.receipt.proofRef',
],
'managed-schedule': ['promptRef'],
'managed-automation_run': ['gateOutputRef'],
'managed-session_message': ['contentRef'],
'managed-session_command': ['argsRef', 'replyRef'],
'managed-team_task': ['descriptionRef', 'metadataRef'],
'managed-team_message': ['contentRef'],
'managed-team_plan': ['planRef', 'feedbackRef'],
'managed-definition': [
'sessionSkillsRef',
'agentDefinition.instructionsRef',
'agentDefinition.skillsRef',
'publicChildExtensionAgent.effectiveConfigRef',
'publicChildExtensionAgent.extensionAgent.bundleRef',
'publicChildExtensionAgent.extensionAgent.contributionsRef',
],
'managed-action-options': [
'definitionRef',
'rootSnapshotRef',
'assistantRef',
'inputRef',
'continuationRef',
'calls.*.inputRef',
'calls.*.definitionRef',
],
'managed-hook-plan': ['messagesRef'],
'managed-hook-message-chunks': ['parts.*'],
'managed-tool-outcome-chunks': ['parts.*'],
'managed-tool-outcome': ['manifestRef', 'envelope.capture.manifest'],
'managed-config_install': [
'previousBundleRef',
'restoresBundleRef',
'bundleRef',
'rootSnapshotRef',
'stageAckRefs.*',
'enableAckRefs.*',
],
'managed-extension-config-bundle': ['packages.*.contributionsRef'],
'managed-extension-contributions': ['parts.*'],
};
const CHECKPOINT_REF_PATHS = [
'resume.apiHistoryRef',
'resume.fileHistoryRef',
'resume.artifactRef',
'resume.goalRecordsRef',
'resume.goalCheckpointWindowRef',
'resume.tokenCountsRef',
'resume.uiTelemetryRef',
'resume.attributionRef',
'attempt.routeRef',
'attempt.capabilityRef',
'attempt.samplingRef',
'attempt.usageRef',
'tools.items.*.outcomeRef',
'approval.optionsRef',
'approval.decisionRef',
'approval.invocationRef',
'output.llmContentRef',
'output.hookResultRef',
'output.mediaRefs.*',
] as const;
function refsAtPaths(
value: unknown,
paths: readonly string[],
): ManagedSessionDurableRef[] {
const visit = (
current: unknown,
path: readonly string[],
): ManagedSessionDurableRef[] => {
if (current === null || current === undefined) return [];
if (path.length === 0)
return [
assertManagedSessionDurableRef(
current as ManagedSessionJsonValue,
'resource dependency',
),
];
if (typeof current !== 'object') return [];
const [field, ...rest] = path;
return field === '*'
? (Array.isArray(current) ? current : []).flatMap((item) =>
visit(item, rest),
)
: visit((current as Record<string, unknown>)[field], rest);
};
return paths.flatMap((path) => visit(value, path.split('.')));
}
/** Complete schema-declared resource edges, without interpreting opaque JSON. */
export function managedSessionNestedResourceRefs(
ref: ManagedSessionDurableRef,
bytes: Buffer,
): ManagedSessionDurableRef[] {
if (ref.kind === MANAGED_MESSAGE_CHUNKS_KIND)
return managedMessageChunkParts(ref.kind, bytes);
if (ref.kind === CHECKPOINT_HISTORY_KIND)
return refsAtPaths(checkpointHistoryPage(bytes), [
'previous',
'items.*.outcomeRef',
]);
if (ref.kind === 'managed-checkpoint') {
if (isPagedHarnessCheckpoint(bytes))
return refsAtPaths(checkpointStorageBody(bytes), [
...CHECKPOINT_REF_PATHS,
'history.ref',
]);
const parsed = tryParseHarnessCheckpointV1(bytes);
return parsed.ok
? refsAtPaths(parsed.checkpoint, CHECKPOINT_REF_PATHS)
: [];
}
if (ref.kind === 'managed-input') {
try {
return decodeChannelInputEnvelope(bytes).attachments.flatMap(
(attachment) => ('ref' in attachment ? [attachment.ref] : []),
);
} catch {
try {
return refsAtPaths(decodeChildLaunchEnvelope(bytes), [
'publicLaunch.skillsRef',
'publicLaunch.extensionAgent.bundleRef',
'publicLaunch.extensionAgent.contributionsRef',
]);
} catch {
// Remaining input schemas contain text/inline data or opaque args.
return [];
}
}
}
if (HOSTED_RECOVERY_RESOURCE_KINDS.has(ref.kind))
return hostedRecoveryReferences(
parseHostedRecoveryResource(bytes, undefined, ref.kind),
);
if (ref.kind === MANAGED_TOOL_RESULT_KINDS.manifest) {
try {
return refsAtPaths(parseToolResultManifestBytes(bytes), [
'contents.*.body.ref',
'contents.*.body.pages.*.ref',
]);
} catch {
return [];
}
}
const domain = MANAGED_SESSION_DOMAINS.find(
(domain) => ref.kind === `managed-${domain}`,
);
const paths = RESOURCE_REF_PATHS[ref.kind] ?? [];
if (!domain && paths.length === 0) return [];
const record = parseManagedSessionRecordJson(
bytes.toString('utf8'),
bytes.length,
);
return refsAtPaths(record, [
...paths,
...(domain ? ['previousRecordRef'] : []),
...(ref.kind === 'managed-file_history' &&
(record as Record<string, unknown>)['schemaVersion'] === 2 &&
(record as Record<string, unknown>)['profile'] === 'csi-files-retirement/1'
? [
'preparation.intentRef',
'preparation.invocations.*.inputRef',
'preparation.invocations.*.toolDefinitionRef',
]
: []),
]);
}
@@ -7,7 +7,7 @@
import { createHash, randomUUID } from 'node:crypto';
import { constants } from 'node:fs';
import type { Stats } from 'node:fs';
import { lstat, mkdir, open, rename, unlink } from 'node:fs/promises';
import { link, lstat, mkdir, open, rename, unlink } from 'node:fs/promises';
import type { FileHandle } from 'node:fs/promises';
import * as path from 'node:path';
import {
@@ -84,6 +84,37 @@ export class LocalManagedSessionResourceStore
async publish(
kind: string,
bytes: Buffer,
): Promise<ManagedSessionDurableRef> {
if (
kind === 'managed-local-journal-snapshot' ||
kind === 'managed-local-journal-snapshot-page'
) {
throw new ManagedSessionRecordError(
'Native snapshot resources use their dedicated publisher.',
);
}
return this.publishBytes(kind, bytes, randomUUID(), false);
}
publishJournalSnapshot(
kind:
| 'managed-local-journal-snapshot'
| 'managed-local-journal-snapshot-page',
bytes: Buffer,
): Promise<ManagedSessionDurableRef> {
return this.publishBytes(
kind,
bytes,
createHash('sha256').update(bytes).digest('hex'),
true,
);
}
private async publishBytes(
kind: string,
bytes: Buffer,
resourceId: string,
immutable: boolean,
): Promise<ManagedSessionDurableRef> {
const safeKind = assertPathSegment(kind, 'kind');
if (bytes.byteLength > MANAGED_SESSION_RESOURCE_LIMITS.maxResourceBytes) {
@@ -91,20 +122,36 @@ export class LocalManagedSessionResourceStore
`${safeKind} is ${bytes.byteLength} bytes where ${MANAGED_SESSION_RESOURCE_LIMITS.maxResourceBytes} is the maximum.`,
);
}
const resourceId = randomUUID();
const directory = path.join(this.root, safeKind);
const pending = path.join(directory, `.${resourceId}.pending`);
const pending = path.join(
directory,
immutable
? `.${resourceId}.${randomUUID()}.pending`
: `.${resourceId}.pending`,
);
const target = path.join(directory, resourceId);
await ensurePrivateDirectory(path.dirname(this.root));
await ensurePrivateDirectory(this.root);
const checkedDirectory = await ensurePrivateDirectory(directory);
const digest = createHash('sha256');
const resourceRef: ManagedSessionDurableRef = {
resourceId,
kind: safeKind,
schemaVersion: 1,
byteLength: bytes.byteLength,
digest: immutable
? resourceId
: createHash('sha256').update(bytes).digest('hex'),
};
if (immutable && (await lstatOrUndefined(target))) {
await this.read(resourceRef);
await syncDirectory(directory);
return resourceRef;
}
const handle = await open(pending, 'wx', 0o600);
let written: { dev: number; ino: number };
try {
digest.update(bytes);
await handle.writeFile(bytes);
await handle.sync();
written = await handle.stat();
@@ -116,7 +163,20 @@ export class LocalManagedSessionResourceStore
await handle.close();
try {
await rename(pending, target);
if (immutable) {
try {
await link(pending, target);
} catch (cause) {
if ((cause as NodeJS.ErrnoException).code !== 'EEXIST') throw cause;
await this.read(resourceRef);
await unlink(pending);
await syncDirectory(directory);
return resourceRef;
}
await unlink(pending);
} else {
await rename(pending, target);
}
// The directory checks above and this rename are path-based, and Node
// has no `openat`, so a swap between them can land the bytes outside the
// session tree. These checks refuse a publish whose entry is no longer
@@ -149,13 +209,7 @@ export class LocalManagedSessionResourceStore
throw cause;
}
return {
resourceId,
kind: safeKind,
schemaVersion: 1,
byteLength: bytes.byteLength,
digest: digest.digest('hex'),
};
return resourceRef;
}
/**
@@ -2519,7 +2519,10 @@ describe('SessionWriterLease', () => {
lockSchema: { schemaVersion: 3, formatVersion: 2 },
});
expect(replacement.takeoverCommitProof).toEqual(commitProof);
await replacement.release();
await expect(replacement.release()).rejects.toBeInstanceOf(
SessionWriterUnavailableError,
);
await replacement.sealForHandoff(commitProof);
});
it('rebuilds the transcript proof after discarding an uncommitted tail', async () => {
@@ -2623,7 +2626,10 @@ describe('SessionWriterLease', () => {
...fixture.options,
lockSchema: { schemaVersion: 3, formatVersion: 2 },
});
await newer.release();
await expect(newer.release()).rejects.toBeInstanceOf(
SessionWriterUnavailableError,
);
await newer.sealForHandoff(commitProof);
});
it('lets a newer managed format take over an older sealed lock', async () => {
@@ -2649,7 +2655,10 @@ describe('SessionWriterLease', () => {
state: 'active',
format_version: 2,
});
await replacement.release();
await expect(replacement.release()).rejects.toBeInstanceOf(
SessionWriterUnavailableError,
);
await replacement.sealForHandoff(commitProof);
});
});
@@ -335,6 +335,8 @@ export interface AcquireSessionWriterLeaseOptions {
lockSchema?: {
readonly schemaVersion: 3;
readonly formatVersion: number;
/** Local snapshot readers promote a certified seal before replacement. */
readonly resumeFormatVersion?: number;
};
onOwnershipAcquired?: (lease: SessionWriterLease) => void;
}
@@ -1251,6 +1253,21 @@ async function restoreMovedLock(
await fs.unlink(movedPath).catch(() => {});
}
async function syncSnapshotLockDirectory(
lockPath: string,
raw: string,
): Promise<void> {
const record = parseLockRecord(raw);
if (!record || !isManagedLockRecord(record) || record.format_version < 2)
return;
const directory = await openDurableLockDirectory(path.dirname(lockPath));
try {
await syncDurableLockDirectory(directory);
} finally {
await directory.handle.close();
}
}
async function installLockRecord(
lockPath: string,
record: SessionWriterLockRecord,
@@ -1266,10 +1283,14 @@ async function installLockRecord(
handle = undefined;
try {
await fs.link(temporaryPath, lockPath);
await syncSnapshotLockDirectory(lockPath, recordRaw);
return true;
} catch (error) {
const { state } = await inspectExactRecord(lockPath, recordRaw);
if (state === 'exact') return true;
if (state === 'exact') {
await syncSnapshotLockDirectory(lockPath, recordRaw);
return true;
}
if (
state === 'other' ||
(error as NodeJS.ErrnoException).code === 'EEXIST'
@@ -1511,6 +1532,7 @@ async function linkClaimedPrimary(
await assertExactTransitionClaim(claimPath, claimRaw);
try {
await fs.link(sourcePath, lockPath);
await syncSnapshotLockDirectory(lockPath, sourceRaw);
return;
} catch (error) {
const { state } = await inspectClaimedPrimary(
@@ -1518,7 +1540,10 @@ async function linkClaimedPrimary(
sourceRaw,
sessionId,
);
if (state === 'source') return;
if (state === 'source') {
await syncSnapshotLockDirectory(lockPath, sourceRaw);
return;
}
if (state === 'candidate') {
// A schema-v2 acquirer can pass its first claim check immediately
// before this transition creates the claim. It must remove its primary
@@ -1713,6 +1738,7 @@ export class SessionWriterLease {
private readonly claimPath: string;
private readonly managedFormatVersion: number | undefined;
private takenOverCommitProof: SessionWriterCommitProof | undefined;
private replacementFailure: Error | undefined;
private constructor(
private readonly lockPath: string,
@@ -1744,6 +1770,10 @@ export class SessionWriterLease {
* advancing the session; absent when the lease did not take over a Managed
* seal.
*/
get managedJournalFormatVersion(): number | undefined {
return this.managedFormatVersion;
}
get takeoverCommitProof(): SessionWriterCommitProof | undefined {
return this.takenOverCommitProof;
}
@@ -1915,7 +1945,10 @@ export class SessionWriterLease {
const lockSchema = normalizedOptions.lockSchema;
if (
lockSchema !== undefined &&
!isManagedFormatVersion(lockSchema.formatVersion)
(!isManagedFormatVersion(lockSchema.formatVersion) ||
(lockSchema.resumeFormatVersion !== undefined &&
(!isManagedFormatVersion(lockSchema.resumeFormatVersion) ||
lockSchema.resumeFormatVersion < lockSchema.formatVersion)))
) {
throw new SessionWriterUnavailableError({
cause: new Error(
@@ -1923,7 +1956,7 @@ export class SessionWriterLease {
),
});
}
const lockRecord: ActiveLockRecord = {
let lockRecord: ActiveLockRecord = {
...(lockSchema === undefined
? { schema_version: LOCK_SCHEMA_VERSION }
: {
@@ -1976,12 +2009,35 @@ export class SessionWriterLease {
if (normalizedOptions.takeoverPolicy !== 'certified') {
throw new SessionWriterConflictError();
}
return SessionWriterLease.takeOverSealed(
lockPath,
state,
lockRecord,
normalizedOptions,
const resumeFormat = lockSchema?.resumeFormatVersion;
if (
resumeFormat !== undefined &&
isManagedLockRecord(state.record) &&
isManagedLockRecord(lockRecord)
) {
lockRecord = { ...lockRecord, format_version: resumeFormat };
}
return SessionWriterLease.takeOverSealed(lockPath, state, lockRecord, {
...normalizedOptions,
...(resumeFormat === undefined
? {}
: {
lockSchema: { schemaVersion: 3, formatVersion: resumeFormat },
}),
});
}
if (
isManagedLockRecord(state.record) &&
isManagedLockRecord(lockRecord) &&
lockSchema?.resumeFormatVersion !== undefined &&
state.record.format_version <= lockSchema.resumeFormatVersion
) {
const formatVersion = Math.max(
lockRecord.format_version,
state.record.format_version,
);
lockRecord = { ...lockRecord, format_version: formatVersion };
normalizedOptions.lockSchema = { schemaVersion: 3, formatVersion };
}
// A Managed lock may only be continued by a writer that speaks its
// schema: reclaiming or taking it over with a baseline writer would
@@ -2026,6 +2082,40 @@ export class SessionWriterLease {
let primaryInstalled = false;
let staleMoved = false;
const stalePath = `${lockPath}.stale.${process.pid}.${randomUUID()}`;
if (isManagedLockRecord(lockRecord) && lockRecord.format_version >= 2) {
const claimRaw = JSON.stringify(lockRecord);
if (!(await installLockRecord(claimPath, lockRecord))) {
throw new SessionWriterUnavailableError();
}
const current = await SessionWriterLease.inspectExistingLock(
lockPath,
normalizedOptions.sessionId,
);
if (current.kind !== 'stale' || current.raw !== state.raw) {
throw new SessionWriterUnavailableError();
}
await transitionExactPrimary(
lockPath,
current.raw,
claimPath,
claimRaw,
stalePath,
normalizedOptions.sessionId,
claimPath,
claimRaw,
);
const lease = await SessionWriterLease.finishAcquisition(
lockPath,
lockRecord,
normalizedOptions,
);
await releaseTransitionClaim(claimPath, claimRaw);
await removeExactRecord(stalePath, current.raw).catch(() => undefined);
await removeOwnedLock(reclaimPath, lockRecord.owner_id).catch(
() => undefined,
);
return lease;
}
try {
await assertPathMissing(claimPath);
const currentState = await SessionWriterLease.inspectExistingLock(
@@ -2293,6 +2383,7 @@ export class SessionWriterLease {
}
return lease;
} catch (error) {
if ((lease.managedFormatVersion ?? 0) >= 2) throw error;
let releaseError: unknown;
for (let attempt = 0; attempt < 2; attempt++) {
if (attempt === 1) lease.releaseRetryable = false;
@@ -2557,68 +2648,95 @@ export class SessionWriterLease {
}
replaceTranscript(text: string, expectedSha256: string): Promise<void> {
return this.runExclusive(async () => {
return this.runExclusive(() =>
this.replaceTranscriptOnce(text, expectedSha256, false),
);
}
replaceManagedTranscript(
text: string,
expectedSha256: string,
verify: () => void,
): Promise<void> {
return this.runExclusive(() =>
this.replaceTranscriptOnce(text, expectedSha256, true, verify),
);
}
private async replaceTranscriptOnce(
text: string,
expectedSha256: string,
managed: boolean,
verify?: () => void,
): Promise<void> {
await this.assertOwnedAndUnchangedOnce();
if (
(managed
? this.managedFormatVersion !== 2
: this.managedFormatVersion !== undefined) ||
this.expectedTranscriptHasher!.copy().digest('hex') !== expectedSha256
) {
throw new SessionTranscriptChangedError();
}
const directoryPath = path.dirname(this.transcriptPath);
await fs.mkdir(directoryPath, { recursive: true, mode: 0o700 });
const directory = await openDurableLockDirectory(directoryPath);
const temporaryPath = path.join(
directoryPath,
`.${randomUUID()}.jsonl.tmp`,
);
let temporary: fs.FileHandle | undefined;
let replaced = false;
let replacementAttempted = false;
try {
temporary = await fs.open(temporaryPath, 'wx', 0o600);
const mode = this.expectedTranscriptState?.exists
? this.expectedTranscriptState.fingerprint.mode
: undefined;
if (mode !== undefined) await temporary.chmod(mode & 0o777);
await temporary.writeFile(text, 'utf8');
await temporary.sync();
const replacementState = transcriptStateFromStat(await temporary.stat());
await temporary.close();
temporary = undefined;
await syncDurableLockDirectory(directory);
await this.assertOwnedAndUnchangedOnce();
if (
this.managedFormatVersion !== undefined ||
this.expectedTranscriptHasher!.copy().digest('hex') !== expectedSha256
) {
replacementAttempted = true;
await fs.rename(temporaryPath, this.transcriptPath);
replaced = true;
await syncDurableLockDirectory(directory);
const snapshot = await captureTranscriptSnapshot(
this.transcriptPath,
replacementState,
() => this.released,
);
const expectedHash = createHash('sha256').update(text).digest('hex');
if (snapshot.hasher.copy().digest('hex') !== expectedHash) {
throw new SessionTranscriptChangedError();
}
const directoryPath = path.dirname(this.transcriptPath);
await fs.mkdir(directoryPath, { recursive: true, mode: 0o700 });
const directory = await openDurableLockDirectory(directoryPath);
const temporaryPath = path.join(
directoryPath,
`.${randomUUID()}.jsonl.tmp`,
);
let temporary: fs.FileHandle | undefined;
let replaced = false;
try {
temporary = await fs.open(temporaryPath, 'wx', 0o600);
const mode = this.expectedTranscriptState?.exists
? this.expectedTranscriptState.fingerprint.mode
: undefined;
if (mode !== undefined) await temporary.chmod(mode & 0o777);
await temporary.writeFile(text, 'utf8');
await temporary.sync();
const replacementState = transcriptStateFromStat(
await temporary.stat(),
);
await temporary.close();
temporary = undefined;
await syncDurableLockDirectory(directory);
await this.assertOwnedAndUnchangedOnce();
await fs.rename(temporaryPath, this.transcriptPath);
replaced = true;
await syncDurableLockDirectory(directory);
const snapshot = await captureTranscriptSnapshot(
this.transcriptPath,
replacementState,
() => this.released,
);
const expectedHash = createHash('sha256').update(text).digest('hex');
if (snapshot.hasher.copy().digest('hex') !== expectedHash) {
throw new SessionTranscriptChangedError();
}
await this.readOwnedLock();
this.expectedTranscriptState = snapshot.state;
this.expectedTranscriptHasher = snapshot.hasher;
} catch (error) {
if (error instanceof SessionWriterError) throw error;
if (!replaced) {
await this.assertOwnedAndUnchangedOnce();
throw error;
}
throw new SessionWriterUnavailableError({
cause: error instanceof Error ? error : undefined,
});
} finally {
await temporary?.close().catch(() => undefined);
await fs.unlink(temporaryPath).catch(() => undefined);
await directory.handle.close().catch(() => undefined);
await this.readOwnedLock();
this.expectedTranscriptState = snapshot.state;
this.expectedTranscriptHasher = snapshot.hasher;
verify?.();
if (managed) await this.assertOwnedAndUnchangedOnce();
} catch (error) {
if (managed && replacementAttempted) {
this.replacementFailure =
error instanceof Error ? error : new Error(String(error));
}
});
if (error instanceof SessionWriterError) throw error;
if (!replaced) {
await this.assertOwnedAndUnchangedOnce();
throw error;
}
throw new SessionWriterUnavailableError({
cause: error instanceof Error ? error : undefined,
});
} finally {
await temporary?.close().catch(() => undefined);
await fs.unlink(temporaryPath).catch(() => undefined);
await directory.handle.close().catch(() => undefined);
}
}
/**
@@ -2919,6 +3037,15 @@ export class SessionWriterLease {
release(): Promise<void> {
if (this.terminalOperation === 'seal') return this.terminalPromise!;
if ((this.managedFormatVersion ?? 0) >= 2 && !this.released) {
return Promise.reject(
new SessionWriterUnavailableError({
cause: new Error(
'A snapshot writer must preserve its seal or active fence.',
),
}),
);
}
this.terminalOperation = 'release';
if (this.terminalPromise) return this.terminalPromise;
const terminal = this.runExclusive(() => this.releaseOnce()).catch(
@@ -2980,7 +3107,15 @@ export class SessionWriterLease {
}
private runExclusive<T>(operation: () => Promise<T>): Promise<T> {
const pending = this.operationTail.then(operation, operation);
const checked = () => {
if (this.replacementFailure) {
throw new SessionWriterUnavailableError({
cause: this.replacementFailure,
});
}
return operation();
};
const pending = this.operationTail.then(checked, checked);
this.operationTail = pending.then(
() => undefined,
() => undefined,
@@ -0,0 +1,632 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { createHash } from 'node:crypto';
import fs from 'node:fs';
import path from 'node:path';
import { openSyncNoFollow } from './no-follow-open.js';
import {
MANAGED_SESSION_LIMITS,
ManagedSessionRecordError,
} from './managed-session-format.js';
import type {
ManagedSessionDurableRef,
ManagedSessionHeader,
ManagedSessionJsonValue,
ManagedSessionKey,
} from '../managed-runtime/managed-session-records.js';
export const LOCAL_MANAGED_SNAPSHOT_KIND = 'managed-local-journal-snapshot';
export const LOCAL_MANAGED_SNAPSHOT_PAGE_KIND =
'managed-local-journal-snapshot-page';
export const LOCAL_MANAGED_SNAPSHOT_BASE = 'managed_session_snapshot_base_v1';
export const LOCAL_MANAGED_SNAPSHOT_FORMAT =
'qwen-managed-local-journal-snapshot/1';
export const LOCAL_MANAGED_SNAPSHOT_PAGE_BYTES = 16 * 1024 * 1024;
export const LOCAL_MANAGED_JOURNAL_FORMAT = 2;
export interface LocalManagedJournalManifest {
readonly format: typeof LOCAL_MANAGED_SNAPSHOT_FORMAT;
readonly sessionKey: ManagedSessionKey;
readonly pages: readonly ManagedSessionDurableRef[];
readonly byteLength: number;
readonly digest: string;
readonly committedSequence: number;
readonly lastCommitDigest: string;
readonly lastRecordUuid: string;
readonly resources: readonly ManagedSessionDurableRef[];
readonly titleRef: ManagedSessionDurableRef | null;
readonly sourceRef: ManagedSessionDurableRef | null;
}
export interface LocalManagedJournalBase {
readonly manifest: ManagedSessionDurableRef;
readonly header: ManagedSessionHeader;
readonly headerEnd: number;
readonly baseEnd: number;
}
export function localJournalDigest(bytes: Buffer): string {
return createHash('sha256').update(bytes).digest('hex');
}
export function localJournalObject(value: unknown): Record<string, unknown> {
if (value === null || typeof value !== 'object' || Array.isArray(value)) {
throw new ManagedSessionRecordError(
'Local journal snapshot is not an object.',
);
}
return value as Record<string, unknown>;
}
function requireSnapshot(
condition: boolean,
detail: string,
): asserts condition {
if (!condition)
throw new ManagedSessionRecordError(`Local journal snapshot ${detail}.`);
}
export function localJournalRef(value: unknown): ManagedSessionDurableRef {
return assertManagedSessionDurableRef(value as ManagedSessionJsonValue);
}
/** Read one physical generation, including when a frozen byte bound is supplied. */
export function readLocalJournalBytes(
filePath: string,
maxBytes?: number,
expectedBytes?: number,
): Buffer {
const fd = openSyncNoFollow(filePath);
try {
const before = fs.fstatSync(fd);
requireSnapshot(before.isFile(), 'is not a regular file');
requireSnapshot(
expectedBytes === undefined || before.size === expectedBytes,
'resource length differs',
);
const size =
maxBytes === undefined ? before.size : Math.min(maxBytes, before.size);
requireSnapshot(
Number.isSafeInteger(size) && size >= 0,
'byte bound is invalid',
);
const bytes = readDescriptorPrefix(fd, size);
const after = fs.fstatSync(fd);
const current = fs.lstatSync(filePath);
requireSnapshot(
!current.isSymbolicLink() &&
current.isFile() &&
current.dev === before.dev &&
current.ino === before.ino &&
after.size >= size &&
((after.ctimeMs === before.ctimeMs &&
after.mtimeMs === before.mtimeMs) ||
(maxBytes !== undefined &&
after.size > before.size &&
readDescriptorPrefix(fd, size).equals(bytes))),
'file changed while reading',
);
return bytes;
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
throw new ManagedSessionRecordError(
'Local journal generation disappeared while reading.',
);
}
throw error;
} finally {
fs.closeSync(fd);
}
}
function readDescriptorPrefix(fd: number, size: number): Buffer {
const bytes = Buffer.alloc(size);
let offset = 0;
while (offset < size) {
const count = fs.readSync(fd, bytes, offset, size - offset, offset);
requireSnapshot(count > 0, 'file ended before its frozen boundary');
offset += count;
}
return bytes;
}
export function parseLocalJournalBase(
bytes: Buffer,
sessionKey: ManagedSessionKey,
): LocalManagedJournalBase | undefined {
let header: ManagedSessionHeader | undefined;
let headerEnd = 0;
let offset = 0;
while (offset < bytes.length) {
const start = offset;
const end = bytes.indexOf(10, offset);
if (end < 0) break;
const line = bytes.subarray(offset, end).toString('utf8');
if (line.length === 0) return undefined;
let record: Record<string, unknown>;
try {
record = localJournalObject(
parseManagedSessionRecordJson(
line,
MANAGED_SESSION_LIMITS.maxEventBytes,
),
);
} catch {
return undefined;
}
offset = end + 1;
if (record['subtype'] === 'managed_session_header_v1') {
if (
localJournalObject(record['managedSession'])['minimumReader'] !==
'managed-session/2'
)
return undefined;
requireSnapshot(header === undefined, 'header is repeated');
requireSnapshot(
end - start <= MANAGED_SESSION_LIMITS.maxHeaderBytes,
'header exceeds its byte limit',
);
header = parseManagedSessionHeader(record['managedSession']);
requireSnapshot(
managedSessionKeysEqual(header.sessionKey, sessionKey),
'scope differs',
);
headerEnd = offset;
continue;
}
if (record['subtype'] === LOCAL_MANAGED_SNAPSHOT_BASE) {
requireSnapshot(
header !== undefined && headerEnd === start,
'base precedes its header',
);
requireSnapshot(
header.minimumReader === 'managed-session/2',
'base has no reader barrier',
);
const body = localJournalObject(record['managedSessionSnapshot']);
requireSnapshot(
body['format'] === LOCAL_MANAGED_SNAPSHOT_FORMAT &&
Object.keys(body).length === 3 &&
record['type'] === 'system' &&
typeof record['uuid'] === 'string' &&
record['uuid'].length > 0 &&
(typeof record['parentUuid'] === 'string' ||
record['parentUuid'] === null) &&
record['sessionId'] === sessionKey.sessionId,
'base fields are invalid',
);
requireSnapshot(
managedSessionKeysEqual(
assertManagedSessionKey(
body['sessionKey'] as ManagedSessionJsonValue,
),
sessionKey,
),
'base scope differs',
);
const manifest = localJournalRef(body['manifest']);
requireSnapshot(
manifest.kind === LOCAL_MANAGED_SNAPSHOT_KIND &&
manifest.resourceId === manifest.digest,
'manifest ref is invalid',
);
return { manifest, header, headerEnd, baseEnd: offset };
}
if (header !== undefined) {
requireSnapshot(
header.minimumReader !== 'managed-session/2',
'base is missing',
);
return undefined;
}
}
requireSnapshot(
header?.minimumReader !== 'managed-session/2',
'base is incomplete',
);
return undefined;
}
export function readLocalJournalResource(
runtimeBaseDir: string,
sessionKey: ManagedSessionKey,
ref: ManagedSessionDurableRef,
): Buffer {
const root = path.join(runtimeBaseDir, 'resources', sessionKey.sessionId);
for (const segment of [sessionKey.sessionId, ref.kind, ref.resourceId]) {
requireSnapshot(
segment.length > 0 &&
segment !== '.' &&
segment !== '..' &&
!/[/\\]/.test(segment),
'resource path is invalid',
);
}
for (const directory of [
path.dirname(root),
root,
path.join(root, ref.kind),
]) {
const stat = fs.lstatSync(directory);
requireSnapshot(
stat.isDirectory() && !stat.isSymbolicLink(),
'resource directory is not private',
);
}
requireSnapshot(
ref.byteLength <= 64 * 1024 * 1024,
'resource exceeds its byte limit',
);
const bytes = readLocalJournalBytes(
path.join(root, ref.kind, ref.resourceId),
undefined,
ref.byteLength,
);
requireSnapshot(
bytes.length === ref.byteLength && localJournalDigest(bytes) === ref.digest,
'resource length or digest differs',
);
return bytes;
}
export function readLocalJournalManifest(
runtimeBaseDir: string,
sessionKey: ManagedSessionKey,
ref: ManagedSessionDurableRef,
): LocalManagedJournalManifest {
requireSnapshot(
ref.kind === LOCAL_MANAGED_SNAPSHOT_KIND &&
ref.resourceId === ref.digest &&
ref.schemaVersion === 1,
'manifest reference is invalid',
);
const value = localJournalObject(
parseManagedSessionRecordJson(
readLocalJournalResource(runtimeBaseDir, sessionKey, ref).toString(
'utf8',
),
ref.byteLength,
),
);
requireSnapshot(
value['format'] === LOCAL_MANAGED_SNAPSHOT_FORMAT &&
Object.keys(value).length === 11,
'manifest fields are invalid',
);
const key = assertManagedSessionKey(
value['sessionKey'] as ManagedSessionJsonValue,
);
requireSnapshot(
managedSessionKeysEqual(key, sessionKey),
'manifest scope differs',
);
const list = (field: string) => {
const items = value[field];
requireSnapshot(Array.isArray(items), `${field} is not an array`);
return items.map(localJournalRef);
};
const pages = list('pages');
requireSnapshot(
pages.length > 0 &&
pages.every(
(page) =>
page.kind === LOCAL_MANAGED_SNAPSHOT_PAGE_KIND &&
page.schemaVersion === 1 &&
page.resourceId === page.digest &&
page.byteLength <= LOCAL_MANAGED_SNAPSHOT_PAGE_BYTES,
),
'page reference is invalid',
);
requireSnapshot(
Number.isSafeInteger(value['byteLength']) &&
(value['byteLength'] as number) > 0 &&
Number.isSafeInteger(value['committedSequence']) &&
(value['committedSequence'] as number) > 0 &&
typeof value['digest'] === 'string' &&
/^[0-9a-f]{64}$/.test(value['digest']) &&
typeof value['lastCommitDigest'] === 'string' &&
/^[0-9a-f]{64}$/.test(value['lastCommitDigest']) &&
typeof value['lastRecordUuid'] === 'string',
'manifest head is invalid',
);
return {
format: LOCAL_MANAGED_SNAPSHOT_FORMAT,
sessionKey: key,
pages,
byteLength: value['byteLength'] as number,
digest: value['digest'],
committedSequence: value['committedSequence'] as number,
lastCommitDigest: value['lastCommitDigest'],
lastRecordUuid: value['lastRecordUuid'],
resources: list('resources'),
titleRef:
value['titleRef'] === null ? null : localJournalRef(value['titleRef']),
sourceRef:
value['sourceRef'] === null ? null : localJournalRef(value['sourceRef']),
};
}
/** Metadata readers need only the manifest and current physical tail. */
export function readLocalSnapshotDomainBody(
bytes: Buffer,
runtimeBaseDir: string,
domain: 'session_metadata' | 'session_source',
): { body?: unknown } | undefined {
let header: ManagedSessionHeader | undefined;
for (const line of bytes.toString('utf8').split('\n')) {
if (!line) break;
const record = localJournalObject(JSON.parse(line));
if (record['subtype'] === 'managed_session_header_v1') {
header = parseManagedSessionHeader(record['managedSession']);
break;
}
}
if (!header) return undefined;
const base = parseLocalJournalBase(bytes, header.sessionKey);
if (!base) return undefined;
const manifest = readLocalJournalManifest(
runtimeBaseDir,
header.sessionKey,
base.manifest,
);
const pending = new Map<number, ManagedSessionDurableRef>();
let latest =
domain === 'session_metadata' ? manifest.titleRef : manifest.sourceRef;
const lines = bytes.subarray(base.baseEnd).toString('utf8').split('\n');
lines.pop();
for (const line of lines) {
const record = localJournalObject(JSON.parse(line));
const body = localJournalObject(record['managedSession']);
if (
record['subtype'] === 'managed_session_event_v1' &&
body['kind'] === 'domain.committed'
) {
const payload = localJournalObject(body['payload']);
if (payload['domain'] === domain)
pending.set(
body['sequence'] as number,
localJournalRef(payload['recordRef']),
);
}
if (record['subtype'] === 'managed_session_commit_v1') {
for (const [sequence, ref] of pending) {
if (
sequence >= (body['firstSequence'] as number) &&
sequence <= (body['lastSequence'] as number)
)
latest = ref;
}
pending.clear();
}
}
return latest
? {
body: JSON.parse(
readLocalJournalResource(
runtimeBaseDir,
header.sessionKey,
latest,
).toString('utf8'),
) as unknown,
}
: {};
}
/** Ownership and a maximal header need not fit together in a display window. */
export function readLocalJournalHeader(
filePath: string,
): ManagedSessionHeader | undefined {
for (const { line, complete } of localJournalPrefixLines(filePath)) {
if (!complete) return undefined;
const record = localJournalObject(
parseManagedSessionRecordJson(
line.toString('utf8'),
MANAGED_SESSION_LIMITS.maxEventBytes,
),
);
if (record['subtype'] === 'managed_session_header_v1') {
requireSnapshot(
line.length <= MANAGED_SESSION_LIMITS.maxHeaderBytes,
'header exceeds its byte limit',
);
return parseManagedSessionHeader(record['managedSession']);
}
if (record['subtype'] !== 'session_execution_engine') return undefined;
}
return undefined;
}
export function readLocalJournalFirstLine(
filePath: string,
): string | undefined {
for (const { line } of localJournalPrefixLines(filePath))
return line.toString('utf8');
return undefined;
}
function* localJournalPrefixLines(
filePath: string,
): Generator<{ line: Buffer; complete: boolean }> {
const fd = openSyncNoFollow(filePath);
try {
const chunk = Buffer.alloc(64 * 1024);
let pending = Buffer.alloc(0);
for (;;) {
const count = fs.readSync(fd, chunk, 0, chunk.length, null);
if (!count) {
if (pending.length) yield { line: pending, complete: false };
return;
}
pending = Buffer.concat([pending, chunk.subarray(0, count)]);
let end: number;
while ((end = pending.indexOf(10)) >= 0) {
const line = pending.subarray(0, end);
requireSnapshot(
line.length <= MANAGED_SESSION_LIMITS.maxEventBytes,
'ownership record exceeds its byte limit',
);
pending = pending.subarray(end + 1);
yield { line, complete: true };
}
requireSnapshot(
pending.length <= MANAGED_SESSION_LIMITS.maxEventBytes,
'ownership record exceeds its byte limit',
);
}
} finally {
fs.closeSync(fd);
}
}
export function localJournalPhysicalPrefix(
bytes: Buffer,
header: ManagedSessionHeader,
): Buffer {
const lines: string[] = [];
for (const line of bytes.toString('utf8').split('\n')) {
const record = localJournalObject(JSON.parse(line));
if (record['subtype'] === 'managed_session_header_v1') {
lines.push(
JSON.stringify({
...record,
managedSession: { ...header, minimumReader: 'managed-session/2' },
}),
);
return Buffer.from(`${lines.join('\n')}\n`);
}
lines.push(line);
}
throw new ManagedSessionRecordError('Local snapshot source has no header.');
}
function parseManagedSessionRecordJson(text: string, maximum: number): unknown {
requireSnapshot(
Buffer.byteLength(text) <= maximum,
'record exceeds its byte limit',
);
let value: unknown;
try {
value = JSON.parse(text);
} catch {
throw new ManagedSessionRecordError(
'Local journal snapshot JSON is invalid.',
);
}
const pending: Array<{ value: unknown; depth: number }> = [
{ value, depth: 1 },
];
while (pending.length) {
const item = pending.pop()!;
if (typeof item.value === 'number')
requireSnapshot(Number.isFinite(item.value), 'number is not finite');
if (item.value !== null && typeof item.value === 'object') {
requireSnapshot(
item.depth <= MANAGED_SESSION_LIMITS.maxJsonDepth,
'JSON exceeds its depth limit',
);
for (const child of Object.values(item.value))
pending.push({ value: child, depth: item.depth + 1 });
}
}
return value;
}
function assertManagedSessionKey(value: unknown): ManagedSessionKey {
const key = localJournalObject(value);
requireSnapshot(
Object.keys(key).length === 3 &&
[key['tenantId'], key['workspaceId'], key['sessionId']].every(
(part) =>
typeof part === 'string' &&
part.length > 0 &&
Buffer.byteLength(part) <= 512 &&
[...part].every(
(character) =>
character.charCodeAt(0) >= 32 && character.charCodeAt(0) !== 127,
),
),
'scope is invalid',
);
return {
tenantId: key['tenantId'] as string,
workspaceId: key['workspaceId'] as string,
sessionId: key['sessionId'] as string,
};
}
function managedSessionKeysEqual(
a: ManagedSessionKey,
b: ManagedSessionKey,
): boolean {
return (
a.tenantId === b.tenantId &&
a.workspaceId === b.workspaceId &&
a.sessionId === b.sessionId
);
}
function assertManagedSessionDurableRef(
value: unknown,
): ManagedSessionDurableRef {
const ref = localJournalObject(value);
requireSnapshot(
Object.keys(ref).length === 5 &&
[ref['resourceId'], ref['kind']].every(
(part) =>
typeof part === 'string' &&
part.length > 0 &&
Buffer.byteLength(part) <= 512,
) &&
Number.isSafeInteger(ref['schemaVersion']) &&
(ref['schemaVersion'] as number) > 0 &&
Number.isSafeInteger(ref['byteLength']) &&
(ref['byteLength'] as number) >= 0 &&
typeof ref['digest'] === 'string' &&
/^[0-9a-f]{64}$/.test(ref['digest']),
'resource reference is invalid',
);
return {
resourceId: ref['resourceId'] as string,
kind: ref['kind'] as string,
schemaVersion: ref['schemaVersion'] as number,
byteLength: ref['byteLength'] as number,
digest: ref['digest'] as string,
};
}
// The filesystem envelope is checked here; native replay validates the complete source header.
function parseManagedSessionHeader(value: unknown): ManagedSessionHeader {
const header = localJournalObject(value);
requireSnapshot(
header['formatVersion'] === 1 &&
header['engine'] === 'managed' &&
['managed-session/1', 'managed-session/2'].includes(
String(header['minimumReader']),
) &&
typeof header['createdBy'] === 'string',
'physical header is invalid',
);
return {
formatVersion: 1,
minimumReader: header['minimumReader'] as string,
engine: 'managed',
sessionKey: assertManagedSessionKey(header['sessionKey']),
definitionRef: localJournalRef(header['definitionRef']),
rootSnapshotRef: localJournalRef(header['rootSnapshotRef']),
createdBy: header['createdBy'],
...(header['baseTranscriptProof'] === undefined
? {}
: {
baseTranscriptProof: localJournalRef(header['baseTranscriptProof']),
}),
};
}
@@ -0,0 +1,30 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
export const MANAGED_SESSION_LIMITS = {
maxIdBytes: 512,
// Not frozen by the storage spec: a bound for free-form enum-adjacent fields
// such as `source` or `stopReason`, borrowed from its error-message cap.
maxTextBytes: 4096,
maxTimeMs: 8_640_000_000_000_000,
maxJsonDepth: 64,
maxHeaderBytes: 64 * 1024,
maxEventBytes: 1024 * 1024,
maxCommitMarkerBytes: 64 * 1024,
maxTransactionEvents: 256,
maxTransactionBytes: 8 * 1024 * 1024,
defaultReadEvents: 100,
maxReadEvents: 256,
} as const;
export class ManagedSessionRecordError extends Error {
readonly code: string = 'managed_session_invalid_record';
constructor(message: string) {
super(message);
this.name = 'ManagedSessionRecordError';
}
}
+51 -8
View File
@@ -20,6 +20,13 @@ import {
parseLineTolerantWithIntegrity,
} from './jsonl-utils.js';
import { openSyncNoFollow } from './no-follow-open.js';
import {
readLocalJournalBytes,
readLocalJournalFirstLine,
readLocalJournalHeader,
readLocalSnapshotDomainBody,
} from './local-managed-journal-snapshot.js';
import { ManagedSessionRecordError } from './managed-session-format.js';
/** Size of the head/tail buffer for lite metadata reads (64KB). */
export const LITE_READ_BUF_SIZE = 64 * 1024;
@@ -892,7 +899,13 @@ export function isManagedSessionTranscriptSync(
scratchBuffer?: Buffer,
): boolean {
const head = readTranscriptHeadSync(filePath, scratchBuffer);
return head !== undefined && headCarriesManagedHeader(head, filePath);
if (head === undefined) return false;
if (headCarriesManagedHeader(head, filePath)) return true;
try {
return readLocalJournalHeader(filePath) !== undefined;
} catch {
return false;
}
}
/** True for a parsed record that is itself the Managed Session header. */
@@ -930,8 +943,8 @@ function headCarriesManagedHeader(head: string, filePath: string): boolean {
/**
* True when the transcript positively identifies as owned by the Managed
* engine: it carries the Managed Session header, or an execution-engine owner
* record naming `managed`. A Managed owner is written before anything else,
* so the head window holds it.
* record naming `managed`. A large first owner may extend beyond the display
* window, so it is read as one bounded control record when needed.
*
* Lines are parsed the way the owner reader parses them, so an owner record
* that parses whole counts even when another record shares its line. A line
@@ -966,7 +979,7 @@ export function readManagedExecutionEvidenceSync(
// One parse pass serves both gates; a line is parsed only when it could
// name the header subtype (the wrapped marker) or the owner subtype (the
// bare string), literally or through a `\u` escape.
return head.split('\n').some((line) => {
const found = head.split('\n').some((line) => {
if (
!line.includes(MANAGED_HEADER_MARKER) &&
!line.includes(EXECUTION_ENGINE_SUBTYPE) &&
@@ -979,6 +992,20 @@ export function readManagedExecutionEvidenceSync(
records.some(isManagedHeaderRecord) || records.some(isManagedOwnerRecord)
);
});
if (found) return true;
try {
const first = readLocalJournalFirstLine(filePath);
if (first === undefined) return false;
const { records } = parseLineTolerantWithIntegrity(first, filePath);
return (
records.some(isManagedHeaderRecord) || records.some(isManagedOwnerRecord)
);
} catch (error) {
return error instanceof ManagedSessionRecordError ||
(error as NodeJS.ErrnoException).code === 'ENOENT'
? false
: undefined;
}
}
/**
@@ -1050,8 +1077,8 @@ function readTranscriptHeadOrThrowSync(
* renamed, which genuinely has no custom title. Display stays tolerant: an
* unreadable body yields `{}` rather than throwing into the session list.
*
* Like the legacy reader this only scans the tail and head windows, so a rename
* buried in the middle of a very long log is not found.
* Ordinary layouts use head/tail windows; snapshot layouts read their complete
* committed tail and manifest anchor.
*/
export function readManagedSessionTitleInfoSync(
filePath: string,
@@ -1126,8 +1153,8 @@ export function readManagedSessionSourceSync(
* nothing to report" (`{}`). Display stays tolerant: an unreadable body is the
* latter rather than a throw into the session list.
*
* Like the legacy readers this only scans the tail and head windows, so a
* record buried in the middle of a very long log is not found.
* Ordinary layouts use head/tail windows. Snapshot layouts scan the complete
* committed physical tail before falling back to the manifest anchor.
*/
function readManagedDomainBodySync(
filePath: string,
@@ -1154,6 +1181,22 @@ function readManagedDomainBodySync(
const header = findManagedHeaderRecord(headText, filePath) as
| { managedSession?: { sessionKey?: { sessionId?: unknown } } }
| undefined;
let localHeader;
try {
localHeader = readLocalJournalHeader(filePath);
} catch {
if (header === undefined) return undefined;
}
if (localHeader?.minimumReader === 'managed-session/2') {
const snapshotBody = readLocalSnapshotDomainBody(
readLocalJournalBytes(filePath),
runtimeBaseDir,
marker === MANAGED_METADATA_MARKER
? 'session_metadata'
: 'session_source',
);
if (snapshotBody !== undefined) return snapshotBody;
}
if (header === undefined) return undefined;
/* The recorded identity, not the file name, decides where the resources
@@ -113,6 +113,7 @@ const MANAGED_SESSION_RECORD_SUBTYPES = new Set([
'managed_session_header_v1',
'managed_session_event_v1',
'managed_session_commit_v1',
'managed_session_snapshot_base_v1',
]);
const NON_CONVERSATION_RECORD_SUBTYPES = new Set([