diff --git a/packages/agent-loop/tests/properties.spec.ts b/packages/agent-loop/tests/properties.spec.ts index 8e6768d402..23c0f77b46 100644 --- a/packages/agent-loop/tests/properties.spec.ts +++ b/packages/agent-loop/tests/properties.spec.ts @@ -58,13 +58,14 @@ function nextIdle(ctx: Context, agent: LoopAgent): Promise { }) } -/** Record every status transition for the legal-machine assertion. */ -function recordStatus(ctx: Context, agent: LoopAgent): string[] { +/** Record every status transition for the legal-machine assertion. Returns + * the seen list plus a disposer for the listener (per the registry convention). */ +function recordStatus(ctx: Context, agent: LoopAgent): { seen: string[]; dispose: () => void } { const seen: string[] = [] - ctx.on('agent/status', (subject, status) => { + const dispose = ctx.on('agent/status', (subject, status) => { if (subject === agent) seen.push(status) }) - return seen + return { seen, dispose } } function userMessageTexts(agent: LoopAgent): string[] { @@ -95,7 +96,7 @@ describe('agent loop scheduling properties', () => { const ctx = await harness() try { const agent = ctx.agentLoop.create('a', { model: 'mock' }) - const trace = recordStatus(ctx, agent) + const { seen: trace } = recordStatus(ctx, agent) const idle = nextIdle(ctx, agent) // Send all in one synchronous tick: they queue before the loop wakes. for (const text of texts) agent.send([{ type: 'text', text }]) @@ -103,15 +104,14 @@ describe('agent loop scheduling properties', () => { // No message lost: every send appears as a user/message, in order. expect(userMessageTexts(agent)).toEqual(texts) - // Turn numbers strictly increase. - const turns = turnNumbers(agent) - for (let i = 1; i < turns.length; i++) expect(turns[i]!).toBeGreaterThan(turns[i - 1]!) + // A synchronous burst batches into exactly one turn. + expect(turnNumbers(agent)).toEqual([1]) assertLegalStatusTrace(trace) } finally { await ctx.fiber.dispose() } }, - ), { numRuns: 25 }) + ), { numRuns: 25, timeout: 2000 }) }) it('sequential sends each get their own turn with increasing numbers', async () => { @@ -133,6 +133,44 @@ describe('agent loop scheduling properties', () => { await ctx.fiber.dispose() } }, - ), { numRuns: 20 }) + ), { numRuns: 20, timeout: 2000 }) + }) + + it('mixed schedule (send, optionally settle) loses no message and orders turns', async () => { + // Each step is a (text, settle?) pair: settle=true awaits idle before the + // next send (own turn); settle=false sends in the same tick (batches). + const stepArb = fc.record({ text: fc.string({ minLength: 1 }), settle: fc.boolean() }) + await fc.assert(fc.asyncProperty( + fc.array(stepArb, { minLength: 1, maxLength: 6 }), + async (steps) => { + const ctx = await harness() + try { + const agent = ctx.agentLoop.create('a', { model: 'mock' }) + // Capture an idle waiter before EACH send; the last one is guaranteed + // to resolve because the final send always triggers (or joins) a turn + // that ends idle. Awaiting an already-resolved waiter is a no-op, so a + // trailing settle step can't cause a hang. + let lastIdle: Promise | undefined + for (const step of steps) { + const idle = nextIdle(ctx, agent) + lastIdle = idle + agent.send([{ type: 'text', text: step.text }]) + if (step.settle) await idle + } + await lastIdle + + // No message lost or reordered, regardless of batching. + expect(userMessageTexts(agent)).toEqual(steps.map(s => s.text)) + // Turn numbers are a strictly increasing 1..N prefix (N = turn count). + const turns = turnNumbers(agent) + expect(turns).toEqual(turns.map((_, i) => i + 1)) + // Every message landed in some turn; turns never exceed messages. + expect(turns.length).toBeLessThanOrEqual(steps.length) + expect(turns.length).toBeGreaterThanOrEqual(1) + } finally { + await ctx.fiber.dispose() + } + }, + ), { numRuns: 25, timeout: 3000 }) }) }) diff --git a/packages/llm/tests/properties.spec.ts b/packages/llm/tests/properties.spec.ts index 60c546aa39..f0118d9dfd 100644 --- a/packages/llm/tests/properties.spec.ts +++ b/packages/llm/tests/properties.spec.ts @@ -39,9 +39,12 @@ const chunkArb: fc.Arbitrary = indexArb.chain(index => fc.oneof( .map((r): StreamChunk => ({ type: 'tool-call-delta', index, id: CallId(r.id), argumentsDelta: r.argumentsDelta })), blockEndArb(index), fc.constant({ type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } }), + fc.constant({ type: 'finish', reason: { kind: 'stop' } }), + fc.constant({ type: 'finish', reason: { kind: 'tool-calls' } }), + fc.string().map((message): StreamChunk => ({ type: 'finish', reason: { kind: 'error', message } })), )) -/** A stream is a list of chunks; we add the terminal `finish` ourselves. */ +/** A stream is an arbitrary list of chunks (we do NOT force a terminal finish). */ const streamArb = fc.array(chunkArb, { maxLength: 30 }) /** Feed a fresh assembler, return it. */ @@ -115,11 +118,33 @@ describe('BlockAssembler properties', () => { })) }) - it('result().finish defaults to stop when no finish chunk arrives', () => { + it('finish reflects the last finish chunk, or defaults to stop when none arrives', () => { fc.assert(fc.property(streamArb, (chunks) => { const a = feed(chunks) - const hasFinish = chunks.some(c => c.type === 'finish') - if (!hasFinish) expect(a.finish).toEqual({ kind: 'stop' }) + const finishes = chunks.filter(c => c.type === 'finish') + if (finishes.length === 0) { + expect(a.finish).toEqual({ kind: 'stop' }) + } else { + // last-write-wins: the assembler keeps the most recent finish reason. + const last = finishes[finishes.length - 1] + if (last?.type === 'finish') expect(a.finish).toEqual(last.reason) + } + })) + }) + + it('streaming and one-shot assembly agree on usage and finish', () => { + fc.assert(fc.property(streamArb, (chunks) => { + // Streaming consumer: push + flush as it goes. + const streaming = new BlockAssembler() + for (const chunk of chunks) { + streaming.push(chunk) + streaming.flushReady() + } + streaming.flushRemaining() + // One-shot consumer: push all, then read. + const oneShot = feed(chunks) + expect(streaming.usage).toEqual(oneShot.usage) + expect(streaming.finish).toEqual(oneShot.finish) })) }) }) diff --git a/packages/session/tests/properties.spec.ts b/packages/session/tests/properties.spec.ts index 2b31d6e83b..5c4e210c52 100644 --- a/packages/session/tests/properties.spec.ts +++ b/packages/session/tests/properties.spec.ts @@ -74,19 +74,24 @@ describe('Session properties', () => { })) }) - it('non-message events never affect derived history', () => { + it('non-message events never affect derived history (any interleaving)', () => { fc.assert(fc.property( fc.array(messageEventArb, { maxLength: 12 }), fc.array(nonMessageEventArb, { maxLength: 12 }), - (messages, noise) => { - // The same message events, with and without interleaved noise, derive - // the same history (noise is inserted at arbitrary positions). + // An arbitrary merge of the two streams that PRESERVES each stream's + // relative order (a random interleaving, not a fixed alternation). + fc.infiniteStream(fc.boolean()), + (messages, noise, pick) => { const clean = build(messages).deriveMessages() const interleaved: Appendable[] = [] - const maxLen = Math.max(messages.length, noise.length) - for (let i = 0; i < maxLen; i++) { - if (i < noise.length) interleaved.push(noise[i]!) - if (i < messages.length) interleaved.push(messages[i]!) + let mi = 0 + let ni = 0 + const picker = pick[Symbol.iterator]() + while (mi < messages.length || ni < noise.length) { + // take from noise when chosen and available, else from messages + const takeNoise = ni < noise.length && (mi >= messages.length || picker.next().value === true) + if (takeNoise) { interleaved.push(noise[ni]!); ni++ } + else { interleaved.push(messages[mi]!); mi++ } } const withNoise = build(interleaved).deriveMessages() expect(withNoise).toEqual(clean) diff --git a/packages/tools/tests/properties.spec.ts b/packages/tools/tests/properties.spec.ts index a5d44baf2d..49cb3cb640 100644 --- a/packages/tools/tests/properties.spec.ts +++ b/packages/tools/tests/properties.spec.ts @@ -47,7 +47,7 @@ function specArb(depth: number): fc.Arbitrary { function valueForProp(prop: SchemaProp): fc.Arbitrary { switch (prop.type) { case 'string': return prop.enum ? fc.constantFrom(...prop.enum) : fc.string() - case 'number': return fc.double({ noNaN: true }) + case 'number': return fc.double({ noNaN: true, noDefaultInfinity: true }) case 'boolean': return fc.boolean() case 'object': return prop.properties ? validArgsForSpec(prop.properties) : fc.constant({}) case 'array': return prop.items ? fc.array(valueForProp(prop.items), { maxLength: 3 }) : fc.constant([])