fix(pi): stream assistant deltas when pi omits the cumulative message (#2978)
* fix(pi): stream assistant deltas when pi omits the cumulative message pi 0.84.0 removed the cumulative `message` snapshot from JSON/RPC `message_update` events, so `handleMessageUpdate` dereferenced `undefined` and took down the whole daemon on the first assistant delta. Every prompt to a pi agent crash-looped the worker. `message_update` carries an `assistantMessageEvent`, which pi only emits for assistant messages, so a missing snapshot is treated as an assistant update. `message_start` still carries the message, so the response id keeps flowing through `activeAssistantMessageId`. Fixes #2940 Fixes #2946 Fixes #2950 * fix(pi): give the message_update COMPAT tag a version and removal date Follows docs/protocol-compatibility.md: every shim tag needs a name, a version, and a removal condition or date so `rg "COMPAT\("` stays a schedulable cleanup backlog. --------- Co-authored-by: dundunHa <dundunhade@gmail.com>
This commit is contained in:
@@ -662,6 +662,63 @@ describe("PiRpcAgentSession", () => {
|
|||||||
]);
|
]);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("streams assistant text and reasoning when Pi omits the cumulative message", async () => {
|
||||||
|
const { pi, session, events } = await createSession();
|
||||||
|
const fakeSession = pi.latestSession();
|
||||||
|
|
||||||
|
await session.startTurn("hello");
|
||||||
|
fakeSession.emit({
|
||||||
|
type: "message_start",
|
||||||
|
message: { role: "assistant", content: [], responseId: "response-1" },
|
||||||
|
});
|
||||||
|
fakeSession.emit({
|
||||||
|
type: "message_update",
|
||||||
|
assistantMessageEvent: { type: "text_delta", delta: "hel" },
|
||||||
|
});
|
||||||
|
fakeSession.emit({
|
||||||
|
type: "message_update",
|
||||||
|
assistantMessageEvent: { type: "text_delta", delta: "lo" },
|
||||||
|
});
|
||||||
|
fakeSession.emit({
|
||||||
|
type: "message_update",
|
||||||
|
assistantMessageEvent: { type: "thinking_delta", delta: "thinking" },
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(events.timelineItems()).toEqual([
|
||||||
|
{ type: "assistant_message", text: "hel", messageId: "response-1" },
|
||||||
|
{ type: "assistant_message", text: "lo", messageId: "response-1" },
|
||||||
|
{ type: "reasoning", text: "thinking" },
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("generates one message id when Pi omits both message start and the cumulative message", async () => {
|
||||||
|
const { pi, session, events } = await createSession();
|
||||||
|
const fakeSession = pi.latestSession();
|
||||||
|
|
||||||
|
await session.startTurn("hello");
|
||||||
|
fakeSession.emit({
|
||||||
|
type: "message_update",
|
||||||
|
assistantMessageEvent: { type: "text_delta", delta: "hel" },
|
||||||
|
});
|
||||||
|
fakeSession.emit({
|
||||||
|
type: "message_update",
|
||||||
|
assistantMessageEvent: { type: "text_delta", delta: "lo" },
|
||||||
|
});
|
||||||
|
|
||||||
|
const [firstChunk, secondChunk] = events.timelineItems();
|
||||||
|
expect(firstChunk).toMatchObject({
|
||||||
|
type: "assistant_message",
|
||||||
|
text: "hel",
|
||||||
|
messageId: expect.any(String),
|
||||||
|
});
|
||||||
|
const firstMessageId = (firstChunk as { messageId: string }).messageId;
|
||||||
|
expect(secondChunk).toEqual({
|
||||||
|
type: "assistant_message",
|
||||||
|
text: "lo",
|
||||||
|
messageId: firstMessageId,
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
test("emits live user messages with submitted Pi tree entry ids", async () => {
|
test("emits live user messages with submitted Pi tree entry ids", async () => {
|
||||||
const { pi, session, events } = await createSession();
|
const { pi, session, events } = await createSession();
|
||||||
const fakeSession = pi.latestSession();
|
const fakeSession = pi.latestSession();
|
||||||
|
|||||||
@@ -2191,12 +2191,12 @@ export class PiRpcAgentSession implements AgentSession {
|
|||||||
event: Extract<PiAgentSessionEvent, { type: "message_update" }>,
|
event: Extract<PiAgentSessionEvent, { type: "message_update" }>,
|
||||||
turnId: string | undefined,
|
turnId: string | undefined,
|
||||||
): void {
|
): void {
|
||||||
if (event.message.role !== "assistant") {
|
if (event.message && event.message.role !== "assistant") {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (event.assistantMessageEvent.type === "text_delta") {
|
if (event.assistantMessageEvent.type === "text_delta") {
|
||||||
// Pi-compatible runtimes may emit updates without a preceding message_start.
|
// Pi-compatible runtimes may emit updates without a preceding message_start.
|
||||||
this.activeAssistantMessageId ??= event.message.responseId || randomUUID();
|
this.activeAssistantMessageId ??= event.message?.responseId || randomUUID();
|
||||||
this.emit({
|
this.emit({
|
||||||
type: "timeline",
|
type: "timeline",
|
||||||
provider: this.provider,
|
provider: this.provider,
|
||||||
|
|||||||
@@ -160,7 +160,9 @@ export type PiAgentSessionEvent =
|
|||||||
| { type: "message_end"; message: PiAgentMessage }
|
| { type: "message_end"; message: PiAgentMessage }
|
||||||
| {
|
| {
|
||||||
type: "message_update";
|
type: "message_update";
|
||||||
message: PiAgentMessage;
|
// COMPAT(piCumulativeMessageUpdate): added in v0.3.0, remove after 2027-02-07 once the pi
|
||||||
|
// floor is >=0.84. pi <=0.83 repeats the cumulative assistant message on every update.
|
||||||
|
message?: PiAgentMessage;
|
||||||
assistantMessageEvent: PiAssistantMessageEvent;
|
assistantMessageEvent: PiAssistantMessageEvent;
|
||||||
}
|
}
|
||||||
| {
|
| {
|
||||||
|
|||||||
Reference in New Issue
Block a user