fix(compact): preserve replay ownership (PR2 round 2)

packages/compact/compact-basic/README.md

@@ -13,7 +13,7 @@ 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 model and cap without running the loop-only `agent/request` seam. The input transcript preserves non-text blocks as tagged placeholders; only returned text enters the checkpoint, excluding reasoning and tool calls that would leak private reasoning or create an orphaned call.
 - **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()` records its start, summary, replacement, and end. The serial `agent/pre-step` listener checks pressure before every step, outside an open step, so a tool-heavy turn remains compactable and the loop derives history once after mutation.
+- **Lifecycle** — `compactRegion()` requires its agent to own the exact target session and rejects mismatch before resolution or mutation; a valid call records its start, summary, replacement, and end. The serial `agent/pre-step` listener checks pressure before every step, outside an open step, so a tool-heavy turn remains compactable and the loop derives history once after mutation.
 - **Failure handling** — an unmatched `compact/start` is an inert crash marker because no replacement landed. Recoverable failure records an error end and leaves the surface unchanged.
 
 `summarize()` 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 the conversation model's meter. The hook returns the summary blocks together with the call envelope it used (`{ summary, model, maxTokens? }`), which is logged on `compact/summary`.

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

@@ -159,11 +159,12 @@ export class BasicCompactService extends CompactService {
 
   /**
    * Compact one inclusive positional surface range using the effective
-   * conversation model for all retention and shrink pricing.
-   * @param session - session whose surface is mutated.
+   * conversation model for all retention and shrink pricing. Reject an agent
+   * that does not own the exact target before any resolution or mutation.
+   * @param session - session whose surface is mutated; must equal `agent.session`.
    * @param start - inclusive first surface-node seq.
    * @param end - inclusive last surface-node seq.
-   * @param agent - agent used by the summarizer and model resolver.
+   * @param agent - owner of the target session, used by the summarizer and model resolver.
    * @param signal - optional summarization cancellation signal.
    * @returns the successful durable compaction result.
    */
@@ -174,6 +175,9 @@ export class BasicCompactService extends CompactService {
     agent: Agent,
     signal?: AbortSignal,
   ): Promise<CompactionResult> {
+    if (session !== agent.session) {
+      throw new Error('compactRegion: agent.session must be the exact target session')
+    }
     const model = effectiveModel(agent)
     if (model === undefined || model.length === 0) {
       throw new Error('compactRegion: no routed or configured conversation model is available for token pricing')

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

@@ -361,6 +361,26 @@ describe('pressure measurement and retention', () => {
 })
 
 describe('compaction region transaction', () => {
+  it('rejects an agent that does not own the exact target session before mutation', async () => {
+    const compact = service()
+    const target = conversation(2)
+    const owner = conversation(1)
+    const targetEvents = [...target.events]
+    const ownerEvents = [...owner.events]
+    const nodes = target.surface.nodes
+
+    await expect(compact.compactRegion(
+      target,
+      nodes[0]!.seq,
+      nodes[1]!.seq,
+      agent(owner),
+    )).rejects.toThrow('compactRegion: agent.session must be the exact target session')
+
+    expect(target.events).toEqual(targetEvents)
+    expect(owner.events).toEqual(ownerEvents)
+    expect(compact.calls).toEqual([])
+  })
+
   it('lands a framed, replayable checkpoint with exact pricing provenance', async () => {
     const compact = service()
     const session = conversation(3)

packages/compact/compact/README.md

@@ -19,9 +19,9 @@ Both methods are **abstract** — the backend owns trigger policy, retention, ev
 | Member | Semantics |
 |---|---|
 | `compactIfNeeded(agent, fullSystemPrompt, sessionPrefix, signal)` | Estimate the surface-derived history size; if over the backend's threshold, compact an older range via `compactRegion`, keeping recent context intact. Returns the `CompactionResult`, or `null` if nothing needed compacting. All parameters required — the loop's `agent/pre-step` checkpoint supplies the agent, assembled `fullSystemPrompt`, composed `sessionPrefix` (request-only messages every request carries but the derived history omits — the pressure estimate must count them), and turn `signal`. A backend's summarization request is a direct `ctx.llm.stream()` call (not a loop step), so per-call interception happens at `llm/stream`. |
-| `compactRegion(session, start, end, agent, signal?)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start` is positioned after `end` on the surface. The range is a SURFACE-POSITION span, not a numeric seq interval — after a prior replace lands a fresh high-seq summary node at the shadowed range's position, surface order no longer tracks seq order. |
+| `compactRegion(session, start, end, agent, signal?)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. The agent must own the exact target (`session === agent.session`); a backend rejects mismatch before model resolution, lock acquisition, summarization, or log mutation. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start` is positioned after `end` on the surface. The range is a SURFACE-POSITION span, not a numeric seq interval — after a prior replace lands a fresh high-seq summary node at the shadowed range's position, surface order no longer tracks seq order. |
 
-`compactIfNeeded` takes a required `signal`; `compactRegion`'s is optional. A backend that summarizes via `ctx.llm.stream()` **must** forward it into the call's `GenerateOptions.signal`, so an abort or fiber dispose tears down the in-flight summarization instead of leaving an orphaned model call running past the cancellation. The session being compacted comes from the agent context; the turn that the `compact/*` events belong to is recoverable from the log (the currently-open turn), so the backend stamps it from the log rather than trusting a caller-supplied value.
+`compactIfNeeded` takes a required `signal`; `compactRegion`'s is optional. A backend that summarizes via `ctx.llm.stream()` **must** forward it into the call's `GenerateOptions.signal`, so an abort or fiber dispose tears down the in-flight summarization instead of leaving an orphaned model call running past the cancellation. The turn that the `compact/*` events belong to is recoverable from the owned session's log (the currently-open turn), so the backend stamps it from the log rather than trusting a caller-supplied value.
 
 ## Tool-pairing boundaries
 

packages/compact/compact/src/index.ts

@@ -67,17 +67,19 @@ export abstract class CompactService extends Service {
    * `start` and `end` name an inclusive span by surface position, not numeric seq
    * order; replacements can make visible seqs non-monotonic. Both edges must be
    * balanced so assistant tool calls remain paired with their results. A model-
-   * backed implementation forwards cancellation and rejects active, missing,
-   * reversed, or unbalanced ranges.
+   * backed implementation forwards cancellation. The agent must own the exact
+   * target session object; implementations reject an ownership mismatch before
+   * model resolution, lock acquisition, summarization, or log mutation, and
+   * reject active, missing, reversed, or unbalanced ranges.
    * Use {@link toolPairingBalancedBefore} and {@link toolPairingBalancedAfter}
    * for the edge checks.
    *
-   * @param session - session to mutate.
+   * @param session - session to mutate; must be identical to `agent.session`.
    * @param start - first surface seq, inclusive.
    * @param end - last surface seq, inclusive.
-   * @param agent - summarizer context.
+   * @param agent - owner of the target session and summarizer context.
    * @param signal - optional cancellation; model-backed implementations must forward it.
-   * @throws when compaction is active or the range is missing, reversed, or unbalanced.
+   * @throws when the agent does not own `session`, compaction is active, or the range is missing, reversed, or unbalanced.
    * @returns the replaced range and summary.
    */
   abstract compactRegion(

packages/core/agent-loop/README.md

@@ -50,7 +50,7 @@ Configured agents start automatically. `cwd` applies only to fresh sessions; `re
 
 The driver owns one agent for its lifetime. It records turn, step, request, stream, and tool boundaries in the session log; live extension events coordinate policy around those durable facts. The [architecture turn flow](../../../docs/architecture.md#turn-flow) and generated [event catalog](../../../docs/cordis-catalog/events.md) are the authoritative sequence and signatures.
 
-Every provider call that reaches a successful finish appends one `assistant/message` completion anchor after `agent/step-result`, including content-less calls and `max-tokens` finishes. The anchor records exact chunk provenance (`[]` for a stream with no chunks) and usage when available; empty content stays out of derived message history while those replay facts remain durable.
+Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history.
 
 Plugin failure ends the current turn, not the loop. Cancellation clears pending work and aborts the current step without leaking to the next prompt. Terminal continuation stops remain authoritative through turn close and durability flush.
 

packages/core/agent-loop/src/loop.ts

@@ -6,7 +6,7 @@
  */
 
 import type { Context } from 'cordis'
-import type { FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
+import type { FinishReason, GenerateOptions, LlmCallConfig, Message, TokenUsage } from '@deepseek-ai/dsh-llm'
 import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
 import { agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
 import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
@@ -527,29 +527,26 @@ async function runStep(
 
   if (assembler.finish.kind === 'max-tokens') {
     let message: Message = withoutToolCalls(assembler.message())
-    message = withoutToolCalls(await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message)))
-    // Every successful call records its completion anchor. Empty content is
-    // skipped by deriveMessages(), while exact chunk provenance lets replay
-    // distinguish a known empty provider stream from unrecorded provenance.
-    session.append(
-      'assistant/message',
-      { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) },
-      { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
+    message = withoutToolCalls(await processStepResult(
+      events, session, turn, step, message, assembler.usage, chunkSeqs,
+    ))
+    appendAssistantCompletion(
+      session, turn, step, message.content, assembler.usage, chunkSeqs,
     )
     return { hadToolCalls: false, finish: assembler.finish }
   }
 
   // Record the post-waterfall message that tool dispatch uses.
   let message: Message = assembler.message()
-  message = await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message))
+  message = await processStepResult(
+    events, session, turn, step, message, assembler.usage, chunkSeqs,
+  )
 
   // Every successful call records its completion anchor. A present empty
   // source set means the provider stream was known to contain no chunks;
   // omission remains the conservative legacy/unrecorded representation.
-  session.append(
-    'assistant/message',
-    { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) },
-    { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
+  appendAssistantCompletion(
+    session, turn, step, message.content, assembler.usage, chunkSeqs,
   )
 
   // Tool execution stays sequential; recheck abort around each normalized result.
@@ -601,6 +598,42 @@ async function runStep(
   return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
 }
 
+/** Append the single durable completion anchor for one successful provider call. */
+function appendAssistantCompletion(
+  session: Session,
+  turn: number,
+  step: number,
+  content: Message['content'],
+  usage: TokenUsage | undefined,
+  sourceEventSeqs: number[],
+): void {
+  session.append(
+    'assistant/message',
+    { turn, step, content, ...(usage ? { usage } : {}) },
+    { surfaceOp: 'append', sourceEventSeqs },
+  )
+}
+
+/** Preserve successful-call accounting without retaining output that result processing rejected. */
+async function processStepResult(
+  events: AgentEventDispatch,
+  session: Session,
+  turn: number,
+  step: number,
+  message: Message,
+  usage: TokenUsage | undefined,
+  sourceEventSeqs: number[],
+): Promise<Message> {
+  try {
+    return await events.waterfall(
+      'agent/step-result', turn, step, message, () => Promise.resolve(message),
+    )
+  } catch (error: unknown) {
+    appendAssistantCompletion(session, turn, step, [], usage, sourceEventSeqs)
+    throw error
+  }
+}
+
 function withoutToolCalls(message: Message): Message {
   return { ...message, content: message.content.filter(block => block.type !== 'tool-call') }
 }

packages/core/agent-loop/tests/contract-regressions.spec.ts

@@ -8,7 +8,7 @@ import AgentRegistry, { AgentId, type ContinuationDecision } from '@deepseek-ai/
 import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
 import { prepareReactLoopAgent } from '../src/agent.ts'
 import * as Invariants from '@deepseek-ai/dsh-invariants'
-import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
+import { maxTokensResponse, MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
 
 /** Regression tests for agent-loop boundary, identity, and lifecycle contracts. */
 
@@ -89,6 +89,72 @@ describe('session log records what agent/step-result actually produced', () => {
   })
 })
 
+describe('successful provider completion survives agent/step-result failure', () => {
+  async function expectContentlessCompletionAnchor(
+    response: StreamChunk[],
+    id: string,
+    providerText: string,
+  ): Promise<void> {
+    const adapter = new MockAdapter([response])
+    const ctx = await harness(adapter)
+    await ctx.plugin(Invariants)
+    const agent = ctx.agentLoop.create(AgentId(id), { model: 'mock' })
+    const failure = new Error(`${id} result processing failed`)
+    const reported: Error[] = []
+
+    ctx.on('agent/step-result', async () => {
+      throw failure
+    })
+    ctx.on('agent/error', (subject, _turn, _step, error) => {
+      if (subject === agent) reported.push(error)
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    const events = [...agent.session.events]
+    const chunks = events.filter(event => event.type === 'assistant/chunk')
+    const completions = events.filter(event => event.type === 'assistant/message')
+    expect(completions).toHaveLength(1)
+    expect(completions[0]?.type === 'assistant/message' && completions[0].data).toEqual({
+      turn: 1,
+      step: 1,
+      content: [],
+      usage: { inputTokens: 10, outputTokens: providerText.length },
+    })
+    expect(completions[0]?.sourceEventSeqs).toEqual(chunks.map(event => event.seq))
+    expect(agent.session.deriveMessages()).toEqual([
+      { role: 'user', content: [{ type: 'text', text: 'go' }] },
+    ])
+    expect(reported).toHaveLength(1)
+    expect(reported[0]).toBe(failure)
+    const turnEnd = events.findLast(event => event.type === 'turn/end')
+    expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({
+      kind: 'error',
+      step: 1,
+      message: failure.message,
+    })
+  }
+
+  it('records one content-less anchor when ordinary stop result processing rejects', async () => {
+    const providerText = 'ordinary provider output'
+    await expectContentlessCompletionAnchor(
+      textResponse(providerText),
+      'a-step-result-stop-failure',
+      providerText,
+    )
+  })
+
+  it('records one content-less anchor when max-token result processing rejects', async () => {
+    const providerText = 'truncated provider output'
+    await expectContentlessCompletionAnchor(
+      maxTokensResponse(providerText),
+      'a-step-result-max-token-failure',
+      providerText,
+    )
+  })
+})
+
 describe('abort during tool execution ends the turn', () => {
   it('aborting the in-flight step inside a tool prevents both remaining tools and the next model step', async () => {
     const adapter = new MockAdapter([