puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-03 23:23:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(pi): distinguish foreground prompts and label image steers

Browse Source
This commit is contained in:
YEONGUK LEE
2026-11-03 21:37:39 +00:00
committed by paseo-bot[bot]
parent 19247ad6c7
commit 860c55e59e
2 changed files with 132 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/providers/pi/agent.test.ts
+101
View File
@@ -2490,6 +2490,107 @@ describe("PiRpcAgentSession", () => {
});
describe("PiRpcAgentSession steering", () => {
test("does not consume a queued steer when the outstanding foreground prompt has the same text", async () => {
const { pi, session, events } = await createSession();
const { turnId } = await session.startTurn("continue", {
clientMessageId: "client-prompt-1",
});
await session.steerActiveTurn("continue", {
expectedTurnId: turnId,
clientMessageId: "client-steer-1",
});
pi.latestSession().emit({ type: "agent_start" });
pi.latestSession().emit({ type: "turn_start" });
pi.latestSession().finishSubmittedUserMessage({
id: "native-prompt-1",
parentId: null,
text: "continue",
});
await session.interrupt();
expect(events.timelineItems()).toEqual([
{
type: "user_message",
text: "continue",
messageId: "native-prompt-1",
clientMessageId: "client-prompt-1",
},
{
type: "notification",
level: "warning",
message: "Not delivered (stopped): continue",
},
]);
});
test("keeps a consumed steer distinct from an identical foreground prompt", async () => {
const { pi, session, events } = await createSession();
const { turnId } = await session.startTurn("continue", { clientMessageId: "client-prompt-1" });
await session.steerActiveTurn("continue", {
expectedTurnId: turnId,
clientMessageId: "client-steer-1",
});
const native = pi.latestSession();
native.finishSubmittedUserMessage({ id: "native-prompt-1", parentId: null, text: "continue" });
native.finishSubmittedUserMessage({
id: "native-steer-1",
parentId: "native-prompt-1",
text: "continue",
});
await session.interrupt();
expect(events.timelineItems()).toEqual([
{
type: "user_message",
text: "continue",
messageId: "native-prompt-1",
clientMessageId: "client-prompt-1",
},
{
type: "user_message",
text: "continue",
messageId: "native-steer-1",
clientMessageId: "client-steer-1",
},
]);
});
test.each([1, 2])(
"labels a stopped image-only steer containing %i image(s)",
async (imageCount) => {
const { pi, session, events } = await createSession();
const native = pi.latestSession();
native.setModelResult = {
provider: "openai",
id: "gpt-4o",
name: "GPT-4o",
input: ["text", "image"],
};
await session.setModel("openai/gpt-4o");
const { turnId } = await session.startTurn([
{ type: "image", data: ONE_BY_ONE_PNG_BASE64, mimeType: "image/png" },
]);
await session.steerActiveTurn(
Array.from({ length: imageCount }, () => ({
type: "image" as const,
data: ONE_BY_ONE_PNG_BASE64,
mimeType: "image/png",
})),
{ expectedTurnId: turnId, clientMessageId: "client-steer-1" },
);
native.emit({ type: "message_end", message: { role: "user", content: [] } });
await session.interrupt();
expect(native.steerCalls).toEqual([{ message: "", imageCount }]);
expect(events.timelineItems()).toEqual([
{
type: "notification",
level: "warning",
message: `Not delivered (stopped): ${imageCount} ${imageCount === 1 ? "image" : "images"}`,
},
]);
},
);
test("reports an unconsumed steer as not delivered when Stop clears the queue", async () => {
const { session, events } = await createSession();
const { turnId } = await session.startTurn("work");
packages/server/src/server/agent/providers/pi/agent.ts
+31 -1
View File
@@ -232,10 +232,16 @@ interface StartTurnResult {
interface PiPendingSteerSubmission {
text: string;
label: string;
clientMessageId: string | null;
consumed: boolean;
}
interface PiPendingForegroundSubmission {
text: string;
consumed: boolean;
}
// COMPAT(piSteerFallback): added in v0.5.0, remove after 2027-03-01 once the pi floor
// supports the steer RPC. Older binaries answer with "Unknown command: steer".
function isPiDefinitiveSteerRejection(error: unknown): boolean {
@@ -1211,6 +1217,7 @@ export class PiRpcAgentSession implements AgentSession {
private activePromptRequestId: string | null = null;
private readonly pendingPromptResults = new Map<string, boolean>();
private readonly pendingSteerSubmissions: PiPendingSteerSubmission[] = [];
private pendingForegroundSubmission: PiPendingForegroundSubmission | null = null;
currentLeafOverrideId: string | null | undefined;
private readonly treeUserEntriesById = new Map<string, PiCapturedEntry>();
private readonly pendingExtensionResults = new Map<string, PendingExtensionResult>();
@@ -1320,6 +1327,7 @@ export class PiRpcAgentSession implements AgentSession {
this.pendingSettledMessages = null;
this.activePromptRequestId = null;
this.pendingSteerSubmissions.length = 0;
this.pendingForegroundSubmission = { text: payload.text, consumed: false };
this.clearNoTurnBuffers();
const payload = await convertPromptInput(prompt, { model: this.state.model });
if (this.closed) throw new Error("Pi session is closed");
@@ -1366,6 +1374,7 @@ export class PiRpcAgentSession implements AgentSession {
this.pendingSettledMessages = null;
this.activeAssistantMessageId = null;
this.pendingSteerSubmissions.length = 0;
this.pendingForegroundSubmission = null;
this.clearNoTurnBuffers();
if (isPiRequestAbortError(error)) {
this.emit({
@@ -1404,8 +1413,14 @@ export class PiRpcAgentSession implements AgentSession {
return { status: "unavailable" };
}
const imageCount = payload.images?.length ?? 0;
const label =
payload.text.trim().length === 0 && imageCount > 0
? `${imageCount} ${imageCount === 1 ? "image" : "images"}`
: payload.text;
const submission: PiPendingSteerSubmission = {
text: payload.text,
label,
clientMessageId: options.clientMessageId ?? null,
consumed: false,
};
@@ -1444,6 +1459,11 @@ export class PiRpcAgentSession implements AgentSession {
}
private takePendingSteerSubmission(text: string): PiPendingSteerSubmission | undefined {
// The original prompt is also echoed, possibly after a matching steer is queued.
if (this.pendingForegroundSubmission?.text === text) {
this.pendingForegroundSubmission = null;
return undefined;
}
const index = this.pendingSteerSubmissions.findIndex((submission) => submission.text === text);
if (index < 0) {
return undefined;
@@ -1561,7 +1581,7 @@ export class PiRpcAgentSession implements AgentSession {
item: {
type: "notification",
level: "warning",
message: `Not delivered (stopped): ${submission.text}`,
message: `Not delivered (stopped): ${submission.label}`,
},
});
}
@@ -1587,6 +1607,7 @@ export class PiRpcAgentSession implements AgentSession {
this.pendingSettledMessages = null;
this.activeAssistantMessageId = null;
this.pendingSteerSubmissions.length = 0;
this.pendingForegroundSubmission = null;
this.clearNoTurnBuffers();
this.emit({
type: "turn_failed",
@@ -1611,6 +1632,7 @@ export class PiRpcAgentSession implements AgentSession {
this.pendingSettledMessages = null;
this.activeAssistantMessageId = null;
this.pendingSteerSubmissions.length = 0;
this.pendingForegroundSubmission = null;
this.clearNoTurnBuffers();
this.emit({
type: "turn_canceled",
@@ -2226,6 +2248,7 @@ export class PiRpcAgentSession implements AgentSession {
this.activeTurnStartedEmitted = false;
this.pendingSettledMessages = null;
this.pendingSteerSubmissions.length = 0;
this.pendingForegroundSubmission = null;
this.clearNoTurnBuffers();
this.emit({
type: "turn_failed",
@@ -2507,6 +2530,12 @@ export class PiRpcAgentSession implements AgentSession {
if (event.message.role === "user") {
// Entry IDs arrive through the extension later; consumption is already known.
const text = getUserMessageText(event.message.content);
const foreground = this.pendingForegroundSubmission;
// prompt() runs asynchronously, so its first user event may follow steer().
if (foreground && !foreground.consumed && foreground.text === text) {
foreground.consumed = true;
return;
}
const submission = this.pendingSteerSubmissions.find(
(pending) => !pending.consumed && pending.text === text,
);
@@ -2595,6 +2624,7 @@ export class PiRpcAgentSession implements AgentSession {
this.activeTurnStartedEmitted = false;
this.pendingSettledMessages = null;
this.pendingSteerSubmissions.length = 0;
this.pendingForegroundSubmission = null;
this.clearNoTurnBuffers();
if (typeof errorMessage === "string" && errorMessage.length > 0) {
this.usagePoller.stopTurn();