mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-04 07:33:18 +00:00
fix(muse): retry child history after the parent finishes
Finishing the parent clears its silence timer, so a failed final child read had no future retry. Keep retries with the child projection, share concurrent reads, and cancel pending retries when its session closes. Co-authored-by: kb <kb@wardbitz.com>
This commit is contained in:
@@ -10,10 +10,15 @@ import {
|
||||
type WireItem,
|
||||
} from "./wire.js";
|
||||
|
||||
// Match the existing silence-recovery cadence, including after the parent becomes idle.
|
||||
const READ_RETRY_MS = 2 * 60 * 1000;
|
||||
|
||||
interface Child {
|
||||
timeline: Timeline;
|
||||
cursor: string | undefined;
|
||||
ended: boolean;
|
||||
reading: Promise<void> | undefined;
|
||||
retry: ReturnType<typeof setTimeout> | undefined;
|
||||
}
|
||||
export class Children {
|
||||
private readonly children = new Map<string, Child>();
|
||||
@@ -69,15 +74,30 @@ export class Children {
|
||||
timeline: new Timeline(this.host, id, id, this.emit),
|
||||
cursor: undefined,
|
||||
ended: false,
|
||||
reading: undefined,
|
||||
retry: undefined,
|
||||
});
|
||||
}
|
||||
// The next item update or refresh() reads a failed child again from the same cursor.
|
||||
private async read(id: string): Promise<void> {
|
||||
try {
|
||||
await this.page(id);
|
||||
} catch (error) {
|
||||
this.readFailed(id, error);
|
||||
}
|
||||
// A failed final read must recover even after the parent finishes and stops its silence timer.
|
||||
// Item updates and timer retries share one read so their cursors cannot race.
|
||||
private read(id: string): Promise<void> {
|
||||
const child = this.children.get(id);
|
||||
if (!child) return Promise.resolve();
|
||||
clearTimeout(child.retry);
|
||||
child.retry = undefined;
|
||||
if (child.reading) return child.reading;
|
||||
child.reading = this.page(id)
|
||||
.catch((error: unknown) => {
|
||||
this.readFailed(id, error);
|
||||
if (this.children.get(id) === child)
|
||||
child.retry = setTimeout(() => {
|
||||
void this.read(id);
|
||||
}, READ_RETRY_MS);
|
||||
})
|
||||
.finally(() => {
|
||||
child.reading = undefined;
|
||||
});
|
||||
return child.reading;
|
||||
}
|
||||
private readFailed(id: string, error: unknown): void {
|
||||
process.stderr.write(
|
||||
@@ -132,7 +152,10 @@ export class Children {
|
||||
} while (next !== null);
|
||||
}
|
||||
close(): void {
|
||||
for (const sessionId of this.children.keys()) this.emit({ type: "session.closed", sessionId });
|
||||
for (const [sessionId, child] of this.children) {
|
||||
clearTimeout(child.retry);
|
||||
this.emit({ type: "session.closed", sessionId });
|
||||
}
|
||||
this.children.clear();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -155,6 +155,12 @@ function failedChildRead(frame) {
|
||||
const failure = process.env[childReadFailures[frame.method]];
|
||||
if (!failure || frame.params.sessionId !== "fixture-child") return false;
|
||||
if (failure === "once" && failedChildReads.has(frame.method)) return false;
|
||||
if (failure === "twice") {
|
||||
const key = `child-fail:${frame.method}`;
|
||||
const count = counts.get(key) ?? 0;
|
||||
if (count >= 2) return false;
|
||||
counts.set(key, count + 1);
|
||||
}
|
||||
failedChildReads.add(frame.method);
|
||||
rpcError(frame, -32603, "internal", "view range read failed");
|
||||
return true;
|
||||
|
||||
@@ -1932,3 +1932,48 @@ test("Muse reports belong only to the session's account and remain available glo
|
||||
expect(await registry.listReports({ agentId: "muse-unknown" })).toEqual([]);
|
||||
expect((await registry.listReports()).map((report) => report.id)).toEqual(accountIds);
|
||||
});
|
||||
|
||||
test("a child history read recovers after the parent finishes", async () => {
|
||||
const h = await harness("subagent", {
|
||||
MUSE_TEST_CHILD: "dedicated",
|
||||
MUSE_TEST_CHILD_PAGE_FAIL: "twice",
|
||||
});
|
||||
await h.open();
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
await h.prompt();
|
||||
await h.wait(
|
||||
(e) => e.type === "session.turn" && e.sessionId === "paseo-session" && e.state === "completed",
|
||||
);
|
||||
expect(h.events.filter((e) => e.type === "session.runtime_failed")).toEqual([]);
|
||||
expect(
|
||||
h.events.some(
|
||||
(e) =>
|
||||
e.type === "timeline.item" &&
|
||||
e.sessionId === "fixture-child" &&
|
||||
e.item.type === "assistant_message",
|
||||
),
|
||||
).toBe(false);
|
||||
await vi.advanceTimersByTimeAsync(120000);
|
||||
await h.wait(
|
||||
(e) =>
|
||||
e.type === "timeline.item" &&
|
||||
e.sessionId === "fixture-child" &&
|
||||
e.item.type === "assistant_message",
|
||||
);
|
||||
}, 10000);
|
||||
|
||||
test("closing a session cancels pending child history retries", async () => {
|
||||
const h = await harness("subagent", {
|
||||
MUSE_TEST_CHILD: "dedicated",
|
||||
MUSE_TEST_CHILD_PAGE_FAIL: "always",
|
||||
});
|
||||
await h.open();
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
await h.prompt();
|
||||
await h.wait(
|
||||
(e) => e.type === "session.turn" && e.sessionId === "paseo-session" && e.state === "completed",
|
||||
);
|
||||
await h.send({ type: "session.close", sessionId: "paseo-session" });
|
||||
await h.wait((e) => e.type === "session.closed" && e.sessionId === "paseo-session");
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user