From 4c48784b1dbdba97d1dc3e4a7f4ab74eeace94a8 Mon Sep 17 00:00:00 2001 From: BK Date: Fri, 23 Oct 2026 13:07:16 +0700 Subject: [PATCH 1/2] fix(transport,loop): Stop records turn.aborted; keep non-streamed tool narration - conversation transport: an explicit Stop / client disconnect (LoopError cause 'caller-abort') finalizes with turn.abort {reason} instead of turn.fail, on both the authority and sink-only paths. Envelope code, ledger settle code and session accounting are unchanged. - ExecutionAuthority.finalizeOwned accepts kind 'turn.abort' (+ reason) and counts turn.aborted as a written terminal. - loop: a hop that streamed nothing but returns text beside its tool calls emits that text once as a delta (block start/delta/end before the calls) - previously a non-streaming provider's narration never reached the durable log. Streamed hops and final answers are unchanged. Tests: Stop -> turn.aborted on both paths, provider failure stays turn.failed, non-streamed narration lands once and before its call, streamed narration is not duplicated (10 source mutations, all caught). --- src/loop.ts | 13 +++++++++ src/orchestrate/execution-authority.ts | 10 ++++--- src/transport/conversation.ts | 12 +++++++-- test/abort.test.ts | 28 +++++++++++++++++++ test/activity-blocks.test.ts | 37 ++++++++++++++++++++++++++ test/conversation-authority.test.ts | 6 +++++ 6 files changed, 101 insertions(+), 5 deletions(-) diff --git a/src/loop.ts b/src/loop.ts index dd6971a..8d2ff0d 100644 --- a/src/loop.ts +++ b/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 diff --git a/src/orchestrate/execution-authority.ts b/src/orchestrate/execution-authority.ts index e0af6aa..acec228 100644 --- a/src/orchestrate/execution-authority.ts +++ b/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); diff --git a/src/transport/conversation.ts b/src/transport/conversation.ts index 68b80c7..a49adda 100644 --- a/src/transport/conversation.ts +++ b/src/transport/conversation.ts @@ -541,6 +541,10 @@ 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 @@ -548,7 +552,7 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver // 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')); } diff --git a/test/abort.test.ts b/test/abort.test.ts index c0c19dd..6f07ba5 100644 --- a/test/abort.test.ts +++ b/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, { 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(); + } + }); }); diff --git a/test/activity-blocks.test.ts b/test/activity-blocks.test.ts index 4a146a9..ab55af0 100644 --- a/test/activity-blocks.test.ts +++ b/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 diff --git a/test/conversation-authority.test.ts b/test/conversation-authority.test.ts index 7837d80..a1ec947 100644 --- a/test/conversation-authority.test.ts +++ b/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(); From 16d1b03eba12a834da63917cbb0468f57b0de0b4 Mon Sep 17 00:00:00 2001 From: BK Date: Fri, 23 Oct 2026 13:07:16 +0700 Subject: [PATCH 2/2] docs(changelog): stop as abort, non-streamed tool narration --- CHANGELOG.md | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8443bc7..5962d17 100644 --- a/CHANGELOG.md +++ b/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__n`, 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)