@@ -0,0 +1,249 @@
1+import { describe, expect, it, vi } from "vitest";
2+import { __testing, type EmbeddedPiActiveSessionSteerTarget } from "./attempt.js";
3+4+describe("embedded Pi queued steering cancellation", () => {
5+it("waits for the queued user message_end transcript boundary", async () => {
6+let emit!: (event: unknown) => void;
7+const activeSession: EmbeddedPiActiveSessionSteerTarget = {
8+getSteeringMessages: () => [],
9+steer: async () => {},
10+subscribe: (listener) => {
11+emit = listener;
12+return () => {};
13+},
14+};
15+const wait = __testing.steerAndWaitForTranscriptCommit(
16+activeSession,
17+"queued completion",
18+10_000,
19+);
20+let settled = false;
21+void wait.then(() => {
22+settled = true;
23+});
24+25+emit({
26+type: "message_start",
27+message: {
28+role: "user",
29+content: [{ type: "text", text: "queued completion" }],
30+},
31+});
32+await Promise.resolve();
33+expect(settled).toBe(false);
34+35+emit({
36+type: "message_end",
37+message: {
38+role: "user",
39+content: [{ type: "text", text: "queued completion" }],
40+},
41+});
42+43+await expect(wait).resolves.toBeUndefined();
44+expect(settled).toBe(true);
45+});
46+47+it("removes only the timed-out steering message and preserves unrelated payloads", async () => {
48+const unrelatedImage = {
49+type: "image",
50+source: { type: "base64", data: "abc", media_type: "image/png" },
51+};
52+const unrelatedMessage = {
53+role: "user",
54+content: [{ type: "text", text: "keep this rich payload" }, unrelatedImage],
55+timestamp: 1,
56+};
57+const targetMessage = {
58+role: "user",
59+content: [{ type: "text", text: "timed-out completion announce" }],
60+timestamp: 2,
61+};
62+const trailingMessage = {
63+role: "custom",
64+customType: "notice",
65+content: "preserve custom queued message",
66+timestamp: 3,
67+};
68+const steeringUiMessages = ["keep this rich payload", "timed-out completion announce"];
69+const queueMessages = [unrelatedMessage, targetMessage, trailingMessage];
70+const activeSession: EmbeddedPiActiveSessionSteerTarget = {
71+agent: {
72+steeringQueue: {
73+messages: queueMessages,
74+},
75+},
76+getSteeringMessages: () => steeringUiMessages,
77+steer: async () => {},
78+subscribe: () => () => {},
79+};
80+81+await expect(
82+__testing.cancelQueuedSteeringMessage(activeSession, "timed-out completion announce"),
83+).resolves.toBe(true);
84+85+expect(queueMessages).toEqual([unrelatedMessage, trailingMessage]);
86+expect(queueMessages[0]).toBe(unrelatedMessage);
87+expect(queueMessages[0]?.content[1]).toBe(unrelatedImage);
88+expect(queueMessages[1]).toBe(trailingMessage);
89+expect(steeringUiMessages).toEqual(["keep this rich payload"]);
90+});
91+92+it("rejects and removes the queued steering message when the session ends first", async () => {
93+vi.useFakeTimers();
94+let emit!: (event: unknown) => void;
95+const targetMessage = {
96+role: "user",
97+content: [{ type: "text", text: "completion after parent stopped" }],
98+timestamp: 2,
99+};
100+const keepMessage = {
101+role: "user",
102+content: [{ type: "text", text: "keep unrelated queue entry" }],
103+timestamp: 3,
104+};
105+const steeringUiMessages = ["completion after parent stopped", "keep unrelated queue entry"];
106+const queueMessages = [targetMessage, keepMessage];
107+let unsubscribed = false;
108+const activeSession: EmbeddedPiActiveSessionSteerTarget = {
109+agent: {
110+steeringQueue: {
111+messages: queueMessages,
112+},
113+},
114+getSteeringMessages: () => steeringUiMessages,
115+steer: async () => {},
116+subscribe: (listener) => {
117+emit = listener;
118+return () => {
119+unsubscribed = true;
120+};
121+},
122+};
123+124+const wait = __testing.steerAndWaitForTranscriptCommit(
125+activeSession,
126+"completion after parent stopped",
127+10_000,
128+);
129+const rejection = expect(wait).rejects.toThrow(
130+"active session ended before queued steering message was committed to the transcript",
131+);
132+133+emit({ type: "agent_end", messages: [] });
134+await vi.advanceTimersByTimeAsync(0);
135+136+try {
137+await rejection;
138+expect(queueMessages).toEqual([keepMessage]);
139+expect(steeringUiMessages).toEqual(["keep unrelated queue entry"]);
140+expect(unsubscribed).toBe(true);
141+} finally {
142+vi.useRealTimers();
143+}
144+});
145+146+it("keeps queued steering pending when Pi auto-retry starts after agent_end", async () => {
147+vi.useFakeTimers();
148+try {
149+let emit!: (event: unknown) => void;
150+const targetMessage = {
151+role: "user",
152+content: [{ type: "text", text: "completion survives retry" }],
153+timestamp: 2,
154+};
155+const steeringUiMessages = ["completion survives retry"];
156+const queueMessages = [targetMessage];
157+const activeSession: EmbeddedPiActiveSessionSteerTarget = {
158+agent: {
159+steeringQueue: {
160+messages: queueMessages,
161+},
162+},
163+getSteeringMessages: () => steeringUiMessages,
164+steer: async () => {},
165+subscribe: (listener) => {
166+emit = listener;
167+return () => {};
168+},
169+};
170+171+const wait = __testing.steerAndWaitForTranscriptCommit(
172+activeSession,
173+"completion survives retry",
174+10_000,
175+);
176+177+emit({ type: "agent_end", messages: [] });
178+emit({ type: "auto_retry_start", attempt: 1, maxAttempts: 3, delayMs: 1_000 });
179+await vi.advanceTimersByTimeAsync(0);
180+181+expect(queueMessages).toEqual([targetMessage]);
182+expect(steeringUiMessages).toEqual(["completion survives retry"]);
183+184+emit({
185+type: "message_end",
186+message: {
187+role: "user",
188+content: [{ type: "text", text: "completion survives retry" }],
189+},
190+});
191+192+await expect(wait).resolves.toBeUndefined();
193+} finally {
194+vi.useRealTimers();
195+}
196+});
197+198+it("keeps queued steering pending when Pi auto-compaction starts after agent_end", async () => {
199+vi.useFakeTimers();
200+try {
201+let emit!: (event: unknown) => void;
202+const targetMessage = {
203+role: "user",
204+content: [{ type: "text", text: "completion survives compaction" }],
205+timestamp: 2,
206+};
207+const steeringUiMessages = ["completion survives compaction"];
208+const queueMessages = [targetMessage];
209+const activeSession: EmbeddedPiActiveSessionSteerTarget = {
210+agent: {
211+steeringQueue: {
212+messages: queueMessages,
213+},
214+},
215+getSteeringMessages: () => steeringUiMessages,
216+steer: async () => {},
217+subscribe: (listener) => {
218+emit = listener;
219+return () => {};
220+},
221+};
222+223+const wait = __testing.steerAndWaitForTranscriptCommit(
224+activeSession,
225+"completion survives compaction",
226+10_000,
227+);
228+229+emit({ type: "agent_end", messages: [] });
230+emit({ type: "compaction_start", reason: "threshold" });
231+await vi.advanceTimersByTimeAsync(0);
232+233+expect(queueMessages).toEqual([targetMessage]);
234+expect(steeringUiMessages).toEqual(["completion survives compaction"]);
235+236+emit({
237+type: "message_end",
238+message: {
239+role: "user",
240+content: [{ type: "text", text: "completion survives compaction" }],
241+},
242+});
243+244+await expect(wait).resolves.toBeUndefined();
245+} finally {
246+vi.useRealTimers();
247+}
248+});
249+});
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。