fix(transport,loop): Stop records turn.aborted; keep non-streamed tool narration

CHANGELOG.md

@@ -1,4 +1,4 @@
-## [Unreleased] (detached send, server-owned queue, steering, push-on-append; per-source chunk budget)
+## [Unreleased] (detached send, server-owned queue, steering, push-on-append; per-source chunk budget; stop as abort, non-streamed tool narration)
 
 ### Added
 - **Detached send (opt-in)** — `POST /api/agent/sessions/:id/messages` with `detach: true`
@@ -38,6 +38,13 @@
 - Tool events without an invocation id get a per-turn counter id (`tc_<turn>_n<k>`, paired FIFO
   by name) instead of the bare tool name, in the conversation transport and workflow runs.
 - `TurnEventBase.seq` comment: per-session, not per-turn (the schema is `UNIQUE(session_id, seq)`).
+- **A caller Stop is recorded as `turn.aborted {reason}`, not `turn.failed`**: the conversation
+  transport (authority and sink paths) on an explicit abort or client disconnect.
+  `ExecutionAuthority.finalizeOwned` accepts `kind: 'turn.abort'` (+ `reason`). The response
+  envelope (`code: 'ABORTED'`), the ledger settle code and session accounting are unchanged.
+- **Text written beside tool calls by a non-streaming provider is no longer lost**: the loop emits
+  it as one delta (one text block, before the calls), the same shape a streaming provider
+  produces. Streamed hops and final answers are unchanged.
 
 ## [0.8.0] — 2026-10-21 (Stage 3: tree-sitter symbols + watchRoots + structured search)
 

src/loop.ts

@@ -196,6 +196,19 @@ export async function runLoop(systemPrompt: string, userMessage: string, deps: L
     // aborted; ordinary success passes untouched.
     if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages, 'caller-abort');
 
+    // A non-streaming provider returns narration beside its tool calls as
+    // message content only — no deltas, so it never reached the delta/block
+    // rail and replayed logs lost it. Emit it as ONE delta, exactly what a
+    // streaming provider would have sent. Final answers (no tool calls) are
+    // left alone: their text rides the turn's answer.
+    const narration = res.message.content ?? '';
+    if (streamedText === '' && (res.message.tool_calls?.length ?? 0) > 0 && narration.trim() !== '') {
+      streamedText = narration;
+      const bid = tracker?.onDelta(hop, narration);
+      deps.onDelta?.(narration);
+      deps.onEvent?.({ type: 'turn.delta', hop, detail: { delta: narration, ...(bid !== undefined ? { blockId: bid } : {}) } });
+    }
+
     // F2: close the hop's text block BEFORE the message lands. text.end prefers
     // the STREAMED accumulation when any deltas flowed (byte-exact to what the
     // UI saw); content is the fallback for snapshot-only hops (which usually

src/orchestrate/execution-authority.ts

@@ -333,9 +333,11 @@ export class ExecutionAuthority {
   async finalizeOwned(
     attempt: ExecutionAttempt,
     input: {
-      kind: 'turn.complete' | 'turn.fail';
+      kind: 'turn.complete' | 'turn.fail' | 'turn.abort';
       answer?: string;
       error?: string;
+      /** turn.abort only: the recorded reason (e.g. a caller Stop) */
+      reason?: string;
       commandKey: string;
       settleCode: string;
       accounting?: { status?: 'active' | 'ended' | 'error'; tokensInDelta?: number; tokensOutDelta?: number };
@@ -351,9 +353,11 @@ export class ExecutionAuthority {
         if (!auth || auth.epoch !== this.epoch) throw new Error('stale authority epoch');
         const cmd: TurnCommand = (input.kind === 'turn.complete'
           ? { kind: 'turn.complete', sessionId: attempt.sessionId, turnId: attempt.turnId, answer: input.answer ?? '' }
-          : { kind: 'turn.fail', sessionId: attempt.sessionId, turnId: attempt.turnId, error: input.error ?? 'turn failed' }) as TurnCommand;
+          : input.kind === 'turn.abort'
+            ? { kind: 'turn.abort', sessionId: attempt.sessionId, turnId: attempt.turnId, reason: input.reason ?? 'aborted' }
+            : { kind: 'turn.fail', sessionId: attempt.sessionId, turnId: attempt.turnId, error: input.error ?? 'turn failed' }) as TurnCommand;
         const { events } = this.sink.handleInTransaction(attempt.sessionId, cmd, input.commandKey, undefined);
-        wroteTerminal = events.some((e) => e.kind === 'turn.completed' || e.kind === 'turn.failed');
+        wroteTerminal = events.some((e) => e.kind === 'turn.completed' || e.kind === 'turn.failed' || e.kind === 'turn.aborted');
         if (input.stash) this.storage.stashPartialSync?.(attempt.sessionId, attempt.turnId, input.stash.json);
         if (input.accounting) {
           this.storage.applyAccountingSync?.(attempt.sessionId, input.accounting);

src/transport/conversation.ts

@@ -541,14 +541,18 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver
         outcome.error = e.message ?? String(err);
       }
       outcome.code = e.code ?? undefined;
+      // an explicit Stop (or client disconnect) is recorded as turn.aborted,
+      // not turn.failed — readers tell "stopped" from "broke". The envelope
+      // and session accounting are unchanged (ABORTED keeps poisoning).
+      const callerAbort = err instanceof LoopError && err.cause === 'caller-abort';
       if (exec && ownedAttempt) {
         // §3.5 owned ATOMIC finalization: terminal event + budget stash +
         // session accounting + ledger settle in ONE commit. A storage failure
         // leaves the attempt reserved for an idempotent retry and reports
         // outcome-pending — never a false success.
         try {
           const fin = await finalize(ownedAttempt, {
-            kind: 'turn.fail',
+            ...(callerAbort ? { kind: 'turn.abort' as const, reason: outcome.error } : { kind: 'turn.fail' as const }),
             error: outcome.error,
             commandKey: `chat:${durableTurnId}:fail`,
             settleCode: budgetExhausted ? 'BUDGET_EXCEEDED' : (e.code ?? 'FAILED'),
@@ -561,7 +565,11 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver
           return { code: 503, payload: { error: `finalization storage failure — outcome pending, the turn stays reserved for idempotent retry: ${finErr instanceof Error ? finErr.message : String(finErr)}`, code: 'FINALIZATION_PENDING' } };
         }
       } else {
-        if (deps.sink || exec) durable({ kind: 'turn.fail', sessionId: id, turnId: durableTurnId, error: outcome.error } as TurnCommand, `chat:${durableTurnId}:fail`);
+        if (deps.sink || exec) {
+          durable((callerAbort
+            ? { kind: 'turn.abort', sessionId: id, turnId: durableTurnId, reason: outcome.error }
+            : { kind: 'turn.fail', sessionId: id, turnId: durableTurnId, error: outcome.error }) as TurnCommand, `chat:${durableTurnId}:fail`);
+        }
         await storage.updateSession(id, { status: budgetExhausted ? 'active' : 'error' });
         await releaseOwnership(budgetExhausted ? 'BUDGET_EXCEEDED' : (e.code ?? 'FAILED'));
       }

test/abort.test.ts

@@ -1,7 +1,9 @@
 import { describe, expect, it } from 'vitest';
 import net from 'node:net';
 import Fastify from 'fastify';
+import Database from 'better-sqlite3';
 import { SqliteStorage } from '../src/memory/sqlite.ts';
+import { EventSink } from '../src/orchestrate/event-sink.ts';
 import { MockProvider } from '../src/provider/mock.ts';
 import { ToolRegistry } from '../src/tools.ts';
 import { registerConversationApi } from '../src/transport/conversation.ts';
@@ -47,6 +49,32 @@ describe('abort mid-flight', () => {
     await app.close();
     storage.close();
   });
+
+  it('durable log (sink, no authority): a Stop records turn.aborted, never turn.failed', async () => {
+    const storage = new SqliteStorage(':memory:');
+    const sink = new EventSink(storage['db'] as InstanceType<typeof Database>, { passive: true, enqueueRunTurn: false });
+    const app = Fastify();
+    await registerConversationApi(app, {
+      storage, provider: new MockProvider([{ kind: 'hang', ms: 30_000 }]), registry: new ToolRegistry(), policy: { name: 'draft', allows: 'draft' },
+      systemPrompt: 'sys', budget: { wallMs: 60_000 }, emit: () => undefined, sink,
+    });
+    await app.listen({ port: 0, host: '127.0.0.1' });
+    const baseUrl = `http://127.0.0.1:${(app.server.address() as { port: number }).port}`;
+    try {
+      const sid = ((await (await fetch(`${baseUrl}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json()) as { session: { id: string } }).session.id;
+      const turnPromise = fetch(`${baseUrl}/api/agent/sessions/${sid}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content: 'hang please' }) })
+        .then((r) => r.json() as Promise<{ turn: { ok: boolean; code?: string } }>);
+      await new Promise((r) => setTimeout(r, 300));
+      await fetch(`${baseUrl}/api/agent/sessions/${sid}/abort`, { method: 'POST' });
+      expect((await turnPromise).turn.code).toBe('ABORTED');
+      const terminals = sink.events(sid).filter((e) => e.kind === 'turn.completed' || e.kind === 'turn.failed' || e.kind === 'turn.aborted');
+      expect(terminals.map((e) => [e.kind, e.detail])).toEqual([['turn.aborted', { reason: 'aborted by caller' }]]);
+      expect((await storage.getSession(sid))?.status).toBe('error'); // accounting unchanged: ABORTED still poisons
+    } finally {
+      await app.close();
+      storage.close();
+    }
+  });
 });
 
 

test/activity-blocks.test.ts

@@ -166,6 +166,43 @@ describe('text block lifecycle durability (F2)', () => {
     }
   });
 
+  it('a non-streamed hop: text written beside tool calls lands as ONE block before the call; the final answer stays block-less', async () => {
+    const provider = new StreamingProvider([
+      { content: 'Looking at the probe first.', toolCalls: [{ id: 'c1', type: 'function', function: { name: 'probe', arguments: '{"q":"x"}' } }] }, // no deltas
+      { content: 'done' }, // no deltas: rides the turn answer only
+    ]);
+    const rig = await boot(provider);
+    try {
+      const sid = await runTurn(rig, 'run probe');
+      const chunks = chunkEvents(rig.sink, sid);
+      expect(chunks.map((c) => c.detail.block?.phase)).toEqual(['start', 'delta', 'end']);
+      expect(chunks.map((c) => c.detail.delta).join('')).toBe('Looking at the probe first.'); // scalar rail carries it once
+      expect(chunks[2]!.detail.block!.text).toBe('Looking at the probe first.');
+      expect(new Set(chunks.map((c) => c.detail.block!.id)).size).toBe(1);
+      // order: the narration precedes the tool call it introduces
+      const kinds = rig.sink.events(sid).map((e) => e.kind).filter((k) => k === 'chunk' || k === 'tool_call');
+      expect(kinds).toEqual(['chunk', 'chunk', 'chunk', 'tool_call']);
+    } finally {
+      await rig.app.close();
+      rig.storage.close();
+    }
+  });
+
+  it('a streamed hop with tool calls is not duplicated by the snapshot', async () => {
+    const provider = new StreamingProvider([
+      { deltas: ['Let me ', 'check.'], content: 'Let me check.', toolCalls: [{ id: 'c1', type: 'function', function: { name: 'probe', arguments: '{"q":"x"}' } }] },
+      { content: 'done' },
+    ]);
+    const rig = await boot(provider);
+    try {
+      const sid = await runTurn(rig, 'run probe');
+      expect(chunkEvents(rig.sink, sid).map((c) => c.detail.delta).join('')).toBe('Let me check.');
+    } finally {
+      await rig.app.close();
+      rig.storage.close();
+    }
+  });
+
   it('a hop with no deltas emits no block', async () => {
     const provider = new StreamingProvider([
       { toolCalls: [{ id: 'c1', type: 'function', function: { name: 'probe', arguments: '{"q":"silent"}' } }] }, // no deltas

test/conversation-authority.test.ts

@@ -144,6 +144,8 @@ describe('authority-driven transport', () => {
       const sess = await rig.storage.getSession(sid);
       expect(sess?.status).toBe('error');
       expect(rig.auth.readSession(sid)).toMatchObject({ state: 'idle', admission: 'ready' }); // ownership released
+      // a provider failure stays a failure (only a caller Stop is an abort)
+      expect(rig.sink.events(sid).filter((e) => e.kind === 'turn.failed' || e.kind === 'turn.aborted').map((e) => e.kind)).toEqual(['turn.failed']);
     } finally {
       await rig.app.close();
       await rig.auth.drain();
@@ -187,6 +189,10 @@ describe('authority-driven transport', () => {
       expect(body.turn.ok).toBe(false);
       expect(body.turn.code).toBe('ABORTED');
       expect(rig.auth.readSession(sid)).toMatchObject({ state: 'idle', admission: 'ready' });
+      // a Stop is recorded as an abort, not a failure — readers show "stopped"
+      const terminals = rig.sink.events(sid).filter((e) => e.kind === 'turn.completed' || e.kind === 'turn.failed' || e.kind === 'turn.aborted');
+      expect(terminals.map((e) => [e.kind, e.detail])).toEqual([['turn.aborted', { reason: 'aborted by caller' }]]);
+      expect(rig.sink.state(sid).activeTurnId).toBeNull();
     } finally {
       await rig.app.close();
       await rig.auth.drain();