fix(muse): restore earlier messages after daemon restart (#6159)

MSP view cursors request only a suffix. A fresh provider session rebuilds an empty timeline, so it must resume without a cursor to recover earlier messages. Keep gap recovery cursors in memory and omit them from durable session state.
This commit is contained in:
Mohamed Boudra
2026-10-05 17:47:04 +01:00
committed by GitHub
parent 27a95a5dec
commit bc6ddbfa97
4 changed files with 86 additions and 24 deletions
+3 -8
View File
@@ -67,7 +67,6 @@ export class Session {
private readonly host: MspConnection;
private liveState: Live | undefined;
private providerId: string | undefined;
private cursor: string | undefined;
private config: ProviderSessionConfig;
private catalog: ProviderCatalog = { models: [], modes };
private readonly approvals = new Map<string, WireApproval>();
@@ -110,9 +109,11 @@ export class Session {
if (input.persistence.version !== 1)
throw new MuseError("invalidPersistence", "Unsupported Muse persistence version");
const saved = persistenceSchema.parse(input.persistence.data);
// A fresh timeline needs history. MSP cursor resumes return only the suffix,
// so even legacy persistence cursors cannot restore this session's projection.
response = await this.host.command(
"session/resume",
{ sessionId: saved.sessionId, cursor: saved.cursor, config: { mcpServers } },
{ sessionId: saved.sessionId, config: { mcpServers } },
sessionSchema,
);
this.config = {
@@ -168,7 +169,6 @@ export class Session {
),
};
this.providerId = response.session.providerId ?? undefined;
this.cursor = response.viewCursor;
this.config = {
...this.config,
model: response.session.modelId ?? this.config.model,
@@ -411,10 +411,6 @@ export class Session {
}
await this.processNotification(notification);
this.live().recovery.activity(envelope.viewCursor, this.activeTurnIds().length > 0);
if (envelope.viewCursor) {
this.cursor = envelope.viewCursor;
this.publishPersistence();
}
}
private async processNotification(notification: Notification): Promise<void> {
switch (notification.method) {
@@ -597,7 +593,6 @@ export class Session {
version: 1,
data: {
sessionId: this.live().nativeId,
...(this.cursor ? { cursor: this.cursor } : {}),
...(this.config.model ? { model: this.config.model } : {}),
...(this.config.thinkingOption ? { thinkingOption: this.config.thinkingOption } : {}),
},
-1
View File
@@ -143,7 +143,6 @@ export const sessionSchema = z.object({
});
export const persistenceSchema = z.object({
sessionId: z.string(),
cursor: z.string().optional(),
model: z.string().optional(),
thinkingOption: effortSchema.optional(),
});
+19
View File
@@ -80,6 +80,22 @@ function receiveFrame(line) {
} else for (const row of messages) emitFixtureMessage(row.msg, recorded, frame);
}
function controlResponse(frame) {
if (frame.method === "session/resume" && process.env.MUSE_TEST_HISTORY) {
const result = responseFor(readFixture("resume-without-cursor"), "session/resume");
result.session.sessionId = frame.params.sessionId;
result.history = frame.params.cursor
? { mode: "none", noneReason: "cursorSuffix", items: null, snapshot: null }
: {
mode: "inline",
items: readFileSync(process.env.MUSE_TEST_HISTORY, "utf8")
.trim()
.split("\n")
.map((line) => JSON.parse(line)),
snapshot: null,
};
respond(frame, result);
return true;
}
if (frame.method === "initialized") return true;
if (frame.method === "initialize" && process.env.MUSE_TEST_EXIT) {
process.stderr.write("diagnostic tail, not protocol\n");
@@ -438,6 +454,9 @@ function replace(value) {
return value;
}
function send(frame) {
if (frame.method === "item/completed" && process.env.MUSE_TEST_HISTORY) {
appendFileSync(process.env.MUSE_TEST_HISTORY, JSON.stringify(frame.params.item) + "\n");
}
process.stdout.write(JSON.stringify(frame) + "\n");
}
function respond(frame, result) {
+64 -15
View File
@@ -221,9 +221,9 @@ test("text and reasoning stream, completion overwrites deltas, user echo retains
{ type: "text", text: "hello" },
],
});
expect(h.events.filter((event) => event.type === "session.persistence").length).toBeGreaterThan(
1,
);
expect(h.events.findLast((event) => event.type === "session.persistence")).toMatchObject({
persistence: { version: 1, data: { model: "meta/muse-spark-1.3", thinkingOption: "high" } },
});
});
test("tool calls classify read, shell, and edit with fetched JSON patch converted to unified diff", async () => {
const h = await harness("tools-edit");
@@ -327,23 +327,69 @@ test("interrupt awaits cancelled terminal and reconciles the tool", async () =>
),
).toBeGreaterThan(terminalIndex);
});
test("a fresh connection restores earlier messages from persisted session state", async () => {
const original = await harness();
const historyPath = path.join(original.root, "native-history.ndjson");
original.launch.env.MUSE_TEST_HISTORY = historyPath;
await original.open();
await original.prompt();
await original.wait((event) => event.type === "session.turn" && event.state === "completed");
const messages = [
...new Map(
original.events
.filter((event) => event.type === "timeline.item")
.map((event) => event.item)
.filter((item) => item.type === "user_message" || item.type === "assistant_message")
.map((item) => [item.id, item] as const),
).values(),
];
expect(messages.map((item) => item.type)).toEqual(["user_message", "assistant_message"]);
const saved = original.events.findLast((event) => event.type === "session.persistence");
if (!saved) throw new Error("Expected session persistence");
const persistencePath = path.join(original.root, "persistence.json");
// Older daemon records carry a view cursor even though their timeline is not durable.
await writeFile(
persistencePath,
JSON.stringify({
...saved.persistence,
data: { ...saved.persistence.data, cursor: saved.persistence.data.cursor ?? "v:old:7" },
}),
);
await original.connection.close();
const reopened = await harness("text-reasoning", { MUSE_TEST_HISTORY: historyPath });
expect(await reopened.open(JSON.parse(await readFile(persistencePath, "utf8")))).toMatchObject({
type: "session.ready",
});
const restored = reopened.events
.filter((event) => event.type === "timeline.item")
.map((event) => event.item)
.filter((item) => item.type === "user_message" || item.type === "assistant_message");
expect(restored.map((item) => ({ type: item.type, text: item.text }))).toEqual(
messages.map((item) => ({ type: item.type, text: item.text })),
);
const reopenedState = reopened.events.findLast((event) => event.type === "session.persistence");
expect(reopenedState?.persistence.data).not.toHaveProperty("cursor");
expect(reopened.events.filter((event) => event.type === "session.turn")).toEqual([]);
await reopened.prompt();
await reopened.wait((event) => event.type === "session.turn" && event.state === "completed");
});
for (const cursor of [undefined, "v:old:7"]) {
test(`resume ${cursor ? "with cursor emits suffix and ignores old terminal" : "without cursor replays inline history"}`, async () => {
const h = await harness(cursor ? "resume-with-cursor" : "resume-without-cursor");
test(`resume ${cursor ? "with a legacy persistence cursor" : "without a persistence cursor"} replays inline history`, async () => {
const h = await harness("resume-without-cursor");
await h.open({
version: 1,
data: { sessionId: "saved-session", ...(cursor ? { cursor } : {}) },
});
expect(h.events.filter((event) => event.type === "session.turn")).toEqual([]);
const history = h.events.filter((event) => event.type === "timeline.item");
if (!cursor)
expect(history).toEqual(
expect.arrayContaining([
expect.objectContaining({
item: expect.objectContaining({ type: "assistant_message", text: "RESUME_MARKER_OK" }),
}),
]),
);
expect(history).toEqual(
expect.arrayContaining([
expect.objectContaining({
item: expect.objectContaining({ type: "assistant_message", text: "RESUME_MARKER_OK" }),
}),
]),
);
await h.prompt();
await h.wait((event) => event.type === "session.turn" && event.state === "completed");
expect(
@@ -351,7 +397,10 @@ for (const cursor of [undefined, "v:old:7"]) {
).toHaveLength(1);
expect(
(await h.recorded()).find((frame) => frame.method === "session/resume").params,
).toMatchObject({ sessionId: "saved-session", ...(cursor ? { cursor } : {}) });
).toMatchObject({ sessionId: "saved-session" });
expect(
(await h.recorded()).find((frame) => frame.method === "session/resume").params,
).not.toHaveProperty("cursor");
});
}
test("steer sends ifBusy steer and joins the running turn", async () => {
@@ -862,7 +911,7 @@ test("fingerprint mismatch is logged but does not prevent opening", async () =>
});
for (const restoring of [false, true]) {
test(`MCP servers map to MSP stdio and streamableHttp on ${restoring ? "resume" : "start"}`, async () => {
const h = await harness(restoring ? "resume-with-cursor" : "text-reasoning");
const h = await harness(restoring ? "resume-without-cursor" : "text-reasoning");
await h.send({
type: "session.open",
sessionId: "paseo-session",