Abort/failure materialization silently truncates accepted stream deltas after 1,000 documents

Published `@convex-dev/agent@0.7.1` can expose accepted stream deltas to the UI and then silently omit their tail when abort/failure materializes the pending canonical message. `getStreamingMessagesWithMetadata` reads only 1,000 delta documents. The reproduction uses 1,002: one text-start, 1,000 text deltas and a Unicode suffix.

The UI delta query contains the suffix before abort, but the failed canonical message has only 999 `x` characters afterward. The expected persisted result is all 1,000 `x` characters plus the suffix. Both Stop and provider-failure variants fail on pristine 0.7.1.

## Reproduction

Use a fresh directory with Node 22 and these exact dependencies (no app code, deployment or provider credentials):

```bash
npm init -y
npm install --save-exact @convex-dev/agent@0.7.1 convex@1.45.0 convex-test@0.0.56 convex-helpers@0.1.109 ai@7.0.93 @ai-sdk/provider@4.0.10 @ai-sdk/provider-utils@5.0.36 zod@3.25.76 react@19.2.3 vitest@3.2.4
npx vitest run --config vitest.config.mjs
```

`vitest.config.mjs`:

```js
export default { test: { environment: 'node', include: ['repro.test.ts'], fileParallelism: false, maxWorkers: 1, testTimeout: 30000 } };
```

`repro.test.ts`:

```ts
import { afterEach, expect, it, vi } from 'vitest';
import { convexTest } from 'convex-test';
import componentTest from '@convex-dev/agent/test';
import { componentsGeneric, defineSchema } from 'convex/server';
import { createThread, saveMessage, type AgentComponent } from '@convex-dev/agent';

const component = componentsGeneric().agent as unknown as AgentComponent;
function setup() {
const t = convexTest({
schema: defineSchema({}),
modules: { './_generated/server.ts': async () => ({}) },
transactionLimits: true,
});
componentTest.register(t);
return t;
}
afterEach(() => vi.useRealTimers());

it.each(['Stopped', 'Provider failed'])('preserves all 1,002 accepted delta documents on %s', async (reason) => {
vi.useFakeTimers();
const t = setup();
const threadId = await t.mutation(ctx => createThread(ctx, component));
const pending = await t.mutation(component.messages.addMessages, {
threadId, messages: [{ message: { role: 'assistant', content: [] }, status: 'pending' }],
});
const streamId = await t.mutation(component.streams.create, { threadId, order: 0, stepOrder: 0, format: 'UIMessageChunk' });
await t.mutation(component.streams.addDelta, { streamId, start: 0, end: 1, parts: [{ type: 'text-start', id: 'text' }] });
for (let start = 1; start <= 1000; start += 100) {
await t.mutation(async ctx => {
for (let i = start; i < start + 100; i++) {
await ctx.runMutation(component.streams.addDelta, { streamId, start: i, end: i + 1, parts: [{ type: 'text-delta', id: 'text', delta: 'x' }] });
}
});
}
const tail = 'accepted 🦉 tail';
await t.mutation(component.streams.addDelta, { streamId, start: 1001, end: 1002, parts: [{ type: 'text-delta', id: 'text', delta: tail }] });
const visible = await t.query(component.streams.listDeltas, { threadId, cursors: [{ streamId, cursor: 1000 }] });
expect(JSON.stringify(visible)).toContain(tail);
await t.mutation(component.streams.abort, { streamId, reason });
await t.mutation(component.messages.finalizeMessage, { messageId: pending.messages[0]._id, result: { status: 'failed', error: reason } });
const [after] = await t.query(component.messages.getMessagesByIds, { messageIds: [pending.messages[0]._id] });
expect(after?.text).toBe('x'.repeat(1000) + tail);
expect(after?.status).toBe('failed');
});

```

## Related bounded-selection problem

The same recovery helper selects only ten matching stream records. A deterministic source/distribution helper test separately reproduces omitted records when more than ten streams match. This is separate from any application's attempt-correlation policy.

## Requested behavior

Materialize the accepted prefix completely within explicit document/byte bounds. If a bound is reached, retain the recoverable accepted prefix with a failed canonical status and an explicit materialization error; do not silently present a truncated successful result or throw away the accepted prefix by rolling back finalization. The read-only UI query should remain usable with bounded degradation.

Our local source/distribution patch uses 100 streams/1 MiB metadata and 8,192 delta documents/8 MiB aggregate deltas, with a separate application admission bound. Those numeric limits are application defensive choices, not a proposed universal Agent cap. Regression checks cover source and distribution, Stop/provider failure, Unicode, exact canonical IDs, metadata/delta bounds and prefix preservation.

Start by running the supplied repro.test.ts with the exact dependencies, then trace getStreamingMessagesWithMetadata through streams.listDeltas, streams.abort, and messages.finalizeMessage. Compare Stop and Provider failed behavior against the requested bounded materialization rules; done means the full accepted prefix, including Unicode, is preserved without unbounded reads or silent truncation.

Written by the indexing model from the issue text.