feat(core): Add durable terminal identity lookup (#15566)

Add journal-validated terminal query records for local activation and inbox duplicate replay, preserving terminal commit acknowledgements.

Co-authored-by: Qwen-Coder <[email protected]>

docs/design/2026-10-19-managed-session-bounded-retention.md

@@ -154,6 +154,11 @@ These are required serial follow-ups, not implemented by slice 1:
    wake/cancel recovery, tool results, tasks, hooks, automations, channels,
    transcript projection, and SSE must preserve their historical answers.
 
+The [local terminal-identity lookup slice](2026-10-26-managed-terminal-identity-lookup.md)
+adds a derived per-key query used by terminal duplicate admission. It keeps the
+journal authoritative and retains resident maps; an independently authoritative
+archive and its completeness boundary remain prerequisites for eviction.
+
 ## Validation plan
 
 - Reproduce the baseline using the globally installed CLI where supported and

docs/design/2026-10-19-managed-session-bounded-retention.zh-CN.md

@@ -124,6 +124,10 @@ Rename 前崩溃保留旧 journal，之后崩溃保留新快照；两者描述
    Approval/model/Plan 选择、wake/cancel 恢复、工具结果、task、hook、automation、
    channel、transcript 投影和 SSE 都必须保持历史查询语义。
 
+[本地终态身份查询切片](2026-10-26-managed-terminal-identity-lookup.zh-CN.md)
+增加派生的按 key 查询，并用于终态重复准入。它保留 journal 权威与驻留 Map；
+独立权威的归档及完整性边界仍是淘汰前置条件。
+
 ## 验证计划
 
 - 在全局 CLI 支持的范围内复现基线；CLI 不暴露这些适配器方法时使用直接存储脚本。

docs/design/2026-10-26-managed-terminal-identity-lookup.md

@@ -0,0 +1,81 @@
+# Managed activation and inbox terminal identity lookup
+
+[English](2026-10-26-managed-terminal-identity-lookup.md) | [简体中文](2026-10-26-managed-terminal-identity-lookup.zh-CN.md)
+
+## Problem and current state
+
+Issue [#13184](https://github.com/QwenLM/qwen-code/issues/13184) tracks growth of Managed state. The [bounded-retention design](2026-10-19-managed-session-bounded-retention.md) requires durable historical identity queries before any terminal state can leave memory. This is PR1 of that remaining work.
+
+`FileManagedActivationStore` and `FileManagedSessionInbox` already preserve terminal results in their authoritative journals and snapshots. Opening them rebuilds every identity in memory. Duplicate admission consults those resident maps; there is no per-identity durable terminal lookup. Activation/inbox transition compaction does not bound the number of terminal identities.
+
+These are single-process local adapters. Their consumers are `EmbeddedHarnessScheduler` and `ManagedPromptAdmissionController`; neither is wired into the public Hosted application. The native HTTP Session journal is a different store. This change does not claim to bound Hosted service memory or startup reads.
+
+## Goals and scope
+
+- Materialize and read a complete immutable terminal result by its tenant, Session and activation/message identity.
+- Preserve original normalized data, queue/admission position, timestamps, outcome, cancellation state and processing fence.
+- Keep exact duplicate replay and conflicting identity refusal unchanged, including when current admission limits are full.
+- Establish a usable durable query path without changing terminal commit acknowledgements.
+
+Memory eviction, identity expiration, journal deletion, native Session snapshots, Java/HTTP APIs and automatic inbox completion are follow-up work. Resident maps and existing synchronous reads remain unchanged.
+
+## Design
+
+### Authority and index layout
+
+The existing journal remains the sole authority. A sibling `<journal>.terminal-identities` directory contains derived immutable records. Each filename is the SHA-256 of the JSON-encoded identity tuple `[tenantId, sessionId, activationId/messageId]`. Activation and inbox records live beside their respective journals and include distinct record kinds.
+
+Each version-1 record contains its kind, full identity tuple, original terminal snapshot and the snapshot's SHA-256 digest. Encoding uses the original journal-derived snapshot, never the retry's property order. A read must match the complete canonical expected record, including identity, type and digest. Missing files may be rebuilt only from an already validated journal terminal. Corrupt or conflicting files cause the lookup to fail; they are never treated as a new identity or silently overwritten.
+
+An orphan side record cannot establish an identity absent from the journal, override an active state or restore a missing journal. The index is not independently authoritative. Future eviction must first introduce a durable completeness/publication boundary; these files alone do not permit eviction.
+
+### Read-through and real consumers
+
+Both adapters expose an asynchronous `readTerminal` lookup. It runs in the adapter's existing serial queue, validates and captures the requested identity, and returns no result for missing or non-terminal journal state. For a known terminal state it reads or materializes the per-key record and returns the original cloned snapshot.
+
+The existing duplicate paths in `enqueue` and `admit` use the same read-through for terminal records after their current normalization, limit validation and deep-equality conflict checks. Non-terminal duplicates keep their existing path. Reordered object keys and normalized numbers continue to compare by the existing semantic rules; request JSON bytes are not an idempotency key.
+
+This also preserves reconciliation: resubmitting an unfinished inbox entry whose activation is already released returns that original activation and cannot queue the effect again. Reconciliation still does not finish the inbox automatically.
+
+### Publication and failure behavior
+
+Materialization happens only during terminal lookup or duplicate replay. `release`, `cancelQueued`, `finish` and cancellation continue committing only their original journals. An index write failure therefore cannot turn an already committed terminal mutation into a reported release/finish failure.
+
+The index directory is created with mode `0700`; records use `0600`. Parent/index directories must be real directories and leaf reads use the existing no-follow opener. A bounded read checks the expected record length before allocation. These checks preserve the trusted single-owner boundary; they do not establish isolation against a hostile same-UID process.
+
+A record is written to a unique pending file, flushed, linked exclusively to its final name and followed by a directory sync on supported platforms. An existing final name is read and validated instead of replaced. Pending files are removed in normal cleanup and never consulted as results. A crash may leave an ignored pending file; repeated process crashes can accumulate these files, and orphan reclamation remains follow-up work.
+
+A failed query rejects without changing the authoritative journal, admission counters or resident result. A missing derived file is rebuilt on a later query; conflicting/corrupt data continues to reject. No new availability fallback or terminal-transition retry behavior is introduced.
+
+## Affected components
+
+| Component                            | Change                                                                   |
+| ------------------------------------ | ------------------------------------------------------------------------ |
+| `managed-terminal-identity-store.ts` | Shared derived index, bounded validation and immutable publication.      |
+| `managed-activation-store.ts`        | Serialized terminal lookup and released-identity duplicate read-through. |
+| `managed-session-inbox.ts`           | Serialized terminal lookup and finished-message duplicate read-through.  |
+| Focused store/admission tests        | Persistence, normalization, refusal, faults and reconciliation.          |
+
+No dependency, configuration switch, migration or public protocol is added. Existing version-1 journals and version-2 transition snapshots remain readable by old and new code. Old code ignores the sibling index, which is safe because PR1 retains all authoritative identities.
+
+## Validation and acceptance
+
+- Confirm the installed global CLI cannot expose this internal adapter API, then use a direct adapter script for the baseline and local verification.
+- Verify lookup and original-result replay after process restart for completed/failed activations, queued cancellation, finished inbox messages and cancellation without a processing fence.
+- Verify normalized/reordered payload retries, conflicting payload/descriptor reuse, tenant/Session identity separation and duplicates while admission is full.
+- Remove an index record and prove it is rebuilt from the original journal terminal. Corrupt or substitute a record and prove lookup rejects without another admission or effect.
+- Inject pending write, file-sync, publication and directory-sync failures. Verify terminal journal mutations remain acknowledged independently, retry preserves the original result and pending records never become authority.
+- Verify reconciliation against an already released activation cannot create another queued activation, and preserve existing FIFO/epoch/fence tests.
+- Run affected package tests with strict unhandled-rejection handling, build, typecheck, bundle, changed-file lint/format and two clean self-audit passes followed by independent review.
+
+Acceptance is a durable per-key terminal query used by actual duplicate paths. Constant service memory, bounded startup replay, physical history reclamation and independently authoritative archive queries are not acceptance claims for this PR.
+
+## Risks and follow-up
+
+The index duplicates terminal bytes on disk and adds I/O to terminal duplicate replay. Its records include private inbox payloads, so the directory remains local and private. Index corruption can make a duplicate query fail while its journal remains valid; there is no automatic destructive repair.
+
+Before memory eviction, persist a completeness boundary that proves which terminal identities have safe durable query results, migrate historical consumers, and define crash recovery and missing-index behavior after eviction. Native snapshot/HTTP/local replay and the authority window remain the later PRs. The complete #13184 tracker stays open.
+
+## Open questions
+
+None for this bounded local-index slice. Archive authority, eviction and reclamation are intentionally deferred decisions.

docs/design/2026-10-26-managed-terminal-identity-lookup.zh-CN.md

@@ -0,0 +1,81 @@
+# Managed activation 与 inbox 终态身份查询
+
+[English](2026-10-26-managed-terminal-identity-lookup.md) | [简体中文](2026-10-26-managed-terminal-identity-lookup.zh-CN.md)
+
+## 问题与当前状态
+
+Issue [#13184](https://github.com/QwenLM/qwen-code/issues/13184) 跟踪 Managed 状态增长。[有界保留设计](2026-10-19-managed-session-bounded-retention.zh-CN.md) 要求先提供持久历史身份查询，再允许终态离开内存。本次是该剩余工作的 PR1。
+
+`FileManagedActivationStore` 与 `FileManagedSessionInbox` 已在权威 journal 和快照中保留终态结果。打开存储会在内存重建全部身份。重复准入查询这些驻留 Map，尚无按身份读取持久终态的路径。Activation/inbox 转换压缩并不约束终态身份数量。
+
+这两个适配器要求单进程、本地 owner。消费者为 `EmbeddedHarnessScheduler` 与 `ManagedPromptAdmissionController`，两者均未接入公开 Hosted 应用。原生 HTTP Session journal 是另一个存储。本次不宣称 Hosted 服务内存或启动读取量已经有界。
+
+## 目标与范围
+
+- 按 tenant、Session 和 activation/message 身份物化并读取完整、不可变的终态结果。
+- 保留原始规范化数据、排队/准入位置、时间、结果、取消状态和 processing fence。
+- 保持准确重复重放与身份冲突拒绝，包括当前准入容量已满的情况。
+- 在不改变终态提交确认的前提下，建立有实际消费者的持久查询路径。
+
+内存淘汰、身份过期、journal 删除、原生 Session 快照、Java/HTTP API 与自动完成 inbox 留给后续工作。驻留 Map 和现有同步查询保持不变。
+
+## 方案
+
+### 权威与索引布局
+
+现有 journal 仍是唯一权威。相邻的 `<journal>.terminal-identities` 目录保存派生的不可变记录。文件名为 JSON 编码身份元组 `[tenantId, sessionId, activationId/messageId]` 的 SHA-256。Activation 与 inbox 记录分别位于各自 journal 旁，并包含不同的记录 kind。
+
+每条 version-1 记录包含 kind、完整身份元组、原始终态快照和快照的 SHA-256 摘要。编码使用 journal 派生的原始快照，不使用重试请求的属性顺序。读取必须匹配完整的规范化预期记录，包括身份、类型与摘要。只有已验证的 journal 终态才能重建缺失文件。损坏或冲突文件使查询失败，不能视为新身份或静默覆盖。
+
+孤立的旁路记录不能建立 journal 中不存在的身份、覆盖活跃状态或恢复缺失的 journal。索引不是独立权威。未来淘汰前必须先引入持久完整性/发布边界；仅有这些文件还不允许淘汰。
+
+### Read-through 与实际消费者
+
+两个适配器均暴露异步 `readTerminal` 查询。查询进入现有串行队列，验证并捕获请求身份，对 journal 中缺失或非终态的身份返回无结果。对已知终态读取或物化按 key 记录，再返回原始快照的克隆。
+
+`enqueue` 与 `admit` 的既有重复路径，在完成现有规范化、容量参数校验和深相等冲突判断后，对终态使用同一 read-through。非终态重复保持原路径。属性重排与数值规范化仍使用现有语义比较，请求 JSON 字节不充当幂等 key。
+
+这也保持 reconciliation：未完成 inbox 重新提交一个已释放 activation 时返回原 activation，不会再次排队执行副作用。Reconciliation 仍不自动完成 inbox。
+
+### 发布与失败行为
+
+只在终态查询或重复重放期间物化索引。`release`、`cancelQueued`、`finish` 和取消操作继续只提交原 journal。因此，索引写失败不能把已提交的终态操作转成一次对外报告的 release/finish 失败。
+
+索引目录使用 `0700`，记录使用 `0600`。父目录与索引目录必须是真实目录，叶子读取复用现有 no-follow opener。有界读取先检查预期记录长度，再分配缓冲区。这些检查保持可信单 owner 边界，不证明能隔离恶意同 UID 进程。
+
+记录先写唯一 pending 文件并 flush，再以独占 link 发布最终文件名，随后在受支持平台同步目录。最终文件已存在时读取并验证，不能覆盖。正常清理删除 pending 文件，查询从不把它当作结果。崩溃可遗留被忽略的 pending 文件；反复进程崩溃可能积累这些文件，孤儿回收留给后续工作。
+
+查询失败直接拒绝，不改变权威 journal、准入计数或驻留结果。派生文件缺失时后续查询可重建；冲突/损坏仍持续拒绝。不引入新的可用性降级或终态转换重试行为。
+
+## 受影响组件
+
+| 组件                                 | 改动                                          |
+| ------------------------------------ | --------------------------------------------- |
+| `managed-terminal-identity-store.ts` | 共享派生索引、有界校验与不可变发布。          |
+| `managed-activation-store.ts`        | 串行终态查询与已释放身份的重复 read-through。 |
+| `managed-session-inbox.ts`           | 串行终态查询与已完成消息的重复 read-through。 |
+| 定向 store/admission 测试            | 持久化、规范化、拒绝、故障与 reconciliation。 |
+
+不增加依赖、配置开关、迁移或公开协议。既有 version-1 journal 与 version-2 转换快照仍可由新旧代码读取。旧代码忽略相邻索引；PR1 保留全部权威身份，因此兼容。
+
+## 验证与验收
+
+- 先确认全局安装的 CLI 无法暴露这些内部适配器 API，再以直接适配器脚本执行基线与本地验证。
+- 对 completed/failed activation、排队取消、已完成 inbox 消息和无 processing fence 的取消，验证进程重启后的查询与原结果重放。
+- 验证规范化/属性重排的 payload 重试、不同 payload/descriptor 的身份冲突、tenant/Session 身份隔离及准入满时的重复请求。
+- 删除索引记录，证明由原 journal 终态重建；损坏或替换记录，证明拒绝查询且不产生第二次准入或副作用。
+- 注入 pending 写、file-sync、发布和 directory-sync 失败。验证终态 journal 操作的确认独立不变、重试保留原结果、pending 从不成为权威。
+- 验证对已释放 activation 的 reconciliation 不会产生第二个排队 activation，并保持现有 FIFO/epoch/fence 测试。
+- 以严格未处理拒绝检查运行受影响包测试，完成 build、typecheck、bundle、改动文件 lint/format、连续两轮干净自审和独立审查。
+
+验收目标是具有实际重复路径消费者的持久按 key 终态查询。本 PR 不宣称服务内存为常量、启动重放有界、物理历史回收完成或归档查询可独立充当权威。
+
+## 风险与后续
+
+索引在磁盘上重复保存终态字节，并为终态重复重放增加 I/O。记录含有私有 inbox payload，因此目录保持本地私有。索引损坏可使重复查询失败，即便 journal 仍有效；本次不做自动破坏性修复。
+
+内存淘汰前，需持久化完整性边界，证明哪些终态身份已具备安全持久查询结果，并迁移历史消费者，定义淘汰后的崩溃恢复与索引缺失行为。原生快照/HTTP/本地重放及 authority 窗口仍属后续 PR。完整 #13184 保持开放。
+
+## 开放问题
+
+该本地索引切片没有待定选择。归档权威、淘汰和回收的决策留给后续。

packages/core/src/managed-runtime/embedded-harness-scheduler.ts

@@ -193,8 +193,8 @@ export class EmbeddedHarnessScheduler {
       if (this.store.haltedError) this.halt(this.store.haltedError);
       throw error;
     }
-    // A duplicate enqueue performs no store I/O, so it must not reset the
-    // consecutive transient-failure streak.
+    // Only a new journal write resets its transient-failure streak; a
+    // terminal duplicate can succeed through the separate derived index.
     if (result.created) this.noteStoreOutcome();
     if (this.started) {
       // A genuinely new queued activation restarts the blocked poll cadence;

packages/core/src/managed-runtime/managed-activation-store.ts

@@ -7,6 +7,7 @@
 import { appendFile, mkdir, readFile, stat, truncate } from 'node:fs/promises';
 import path from 'node:path';
 import { isDeepStrictEqual } from 'node:util';
+import { FileManagedTerminalIdentityStore } from './managed-terminal-identity-store.js';
 import {
   encodeManagedJournalSnapshot,
   MANAGED_JOURNAL_COMPACT_BYTES,
@@ -291,11 +292,17 @@ export class FileManagedActivationStore {
   /** File bytes known to end at a complete event boundary. */
   private syncedBytes = 0;
   private snapshotBytes = 0;
+  private readonly terminals: FileManagedTerminalIdentityStore<ManagedActivationSnapshot>;
 
   private constructor(
     readonly filePath: string,
     private readonly clock: () => number,
-  ) {}
+  ) {
+    this.terminals = new FileManagedTerminalIdentityStore(
+      filePath,
+      'activation',
+    );
+  }
 
   static async open(
     filePath: string,
@@ -373,7 +380,13 @@ export class FileManagedActivationStore {
             `Managed activation '${activation.activationId}' was reused with different data.`,
           );
         }
-        return { created: false, activation: snapshot(existing) };
+        return {
+          created: false,
+          activation:
+            existing.status === 'released'
+              ? await this.readTerminalState(existing)
+              : snapshot(existing),
+        };
       }
 
       const queued = [...this.activations.values()].filter(
@@ -523,6 +536,28 @@ export class FileManagedActivationStore {
     });
   }
 
+  readTerminal(
+    input: ManagedActivationIdentity,
+  ): Promise<ManagedActivationSnapshot | undefined> {
+    const captured = structuredClone(input);
+    return this.serial(async () => {
+      const state = this.activations.get(activationKey(identity(captured)));
+      return state?.status === 'released'
+        ? this.readTerminalState(state)
+        : undefined;
+    });
+  }
+
+  private readTerminalState(
+    state: ActivationState,
+  ): Promise<ManagedActivationSnapshot> {
+    const { tenantId, sessionId, activationId } = state.descriptor;
+    return this.terminals.read(
+      [tenantId, sessionId, activationId],
+      snapshot(state),
+    );
+  }
+
   get(input: ManagedActivationIdentity): ManagedActivationSnapshot | undefined {
     const state = this.activations.get(activationKey(input));
     return state ? snapshot(state) : undefined;

packages/core/src/managed-runtime/managed-session-inbox.ts

@@ -8,6 +8,7 @@ import { createHash } from 'node:crypto';
 import { appendFile, mkdir, readFile, truncate } from 'node:fs/promises';
 import path from 'node:path';
 import { isDeepStrictEqual } from 'node:util';
+import { FileManagedTerminalIdentityStore } from './managed-terminal-identity-store.js';
 import type {
   ManagedActivationFence,
   ManagedActivationOutcome,
@@ -406,12 +407,15 @@ export class FileManagedSessionInbox {
   private fatalError: Error | undefined;
   private syncedBytes = 0;
   private snapshotBytes = 0;
+  private readonly terminals: FileManagedTerminalIdentityStore<ManagedSessionMessageSnapshot>;
 
   private constructor(
     readonly filePath: string,
     private readonly clock: () => number,
     private readonly maxPayloadBytes: number,
-  ) {}
+  ) {
+    this.terminals = new FileManagedTerminalIdentityStore(filePath, 'inbox');
+  }
 
   static async open(
     filePath: string,
@@ -499,7 +503,13 @@ export class FileManagedSessionInbox {
             `Managed Session message ${JSON.stringify(admittedMessage.messageId)} was reused with different data.`,
           );
         }
-        return { created: false, message: snapshot(existing) };
+        return {
+          created: false,
+          message:
+            existing.state === 'finished'
+              ? await this.readTerminalState(existing)
+              : snapshot(existing),
+        };
       }
 
       const pending = [...this.messages.values()].filter(
@@ -681,6 +691,28 @@ export class FileManagedSessionInbox {
     });
   }
 
+  readTerminal(
+    input: ManagedSessionMessageIdentity,
+  ): Promise<ManagedSessionMessageSnapshot | undefined> {
+    const captured = structuredClone(input);
+    return this.serial(async () => {
+      const state = this.messages.get(messageKey(identity(captured)));
+      return state?.state === 'finished'
+        ? this.readTerminalState(state)
+        : undefined;
+    });
+  }
+
+  private readTerminalState(
+    state: MessageState,
+  ): Promise<ManagedSessionMessageSnapshot> {
+    const { tenantId, sessionId, messageId } = state.message;
+    return this.terminals.read(
+      [tenantId, sessionId, messageId],
+      snapshot(state),
+    );
+  }
+
   get(
     input: ManagedSessionMessageIdentity,
   ): ManagedSessionMessageSnapshot | undefined {

packages/core/src/managed-runtime/managed-terminal-identity-store.test.ts

@@ -0,0 +1,389 @@
+/**
+ * @license
+ * Copyright 2026 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import {
+  copyFile,
+  mkdir,
+  mkdtemp,
+  readFile,
+  readdir,
+  rm,
+  stat,
+  symlink,
+  writeFile,
+} from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import {
+  FileManagedActivationStore,
+  type ManagedActivationDescriptor,
+} from './managed-activation-store.js';
+import { ManagedPromptAdmissionController } from './managed-prompt-admission.js';
+import {
+  FileManagedSessionInbox,
+  managedSessionPayloadRef,
+} from './managed-session-inbox.js';
+
+const fault = vi.hoisted(() => ({
+  phase: undefined as
+    | 'write'
+    | 'file-sync'
+    | 'link'
+    | 'directory-sync'
+    | undefined,
+}));
+
+vi.mock('node:fs/promises', async (importOriginal) => {
+  const original = await importOriginal<typeof import('node:fs/promises')>();
+  return {
+    ...original,
+    open: async (...args: Parameters<typeof original.open>) => {
+      const handle = await original.open(...args);
+      const pending =
+        path.basename(String(args[0])).startsWith('.pending-') &&
+        path.dirname(String(args[0])).endsWith('.terminal-identities');
+      const directory = String(args[0]).endsWith('.terminal-identities');
+      const write = handle.writeFile.bind(handle);
+      const sync = handle.sync.bind(handle);
+      handle.writeFile = async (...writeArgs) => {
+        if (pending && fault.phase === 'write') {
+          await write('torn');
+          throw new Error('terminal write failed');
+        }
+        return write(...writeArgs);
+      };
+      handle.sync = async () => {
+        if (
+          (pending && fault.phase === 'file-sync') ||
+          (directory && fault.phase === 'directory-sync')
+        ) {
+          throw new Error(`terminal ${fault.phase} failed`);
+        }
+        return sync();
+      };
+      return handle;
+    },
+    link: async (...args: Parameters<typeof original.link>) => {
+      if (fault.phase === 'link') throw new Error('terminal link failed');
+      return original.link(...args);
+    },
+  };
+});
+
+const activationLimits = { maxQueued: 10, maxQueuedPerTenant: 10 };
+const inboxLimits = { maxPending: 10, maxPendingPerTenant: 10 };
+const input = { tenantId: 'tenant', sessionId: 'session', messageId: 'first' };
+const activation: ManagedActivationDescriptor = {
+  tenantId: input.tenantId,
+  sessionId: input.sessionId,
+  activationId: input.messageId,
+  payloadRef: managedSessionPayloadRef(input),
+  reason: 'user_message',
+  recovery: 'replay_safe',
+};
+
+describe('durable terminal identity lookup', () => {
+  let root: string;
+  let activationPath: string;
+  let inboxPath: string;
+
+  beforeEach(async () => {
+    root = await mkdtemp(path.join(os.tmpdir(), 'managed-terminal-'));
+    activationPath = path.join(root, 'activations.jsonl');
+    inboxPath = path.join(root, 'inbox.jsonl');
+  });
+
+  afterEach(async () => {
+    fault.phase = undefined;
+    await rm(root, { recursive: true, force: true });
+  });
+
+  async function released() {
+    const store = await FileManagedActivationStore.open(activationPath, {
+      clock: () => 100,
+    });
+    await store.enqueue(activation, activationLimits);
+    const lease = (await store.claim(activation, 'worker', 1000))!;
+    await store.release(lease, 'completed');
+    return store;
+  }
+
+  async function recordPath(file: string) {
+    const directory = `${file}.terminal-identities`;
+    const names = await readdir(directory);
+    expect(names).toHaveLength(1);
+    expect(names[0]).toMatch(/^[a-f0-9]{64}\.json$/);
+    return path.join(directory, names[0]!);
+  }
+
+  it('materializes an original terminal and reads it after reopening', async () => {
+    const store = await released();
+    const original = store.get(activation);
+    const journal = await readFile(activationPath);
+    const result = await store.readTerminal(activation);
+    expect(result).toEqual(original);
+    expect(result).not.toHaveProperty('lease');
+    const file = await recordPath(activationPath);
+    const record = JSON.parse(await readFile(file, 'utf8'));
+    expect(record).toMatchObject({
+      v: 1,
+      kind: 'activation',
+      identity: ['tenant', 'session', 'first'],
+      terminal: original,
+    });
+    if (process.platform !== 'win32') {
+      expect((await stat(file)).mode & 0o777).toBe(0o600);
+      expect((await stat(path.dirname(file))).mode & 0o777).toBe(0o700);
+    }
+    (result!.descriptor as { payloadRef: string }).payloadRef =
+      'caller mutation';
+    const reopened = await FileManagedActivationStore.open(activationPath);
+    expect(await reopened.readTerminal(activation)).toEqual(original);
+    expect(await reopened.enqueue(activation, activationLimits)).toEqual({
+      created: false,
+      activation: original,
+    });
+    expect(await readFile(activationPath)).toEqual(journal);
+  });
+
+  it('does not expose active, missing or orphan identities as terminals', async () => {
+    const store = await released();
+    await store.readTerminal(activation);
+    const orphan = await recordPath(activationPath);
+    const otherPath = path.join(root, 'other.jsonl');
+    const other = await FileManagedActivationStore.open(otherPath);
+    await mkdir(`${otherPath}.terminal-identities`);
+    await copyFile(
+      orphan,
+      path.join(`${otherPath}.terminal-identities`, path.basename(orphan)),
+    );
+    expect(await other.readTerminal(activation)).toBeUndefined();
+    expect(await other.enqueue(activation, activationLimits)).toMatchObject({
+      created: true,
+    });
+    expect(await other.readTerminal(activation)).toBeUndefined();
+    expect(
+      await store.readTerminal({ ...activation, tenantId: 'another' }),
+    ).toBeUndefined();
+    expect(
+      await store.readTerminal({ ...activation, sessionId: 'another' }),
+    ).toBeUndefined();
+  });
+
+  it('rebuilds a missing index without admitting or appending again', async () => {
+    const store = await released();
+    const original = await store.readTerminal(activation);
+    const file = await recordPath(activationPath);
+    const bytes = await readFile(file);
+    await rm(file);
+    const journal = await readFile(activationPath);
+    expect(await store.enqueue(activation, activationLimits)).toEqual({
+      created: false,
+      activation: original,
+    });
+    expect(await readFile(file)).toEqual(bytes);
+    expect(await readFile(activationPath)).toEqual(journal);
+    expect(store.listRunnable()).toEqual([]);
+  });
+
+  it.each(['size', 'digest', 'identity', 'outcome'] as const)(
+    'refuses a conflicting %s without another admission or journal mutation',
+    async (damage) => {
+      const store = await released();
+      await store.readTerminal(activation);
+      const file = await recordPath(activationPath);
+      const original = await readFile(file);
+      const record = JSON.parse(original.toString());
+      if (damage === 'size') {
+        await writeFile(file, Buffer.alloc(original.length + 1));
+      } else {
+        if (damage === 'digest') record.digest = '0'.repeat(64);
+        if (damage === 'identity') record.identity[0] = 'xxxxxx';
+        if (damage === 'outcome') record.terminal.outcome = 'failed';
+        await writeFile(file, `${JSON.stringify(record)}\n`);
+      }
+      const journal = await readFile(activationPath);
+      await expect(store.readTerminal(activation)).rejects.toThrow(
+        /Terminal identity/,
+      );
+      await expect(store.enqueue(activation, activationLimits)).rejects.toThrow(
+        /Terminal identity/,
+      );
+      expect(await readFile(activationPath)).toEqual(journal);
+      expect(store.listRunnable()).toEqual([]);
+      await writeFile(file, original);
+      expect(await store.readTerminal(activation)).toEqual(
+        store.get(activation),
+      );
+    },
+  );
+
+  it.each(
+    (['write', 'file-sync', 'link', 'directory-sync'] as const).filter(
+      (phase) => phase !== 'directory-sync' || process.platform !== 'win32',
+    ),
+  )(
+    'preserves terminal acknowledgement and replay when %s publication fails',
+    async (phase) => {
+      fault.phase = phase;
+      const store = await released();
+      const original = store.get(activation);
+      const journal = await readFile(activationPath);
+      await expect(store.readTerminal(activation)).rejects.toThrow(
+        /terminal .* failed/,
+      );
+      expect(store.haltedError).toBeUndefined();
+      expect(await readFile(activationPath)).toEqual(journal);
+      expect(
+        await readdir(`${activationPath}.terminal-identities`),
+      ).not.toEqual(
+        expect.arrayContaining([expect.stringMatching(/^\.pending-/)]),
+      );
+      fault.phase = undefined;
+      const reopened = await FileManagedActivationStore.open(activationPath);
+      expect(await reopened.enqueue(activation, activationLimits)).toEqual({
+        created: false,
+        activation: original,
+      });
+      expect(await readFile(activationPath)).toEqual(journal);
+    },
+  );
+
+  it('keeps terminal mutations independent of an unavailable index', async () => {
+    await writeFile(`${activationPath}.terminal-identities`, 'not a directory');
+    const store = await released();
+    await expect(store.readTerminal(activation)).rejects.toThrow(
+      /real directory/,
+    );
+    expect(store.get(activation)).toMatchObject({
+      status: 'released',
+      outcome: 'completed',
+    });
+    const reopened = await FileManagedActivationStore.open(activationPath);
+    expect(reopened.get(activation)).toEqual(store.get(activation));
+
+    await writeFile(`${inboxPath}.terminal-identities`, 'not a directory');
+    const inbox = await FileManagedSessionInbox.open(inboxPath);
+    await inbox.admit({ ...input, payload: {} }, inboxLimits);
+    await expect(inbox.requestCancel(input)).resolves.toMatchObject({
+      state: 'finished',
+    });
+    await expect(inbox.readTerminal(input)).rejects.toThrow(/real directory/);
+  });
+
+  it('replays canonical inbox payloads and preserves cancellation without a fence', async () => {
+    const inbox = await FileManagedSessionInbox.open(inboxPath, {
+      clock: () => 100,
+    });
+    await inbox.admit({ ...input, payload: { a: -0, b: 1 } }, inboxLimits);
+    const original = await inbox.requestCancel(input);
+    expect(await inbox.readTerminal(input)).toEqual(original);
+    expect(original).not.toHaveProperty('fence');
+    expect(original.activationReady).toBe(false);
+    const reopened = await FileManagedSessionInbox.open(inboxPath);
+    expect(
+      await reopened.admit({ ...input, payload: { b: 1, a: 0 } }, inboxLimits),
+    ).toEqual({
+      created: false,
+      message: original,
+    });
+    await expect(
+      reopened.admit({ ...input, payload: { a: 1, b: 1 } }, inboxLimits),
+    ).rejects.toThrow(/different data/);
+    expect(await reopened.readTerminal(input)).toEqual(original);
+  });
+
+  it('reconciles an unfinished inbox against the original released activation', async () => {
+    const store = await released();
+    const inbox = await FileManagedSessionInbox.open(inboxPath);
+    await inbox.admit(
+      { ...input, payload: { prompt: 'original' } },
+      inboxLimits,
+    );
+    await inbox.markActivationReady(input);
+    const reopened = await FileManagedActivationStore.open(activationPath);
+    const controller = new ManagedPromptAdmissionController(
+      await FileManagedSessionInbox.open(inboxPath),
+      { submit: (value) => reopened.enqueue(value, activationLimits) },
+      inboxLimits,
+    );
+    expect(await controller.reconcile()).toEqual({ inspected: 1, readied: 0 });
+    expect(reopened.listRunnable()).toEqual([]);
+    expect(await reopened.readTerminal(activation)).toEqual(
+      store.get(activation),
+    );
+    expect(inbox.listPending()).toHaveLength(1);
+    await recordPath(activationPath);
+  });
+
+  it('uses the durable inbox record for duplicate replay and refuses corruption', async () => {
+    const inbox = await FileManagedSessionInbox.open(inboxPath);
+    const message = { ...input, payload: { prompt: 'original' } };
+    await inbox.admit(message, inboxLimits);
+    const original = await inbox.requestCancel(input);
+    expect(await inbox.admit(message, inboxLimits)).toEqual({
+      created: false,
+      message: original,
+    });
+    const file = await recordPath(inboxPath);
+    const record = JSON.parse(await readFile(file, 'utf8'));
+    expect(record.kind).toBe('inbox');
+    record.digest = '0'.repeat(64);
+    await writeFile(file, `${JSON.stringify(record)}\n`);
+    const journal = await readFile(inboxPath);
+    await expect(inbox.admit(message, inboxLimits)).rejects.toThrow(
+      /Terminal identity/,
+    );
+    expect(await readFile(inboxPath)).toEqual(journal);
+    expect(inbox.get(input)).toEqual(original);
+    expect(inbox.listPending()).toEqual([]);
+  });
+
+  it('orders lookup after pending mutations and captures the call-time identity', async () => {
+    const store = await FileManagedActivationStore.open(activationPath);
+    await store.enqueue(activation, activationLimits);
+    const cancelling = store.cancelQueued(activation);
+    const identity = { ...activation };
+    const lookup = store.readTerminal(identity);
+    identity.activationId = 'different';
+    await cancelling;
+    expect(await lookup).toEqual(store.get(activation));
+
+    const inbox = await FileManagedSessionInbox.open(inboxPath);
+    await inbox.admit({ ...input, payload: {} }, inboxLimits);
+    const finishing = inbox.requestCancel(input);
+    const messageIdentity = { ...input };
+    const messageLookup = inbox.readTerminal(messageIdentity);
+    messageIdentity.messageId = 'different';
+    expect(await messageLookup).toEqual(await finishing);
+  });
+
+  it.skipIf(process.platform === 'win32')(
+    'refuses symlinked index directories and leaf records',
+    async () => {
+      const store = await released();
+      await store.readTerminal(activation);
+      const file = await recordPath(activationPath);
+      const outside = path.join(root, 'outside.json');
+      await copyFile(file, outside);
+      await rm(file);
+      await symlink(outside, file);
+      await expect(store.readTerminal(activation)).rejects.toThrow();
+      await rm(`${activationPath}.terminal-identities`, { recursive: true });
+      const outsideDirectory = path.join(root, 'outside');
+      await mkdir(outsideDirectory);
+      await symlink(
+        outsideDirectory,
+        `${activationPath}.terminal-identities`,
+        'dir',
+      );
+      await expect(store.readTerminal(activation)).rejects.toThrow(
+        /real directory/,
+      );
+    },
+  );
+});

packages/core/src/managed-runtime/managed-terminal-identity-store.ts

@@ -0,0 +1,138 @@
+/**
+ * @license
+ * Copyright 2026 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { createHash, randomUUID } from 'node:crypto';
+import {
+  link,
+  lstat,
+  mkdir,
+  open,
+  unlink,
+  type FileHandle,
+} from 'node:fs/promises';
+import path from 'node:path';
+import { openNoFollow } from '../utils/no-follow-open.js';
+import { syncManagedJournalDirectory } from './managed-journal-snapshot.js';
+
+/** A derived index; callers must supply the journal's validated terminal. */
+export class FileManagedTerminalIdentityStore<T extends object> {
+  private readonly directory: string;
+
+  constructor(
+    journalPath: string,
+    private readonly kind: 'activation' | 'inbox',
+  ) {
+    this.directory = `${journalPath}.terminal-identities`;
+  }
+
+  async read(
+    identity: readonly [string, string, string],
+    terminal: T,
+  ): Promise<T> {
+    const state = JSON.stringify(terminal);
+    const bytes = Buffer.from(
+      `${JSON.stringify({
+        v: 1,
+        kind: this.kind,
+        identity,
+        terminal,
+        digest: createHash('sha256').update(state).digest('hex'),
+      })}\n`,
+    );
+    const name = createHash('sha256')
+      .update(JSON.stringify(identity))
+      .digest('hex');
+    const target = path.join(this.directory, `${name}.json`);
+
+    await this.ensureDirectory();
+    let stored = await this.readExisting(target, bytes);
+    if (stored === undefined) {
+      const pending = path.join(this.directory, `.pending-${randomUUID()}`);
+      try {
+        const handle = await open(pending, 'wx', 0o600);
+        try {
+          await handle.writeFile(bytes);
+          await handle.sync();
+        } finally {
+          await handle.close();
+        }
+        try {
+          await link(pending, target);
+        } catch (error) {
+          if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error;
+        }
+        stored = await this.readExisting(target, bytes);
+        if (stored === undefined) {
+          throw new Error('Terminal identity publication is missing.');
+        }
+      } finally {
+        await unlink(pending).catch(() => undefined);
+      }
+    }
+    await syncManagedJournalDirectory(target);
+    return (JSON.parse(stored.toString('utf8')) as { terminal: T }).terminal;
+  }
+
+  private async ensureDirectory(): Promise<void> {
+    await this.assertDirectory(path.dirname(this.directory));
+    try {
+      await mkdir(this.directory, { mode: 0o700 });
+    } catch (error) {
+      if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error;
+    }
+    await this.assertDirectory(this.directory);
+    await syncManagedJournalDirectory(this.directory);
+  }
+
+  private async assertDirectory(directory: string): Promise<void> {
+    const stat = await lstat(directory);
+    if (stat.isSymbolicLink() || !stat.isDirectory()) {
+      throw new Error('Terminal identity index requires a real directory.');
+    }
+  }
+
+  private async readExisting(
+    target: string,
+    expected: Buffer,
+  ): Promise<Buffer | undefined> {
+    let handle: FileHandle;
+    try {
+      handle = await openNoFollow(target);
+    } catch (error) {
+      if ((error as NodeJS.ErrnoException).code === 'ENOENT') return undefined;
+      throw error;
+    }
+    try {
+      const stat = await handle.stat();
+      if (!stat.isFile() || stat.size !== expected.byteLength) {
+        throw new Error('Terminal identity record size is invalid.');
+      }
+      const bytes = Buffer.alloc(expected.byteLength);
+      let offset = 0;
+      while (offset < bytes.byteLength) {
+        const { bytesRead } = await handle.read(
+          bytes,
+          offset,
+          bytes.byteLength - offset,
+          offset,
+        );
+        if (bytesRead === 0) {
+          throw new Error('Terminal identity record is truncated.');
+        }
+        offset += bytesRead;
+      }
+      if (
+        (await handle.stat()).size !== expected.byteLength ||
+        !bytes.equals(expected)
+      ) {
+        throw new Error('Terminal identity record conflicts with its journal.');
+      }
+      return bytes;
+    } finally {
+      await handle.close();
+    }
+  }
+}