fix(harness-pi): stream between-turn compaction on the next turn - #24407

Background

Pi compaction requested between turns runs successfully but its observation is buffered indefinitely. The next turn never delivers the summary or token count to the application.

Summary

Drain pending compaction observations after the next turn's stream-start, before the model emits new content. Regression coverage checks one or multiple observations, preserves their order/summary/token count, and verifies they are delivered only once. Updated on current main and signed the rebased contributor commit while retaining its original author. The regression mock now returns the current Pi compaction result type and explicitly declares a new session. Opus found no blockers; added its suggested stream-order assertion.

Validation: all 360 Pi harness tests pass. pnpm type-check:full and pnpm check pass. Includes a patch changeset for @ai-sdk/harness-pi.

Contributor Credit

@dannote reported #24359 with a real Pi/Codex reproduction and root cause analysis. @pei711 authored the fix and original regression test. @gr2m added delivery-once coverage and the public application verification below.

End-to-End Verification

Ran the same temporary application using the public HarnessAgent, createPi, and local just-bash sandbox. It executes the actual installed Pi runtime, creates conversation history, calls session.compact(), and consumes two subsequent fullStreams. A loopback HTTP server supplies deterministic OpenAI-compatible model responses, including the compaction summarization request. There are no mocked Pi modules.

Before (5c1f2c6027, whose Pi implementation matches main): seven local model requests complete, including compaction, but neither subsequent stream contains a compaction observation.

After: the first subsequent stream includes tool-call:compaction and tool-result:compaction with { trigger: 'manual', summary: 'The cat is called Brioche.', tokensBefore: 40008 }, followed by the normal model response. The following stream has no duplicate compaction.

Prerequisites: repository dependencies, supported Node and pnpm. Save the complete script below outside the checkout as /tmp/ai-pi-compaction/repro.mjs. From each checkout run:

mkdir -p /tmp/ai-pi-compaction
pnpm exec turbo build --filter=@ai-sdk/harness-pi... --filter=@ai-sdk/sandbox-just-bash...
ln -sfn "$PWD/examples/ai-functions/node_modules" /tmp/ai-pi-compaction/node_modules
node /tmp/ai-pi-compaction/repro.mjs

Both runs resolve the checkout's built workspace packages. No reproduction-only files are committed.

Verification limit: this is a local application E2E flow through the real Pi runtime and sandbox, with synthetic model responses. No live model or Pi subscription credentials are configured, so the reporter's cloud-model reproduction could not be rerun. This does not validate remote model behavior or semantic memory retention.

import { createServer } from 'node:http';
import { once } from 'node:events';
import { createPi } from '@ai-sdk/harness-pi';
import { HarnessAgent } from '@ai-sdk/harness/agent';
import { createJustBashNetworkSandboxSession } from '@ai-sdk/sandbox-just-bash';

let requests = 0;
const server = createServer(async (request, response) => {
  let body = '';
  for await (const chunk of request) body += chunk;
  const input = JSON.parse(body);
  requests++;
  console.log('local model request:', requests, request.url, 'messages:', input.messages.length);
  const chunk = { id: `local-${requests}`, object: 'chat.completion.chunk', created: 1, model: 'local', choices: [{ index: 0, delta: { role: 'assistant', content: 'The cat is called Brioche.' }, finish_reason: null }] };
  response.writeHead(200, { 'content-type': 'text/event-stream' });
  response.write(`data: ${JSON.stringify(chunk)}\n\n`);
  response.end(`data: ${JSON.stringify({ ...chunk, choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], usage: { prompt_tokens: 40000, completion_tokens: 8, total_tokens: 40008 } })}\n\ndata: [DONE]\n\n`);
});
server.listen(0, '127.0.0.1');
await once(server, 'listening');
const baseUrl = `http://127.0.0.1:${server.address().port}/v1`;
const agent = new HarnessAgent({
  harness: createPi({
    auth: { LOCAL_API_KEY: 'dummy', LOCAL_BASE_URL: baseUrl },
    providers: { local: { api: 'openai-completions', baseUrl, apiKey: 'dummy', models: [{ id: 'local', name: 'Local replay', reasoning: false, input: ['text'], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 128000, maxTokens: 4096 }] } },
  }),
  model: 'local/local',
  instructions: 'Reply briefly.',
});
const sandboxSession = await createJustBashNetworkSandboxSession();
let session;
try {
  session = await agent.createSession({ sessionId: 'local-compaction-replay', sandboxSession });
  async function turn(prompt) {
    const result = await agent.stream({ session, prompt });
    const parts = [];
    for await (const part of result.fullStream) {
      if (part.type === 'tool-call' || part.type === 'tool-result') parts.push({ type: part.type, toolName: part.toolName, input: part.input, output: part.output });
      else if (!part.type.endsWith('-delta')) parts.push({ type: part.type });
    }
    return parts;
  }
  await turn('My cat is called Brioche. Reply OK.');
  const notes = Array.from({ length: 2400 }, (_, i) => `Note ${i}: design spacing step ${i % 13}.`).join(' ');
  for (let i = 0; i < 3; i++) await turn(`Read these notes and reply DONE.\n${notes}`);
  await session.compact();
  console.log('next turn:', JSON.stringify(await turn('What is my cat called?')));
  console.log('following turn:', JSON.stringify(await turn('Repeat the name.')));
} finally {
  await session?.destroy();
  await sandboxSession.destroy();
  server.close();
  server.closeAllConnections();
}

Checklist

• All commits are signed (PRs with unsigned commits cannot be merged)
• Tests have been added / updated (for bug fixes / features)
• Documentation has been added / updated (for bug fixes / features)
• A patch changeset for relevant packages has been added (for bug fixes / features - run pnpm changeset in the project root)
• I have reviewed this pull request (self-review)

Future Work

This fixes queued observations on the next turn of a live session. Persisting undelivered observations across stop/detach and ordering compaction that completes concurrently with turn setup are existing limitations outside this change.

Related Issues

Fixes #24359.