Merge latest master into PR 555

# Conflicts:
#	docs/architecture.i18n.yaml
#	docs/core-data-structures/core.i18n.yaml
#	examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl
#	packages/client/runtime/README.i18n.yaml
#	packages/client/ui-conversation/README.i18n.yaml
#	packages/client/ui-conversation/src/client/chat/MessageItem.tsx
#	packages/compact/compact-basic/README.i18n.yaml
#	packages/ui/tui/README.i18n.yaml

packages/compact/compact-basic/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/compact/compact-basic/README.md
-README.md: 397d984676add2dd504f22fbc532184e6542ac61
-README.zh.md: 3e8a2192e22dd0ab36c7b2d827e600677d21e89e
+README.md: 22ee4c00df8ab9f52ebe86bbda540b912dfe16e2
+README.zh.md: 555d3ce7792e2978cf83ee44bea748578d3558d1

packages/compact/compact-basic/README.md

@@ -17,11 +17,11 @@ This backend owns the compaction policy:
 - **Convergence** — retry head-checkpoint compaction up to `compactionRetries`; reject a summary that does not shrink its source, and throw if retries cannot return below threshold.
 - **Summarization** — a direct `llm/stream` call uses the configured provider/model pair and cap, falling back to the latest logged request target and then the agent target, without running the loop-only `agent/request` seam. The call replays the conversation's own system prompt, tools, and shadowed-region messages verbatim, including image references, and appends the compaction instruction as the final user message, so it reuses the provider's warm prefix cache instead of invalidating it. The selected adapter must resolve or explicitly reject those images. It sets `GenerateOptions.purpose` to `compaction`, which adapters may forward as request attribution (the DeepSeek adapter sends `x-deepseek-harness-compact: 1`) without touching the model-visible body. Only returned text enters the checkpoint, excluding reasoning and tool calls that would leak private reasoning or create an orphaned call; image output fails with `UNSUPPORTED_CONTENT` rather than disappearing.
 - **Framing** — the replacement user message marks established checkpoint context with `<compacted-summary>` tags. The raw summary remains on the provenance event, and later automatic cycles merge the prior checkpoint.
-- **Lifecycle** — `compactRegion()` mutates `agent.session` and records its start, summary, replacement, and end. After asynchronous summarization it rejects a changed surface-node snapshot, while unrelated log-only events may append without invalidating the selected span. The serial `agent/step` listener checks pressure before request derivation. A canonical provider overflow is offered through `agent/request-error` after the failed step; the plugin compacts there and returns a retry action only after durable surface progress.
+- **Lifecycle** — all entry points share one bracket-first region transaction. It validates the range and live lock, appends `compact/start` synchronously, prepares and awaits the summary, revalidates, appends provenance plus the replacement, and makes exactly one closing attempt. Automatic and explicit-region calls require a numeric open-turn owner and whole-surface stability. `compactNow()` reserves idle admission, uses `turn: null`, accepts append-only context outside its selected span, flushes every closed attempt, and releases admission in `finally`.
 - **Overflow recovery** — provider-confirmed overflow needs no capacity metadata: it bypasses normal pressure and retention, prunes, then attempts one maximal balanced head reduction while leaving the newest indivisible unit. Retry is authorized whenever `surface.replaceGeneration` advances, including when pruning lands before later summary work throws. No replacement, an exhausted target-specific cap, cancellation, or an unknown/noncanonical error preserves the original provider failure.
-- **Failure handling** — an unmatched `compact/start` is an inert crash marker because no summary replacement landed. A region failure records an error end; the surface remains unchanged unless pruning already landed. Operational pressure failures warn and continue, while overflow-recovery failure preserves the original provider error only when no earlier replacement advanced the surface. Cancellation remains authoritative after any progress.
+- **Failure handling** — a live unmatched `compact/start` is the durable lock. An unmatched marker before a newer `session/end-seed` is stale evidence from a prior lifecycle and does not block; one after that boundary reports `busy`. Summary and changed-span failures close with an error and leave the conversation surface untouched, though the attempt remains in the log. A failed close deliberately leaves a blocking orphan. Operational pressure failures warn and continue, while overflow-recovery failure preserves the original provider error only when no earlier replacement advanced the surface. Cancellation remains authoritative after cleanup and durability.
 
-The protected `summarize()` method is the sole subclass hook. A template- or remote-summarizer subclass can override it while pressure, retention, provenance, shrink validation, and shadowed-token accounting stay on `ctx.tokenMeter`. The hook returns the summary blocks together with the call envelope it used (`{ summary, provider, model, maxTokens? }`), which is logged on `compact/summary`.
+The protected `summarize()` method is the sole subclass hook. A template- or remote-summarizer subclass can override it while pressure, retention, provenance, shrink validation, and shadowed-token accounting stay on `ctx.tokenMeter`. The hook returns the safe summary plus the complete provider output, call envelope, and usage when available (`{ summary, rawOutput?, provider, model, maxTokens?, usage? }`); the transaction preserves those fields on `compact/summary`.
 
 ## Config (`BasicCompactConfig`)
 
@@ -46,21 +46,25 @@ An adapter may return no capacity for a valid dynamic route, and resolved capaci
 
 ## Usage
 
+`BasicCompactService` requires `ctx.llm`, `ctx.tokenMeter`, and `ctx.sessions`. The composition below receives `ctx.llm` from its host and installs the other two services:
+
 ```ts
 import type { Context } from 'cordis'
 import { BasicCompactService } from '@deepseek-ai/dsh-compact-basic'
+import SessionStore from '@deepseek-ai/dsh-session'
 import TokenMeterService from '@deepseek-ai/dsh-token-meter'
 
 export const name = 'compact-basic'
-export const inject = ['llm', 'tokenMeter']
+export const inject = ['llm']
 
 export function apply(ctx: Context): void {
+  ctx.plugin(SessionStore)
   ctx.plugin(TokenMeterService)
   ctx.plugin(BasicCompactService)
 }
 ```
 
-Loading the plugin registers `ctx.compact`. Add [`dsh-compact-tool-result-prune`](../compact-tool-result-prune/README.md) as a sibling before this plugin to enable the optional model-free pass. With `auto: true` (the default) it compacts automatically under token pressure; a consumer (a future `/compact` tool) can also call `ctx.compact.compactIfNeeded(...)` or `ctx.compact.compactRegion(...)` directly.
+Loading the plugin registers `ctx.compact`. Add [`dsh-compact-tool-result-prune`](../compact-tool-result-prune/README.md) as a sibling before this plugin to enable the optional model-free pass. With `auto: true` (the default) it compacts automatically under token pressure. The sibling [`dsh-command-compact`](../command-compact/README.md) calls `ctx.compact.compactNow(...)`; programmatic callers may also use any seam operation directly.
 
 For example, the same compact plugin can safely serve models with different capacities and one target-specific policy:
 

packages/compact/compact-basic/README.zh.md

@@ -17,11 +17,11 @@
 - **收敛**：最多按 `compactionRetries` 重试头部检查点压缩；拒绝不能缩小源内容的摘要，如果重试仍无法回到阈值以下，则抛出异常。
 - **摘要**：直接 `llm/stream` 调用使用已配置的提供方／模型对与上限，回退到最新已记录请求目标，然后再回退到 agent 目标，而不运行仅用于 agent loop 的 `agent/request` seam。该调用会逐字回放会话自身的系统提示词、工具与已遮蔽区域消息（包括图片引用），并将压缩指令作为最后一条 user 消息追加，从而复用提供方的热前缀 cache，而非使它失效。所选适配器必须解析或明确拒绝这些图片。它将 `GenerateOptions.purpose` 设为 `compaction`，适配器可将其作为请求归因转发（DeepSeek 适配器发送 `x-deepseek-harness-compact: 1`），但不会触碰模型可见的请求体。只有返回的文本会进入检查点；推理（reasoning）和工具调用都会被排除，以免泄露私有推理或产生遗留调用；图片输出会以 `UNSUPPORTED_CONTENT` 失败，而不是消失。
 - **框定**：替换 user 消息使用 `<compacted-summary>` 标签标记已建立的检查点上下文。原始摘要保留在溯源事件上，后续自动周期会合并之前的检查点。
-- **生命周期**：`compactRegion()` 会更改 `agent.session`，并记录开始、摘要、替换与结束。异步摘要后，如果表层节点快照已改变，它会拒绝操作，而不相关的仅日志事件可以追加，不会使已选 span 失效。串行 `agent/step` listener 会在派生请求之前检查压力。规范提供方溢出会在失败步骤之后经由 `agent/request-error` 交给本插件；插件在此执行压缩，并且只在表层取得持久进展后才返回重试动作。
+- **生命周期**：所有入口点共享一个先记录标记的区域事务。它会验证范围与活动锁，同步追加 `compact/start`，准备并等待摘要，重新验证，再追加溯源信息和替换，最后恰好进行一次闭合尝试。自动调用和显式范围调用要求数字标识的开放轮次归属，并要求整个表层保持稳定。`compactNow()` 会预留空闲接纳，使用 `turn: null`，允许所选 span 之外追加仅追加上下文，flush 每次已闭合尝试，并在 `finally` 中释放接纳预留。
 - **溢出恢复**：提供方已确认的溢出不需容量元数据。它会绕过常规压力与保留，执行剪枝，再尝试一次最大平衡头部缩减，并留下最新不可分单元。只要 `surface.replaceGeneration` 前进，就允许重试，包括剪枝在后续摘要工作抛出异常前已落地的情况。如果没有替换、目标特定上限已耗尽、已取消，或遇到未知／非规范错误，则保留原始提供方失败。
-- **失败处理**：未配对的 `compact/start` 是不起作用的崩溃标记，因为没有摘要替换落地。区域失败会记录错误结束；除非剪枝已落地，否则表层保持不变。压力检查中的运行故障会发出警告并继续；只有此前没有替换推进表层时，溢出恢复失败才保留原始提供方错误。即使已经取得进展，取消仍具有最终决定权。
+- **失败处理**：活动的未匹配 `compact/start` 是持久锁。位于较新 `session/end-seed` 之前的未匹配标记，是先前生命周期留下的陈旧证据，不会阻塞；位于该边界之后的标记报告 `busy`。摘要和 span 变更失败会以错误闭合，并保持会话表层不变，但日志中仍保留该尝试。闭合失败会有意留下阻塞性的未匹配标记。压力检查中的运行故障会发出警告并继续；只有此前没有替换推进表层时，溢出恢复失败才保留原始提供方错误。完成清理与持久化后，取消仍具有最终决定权。
 
-受保护的 `summarize()` 方法是唯一的子类钩子。基于模板或远程摘要器的子类可以覆盖该方法，同时压力、保留、溯源、缩减验证与已遮蔽 token 计量仍由 `ctx.tokenMeter` 负责。钩子会将摘要块与它使用的调用 envelope 一并返回（`{ summary, provider, model, maxTokens? }`），并记录在 `compact/summary` 上。
+受保护的 `summarize()` 方法是唯一的子类钩子。基于模板或远程摘要器的子类可以覆盖该方法，同时压力、保留、溯源、缩减验证与已遮蔽 token 计量仍由 `ctx.tokenMeter` 负责。钩子返回安全摘要，以及完整提供方输出、调用 envelope 和可用时的 usage（`{ summary, rawOutput?, provider, model, maxTokens?, usage? }`）；事务会在 `compact/summary` 上保留这些字段。
 
 ## 配置（`BasicCompactConfig`）
 
@@ -46,21 +46,25 @@
 
 ## 用法
 
+`BasicCompactService` 需要 `ctx.llm`、`ctx.tokenMeter` 和 `ctx.sessions`。以下组合从其宿主接收 `ctx.llm`，并安装另外两项服务：
+
 ```ts
 import type { Context } from 'cordis'
 import { BasicCompactService } from '@deepseek-ai/dsh-compact-basic'
+import SessionStore from '@deepseek-ai/dsh-session'
 import TokenMeterService from '@deepseek-ai/dsh-token-meter'
 
 export const name = 'compact-basic'
-export const inject = ['llm', 'tokenMeter']
+export const inject = ['llm']
 
 export function apply(ctx: Context): void {
+  ctx.plugin(SessionStore)
   ctx.plugin(TokenMeterService)
   ctx.plugin(BasicCompactService)
 }
 ```
 
-加载插件会注册 `ctx.compact`。在该插件之前添加同级 [`dsh-compact-tool-result-prune`](../compact-tool-result-prune/README.md) 以启用可选的不依赖模型的处理阶段。当 `auto: true`（默认）时，它会在 token 压力下自动压缩；消费方（未来的 `/compact` 工具）也可直接调用 `ctx.compact.compactIfNeeded(...)` 或 `ctx.compact.compactRegion(...)`。
+加载插件会注册 `ctx.compact`。在该插件之前添加同级 [`dsh-compact-tool-result-prune`](../compact-tool-result-prune/README.md) 以启用可选的不依赖模型的处理阶段。当 `auto: true`（默认）时，它会在 token 压力下自动压缩。同级 [`dsh-command-compact`](../command-compact/README.md) 调用 `ctx.compact.compactNow(...)`；编程调用方也可以直接使用任一 seam 操作。
 
 例如，同一个压缩插件可以安全服务于容量不同的模型，并应用一项目标特定策略：
 

packages/compact/compact-basic/src/index.ts

@@ -6,11 +6,12 @@
 
 import { Context } from 'cordis'
 import z from 'schemastery'
-import { CompactService } from '@deepseek-ai/dsh-compact'
+import { CompactService, ManualCompactionError } from '@deepseek-ai/dsh-compact'
 import type { CompactionResult, CompactionTrigger } from '@deepseek-ai/dsh-compact'
+import type { TokenMeterService } from '@deepseek-ai/dsh-token-meter'
 import type { Session } from '@deepseek-ai/dsh-session'
 import { CONTEXT_WINDOW_EXCEEDED_CODE, assertNever } from '@deepseek-ai/dsh-llm'
-import type { ContentBlock, LlmCallConfig } from '@deepseek-ai/dsh-llm'
+import type { LlmCallConfig } from '@deepseek-ai/dsh-llm'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 // Type-only: makes the optional sibling service available to `ctx.get()`.
 import type {} from '@deepseek-ai/dsh-compact-tool-result-prune'
@@ -20,9 +21,13 @@ import {
   resolveTargetPolicy,
   TargetPressureConfigError,
 } from './config.ts'
-import { compactSurfaceRegion, selectCompactableRange } from './region.ts'
+import {
+  assertNoActiveCompaction,
+  compactSurfaceRegion,
+  selectCompactableRange,
+} from './region.ts'
 import { summarizeWithLlm } from './summarizer.ts'
-import type { SummarizationInput } from './summarizer.ts'
+import type { SummarizationInput, SummaryResult } from './summarizer.ts'
 import type {
   BasicCompactConfig,
   ModelCompactPolicyConfig,
@@ -39,6 +44,9 @@ export type {
   ResolvedTargetPolicy,
 } from './types.ts'
 
+/** The region transaction's view of this service's dynamically dispatched summarizer. */
+type RegionSummarize = (input: SummarizationInput, agent: Agent, signal?: AbortSignal) => Promise<SummaryResult>
+
 /** Resolve the exact provider/model durably routed for the latest request. */
 function routedTarget(
   session: Session,
@@ -92,7 +100,7 @@ const modelPolicy: z<ModelCompactPolicyConfig> = z.object({
  * token meter.
  */
 export class BasicCompactService extends CompactService {
-  static inject = ['llm', 'tokenMeter']
+  static inject = ['llm', 'tokenMeter', 'sessions']
 
   static Config: z<BasicCompactConfig> = z.object({
     thresholdRatio: thresholdRatioSchema,
@@ -235,7 +243,7 @@ export class BasicCompactService extends CompactService {
     input: SummarizationInput,
     agent: Agent,
     signal?: AbortSignal,
-  ): Promise<{ summary: ContentBlock[]; provider: string; model: string; maxTokens?: number }> {
+  ): Promise<SummaryResult> {
     const target = conversationTarget(agent)
     const config = target === undefined
       ? this.config
@@ -289,6 +297,7 @@ export class BasicCompactService extends CompactService {
     }
 
     const context = (await this.ctx.llm.resolveModelInfo(target.provider, target.model, signal)).context
+    assertNoActiveCompaction(agent.session, 'automatic pressure compaction')
     const targetKey = `${target.provider}/${target.model}`
     if (context === undefined) {
       throw new TargetPressureConfigError(
@@ -343,11 +352,67 @@ export class BasicCompactService extends CompactService {
     agent: Agent,
     signal?: AbortSignal,
   ): Promise<CompactionResult> {
-    const session = agent.session
-    return compactSurfaceRegion({
+    return compactSurfaceRegion(
+      this.regionDependencies(),
+      agent.session,
+      start,
+      end,
+      agent,
+      { owner: 'current-turn', stability: 'whole-surface' },
+      signal,
+    )
+  }
+
+  /**
+   * Force one useful idle-session compaction below the pressure threshold, and
+   * resolve only after its standalone marker pair is durably checkpointed.
+   * @param agent - idle agent whose next-turn admission this call reserves.
+   * @param signal - command-owned cancellation forwarded to summarization.
+   * @returns the committed result, or `null` when no safe useful range exists.
+   */
+  override async compactNow(
+    agent: Agent,
+    signal: AbortSignal,
+  ): Promise<CompactionResult | null> {
+    signal.throwIfAborted()
+    const releaseTurnAdmission = agent.reserveTurnAdmission()
+    if (releaseTurnAdmission === undefined) {
+      throw new ManualCompactionError(
+        'busy',
+        'manual compaction requires an idle agent with no waking queued work',
+      )
+    }
+    try {
+      const range = selectCompactableRange(
+        agent.session,
+        this.ctx.tokenMeter.measure(agent.session),
+        0,
+      )
+      if (range === null) return null
+      return await compactSurfaceRegion(
+        this.regionDependencies(),
+        agent.session,
+        range.start,
+        range.end,
+        agent,
+        {
+          owner: null,
+          stability: 'selected-span',
+          flush: () => this.ctx.sessions.flush(agent.session),
+        },
+        signal,
+      )
+    } finally {
+      releaseTurnAdmission()
+    }
+  }
+
+  /** Bind the effective token meter and dynamically dispatched summarizer hook. */
+  private regionDependencies(): { meter: TokenMeterService; summarize: RegionSummarize } {
+    return {
       meter: this.ctx.tokenMeter,
       summarize: (input, owner, abort) => this.summarize(input, owner, abort),
-    }, session, start, end, agent, signal)
+    }
   }
 }
 

packages/compact/compact-basic/src/region.ts

@@ -1,5 +1,6 @@
 /**
- * Surface retention selection and the log-recorded compaction transaction.
+ * Surface retention selection and the shared log-recorded compaction
+ * transaction for automatic open-turn and manual idle-session compaction.
  *
  * @module @deepseek-ai/dsh-compact-basic/region
  */
@@ -7,12 +8,13 @@
 import { isDeepStrictEqual } from 'node:util'
 import {
   COMPACT_CHECKPOINT_SOURCE,
+  ManualCompactionError,
   toolPairingBalancedAfter,
   toolPairingBalancedBefore,
 } from '@deepseek-ai/dsh-compact'
 import type { CompactionResult } from '@deepseek-ai/dsh-compact'
-import { createUserMessage } from '@deepseek-ai/dsh-llm'
-import type { Message } from '@deepseek-ai/dsh-llm'
+import { createUserMessage, errorChain } from '@deepseek-ai/dsh-llm'
+import type { Message, UserMessage } from '@deepseek-ai/dsh-llm'
 import type { TokenMeasurement, TokenMeterService } from '@deepseek-ai/dsh-token-meter'
 import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
 import type { Agent } from '@deepseek-ai/dsh-agent'
@@ -24,6 +26,62 @@ interface RegionDependencies {
   summarize(input: SummarizationInput, agent: Agent, signal?: AbortSignal): Promise<SummaryResult>
 }
 
+/** One validated inclusive span of current surface positions. */
+interface SurfaceSelection {
+  readonly start: number
+  readonly end: number
+  readonly startIdx: number
+  readonly endIdx: number
+  readonly shadowedSeqs: readonly number[]
+}
+
+/** A selection with its priced snapshot and the replay input built from it. */
+interface PreparedCompaction extends SurfaceSelection {
+  readonly measurement: TokenMeasurement
+  readonly selectedNodes: TokenMeasurement['nodes']
+  readonly shadowedTokenCount: number
+  readonly input: SummarizationInput
+}
+
+interface SummarizedCompaction extends PreparedCompaction, SummaryResult {
+  readonly checkpointMessage: UserMessage
+}
+
+interface CompactionTransactionOptions {
+  /** `current-turn` derives a numbered owner; `null` writes a standalone bracket. */
+  readonly owner: 'current-turn' | null
+  /** Surface relationship that must survive asynchronous summarization. */
+  readonly stability: 'whole-surface' | 'selected-span'
+  /** Optional durability checkpoint after a successfully closed bracket. */
+  readonly flush?: () => Promise<void>
+}
+
+interface CompactionEntryState {
+  readonly openTurn: number | null
+  readonly unmatchedCompactionStart: SessionEvent<'compact/start'> | undefined
+  readonly latestEndSeedSeq: number | undefined
+}
+
+/**
+ * Rejects a summary whose replacement boundaries are no longer the ones it was
+ * built from, distinguished from summarizer and shrink failures so a manual
+ * caller can report the two causes differently.
+ */
+class SurfaceChangedError extends Error {}
+
+/** Whether the summary may still replace the span it was built from. */
+type StabilityCheck = (
+  dependencies: RegionDependencies,
+  session: Session,
+  prepared: PreparedCompaction,
+) => void
+
+/** Failure captured after `compact/start` has committed. */
+interface TransactionFailure {
+  readonly error: unknown
+  readonly stage: 'summary' | 'commit'
+}
+
 /**
  * Resolve the next head-anchored range while retaining a priced recent tail
  * and never splitting an assistant tool-call/result pair.
@@ -71,12 +129,18 @@ export function selectCompactableRange(
 }
 
 /**
- * Validate and compact one positional surface span.
+ * Run the single compaction transaction over one selected positional span.
+ * Selection and validation are read-only. Idle/log validation and
+ * `compact/start` are synchronously adjacent, so the durable opening marker is
+ * the compaction lock before summarization yields. Every later failure makes
+ * exactly one `compact/end` attempt; a failed close deliberately leaves the
+ * unmatched start detectable.
  * @param dependencies - conversation meter and dynamically dispatched summarizer hook.
  * @param session - session whose surface is mutated.
  * @param start - inclusive first surface-node seq.
  * @param end - inclusive last surface-node seq.
  * @param agent - agent used by the summarizer.
+ * @param options - bracket owner, stability rule, and optional durability checkpoint.
  * @param signal - optional summarization cancellation signal.
  * @returns the successful durable compaction result.
  */
@@ -86,8 +150,151 @@ export async function compactSurfaceRegion(
   start: number,
   end: number,
   agent: Agent,
+  options: CompactionTransactionOptions,
   signal?: AbortSignal,
 ): Promise<CompactionResult> {
+  if (options.owner === null) signal?.throwIfAborted()
+  const selection = validateSurfaceRegion(session, start, end)
+  const entryState = inspectCompactionEntryState(session.events)
+  assertCompactionInactive(
+    entryState.unmatchedCompactionStart,
+    entryState.latestEndSeedSeq,
+    'compaction',
+  )
+
+  let owner: number | null
+  if (options.owner === null) {
+    if (entryState.openTurn !== null) {
+      throw new ManualCompactionError('busy', 'manual compaction: the session already has an open turn')
+    }
+    owner = null
+  } else {
+    if (entryState.openTurn === null) {
+      throw new Error('compactRegion: no open turn — automatic compaction events must be enclosed in a turn')
+    }
+    owner = entryState.openTurn
+  }
+
+  const startEvent = session.append('compact/start', { turn: owner })
+  const assertStable: StabilityCheck = options.stability === 'whole-surface'
+    ? assertWholeSurfaceUnchanged
+    : assertSelectedSpanStable
+  let failure: TransactionFailure | undefined
+  let flushFailure: unknown
+  let result: CompactionResult | undefined
+  let closed = false
+  let closing = false
+  let stage: TransactionFailure['stage'] = 'summary'
+
+  try {
+    const prepared = prepareCompaction(dependencies, session, selection)
+    const summarized = await summarizeCompaction(dependencies, prepared, agent, signal)
+    if (options.owner === null) signal?.throwIfAborted()
+    assertStable(dependencies, session, summarized)
+    stage = 'commit'
+    const pending = commitCompactionBody(session, startEvent, summarized)
+    closing = true
+    const endEvent = session.append('compact/end', { turn: owner })
+    closed = true
+    result = completeCompaction(pending, endEvent)
+  } catch (error: unknown) {
+    failure = { error, stage: closing ? 'commit' : stage }
+    if (!closing) {
+      closing = true
+      try {
+        session.append('compact/end', { turn: owner, error: errorChain(error) })
+        closed = true
+      } catch (closeError: unknown) {
+        failure = { error: closeError, stage: 'commit' }
+      }
+    }
+  }
+
+  if (closed && options.flush !== undefined) {
+    try {
+      await options.flush()
+    } catch (error: unknown) {
+      flushFailure = error
+    }
+  }
+
+  if (options.owner === null) signal?.throwIfAborted()
+  if (failure !== undefined) {
+    if (options.owner === null) throwManualFailure(failure)
+    throw failure.error
+  }
+  if (flushFailure !== undefined) {
+    throw new ManualCompactionError(
+      'persistence',
+      'manual compaction durability checkpoint failed',
+      { cause: flushFailure },
+    )
+  }
+  /* v8 ignore next -- every path without a result records and throws a failure above. */
+  if (result === undefined) throw new Error('compaction committed without a result')
+  return result
+}
+
+/** Classify one closed manual attempt without weakening cancellation precedence. */
+function throwManualFailure(failure: TransactionFailure): never {
+  if (failure.stage === 'commit') {
+    throw new ManualCompactionError(
+      'commit',
+      'manual compaction did not commit cleanly',
+      { cause: failure.error },
+    )
+  }
+  if (failure.error instanceof SurfaceChangedError) {
+    throw new ManualCompactionError(
+      'changed',
+      'the compacted history changed during manual compaction',
+      { cause: failure.error },
+    )
+  }
+  throw new ManualCompactionError(
+    'summary',
+    'manual compaction could not produce a smaller summary',
+    { cause: failure.error },
+  )
+}
+
+/**
+ * Reject a durable unmatched compaction marker unless a later constructor-seed
+ * boundary proves that its owner belongs to an earlier session lifecycle.
+ * @param unmatchedCompactionStart - latest unmatched opening marker, if any.
+ * @param latestEndSeedSeq - newest constructor-seed boundary, if any.
+ * @param stage - operation label included in the busy diagnostic.
+ */
+function assertCompactionInactive(
+  unmatchedCompactionStart: SessionEvent<'compact/start'> | undefined,
+  latestEndSeedSeq: number | undefined,
+  stage: string,
+): void {
+  if (unmatchedCompactionStart === undefined
+    || (latestEndSeedSeq !== undefined
+      && latestEndSeedSeq > unmatchedCompactionStart.seq)) return
+  throw new ManualCompactionError(
+    'busy',
+    `${stage}: compaction already in progress; the session compaction lock is already active`,
+  )
+}
+
+/**
+ * Recheck the durable compaction lock after an asynchronous policy decision.
+ * @param session - session whose latest marker state is inspected.
+ * @param stage - operation label included in the busy diagnostic.
+ */
+export function assertNoActiveCompaction(session: Session, stage: string): void {
+  const entryState = inspectCompactionEntryState(session.events)
+  assertCompactionInactive(
+    entryState.unmatchedCompactionStart,
+    entryState.latestEndSeedSeq,
+    stage,
+  )
+}
+
+/** Validate one requested surface-position span before asynchronous work begins. */
+function validateSurfaceRegion(session: Session, start: number, end: number): SurfaceSelection {
   const nodes = session.surface.nodes
   const startIdx = nodes.indexOf(start)
   const endIdx = nodes.indexOf(end)
@@ -107,75 +314,145 @@ export async function compactSurfaceRegion(
     throw new Error(`compactRegion: end seq ${end} is not a balanced boundary (would split a step, or the step is still open)`)
   }
 
-  const tail = inspectTurnTail(session.events)
-  if (tail.compactionInProgress) throw new Error('compaction already in progress')
-  if (tail.turn === null) {
-    throw new Error('compactRegion: no open turn — compaction events must be enclosed in a turn')
-  }
+  return { start, end, startIdx, endIdx, shadowedSeqs: nodes.slice(startIdx, endIdx + 1) }
+}
 
-  const shadowedSeqs = nodes.slice(startIdx, endIdx + 1)
-  const startEvent = session.append('compact/start', { turn: tail.turn })
+/** Snapshot pricing and replay input for a validated surface range. */
+function prepareCompaction(
+  dependencies: RegionDependencies,
+  session: Session,
+  selection: SurfaceSelection,
+): PreparedCompaction {
+  const measurement = dependencies.meter.measure(session)
+  const selectedNodes = measurement.nodes.slice(selection.startIdx, selection.endIdx + 1)
+  if (selectedNodes.length !== selection.shadowedSeqs.length
+    || selectedNodes.some((node, index) => node.seq !== selection.shadowedSeqs[index])) {
+    throw new SurfaceChangedError('compaction: selected surface changed before summarization began')
+  }
+  return {
+    ...selection,
+    measurement,
+    selectedNodes,
+    shadowedTokenCount: selectedNodes.reduce((total, node) => total + node.tokens, 0),
+    input: buildSummarizationInput(session, selection.shadowedSeqs),
+  }
+}
+
+/** Run the summarizer and frame its replacement checkpoint. */
+async function summarizeCompaction(
+  dependencies: RegionDependencies,
+  prepared: PreparedCompaction,
+  agent: Agent,
+  signal?: AbortSignal,
+): Promise<SummarizedCompaction> {
+  const summaryResult = await dependencies.summarize(prepared.input, agent, signal)
+  const checkpointMessage = createUserMessage({
+    content: frameSummary(summaryResult.summary),
+    source: COMPACT_CHECKPOINT_SOURCE,
+  })
+  const framedSummaryTokenCount = dependencies.meter.estimateMessage(checkpointMessage)
+  if (framedSummaryTokenCount >= prepared.shadowedTokenCount) {
+    throw new Error(
+      `summary is not smaller than the shadowed content (${framedSummaryTokenCount} estimated framed tokens >= ${prepared.shadowedTokenCount})`,
+    )
+  }
+  return {
+    ...prepared,
+    ...summaryResult,
+    checkpointMessage,
+  }
+}
+
+/** Reject a summary prepared against any earlier surface generation. */
+function assertWholeSurfaceUnchanged(
+  dependencies: RegionDependencies,
+  session: Session,
+  prepared: PreparedCompaction,
+): void {
+  const current = dependencies.meter.measure(session)
+  if (!isDeepStrictEqual(current.nodes, prepared.measurement.nodes)) {
+    throw new SurfaceChangedError('compaction: session surface changed during summarization')
+  }
+}
+
+/**
+ * Require only that the selected span remain the same present, contiguous,
+ * equally priced, balanced replacement target. Nodes added outside it remain
+ * visible and do not invalidate the summary.
+ */
+function assertSelectedSpanStable(
+  dependencies: RegionDependencies,
+  session: Session,
+  prepared: PreparedCompaction,
+): void {
+  let current: SurfaceSelection
   try {
-    // Capture after the lock event so a later surface mutation invalidates the
-    // async selection before replacement. Unrelated log-only facts may append.
-    const lockedMeasurement = dependencies.meter.measure(session)
-    const selected = lockedMeasurement.nodes.slice(startIdx, endIdx + 1)
-    if (selected.length !== shadowedSeqs.length
-      || selected.some((node, index) => node.seq !== shadowedSeqs[index])) {
-      throw new Error('compaction: selected surface changed before summarization began')
-    }
-    const shadowedTokenCount = selected.reduce((total, node) => total + node.tokens, 0)
-    const summarizationInput = buildSummarizationInput(session, shadowedSeqs)
-    const {
-      summary, rawOutput, provider, model, maxTokens, usage,
-    } = await dependencies.summarize(summarizationInput, agent, signal)
-
-    const currentMeasurement = dependencies.meter.measure(session)
-    if (!isDeepStrictEqual(currentMeasurement.nodes, lockedMeasurement.nodes)) {
-      throw new Error('compaction: session surface changed during summarization')
-    }
-    const framedSummary = frameSummary(summary)
-    const checkpointMessage = createUserMessage({
-      content: framedSummary,
-      source: COMPACT_CHECKPOINT_SOURCE,
-    })
-    const framedSummaryTokenCount = dependencies.meter.estimateMessage(checkpointMessage)
-    if (framedSummaryTokenCount >= shadowedTokenCount) {
-      throw new Error(
-        `summary is not smaller than the shadowed content (${framedSummaryTokenCount} estimated framed tokens >= ${shadowedTokenCount})`,
-      )
-    }
-
-    const summaryEvent = session.append('compact/summary', {
-      summary,
-      ...rawOutput === undefined ? {} : { rawOutput },
-      shadowedRange: { start, end },
-      shadowedSeqs,
-      shadowedTokenCount,
-      provider,
-      model,
-      ...maxTokens === undefined ? {} : { maxTokens },
-      ...usage === undefined ? {} : { usage },
-    })
-    session.append('user/message', checkpointMessage, {
-      surfaceOp: { op: 'replace', start, end },
-      sourceEventSeqs: [startEvent.seq, summaryEvent.seq, ...shadowedSeqs],
-    })
-    const endEvent = session.append('compact/end', { turn: tail.turn })
-    return {
-      startSeq: startEvent.seq,
-      summarySeq: summaryEvent.seq,
-      endSeq: endEvent.seq,
-      summary,
-      shadowedRange: { start, end },
-      shadowedSeqs,
-      shadowedTokenCount,
-    }
+    current = validateSurfaceRegion(session, prepared.start, prepared.end)
   } catch (error: unknown) {
-    const message = error instanceof Error ? error.message : String(error)
-    session.append('compact/end', { turn: tail.turn, error: message })
-    throw error
+    throw new SurfaceChangedError(
+      'compaction: the selected span is no longer a valid replacement target',
+      { cause: error },
+    )
   }
+  if (!isDeepStrictEqual([...current.shadowedSeqs], [...prepared.shadowedSeqs])) {
+    throw new SurfaceChangedError('compaction: the selected span changed during summarization')
+  }
+  const measured = dependencies.meter.measure(session).nodes.slice(current.startIdx, current.endIdx + 1)
+  if (!isDeepStrictEqual(measured, prepared.selectedNodes)) {
+    throw new SurfaceChangedError('compaction: the selected span was rewritten during summarization')
+  }
+}
+
+/** Append one already-summarized provenance and replacement body without yielding. */
+function commitCompactionBody(
+  session: Session,
+  startEvent: SessionEvent<'compact/start'>,
+  summarized: SummarizedCompaction,
+): Omit<CompactionResult, 'endSeq'> {
+  const {
+    start,
+    end,
+    shadowedSeqs,
+    shadowedTokenCount,
+    summary,
+    rawOutput,
+    provider,
+    model,
+    maxTokens,
+    usage,
+    checkpointMessage,
+  } = summarized
+  const summaryEvent = session.append('compact/summary', {
+    summary,
+    ...rawOutput === undefined ? {} : { rawOutput },
+    shadowedRange: { start, end },
+    shadowedSeqs: [...shadowedSeqs],
+    shadowedTokenCount,
+    provider,
+    model,
+    ...maxTokens === undefined ? {} : { maxTokens },
+    ...usage === undefined ? {} : { usage },
+  })
+  session.append('user/message', checkpointMessage, {
+    surfaceOp: { op: 'replace', start, end },
+    sourceEventSeqs: [startEvent.seq, summaryEvent.seq, ...shadowedSeqs],
+  })
+  return {
+    startSeq: startEvent.seq,
+    summarySeq: summaryEvent.seq,
+    summary,
+    shadowedRange: { start, end },
+    shadowedSeqs: [...shadowedSeqs],
+    shadowedTokenCount,
+  }
+}
+
+/** Attach the successfully appended close event to a pending result. */
+function completeCompaction(
+  pending: Omit<CompactionResult, 'endSeq'>,
+  endEvent: SessionEvent<'compact/end'>,
+): CompactionResult {
+  return { ...pending, endSeq: endEvent.seq }
 }
 
 /**
@@ -206,25 +483,38 @@ function buildSummarizationInput(
   }
 }
 
-/** Inspect the current turn boundary and latest compaction bracket once. */
-function inspectTurnTail(
-  events: readonly SessionEvent[],
-): { turn: number | null; compactionInProgress: boolean } {
-  let compactionInProgress = false
-  let compactionStateKnown = false
+/** Inspect open-turn, unmatched-compaction, and latest seed-boundary state independently. */
+function inspectCompactionEntryState(events: readonly SessionEvent[]): CompactionEntryState {
+  let openTurn: number | null = null
+  let openTurnStateKnown = false
+  let unmatchedCompactionStart: SessionEvent<'compact/start'> | undefined
+  let compactionEntryStateKnown = false
+  let latestEndSeedSeq: number | undefined
   for (let index = events.length - 1; index >= 0; index -= 1) {
     // oxlint-disable-next-line typescript/no-non-null-assertion
     const event = events[index]!
-    if (!compactionStateKnown) {
+    if (latestEndSeedSeq === undefined && event.type === 'session/end-seed') {
+      latestEndSeedSeq = event.seq
+    }
+    if (!compactionEntryStateKnown) {
       if (event.type === 'compact/start') {
-        compactionInProgress = true
-        compactionStateKnown = true
+        unmatchedCompactionStart = event
+        compactionEntryStateKnown = true
       } else if (event.type === 'compact/end') {
-        compactionStateKnown = true
+        compactionEntryStateKnown = true
       }
     }
-    if (event.type === 'turn/start') return { turn: event.data.turn, compactionInProgress }
-    if (event.type === 'turn/end') return { turn: null, compactionInProgress }
+    if (!openTurnStateKnown) {
+      if (event.type === 'turn/start') {
+        openTurn = event.data.turn
+        openTurnStateKnown = true
+      } else if (event.type === 'turn/end') {
+        openTurnStateKnown = true
+      }
+    }
+    if (openTurnStateKnown
+      && compactionEntryStateKnown
+      && latestEndSeedSeq !== undefined) break
   }
-  return { turn: null, compactionInProgress }
+  return { openTurn, unmatchedCompactionStart, latestEndSeedSeq }
 }

packages/compact/compact-basic/tests/compact-basic.spec.ts

@@ -22,7 +22,7 @@ import type {
   StreamChunk,
   TokenUsage,
 } from '@deepseek-ai/dsh-llm'
-import { Session, SessionId } from '@deepseek-ai/dsh-session'
+import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
 import TokenMeterService from '@deepseek-ai/dsh-token-meter'
 import { agentEvents, type Agent, type RequestErrorAction } from '@deepseek-ai/dsh-agent'
 import ToolResultPruneService from '@deepseek-ai/dsh-compact-tool-result-prune'
@@ -1832,6 +1832,7 @@ describe('automatic listener and loader composition', () => {
   it('loads and disposes the real zero-config service stack', async () => {
     const ctx = new Context()
     await ctx.plugin(LlmService)
+    await ctx.plugin(SessionStore)
     const meterFiber = await ctx.plugin(TokenMeterService)
     const compactFiber = await ctx.plugin(BasicCompactService, { auto: false })
 

packages/compact/compact-basic/tests/loader-composition.spec.ts

@@ -7,6 +7,7 @@ import { Context } from 'cordis'
 import Loader from '@cordisjs/plugin-loader'
 import Include from '@cordisjs/plugin-include'
 import LlmService from '@deepseek-ai/dsh-llm'
+import SessionStore from '@deepseek-ai/dsh-session'
 import TokenMeterService from '@deepseek-ai/dsh-token-meter'
 import BasicCompactService from '@deepseek-ai/dsh-compact-basic'
 import ToolResultPruneService from '@deepseek-ai/dsh-compact-tool-result-prune'
@@ -32,6 +33,7 @@ async function loadYaml(lines: readonly string[]): Promise<Context> {
   context.loader.builtins.include = Include
   const modules = new Map<string, unknown>([
     ['@deepseek-ai/dsh-llm', LlmService],
+    ['@deepseek-ai/dsh-session', SessionStore],
     ['@deepseek-ai/dsh-token-meter', TokenMeterService],
     ['@deepseek-ai/dsh-compact-tool-result-prune', ToolResultPruneService],
     ['@deepseek-ai/dsh-compact-basic', BasicCompactService],
@@ -55,6 +57,7 @@ describe('real Loader composition', () => {
   it('loads the shipped token-meter, pruning, and compact-basic YAML order', async () => {
     const loaded = await loadYaml([
       "- name: '@deepseek-ai/dsh-llm'",
+      "- name: '@deepseek-ai/dsh-session'",
       "- name: '@deepseek-ai/dsh-token-meter'",
       "- name: '@deepseek-ai/dsh-compact-tool-result-prune'",
       '  config:',
@@ -91,6 +94,7 @@ describe('real Loader composition', () => {
   it('rejects stale compact-basic config after Schemastery normalization', async () => {
     context = new Context()
     await context.plugin(LlmService)
+    await context.plugin(SessionStore)
     await context.plugin(TokenMeterService)
     await expect(context.plugin(BasicCompactService, {
       models: { legacy: { thresholdRatio: 0.5 } },
@@ -100,6 +104,7 @@ describe('real Loader composition', () => {
   it('rejects a capacity-independent merged ratio conflict during plugin load', async () => {
     context = new Context()
     await context.plugin(LlmService)
+    await context.plugin(SessionStore)
     await context.plugin(TokenMeterService)
     await expect(context.plugin(BasicCompactService, {
       retainRatio: 0.2,
@@ -114,6 +119,7 @@ describe('real Loader composition', () => {
   it('rejects an incomplete model-policy summarization pair during plugin load', async () => {
     context = new Context()
     await context.plugin(LlmService)
+    await context.plugin(SessionStore)
     await context.plugin(TokenMeterService)
     await expect(context.plugin(BasicCompactService, {
       summarizationProvider: 'default-provider',

packages/compact/compact-basic/tests/manual-compact.spec.ts

@@ -0,0 +1,831 @@
+import { describe, expect, it, vi } from 'vitest'
+import { Context } from 'cordis'
+import AgentLoop from '@deepseek-ai/dsh-agent-loop'
+import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
+import InvariantService from '@deepseek-ai/dsh-invariants'
+import * as SessionInvariant from '@deepseek-ai/dsh-session/invariant'
+import * as AgentInvariant from '@deepseek-ai/dsh-agent/invariant'
+import * as AgentLoopInvariant from '@deepseek-ai/dsh-agent-loop/invariant'
+import * as CompactInvariant from '@deepseek-ai/dsh-compact/invariant'
+import * as CompactBasicInvariant from '@deepseek-ai/dsh-compact-basic/invariant'
+import { BasicCompactService } from '@deepseek-ai/dsh-compact-basic'
+import { isCompactCheckpointSource, ManualCompactionError } from '@deepseek-ai/dsh-compact'
+import type { CompactionResult } from '@deepseek-ai/dsh-compact'
+import {
+  createAssistantMessage,
+  createUserMessage,
+  LlmAdapter,
+} from '@deepseek-ai/dsh-llm'
+import type {
+  ContentBlock,
+  LlmResolvedModelInfo,
+  Message,
+  StreamChunk,
+  TokenUsage,
+} from '@deepseek-ai/dsh-llm'
+import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
+import LlmService from '@deepseek-ai/dsh-llm'
+import TokenMeterService from '@deepseek-ai/dsh-token-meter'
+import type { Agent } from '@deepseek-ai/dsh-agent'
+import type {
+  SummarizationInput,
+  SummaryResult,
+} from '@deepseek-ai/dsh-compact-basic/src/summarizer.ts'
+
+const MODEL = 'mock'
+const SIGNAL = new AbortController().signal
+const PROMPT = 'older conversation history '.repeat(60)
+
+/** A summarizer under test control: it can block, fail, or mutate mid-call. */
+class GatedCompactService extends BasicCompactService {
+  summary: ContentBlock[] = [{ type: 'text', text: 'checkpoint' }]
+  rawOutput: ContentBlock[] | undefined
+  usage: TokenUsage | undefined
+  error: unknown
+  gate: Promise<undefined> | undefined
+  duringSummary: (() => void) | undefined
+  calls: SummarizationInput[] = []
+
+  override async summarize(
+    input: SummarizationInput,
+    _agent: Agent,
+    _signal?: AbortSignal,
+  ): Promise<SummaryResult> {
+    this.calls.push(input)
+    this.duringSummary?.()
+    if (this.gate !== undefined) await this.gate
+    if (this.error !== undefined) throw this.error
+    return {
+      summary: this.summary,
+      ...this.rawOutput === undefined ? {} : { rawOutput: this.rawOutput },
+      provider: 'summary-provider',
+      model: 'summary-model',
+      ...this.usage === undefined ? {} : { usage: this.usage },
+    }
+  }
+}
+
+/** One text answer per request, with a context window large enough to avoid pressure. */
+class TextAdapter extends LlmAdapter {
+  readonly requests: Message[][] = []
+
+  override resolveModel(provider: string, model: string): Promise<LlmResolvedModelInfo> {
+    return Promise.resolve({
+      provider,
+      id: model,
+      name: model,
+      context: { contextWindow: 100_000 },
+    })
+  }
+
+  override async * stream(options: { messages: readonly Message[] }): AsyncIterable<StreamChunk> {
+    this.requests.push([...options.messages])
+    yield { type: 'block-start', index: 0, blockType: 'text' }
+    yield { type: 'block-end', index: 0, block: { type: 'text', text: 'answer' } }
+    yield { type: 'finish', reason: { kind: 'stop' } }
+  }
+}
+
+interface LoopHarness {
+  readonly ctx: Context
+  readonly agent: Agent
+  readonly compact: GatedCompactService
+  readonly adapter: TextAdapter
+  readonly log: string[]
+}
+
+/** Real loop, session store, and invariant companions around manual compaction. */
+async function loopHarness(): Promise<LoopHarness> {
+  const ctx = new Context()
+  await mountAgentLoopTestDependencies(ctx)
+  await ctx.plugin(InvariantService)
+  await ctx.plugin(SessionInvariant)
+  await ctx.plugin(AgentInvariant)
+  await ctx.plugin(AgentLoopInvariant)
+  await ctx.plugin(CompactInvariant)
+  await ctx.plugin(CompactBasicInvariant)
+  await ctx.plugin(AgentLoop, { agents: [] })
+  await ctx.plugin(TokenMeterService)
+  const adapter = new TextAdapter()
+  ctx.llm.registerAdapter([MODEL], adapter)
+  const compact = new GatedCompactService(ctx, { auto: false })
+  const agent = ctx.agentLoop.create(SessionId('manual-compact'), { provider: MODEL, model: MODEL })
+  const log: string[] = []
+  ctx.on('session/event', (_session, event) => {
+    if (event.type === 'turn/start') log.push(`turn/start:${event.data.trigger.kind}`)
+    if (event.type === 'turn/end') log.push('turn/end')
+    if (event.type === 'compact/start') log.push(`compact/start:${String(event.data.turn)}`)
+    if (event.type === 'compact/summary') log.push('compact/summary')
+    if (event.type === 'compact/end') log.push(`compact/end:${String(event.data.turn)}`)
+    if (event.type === 'user/message') log.push('user/message')
+  })
+  ctx.on('session/flush', () => { log.push('flush') })
+  return { ctx, agent, compact, adapter, log }
+}
+
+/** Drive one real turn so the closed history holds a compactable older span. */
+async function seedHistory(harness: LoopHarness): Promise<void> {
+  harness.agent.followup(createUserMessage({
+    content: [{ type: 'text', text: PROMPT }],
+    source: { kind: 'user' },
+  }))
+  await harness.agent.whenIdle()
+  harness.log.length = 0
+}
+
+/** Text of every derived model-visible message, in request order. */
+function derivedText(session: Session): string[] {
+  return session.deriveMessages().map((message: Message) => message.content
+    .map(block => block.type === 'text' ? block.text : '')
+    .join(''))
+}
+
+/** Await one classified manual-compaction rejection. */
+async function rejection(operation: Promise<unknown>): Promise<ManualCompactionError> {
+  const caught: unknown = await operation.then(
+    (value: unknown) => { throw new Error(`expected a rejection, resolved with ${String(value)}`) },
+    (error: unknown) => error,
+  )
+  if (!(caught instanceof ManualCompactionError)) {
+    throw new Error(`expected a ManualCompactionError, got ${String(caught)}`)
+  }
+  return caught
+}
+
+/** The Error a classified failure wraps. */
+function causeOf(error: ManualCompactionError): Error {
+  const { cause } = error
+  if (!(cause instanceof Error)) throw new Error(`expected an Error cause, got ${String(cause)}`)
+  return cause
+}
+
+function deferred(): { promise: Promise<undefined>; resolve: () => void } {
+  const { promise, resolve } = Promise.withResolvers<undefined>()
+  return { promise, resolve: () => { resolve(undefined) } }
+}
+
+/** A closed-tail session with compactable exchanges and no live agent. */
+function closedConversation(turns = 2, lastTurnNumber = turns): Session {
+  const session = new Session(SessionId(`closed-${turns}-${lastTurnNumber}`))
+  for (let index = 1; index <= turns; index += 1) {
+    const turn = index === turns ? lastTurnNumber : index
+    session.append('turn/start', { turn, trigger: { kind: 'message', source: { kind: 'user' } } })
+    session.append('user/message', createUserMessage({
+      content: [{ type: 'text', text: `${PROMPT} ${turn}` }],
+      source: { kind: 'user' },
+    }), { surfaceOp: 'append' })
+    session.append('step/start', { turn, step: 1 })
+    if (index === 1) {
+      session.append('request/header', {
+        header: { config: { provider: MODEL, model: MODEL } },
+        reason: 'initial',
+      })
+    }
+    session.append('assistant/message', {
+      turn,
+      step: 1,
+      message: createAssistantMessage({
+        content: [{ type: 'text', text: `answer ${turn}` }],
+        source: { provider: MODEL, model: MODEL },
+      }),
+    }, { surfaceOp: 'append' })
+    session.append('step/end', { turn, step: 1 })
+    session.append('turn/end', { turn, reason: { kind: 'completed' } })
+  }
+  return session
+}
+
+/** A fake idle agent whose admission reservation is scripted per test. */
+function fakeAgent(
+  session: Session,
+  reserve: () => (() => void) | undefined,
+): Agent {
+  return {
+    session,
+    options: { provider: MODEL, model: MODEL },
+    reserveTurnAdmission: reserve,
+  } as unknown as Agent
+}
+
+/** Service over a store-detached session for failure classification. */
+function detachedService(): { ctx: Context; compact: GatedCompactService; flushes: () => number } {
+  const ctx = new Context()
+  void new LlmService(ctx)
+  void new SessionStore(ctx)
+  void new TokenMeterService(ctx)
+  ctx.llm.registerAdapter([MODEL], new TextAdapter())
+  let flushes = 0
+  vi.spyOn(ctx.sessions, 'flush').mockImplementation(() => {
+    flushes += 1
+    return Promise.resolve()
+  })
+  return { ctx, compact: new GatedCompactService(ctx, { auto: false }), flushes: () => flushes }
+}
+
+function compactEvents(session: Session): Array<Session['events'][number]> {
+  return session.events.filter(event => event.type.startsWith('compact/'))
+}
+
+describe('compactNow through the real loop', () => {
+  it('holds a prompt accepted during summarization until the standalone bracket is flushed', async () => {
+    const harness = await loopHarness()
+    const { agent, compact, adapter, log } = harness
+    await seedHistory(harness)
+    const gate = deferred()
+    compact.gate = gate.promise
+
+    const running = compact.compactNow(agent, SIGNAL)
+    await Promise.resolve()
+    expect(log).toEqual(['compact/start:null'])
+    agent.followup(createUserMessage({
+      content: [{ type: 'text', text: 'after compaction' }],
+      source: { kind: 'user' },
+    }))
+    await new Promise<void>((resolve) => { setTimeout(resolve, 5) })
+
+    expect(agent.status).toBe('idle')
+    expect(adapter.requests).toHaveLength(1)
+    expect(log).toEqual(['compact/start:null'])
+
+    gate.resolve()
+    const result = await running
+    expect(result).not.toBeNull()
+    await agent.whenIdle()
+
+    const start = log.indexOf('compact/start:null')
+    const summary = log.indexOf('compact/summary')
+    const end = log.indexOf('compact/end:null')
+    const flush = log.indexOf('flush')
+    const nextTurn = log.indexOf('turn/start:message')
+    expect(start).toBeLessThan(summary)
+    expect(summary).toBeLessThan(end)
+    expect(end).toBeLessThan(flush)
+    expect(flush).toBeLessThan(nextTurn)
+    expect(adapter.requests).toHaveLength(2)
+    const second = (adapter.requests[1] ?? []).map(message => message.content
+      .map(block => block.type === 'text' ? block.text : '')
+      .join(''))
+    expect(second[0]).toContain('checkpoint')
+    expect(second.at(-1)).toBe('after compaction')
+    expect(second.some(text => text.includes(PROMPT))).toBe(false)
+  })
+
+  it('keeps context injected during summarization between the markers and after the checkpoint', async () => {
+    const harness = await loopHarness()
+    const { agent, compact } = harness
+    await seedHistory(harness)
+    compact.duringSummary = () => {
+      agent.inject(createUserMessage({
+        content: [{ type: 'text', text: 'INJECTED CONTEXT' }],
+        source: { kind: 'plugin', plugin: 'test' },
+      }))
+    }
+
+    const result = await compact.compactNow(agent, SIGNAL)
+
+    expect(result).not.toBeNull()
+    const start = agent.session.events.findLast(event => event.type === 'compact/start')
+    const injected = agent.session.events.findLast(event => event.type === 'user/message'
+      && event.data.source.kind === 'plugin' && event.data.source.plugin === 'test')
+    const end = agent.session.events.findLast(event => event.type === 'compact/end')
+    expect(start).toBeDefined()
+    expect(injected).toBeDefined()
+    expect(end).toBeDefined()
+    expect(start!.seq).toBeLessThan(injected!.seq)
+    expect(injected!.seq).toBeLessThan(end!.seq)
+    expect(result?.shadowedSeqs).not.toContain(injected?.seq)
+    const messages = derivedText(agent.session)
+    expect(messages[0]).toContain('checkpoint')
+    expect(messages.at(-1)).toContain('INJECTED CONTEXT')
+    expect(messages.filter(text => text.includes('INJECTED CONTEXT'))).toHaveLength(1)
+  })
+
+  it('keeps the marker order when listeners attempt a re-entrant injection', async () => {
+    const harness = await loopHarness()
+    const { ctx, agent, compact } = harness
+    await seedHistory(harness)
+    const attempts: string[] = []
+    ctx.on('session/event', (_session, event) => {
+      if (event.type !== 'compact/start' && event.type !== 'compact/summary') return
+      attempts.push(event.type)
+      agent.inject(createUserMessage({
+        content: [{ type: 'text', text: `from ${event.type}` }],
+        source: { kind: 'plugin', plugin: 'listener' },
+      }))
+    })
+
+    const result = await compact.compactNow(agent, SIGNAL)
+
+    expect(attempts).toEqual(['compact/start', 'compact/summary'])
+    expect(result).not.toBeNull()
+    expect(derivedText(agent.session)[0]).toContain('checkpoint')
+    expect(agent.session.events.filter(event => event.type === 'user/message'
+      && event.data.source.kind === 'plugin' && event.data.source.plugin === 'listener')).toHaveLength(0)
+    const types = compactEvents(agent.session).map(event => event.type)
+    expect(types).toEqual(['compact/start', 'compact/summary', 'compact/end'])
+  })
+
+  it('reports busy without summarizing when a prompt already owns the next turn', async () => {
+    const harness = await loopHarness()
+    const { agent, compact, adapter } = harness
+    await seedHistory(harness)
+
+    agent.followup(createUserMessage({
+      content: [{ type: 'text', text: 'first in line' }],
+      source: { kind: 'user' },
+    }))
+    expect((await rejection(compact.compactNow(agent, SIGNAL))).code).toBe('busy')
+    expect(compact.calls).toHaveLength(0)
+
+    await agent.whenIdle()
+    expect(adapter.requests).toHaveLength(2)
+    expect(agent.session.events.some(event => event.type === 'compact/start')).toBe(false)
+  })
+
+  it('releases turn admission after a summarizer failure and records the failed attempt', async () => {
+    const harness = await loopHarness()
+    const { agent, compact, adapter } = harness
+    await seedHistory(harness)
+    compact.error = new Error('summarizer unavailable')
+    const before = [...agent.session.surface.nodes]
+
+    expect((await rejection(compact.compactNow(agent, SIGNAL))).code).toBe('summary')
+    expect(agent.session.surface.nodes).toEqual(before)
+    const markers = compactEvents(agent.session)
+    expect(markers.map(event => event.type)).toEqual(['compact/start', 'compact/end'])
+    expect(markers[1]?.type === 'compact/end' && markers[1].data.error)
+      .toContain('summarizer unavailable')
+
+    agent.followup(createUserMessage({
+      content: [{ type: 'text', text: 'runs after the failure' }],
+      source: { kind: 'user' },
+    }))
+    await agent.whenIdle()
+    expect(adapter.requests).toHaveLength(2)
+  })
+})
+
+describe('compactNow transaction and failure classification', () => {
+  it('returns null without writing a bracket for history that cannot be compacted', async () => {
+    const { compact } = detachedService()
+    const session = new Session(SessionId('empty'))
+    let released = 0
+    const agent = fakeAgent(session, () => () => { released += 1 })
+
+    expect(await compact.compactNow(agent, SIGNAL)).toBeNull()
+    expect(released).toBe(1)
+    expect(compact.calls).toHaveLength(0)
+    expect(compactEvents(session)).toEqual([])
+  })
+
+  it('commits a standalone bracket without consuming a turn number and checkpoints durability', async () => {
+    const { compact, flushes } = detachedService()
+    const session = closedConversation(2, 7)
+    const agent = fakeAgent(session, () => () => undefined)
+
+    const result = await compact.compactNow(agent, SIGNAL)
+
+    expect(result).not.toBeNull()
+    expect(flushes()).toBe(1)
+    expect(session.events.filter(event => event.type === 'turn/start').at(-1)?.data.turn).toBe(7)
+    expect(session.events.findLast(event => event.type === 'compact/start')?.data)
+      .toEqual({ turn: null })
+    expect(session.events.findLast(event => event.type === 'compact/end')?.data)
+      .toEqual({ turn: null })
+  })
+
+  it('reports a live unmatched bracket as busy without summarizing', async () => {
+    const { compact } = detachedService()
+    const session = closedConversation(2)
+    session.append('compact/start', { turn: null })
+    const agent = fakeAgent(session, () => () => undefined)
+
+    const error = await rejection(compact.compactNow(agent, SIGNAL))
+    expect(error.code).toBe('busy')
+    expect(error.message).toContain('compaction lock is already active')
+    expect(compact.calls).toHaveLength(0)
+  })
+
+  it('ignores an unmatched bracket inherited before a later end-seed marker', async () => {
+    const { compact } = detachedService()
+    const original = closedConversation(2)
+    original.append('compact/start', { turn: null })
+    const reloaded = new Session(SessionId('stale-orphan'), [...original.events])
+    const boundary = reloaded.events.findLast(event => event.type === 'session/end-seed')
+    const orphan = reloaded.events.find(event => event.type === 'compact/start')
+    const agent = fakeAgent(reloaded, () => () => undefined)
+
+    expect(boundary?.seq).toBeGreaterThan(orphan?.seq ?? Number.MAX_SAFE_INTEGER)
+    await expect(compact.compactNow(agent, SIGNAL)).resolves.not.toBeNull()
+    expect(compact.calls).toHaveLength(1)
+  })
+
+  it('scans a stale orphan independently of later repaired turn state', async () => {
+    const { compact } = detachedService()
+    const original = closedConversation(2)
+    original.append('compact/start', { turn: null })
+    original.append('turn/start', { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } })
+    original.append('turn/end', { turn: 3, reason: { kind: 'interrupted' } })
+    const reloaded = new Session(SessionId('reloaded-orphan'), [...original.events])
+    const agent = fakeAgent(reloaded, () => () => undefined)
+
+    await expect(compact.compactNow(agent, SIGNAL)).resolves.not.toBeNull()
+    expect(compact.calls).toHaveLength(1)
+  })
+
+  it('refuses an open turn in the log', async () => {
+    const { compact } = detachedService()
+    const session = closedConversation(2)
+    session.append('turn/start', { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } })
+    const agent = fakeAgent(session, () => () => undefined)
+
+    const error = await rejection(compact.compactNow(agent, SIGNAL))
+    expect(error.code).toBe('busy')
+    expect(error.message).toContain('already has an open turn')
+  })
+
+  it('reports busy and skips summarization when admission is unavailable', async () => {
+    const { compact } = detachedService()
+    const agent = fakeAgent(closedConversation(2), () => undefined)
+
+    expect((await rejection(compact.compactNow(agent, SIGNAL))).code).toBe('busy')
+    expect(compact.calls).toHaveLength(0)
+  })
+
+  it('rejects a selected span replaced during summarization and records an error close', async () => {
+    const { compact, flushes } = detachedService()
+    const session = closedConversation(2)
+    let released = 0
+    const agent = fakeAgent(session, () => () => { released += 1 })
+    compact.duringSummary = () => {
+      const [head] = session.surface.nodes
+      session.append('user/message', createUserMessage({
+        content: [{ type: 'text', text: 'competing replacement' }],
+        source: { kind: 'plugin', plugin: 'rival' },
+      }), {
+        surfaceOp: { op: 'replace', start: head!, end: head! },
+        sourceEventSeqs: [head!],
+      })
+    }
+
+    expect((await rejection(compact.compactNow(agent, SIGNAL))).code).toBe('changed')
+    expect(released).toBe(1)
+    expect(flushes()).toBe(1)
+    expect(compactEvents(session).map(event => event.type)).toEqual(['compact/start', 'compact/end'])
+  })
+
+  it('rejects a selected span whose middle node was replaced during summarization', async () => {
+    const { compact } = detachedService()
+    const session = closedConversation(3)
+    const agent = fakeAgent(session, () => () => undefined)
+    compact.duringSummary = () => {
+      const middle = session.surface.nodes[1]
+      session.append('user/message', createUserMessage({
+        content: [{ type: 'text', text: 'rewritten middle node' }],
+        source: { kind: 'plugin', plugin: 'rival' },
+      }), {
+        surfaceOp: { op: 'replace', start: middle!, end: middle! },
+        sourceEventSeqs: [middle!],
+      })
+    }
+
+    const error = await rejection(compact.compactNow(agent, SIGNAL))
+    expect(error.code).toBe('changed')
+    expect(causeOf(error).message).toContain('span changed during summarization')
+  })
+
+  it('revalidates the selected span after the summarizer continuation settles', async () => {
+    const { compact, flushes } = detachedService()
+    const session = closedConversation(2)
+    const gate = deferred()
+    compact.gate = gate.promise
+    let released = 0
+    const agent = fakeAgent(session, () => () => { released += 1 })
+    const head = session.surface.nodes[0]!
+    const generation = session.surface.replaceGeneration
+
+    const running = compact.compactNow(agent, SIGNAL)
+    await Promise.resolve()
+    expect(compact.calls).toHaveLength(1)
+
+    gate.resolve()
+    queueMicrotask(() => {
+      queueMicrotask(() => {
+        session.append('user/message', createUserMessage({
+          content: [{ type: 'text', text: 'late competing replacement' }],
+          source: { kind: 'plugin', plugin: 'rival' },
+        }), {
+          surfaceOp: { op: 'replace', start: head, end: head },
+          sourceEventSeqs: [head],
+        })
+      })
+    })
+
+    const error = await rejection(running)
+    expect(error.code).toBe('changed')
+    expect(causeOf(error).message).toContain('selected span')
+    expect(released).toBe(1)
+    expect(flushes()).toBe(1)
+    expect(session.surface.replaceGeneration).toBe(generation + 1)
+    expect(session.surface.nodes).not.toContain(head)
+    expect(compactEvents(session).map(event => event.type)).toEqual(['compact/start', 'compact/end'])
+    expect(session.events.some(event => event.type === 'user/message'
+      && isCompactCheckpointSource(event.data.source))).toBe(false)
+  })
+
+  it('classifies a failing compact/end as commit failure and leaves one orphan', async () => {
+    const { compact, flushes } = detachedService()
+    const session = closedConversation(2)
+    const agent = fakeAgent(session, () => () => undefined)
+    const append = session.append.bind(session)
+    vi.spyOn(session, 'append').mockImplementation(((type: string, ...rest: never[]) => {
+      if (type === 'compact/end') throw new Error('boundary rejected')
+      return (append as (...args: never[]) => unknown)(type as never, ...rest)
+    }) as never)
+
+    const error = await rejection(compact.compactNow(agent, SIGNAL))
+    expect(error.code).toBe('commit')
+    expect(causeOf(error).message).toBe('boundary rejected')
+    vi.restoreAllMocks()
+    expect(flushes()).toBe(0)
+    expect(session.events.findLast(event => event.type.startsWith('compact/'))?.type)
+      .toBe('compact/summary')
+    expect(compactEvents(session).filter(event => event.type === 'compact/start')).toHaveLength(1)
+
+    const calls = compact.calls.length
+    expect((await rejection(compact.compactNow(agent, SIGNAL))).code).toBe('busy')
+    expect(compact.calls).toHaveLength(calls)
+  })
+
+  it('keeps a failed error-close as the commit failure and does not flush', async () => {
+    const { compact, flushes } = detachedService()
+    const session = closedConversation(2)
+    let released = 0
+    const agent = fakeAgent(session, () => () => { released += 1 })
+    compact.error = new Error('summary rejected')
+    const append = session.append.bind(session)
+    vi.spyOn(session, 'append').mockImplementation(((type: string, ...rest: never[]) => {
+      if (type === 'compact/end') throw new Error('error boundary rejected')
+      return (append as (...args: never[]) => unknown)(type as never, ...rest)
+    }) as never)
+
+    const error = await rejection(compact.compactNow(agent, SIGNAL))
+    vi.restoreAllMocks()
+    expect(error.code).toBe('commit')
+    expect(causeOf(error).message).toBe('error boundary rejected')
+    expect(released).toBe(1)
+    expect(flushes()).toBe(0)
+    expect(compactEvents(session).map(event => event.type)).toEqual(['compact/start'])
+  })
+
+  it('rejects a selected span whose pricing changed during summarization', async () => {
+    const { ctx, compact } = detachedService()
+    const session = closedConversation(2)
+    const agent = fakeAgent(session, () => () => undefined)
+    const meter = ctx.tokenMeter
+    const original = meter.measure.bind(meter)
+    compact.duringSummary = () => {
+      vi.spyOn(meter, 'measure').mockImplementationOnce((target) => {
+        const measurement = original(target)
+        return {
+          ...measurement,
+          nodes: measurement.nodes.map((node, index) =>
+            index === 0 ? { ...node, tokens: node.tokens + 1 } : node),
+        }
+      })
+    }
+
+    expect((await rejection(compact.compactNow(agent, SIGNAL))).code).toBe('changed')
+    vi.restoreAllMocks()
+  })
+
+  it('classifies a commit-body failure and still releases admission', async () => {
+    const { compact } = detachedService()
+    const session = closedConversation(2)
+    let released = 0
+    const agent = fakeAgent(session, () => () => { released += 1 })
+    const append = session.append.bind(session)
+    vi.spyOn(session, 'append').mockImplementation(((type: string, ...rest: never[]) => {
+      if (type === 'compact/summary') throw new Error('provenance rejected')
+      return (append as (...args: never[]) => unknown)(type as never, ...rest)
+    }) as never)
+
+    const error = await rejection(compact.compactNow(agent, SIGNAL))
+    vi.restoreAllMocks()
+    expect(error.code).toBe('commit')
+    expect(released).toBe(1)
+    const end = session.events.findLast(event => event.type === 'compact/end')
+    expect(end?.type === 'compact/end' && end.data.error).toContain('provenance rejected')
+    expect(end?.type === 'compact/end' && end.data.turn).toBeNull()
+  })
+
+  it('keeps a commit failure when the durability checkpoint also fails', async () => {
+    const { ctx, compact } = detachedService()
+    const session = closedConversation(2)
+    const agent = fakeAgent(session, () => () => undefined)
+    const append = session.append.bind(session)
+    vi.spyOn(session, 'append').mockImplementation(((type: string, ...rest: never[]) => {
+      if (type === 'compact/summary') throw new Error('provenance rejected')
+      return (append as (...args: never[]) => unknown)(type as never, ...rest)
+    }) as never)
+    vi.spyOn(ctx.sessions, 'flush').mockRejectedValueOnce(new Error('disk full'))
+
+    const error = await rejection(compact.compactNow(agent, SIGNAL))
+    expect(error.code).toBe('commit')
+    expect(causeOf(error).message).toBe('provenance rejected')
+    vi.restoreAllMocks()
+  })
+
+  it('compacts a session with no durable turn boundary without creating one', async () => {
+    const { compact } = detachedService()
+    const session = new Session(SessionId('turnless'))
+    for (const text of [PROMPT, 'recent tail']) {
+      session.append('user/message', createUserMessage({
+        content: [{ type: 'text', text }],
+        source: { kind: 'user' },
+      }), { surfaceOp: 'append' })
+    }
+    const agent = fakeAgent(session, () => () => undefined)
+
+    const result = await compact.compactNow(agent, SIGNAL)
+
+    expect(result).not.toBeNull()
+    expect(session.events.some(event => event.type === 'turn/start')).toBe(false)
+    expect(session.events.find(event => event.type === 'compact/start')?.data)
+      .toEqual({ turn: null })
+  })
+
+  it('classifies a durability failure after the standalone bracket committed', async () => {
+    const { ctx, compact } = detachedService()
+    const session = closedConversation(2)
+    const agent = fakeAgent(session, () => () => undefined)
+    vi.spyOn(ctx.sessions, 'flush').mockRejectedValueOnce(new Error('disk full'))
+
+    expect((await rejection(compact.compactNow(agent, SIGNAL))).code).toBe('persistence')
+    vi.restoreAllMocks()
+    expect(session.events.some(event => event.type === 'compact/summary')).toBe(true)
+    expect(session.events.findLast(event => event.type === 'compact/end')?.data)
+      .toEqual({ turn: null })
+  })
+
+  it('lets a pre-aborted signal win before reservation, measurement, or summarization', async () => {
+    const cases = [
+      { name: 'busy', session: closedConversation(2), release: undefined },
+      { name: 'empty', session: new Session(SessionId('pre-aborted-empty')), release: () => undefined },
+      { name: 'compactable', session: closedConversation(2, 9), release: () => undefined },
+    ] as const
+
+    for (const testCase of cases) {
+      const { ctx, compact } = detachedService()
+      const reserve = vi.fn(() => testCase.release)
+      const measure = vi.spyOn(ctx.tokenMeter, 'measure')
+      const agent = fakeAgent(testCase.session, reserve)
+      const before = [...testCase.session.events]
+      const reason = Object.freeze({ kind: 'cancelled', case: testCase.name })
+      const controller = new AbortController()
+      controller.abort(reason)
+
+      await expect(compact.compactNow(agent, controller.signal)).rejects.toBe(reason)
+      expect(reserve).not.toHaveBeenCalled()
+      expect(measure).not.toHaveBeenCalled()
+      expect(compact.calls).toHaveLength(0)
+      expect(testCase.session.events).toEqual(before)
+      vi.restoreAllMocks()
+    }
+  })
+
+  it('preserves the exact cancellation reason when the summarizer also rejects', async () => {
+    const { compact, flushes } = detachedService()
+    const controller = new AbortController()
+    const reason = new Error('cancelled by the caller')
+    let released = 0
+    const session = closedConversation(2)
+    const agent = fakeAgent(session, () => () => { released += 1 })
+    compact.duringSummary = () => { controller.abort(reason) }
+    compact.error = new Error('summarizer aborted')
+
+    await expect(compact.compactNow(agent, controller.signal)).rejects.toBe(reason)
+    expect(released).toBe(1)
+    expect(flushes()).toBe(1)
+    const events = compactEvents(session)
+    expect(events.map(event => event.type)).toEqual(['compact/start', 'compact/end'])
+    expect(events[1]?.type === 'compact/end' && events[1].data.error)
+      .toContain('summarizer aborted')
+  })
+
+  it('aborts before committing when cancellation lands after summarization', async () => {
+    const { compact } = detachedService()
+    const controller = new AbortController()
+    const reason = new Error('cancelled by the caller')
+    const session = closedConversation(2)
+    const agent = fakeAgent(session, () => () => undefined)
+    compact.duringSummary = () => { controller.abort(reason) }
+
+    await expect(compact.compactNow(agent, controller.signal)).rejects.toBe(reason)
+    expect(compactEvents(session).map(event => event.type)).toEqual(['compact/start', 'compact/end'])
+    expect(session.events.some(event => event.type === 'compact/summary')).toBe(false)
+  })
+
+  it('waits for the durability checkpoint before cancellation wins and admission releases', async () => {
+    const { ctx, compact } = detachedService()
+    const controller = new AbortController()
+    const reason = new Error('cancelled during flush')
+    const flushGate = Promise.withResolvers<undefined>()
+    const flush = vi.spyOn(ctx.sessions, 'flush').mockReturnValueOnce(flushGate.promise)
+    const session = closedConversation(2)
+    let released = 0
+    const agent = fakeAgent(session, () => () => { released += 1 })
+
+    const running = compact.compactNow(agent, controller.signal)
+    let settled = false
+    void running.then(
+      () => { settled = true },
+      () => { settled = true },
+    )
+    await vi.waitFor(() => {
+      expect(flush).toHaveBeenCalledWith(session)
+    })
+    controller.abort(reason)
+    await Promise.resolve()
+    expect(settled).toBe(false)
+    expect(released).toBe(0)
+
+    flushGate.resolve(undefined)
+    await expect(running).rejects.toBe(reason)
+    expect(released).toBe(1)
+  })
+
+  it('preserves raw output and usage in the manual summary event', async () => {
+    const { compact } = detachedService()
+    const session = closedConversation(2)
+    const agent = fakeAgent(session, () => () => undefined)
+    compact.rawOutput = [
+      { type: 'text', text: 'checkpoint' },
+      { type: 'reasoning', text: 'hidden reasoning' },
+    ]
+    compact.usage = { inputTokens: 40, outputTokens: 5 }
+
+    await compact.compactNow(agent, SIGNAL)
+
+    const summary = session.events.find(event => event.type === 'compact/summary')
+    expect(summary?.type === 'compact/summary' && summary.data.rawOutput).toEqual(compact.rawOutput)
+    expect(summary?.type === 'compact/summary' && summary.data.usage).toEqual(compact.usage)
+  })
+
+  it('makes duration derivable from the opening and closing marker times', async () => {
+    const { compact } = detachedService()
+    const session = closedConversation(2)
+    const agent = fakeAgent(session, () => () => undefined)
+    compact.gate = new Promise<undefined>((resolve) => {
+      setTimeout(() => { resolve(undefined) }, 5)
+    })
+
+    await compact.compactNow(agent, SIGNAL)
+
+    const start = session.events.findLast(event => event.type === 'compact/start')
+    const end = session.events.findLast(event => event.type === 'compact/end')
+    expect(start).toBeDefined()
+    expect(end).toBeDefined()
+    expect(end!.time - start!.time).toBeGreaterThan(0)
+  })
+
+  it('excludes concurrent automatic and manual compaction of one session', async () => {
+    const { compact } = detachedService()
+    const session = closedConversation(3)
+    const agent = fakeAgent(session, () => () => undefined)
+    const gate = deferred()
+    compact.gate = gate.promise
+
+    const manual = compact.compactNow(agent, SIGNAL)
+    await Promise.resolve()
+    const nodes = session.surface.nodes
+    await expect(compact.compactRegion(
+      nodes[0]!,
+      nodes[1]!,
+      agent,
+    )).rejects.toThrow('compaction lock is already active')
+
+    gate.resolve()
+    compact.gate = undefined
+    const result: CompactionResult | null = await manual
+    expect(result).not.toBeNull()
+  })
+
+  it('excludes a manual request while an explicit region compaction runs', async () => {
+    const { compact } = detachedService()
+    const session = closedConversation(3)
+    session.append('turn/start', { turn: 4, trigger: { kind: 'message', source: { kind: 'user' } } })
+    const agent = fakeAgent(session, () => () => undefined)
+    const gate = deferred()
+    compact.gate = gate.promise
+    const nodes = session.surface.nodes
+    const region = compact.compactRegion(nodes[0]!, nodes[1]!, agent)
+    await Promise.resolve()
+
+    expect((await rejection(compact.compactNow(agent, SIGNAL))).code).toBe('busy')
+
+    gate.resolve()
+    compact.gate = undefined
+    await expect(region).resolves.toMatchObject({ shadowedSeqs: nodes.slice(0, 2) })
+  })
+})