feat(bookkeeping): correction-chain depth guard + Bedrock stream retry (#1581)

* feat(bookkeeping): bypassable chain-depth guard on corrections and stornos

Correcting or reversing an entry that already sits 3+ links deep in a
rattelse chain (correction_of_id/reverses_id walked in the DB, never
description matching) now throws CORRECTION_CHAIN_TOO_DEEP, steering the
caller to book ONE correction expressing the chain's net effect. Agents
looped storno+rattelse 10 deep on a live company (63/193 vouchers noise).

The guard is advisory, never a dead end: allow_deep_chain bypasses it on
every surface (correctEntry/reverseEntry option, REST body, MCP tool arg
staged through pending_operations, and confirm dialogs with Ratta anda /
Aterfor anda in the web UI). MCP staging pre-flight fires the guard at
stage time so the agent reconsiders in the same turn, and the executor
re-checks at commit. tools/list payload ceiling bumped 59K -> 59.5K for
the two bypass properties (trimmed to one sentence first).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(agent): retry the Bedrock stream once on transient failures

A transient stream death (429/5xx, transport cut, or the two known
stream-corruption signatures: 'Unexpected event order' and 'request ended
without sending any chunks') killed the whole chat turn, stranding the
user mid-answer. The turn now retries once per turn after a short backoff:
safe because nothing is persisted until finalMessage() succeeds. A new
stream_restart event carries the pre-attempt text snapshot so the chat
client resets the partial bubble, drops uncompleted tool chips, and shows
'Forsoker igen...' until the retried stream produces text. Non-transient
errors (403, 400) keep the existing immediate-error path.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(api): regenerate accounted-api skill and wire allow_deep_chain through v1

apiskill:check failed: CorrectJournalEntrySchema gained allow_deep_chain,
making references/journal-entries.md stale. Regenerated (hand-applied: the
generator output is deterministic from the registry). While wiring: the v1
correct route validated allow_deep_chain but dropped it, and the v1 reverse
route's strict body schema would have rejected it outright, leaving API
clients no bypass when the chain-depth guard fires. Both now forward the
flag to the engine and document CORRECTION_CHAIN_TOO_DEEP as a pitfall.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* chore: re-trigger CI after Vercel infra hang

The preview for e527e4044 compiled in 91s then hung 40 minutes in the
TypeScript phase and was killed with no error output; a CLI redeploy of
the identical code went Ready in 5m. Empty commit to refresh the git-
triggered deployment status.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(bookkeeping): address CodeRabbit review on the chain-depth guard

- correction-chain: report rootVoucher only when the walk reached a
  genuine parentless root; a broken link, cycle, or hop-cap now yields
  null instead of presenting an intermediate voucher as the chain root.
- recordate: propagate allow_deep_chain end-to-end (recordateEntry
  option, route schema, and a Flytta anda bypass confirm in the dialog);
  a date move is another storno+rattelse layer and carried the guard
  with no override path.
- v1 correct/reverse: run the chain-depth guard before the dry-run
  return so a dry run gives the same verdict as the real execution.
- dashboard reverse route: 400 on malformed JSON or a non-boolean
  allow_deep_chain instead of silently reversing without the override;
  empty body stays the supported no-body case. Tests added.
- AgentChat stream_restart: discard the dead attempt's reasoning and
  re-arm the post-tool paragraph break so a retried turn doesn't render
  thinking twice or glue its continuation onto restored text.
- v1 reverse route doc comment updated for allow_deep_chain.

Not changed: the journal-list reverse flow (flagged as a dead end) can
never receive CORRECTION_CHAIN_TOO_DEEP: the list renders Aterfor only
for entries that are neither storno nor correction, and such entries
have no backward chain links, so their depth is always 0.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* test(bookkeeping): recordate route test expects the new options arg

recordateEntry now takes { allowDeepChain } as a sixth argument; the
route test's called-with assertion predates it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Mattsson
2026-08-13 19:32:41 +02:00
committed by GitHub
co-authored by Claude Fable 5
parent e494662530
commit 05380ddf54
33 changed files with 1355 additions and 148 deletions
@@ -0,0 +1,167 @@
import { describe, it, expect, vi, beforeEach } from 'vitest'
import type { AgentIntent } from '@/lib/agent/intents/types'
import type { StreamEvent } from '../run-turn'
// Anthropic client mock: `finalMessage()` delegates to a queued mock so tests
// can make the first stream attempt fail and the retry succeed. Mirrors the
// adapter in run-turn-memory.test.ts.
const messagesCreate = vi.fn()
vi.mock('@/lib/agent/composer/client', () => ({
getAnthropic: () => ({
messages: {
create: messagesCreate,
stream: (args: unknown) => {
const stream = {
on: () => stream,
finalMessage: () => messagesCreate(args),
}
return stream
},
},
}),
SONNET_MODEL: 'claude-sonnet-5',
MAX_TOKENS_NO_THINKING: 5400,
MAX_TOKENS_STANDARD: 16000,
MAX_TOKENS_DEEP: 24000,
}))
vi.mock('../system-prompt', () => ({
buildSystemPrompt: vi.fn().mockResolvedValue({
blocks: [],
promptHash: 'sha256:test',
atomsLoaded: [],
}),
}))
const getMock = vi.fn()
const getManyMock = vi.fn()
vi.mock('@/lib/agent/tools/registry', () => ({
agentToolRegistry: {
get: (...args: unknown[]) => getMock(...args),
getMany: (...args: unknown[]) => getManyMock(...args),
},
}))
import { runChatTurn, isTransientStreamError } from '../run-turn'
function fakeSupabase() {
const passthrough: Record<string, unknown> = {}
const proxy: unknown = new Proxy(passthrough, {
get(_t, prop) {
if (prop === 'then') {
return (resolve: (v: unknown) => void) => resolve({ data: null, error: null })
}
return () => proxy
},
})
return proxy as unknown as Parameters<typeof runChatTurn>[0]['supabase']
}
function makeIntent(): AgentIntent {
return {
id: 'general.help',
buttonLabel: 'x',
sheetTitle: 'x',
atoms: { mode: 'progressive', horizontal: [], includeCompanyVertical: false, includeCompanyModifiers: false },
tools: [],
model: 'claude-sonnet-5',
capture: async () => ({}),
promptTemplate: () => '',
}
}
async function runTurn(events: StreamEvent[]): Promise<void> {
await runChatTurn({
supabase: fakeSupabase(),
userId: 'user-1',
companyId: 'company-1',
companyName: 'Acme AB',
firstName: 'Anna',
intent: makeIntent(),
conversationId: 'conv-1',
userMessage: 'hej',
persist: false,
emit: (e) => {
events.push(e)
return true
},
})
}
function transientError(message: string, status?: number): Error & { status?: number } {
const err = new Error(message) as Error & { status?: number }
if (status !== undefined) err.status = status
return err
}
beforeEach(() => {
vi.clearAllMocks()
getManyMock.mockResolvedValue([])
})
describe('isTransientStreamError', () => {
it('classifies the known stream-corruption signatures as transient', () => {
expect(isTransientStreamError(new Error('Unexpected event order, got message_stop'))).toBe(true)
expect(isTransientStreamError(new Error('request ended without sending any chunks'))).toBe(true)
})
it('classifies throttling, 5xx, and transport cuts as transient', () => {
expect(isTransientStreamError(transientError('Too many requests', 429))).toBe(true)
expect(isTransientStreamError(transientError('Internal failure', 503))).toBe(true)
expect(isTransientStreamError(new Error('read ECONNRESET'))).toBe(true)
expect(isTransientStreamError(new Error('Request timed out'))).toBe(true)
})
it('classifies auth and validation failures as permanent', () => {
expect(isTransientStreamError(transientError('Forbidden', 403))).toBe(false)
expect(isTransientStreamError(transientError('Invalid model', 400))).toBe(false)
// A 4xx stays permanent even when the message contains a transient token.
expect(isTransientStreamError(transientError('socket auth rejected', 403))).toBe(false)
expect(isTransientStreamError(new Error('validation failed: max_tokens'))).toBe(false)
})
})
describe('runChatTurn: transient stream retry', () => {
it('retries once on a transient failure and completes the turn', async () => {
messagesCreate
.mockRejectedValueOnce(new Error('Unexpected event order, got message_stop'))
.mockResolvedValueOnce({
content: [{ type: 'text', text: 'Hej igen.' }],
stop_reason: 'end_turn',
})
const events: StreamEvent[] = []
await runTurn(events)
const restarts = events.filter((e) => e.kind === 'stream_restart')
expect(restarts).toHaveLength(1)
expect(restarts[0]).toMatchObject({ kind: 'stream_restart', assistant_text: '' })
expect(events.find((e) => e.kind === 'error')).toBeUndefined()
expect(events.find((e) => e.kind === 'turn_complete')).toBeDefined()
expect(messagesCreate).toHaveBeenCalledTimes(2)
}, 15_000)
it('does not retry a non-transient failure (403): error emitted, turn thrown', async () => {
messagesCreate.mockRejectedValueOnce(transientError('Forbidden', 403))
const events: StreamEvent[] = []
await expect(runTurn(events)).rejects.toThrow('Forbidden')
expect(events.filter((e) => e.kind === 'stream_restart')).toHaveLength(0)
expect(events.find((e) => e.kind === 'error')).toBeDefined()
expect(messagesCreate).toHaveBeenCalledTimes(1)
})
it('retries at most once per turn: a second transient failure surfaces as an error', async () => {
messagesCreate
.mockRejectedValueOnce(transientError('Service unavailable', 503))
.mockRejectedValueOnce(transientError('Service unavailable', 503))
const events: StreamEvent[] = []
await expect(runTurn(events)).rejects.toThrow('Service unavailable')
expect(events.filter((e) => e.kind === 'stream_restart')).toHaveLength(1)
expect(events.find((e) => e.kind === 'error')).toBeDefined()
expect(messagesCreate).toHaveBeenCalledTimes(2)
}, 15_000)
})
+185 -94
View File
@@ -51,6 +51,52 @@ export function friendlyModelError(err: unknown): string {
return 'Något gick fel hos assistenten. Försök igen om en stund.'
}
/**
* True when a Bedrock stream failure is transient: the identical request can
* succeed on an immediate retry without any input change. Covers throttling
* (429), server errors (5xx), transport cuts (timeout/reset/socket), and the
* two known SDK stream-corruption signatures observed in prod on the pinned
* 0.29.x SDK ("Unexpected event order", "request ended without sending any
* chunks"). Auth/validation failures (4xx other than 429) are NOT transient:
* retrying them is wasted work and they must keep surfacing immediately.
*/
export function isTransientStreamError(err: unknown): boolean {
const status = (err as { status?: number } | null)?.status
if (status === 429) return true
if (typeof status === 'number' && status >= 500) return true
// A non-429 4xx is permanent regardless of message text.
if (typeof status === 'number' && status >= 400) return false
const name = (err as { name?: string } | null)?.name ?? ''
const raw = err instanceof Error ? err.message : ''
let cause = ''
try {
const c = (err as { cause?: unknown } | null)?.cause
cause = c instanceof Error ? `${c.name} ${c.message}` : c != null ? String(c) : ''
} catch {
cause = ''
}
const text = `${name} ${raw} ${cause}`.toLowerCase()
return (
text.includes('unexpected event order') ||
text.includes('request ended without sending any chunks') ||
text.includes('throttl') ||
text.includes('rate limit') ||
text.includes('rate exceeded') ||
text.includes('too many') ||
text.includes('timeout') ||
text.includes('timed out') ||
text.includes('etimedout') ||
text.includes('econnreset') ||
text.includes('network') ||
text.includes('socket')
)
}
// One automatic retry per turn on a transient stream failure, after a short
// backoff. Per-turn, not per-iteration: a turn that dies twice is not a blip.
const STREAM_RETRY_BACKOFF_MS = 750
// One turn of the chat loop:
//
// 1. Resolve context (company, profile, ranked memory).
@@ -97,6 +143,15 @@ export type StreamEvent =
memory_kind?: 'fact' | 'preference' | 'pattern' | 'correction'
content?: string
}
| {
// The Bedrock stream died on a transient error and the turn is being
// retried once. Text and eager tool chips from the dead stream were
// never persisted; the chat surface must reset the in-progress
// assistant bubble to `assistant_text` (what had accumulated BEFORE the
// failed attempt) and drop un-completed tool chips.
kind: 'stream_restart'
assistant_text: string
}
| { kind: 'turn_complete'; assistant_text: string }
| { kind: 'error'; message: string }
@@ -255,6 +310,12 @@ export async function runChatTurn(args: RunTurnArgs): Promise<void> {
let assistantText = ''
let iterations = 0
// One automatic retry per TURN when the Bedrock stream dies on a transient
// error (throttling, 5xx, transport cut, stream corruption). The failed
// attempt persisted nothing (persist happens after finalMessage() succeeds),
// so a clean retry is safe; the client is told to discard the partial
// bubble via stream_restart.
let streamRetryUsed = false
// Extended thinking ("tänka längre"): when the intent opts in, every model
// call in the loop gets a reasoning channel so the agent reasons BEFORE it
@@ -291,103 +352,133 @@ export async function runChatTurn(args: RunTurnArgs): Promise<void> {
// sees Anna's reply appear word-by-word instead of waiting 1-5 s for
// the full block to land. We still collect the final assembled message
// for tool detection, persistence and stop-reason control flow.
const stream = anthropic.messages.stream({
model,
max_tokens: maxTokens,
system: systemPrompt.blocks,
messages,
tools: tools.length > 0 ? tools.map(toAnthropicTool) : undefined,
...(thinking ? { thinking } : {}),
...(outputConfig ? { output_config: outputConfig } : {}),
})
stream.on('text', (delta) => {
assistantText += delta
emit({ kind: 'text_delta', delta })
})
// Track which tool_use ids have already been announced to the client so
// the dispatch loop below doesn't re-emit them. Eager-emitting on
// `content_block_start` shaves the perceived lag for tool chips: the
// chip appears the moment the LLM commits to a tool call, instead of
// after the entire response is buffered.
const eagerToolIds = new Set<string>()
stream.on('streamEvent', (ev) => {
// The raw stream event shape depends on the SDK; we care about
// content_block_start with a tool_use block, and content_block_delta
// carrying extended-thinking text.
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const e = ev as any
if (
e?.type === 'content_block_delta' &&
e?.delta?.type === 'thinking_delta' &&
typeof e.delta.thinking === 'string'
) {
emit({ kind: 'reasoning_delta', delta: e.delta.thinking })
return
}
if (e?.type === 'content_block_start' && e?.content_block?.type === 'tool_use') {
const block = e.content_block
if (typeof block.id === 'string' && typeof block.name === 'string') {
eagerToolIds.add(block.id)
emit({
kind: 'tool_use',
tool_use_id: block.id,
name: block.name,
// Input is still being streamed at this point; the chip only
// displays the tool name so empty input is fine.
input: {},
})
}
}
})
//
// The attempt loop wraps stream creation + finalMessage() so a transient
// failure (throttling, 5xx, transport cut, stream corruption) gets ONE
// automatic retry per turn. Everything the dead stream emitted is
// reverted: assistantText rolls back to its pre-attempt snapshot and the
// client discards the partial bubble on stream_restart.
let response
try {
response = await stream.finalMessage()
} catch (err) {
// Surface as a chat error so the UI clears its streaming state. Re-throw
// to let the route's outer try/catch persist the failure if needed.
// Normalize Bedrock throttling/timeout/5xx into a friendly Swedish line.
//
// Extract status/code/cause/stack explicitly: the logger keeps only
// name/message/code from an Error and drops the stack in production, so
// the real failure was invisible (every prod log just said "request ended
// without sending any chunks"). These fields tell us whether the empty
// stream is auth (403), bad model/region (400), throttling (429), or a
// genuine transport cut. No secrets: AWS/SDK errors carry none, and the
// logger still redacts personnummer/UUIDs from any string.
const bedrockErr = err as {
status?: number
code?: string
cause?: unknown
stack?: string
}
let errCause: string | undefined
try {
errCause =
bedrockErr?.cause != null
? String(
bedrockErr.cause instanceof Error
? `${bedrockErr.cause.name}: ${bedrockErr.cause.message}`
: bedrockErr.cause,
).slice(0, 300)
: undefined
} catch {
errCause = '[uninspectable cause]'
}
log.error('Bedrock stream failed', err, {
conversationId,
companyId,
let eagerToolIds = new Set<string>()
for (;;) {
const assistantTextBefore = assistantText
const stream = anthropic.messages.stream({
model,
iterations,
errStatus: typeof bedrockErr?.status === 'number' ? bedrockErr.status : undefined,
errCode: typeof bedrockErr?.code === 'string' ? bedrockErr.code : undefined,
errCause,
errStack: typeof bedrockErr?.stack === 'string' ? bedrockErr.stack.slice(0, 1200) : undefined,
max_tokens: maxTokens,
system: systemPrompt.blocks,
messages,
tools: tools.length > 0 ? tools.map(toAnthropicTool) : undefined,
...(thinking ? { thinking } : {}),
...(outputConfig ? { output_config: outputConfig } : {}),
})
emit({ kind: 'error', message: friendlyModelError(err) })
throw err
stream.on('text', (delta) => {
assistantText += delta
emit({ kind: 'text_delta', delta })
})
// Track which tool_use ids have already been announced to the client so
// the dispatch loop below doesn't re-emit them. Eager-emitting on
// `content_block_start` shaves the perceived lag for tool chips: the
// chip appears the moment the LLM commits to a tool call, instead of
// after the entire response is buffered.
eagerToolIds = new Set<string>()
stream.on('streamEvent', (ev) => {
// The raw stream event shape depends on the SDK; we care about
// content_block_start with a tool_use block, and content_block_delta
// carrying extended-thinking text.
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const e = ev as any
if (
e?.type === 'content_block_delta' &&
e?.delta?.type === 'thinking_delta' &&
typeof e.delta.thinking === 'string'
) {
emit({ kind: 'reasoning_delta', delta: e.delta.thinking })
return
}
if (e?.type === 'content_block_start' && e?.content_block?.type === 'tool_use') {
const block = e.content_block
if (typeof block.id === 'string' && typeof block.name === 'string') {
eagerToolIds.add(block.id)
emit({
kind: 'tool_use',
tool_use_id: block.id,
name: block.name,
// Input is still being streamed at this point; the chip only
// displays the tool name so empty input is fine.
input: {},
})
}
}
})
try {
response = await stream.finalMessage()
break
} catch (err) {
if (!streamRetryUsed && isTransientStreamError(err)) {
streamRetryUsed = true
log.warn('Bedrock stream failed transiently, retrying once', {
conversationId,
companyId,
model,
iterations,
errMessage: err instanceof Error ? err.message.slice(0, 300) : String(err).slice(0, 300),
errStatus: (err as { status?: number } | null)?.status,
})
// Roll back what the dead stream produced. Nothing was persisted
// (persist happens after finalMessage() succeeds), so state-wise
// this attempt never happened; the client resets its bubble.
assistantText = assistantTextBefore
emit({ kind: 'stream_restart', assistant_text: assistantText })
await new Promise((resolve) => setTimeout(resolve, STREAM_RETRY_BACKOFF_MS))
continue
}
// Surface as a chat error so the UI clears its streaming state. Re-throw
// to let the route's outer try/catch persist the failure if needed.
// Normalize Bedrock throttling/timeout/5xx into a friendly Swedish line.
//
// Extract status/code/cause/stack explicitly: the logger keeps only
// name/message/code from an Error and drops the stack in production, so
// the real failure was invisible (every prod log just said "request ended
// without sending any chunks"). These fields tell us whether the empty
// stream is auth (403), bad model/region (400), throttling (429), or a
// genuine transport cut. No secrets: AWS/SDK errors carry none, and the
// logger still redacts personnummer/UUIDs from any string.
const bedrockErr = err as {
status?: number
code?: string
cause?: unknown
stack?: string
}
let errCause: string | undefined
try {
errCause =
bedrockErr?.cause != null
? String(
bedrockErr.cause instanceof Error
? `${bedrockErr.cause.name}: ${bedrockErr.cause.message}`
: bedrockErr.cause,
).slice(0, 300)
: undefined
} catch {
errCause = '[uninspectable cause]'
}
log.error('Bedrock stream failed', err, {
conversationId,
companyId,
model,
iterations,
retried: streamRetryUsed,
errStatus: typeof bedrockErr?.status === 'number' ? bedrockErr.status : undefined,
errCode: typeof bedrockErr?.code === 'string' ? bedrockErr.code : undefined,
errCause,
errStack: typeof bedrockErr?.stack === 'string' ? bedrockErr.stack.slice(0, 1200) : undefined,
})
emit({ kind: 'error', message: friendlyModelError(err) })
throw err
}
}
const assistantContent: ContentBlock[] = response.content