From 14a968e2f1d5fb96e2f5ba73ef1dafd5305ce0d2 Mon Sep 17 00:00:00 2001 From: Mark Goldenstein Date: Wed, 29 Jul 2026 18:31:40 -0700 Subject: [PATCH] Fix completion wakes for resumed sessions --- src/completion-summary-coordinator.ts | 12 +++++- src/session-store.ts | 8 ++++ tests/session-notifications.test.ts | 56 ++++++++++++++++++++++----- tests/session-store.test.ts | 44 +++++++++++++++++++++ 4 files changed, 110 insertions(+), 10 deletions(-) diff --git a/src/completion-summary-coordinator.ts b/src/completion-summary-coordinator.ts index 0c08fe75..512cca06 100644 --- a/src/completion-summary-coordinator.ts +++ b/src/completion-summary-coordinator.ts @@ -25,6 +25,8 @@ export type CompletionSummarySession = Pick< name?: string; goalTaskId?: string; sessionId?: string; + startedAt?: number; + createdAt?: number; }; export type PersistedCompletionSummarySession = Pick< @@ -38,8 +40,10 @@ export type PersistedCompletionSummarySession = Pick< | "originThreadId" | "originSessionKey" | "goalTaskId" + | "createdAt" > & { id?: string; + startedAt?: number; }; export interface CompletionSummaryDecision { @@ -385,6 +389,12 @@ export class CompletionSummaryCoordinator { aliasKind: "visible-summary" | "goal-summary", ): string[] { const scope = this.buildVisibleScope(deliveryRef, session); + const lifecycleStartedAt = "startedAt" in session + ? session.startedAt + : session.createdAt; + const lifecycleScope = Number.isFinite(lifecycleStartedAt) + ? `${scope}:cycle:${this.digest(String(lifecycleStartedAt))}` + : scope; const refs = [ "id" in session ? session.id : undefined, "sessionId" in session ? session.sessionId : undefined, @@ -396,7 +406,7 @@ export class CompletionSummaryCoordinator { .map((ref) => ref?.trim()) .filter((ref): ref is string => Boolean(ref)); return [...new Set(refs)] - .map((ref) => `${scope}:session:${this.digest(ref)}:${aliasKind}`); + .map((ref) => `${lifecycleScope}:session:${this.digest(ref)}:${aliasKind}`); } private getDeliveryRef(session: CompletionSummarySession | PersistedCompletionSummarySession): string { diff --git a/src/session-store.ts b/src/session-store.ts index 1c07468d..9af11746 100644 --- a/src/session-store.ts +++ b/src/session-store.ts @@ -253,6 +253,7 @@ export class SessionStore { taskFlowMirror: session.taskFlowMirror, deliveryState: session.deliveryState, notificationDedupe: this.getExistingNotificationDedupe(session), + completionSummaryDedupe: this.getExistingCompletionSummaryDedupe(session), costUsd: 0, originAgentId: routing.originAgentId, originChannel: routing.originChannel, @@ -364,6 +365,7 @@ export class SessionStore { taskFlowMirror: session.taskFlowMirror, deliveryState: session.deliveryState, notificationDedupe: this.getExistingNotificationDedupe(session), + completionSummaryDedupe: this.getExistingCompletionSummaryDedupe(session), killReason: session.killReason, costUsd: session.costUsd, originAgentId: routing.originAgentId, @@ -419,6 +421,12 @@ export class SessionStore { ?? (getBackendConversationId(session) ? this.getPersistedSession(getBackendConversationId(session)!)?.notificationDedupe : undefined); } + private getExistingCompletionSummaryDedupe(session: Session): PersistedSessionInfo["completionSummaryDedupe"] { + return this.getPersistedSession(session.id)?.completionSummaryDedupe + ?? (session.harnessSessionId ? this.getPersistedSession(session.harnessSessionId)?.completionSummaryDedupe : undefined) + ?? (getBackendConversationId(session) ? this.getPersistedSession(getBackendConversationId(session)!)?.completionSummaryDedupe : undefined); + } + /** Return newest persisted entry for a user-facing name, handling name collisions. */ getLatestPersistedByName(name: string): PersistedSessionInfo | undefined { return this.queries.getLatestPersistedByName(name); diff --git a/tests/session-notifications.test.ts b/tests/session-notifications.test.ts index f3f3f19b..6a5b5957 100644 --- a/tests/session-notifications.test.ts +++ b/tests/session-notifications.test.ts @@ -265,7 +265,7 @@ describe("SessionNotificationService", () => { assert.equal(persisted.notificationDedupe?.every((record: { status?: string }) => record.status === "delivered"), true); }); - it("dedupes one terminal completion notification but allows a later terminal cycle for the same session id", () => { + it("delivers one canonical completion and wake for each of two sequential resumed cycles on the same session id", () => { const persisted = { notificationDedupe: undefined } as any; const requests: Array> = []; const fakeDispatcher = { @@ -283,8 +283,9 @@ describe("SessionNotificationService", () => { (_ref, patch) => Object.assign(persisted, patch), { getPersistedSession: () => persisted }, ); - const session = { + const firstSession = { id: "stable-terminal-session", + startedAt: 1780000001000, route: { provider: "telegram", target: "topic", threadId: "13832", sessionKey: "session:terminal" }, } as any; const firstCycleRequest = { @@ -299,7 +300,10 @@ describe("SessionNotificationService", () => { }, completionWakeSummaryRequired: true, completionWakeOutcomeKey: "terminal:stable-terminal-session:completed:1780000001000:backend-thread:4:unknown", - wakeMessageOnNotifySuccess: "terminal follow-up wake", + wakeMessageOnNotifySuccess: [ + "terminal follow-up wake", + 'originRoute: {"provider":"telegram","target":"topic","threadId":"13832","sessionKey":"session:terminal"}', + ].join("\n"), }; const laterCycleRequest = { ...firstCycleRequest, @@ -311,15 +315,49 @@ describe("SessionNotificationService", () => { }, completionWakeOutcomeKey: "terminal:stable-terminal-session:completed:1780000002000:backend-thread:5:unknown", }; + const finalCycleRequest = { + ...firstCycleRequest, + idempotencyKey: "terminal-completed:stable-terminal-session:completed:1780000003000:backend-thread:6:unknown", + completionSummary: { + required: true, + producer: "terminal" as const, + outcomeKey: "terminal:stable-terminal-session:completed:1780000003000:backend-thread:6:unknown", + }, + completionWakeOutcomeKey: "terminal:stable-terminal-session:completed:1780000003000:backend-thread:6:unknown", + }; - service.dispatch(session, firstCycleRequest); - service.dispatch(session, firstCycleRequest); - service.dispatch(session, laterCycleRequest); + const resumedSession = { + ...firstSession, + startedAt: 1780000002000, + } as any; + const resumedAgainSession = { + ...firstSession, + startedAt: 1780000003000, + } as any; - assert.equal(requests.length, 2); - assert.deepEqual(requests.map((request) => request.label), ["completed", "completed"]); + service.dispatch(firstSession, firstCycleRequest); + service.dispatch(firstSession, firstCycleRequest); + service.dispatch(resumedSession, laterCycleRequest); + service.dispatch(resumedSession, laterCycleRequest); + service.dispatch(resumedAgainSession, finalCycleRequest); + service.dispatch(resumedAgainSession, finalCycleRequest); + + assert.equal(requests.length, 3); + assert.deepEqual(requests.map((request) => request.label), ["completed", "completed", "completed"]); assert.notEqual(requests[0]?.idempotencyKey, requests[1]?.idempotencyKey); - assert.equal(persisted.notificationDedupe?.length, 2); + assert.notEqual(requests[1]?.idempotencyKey, requests[2]?.idempotencyKey); + assert.deepEqual( + requests.map((request) => request.completionWakeSummaryRequired), + [true, true, true], + ); + assert.equal( + requests.every((request) => String(request.wakeMessageOnNotifySuccess).includes( + 'originRoute: {"provider":"telegram","target":"topic","threadId":"13832","sessionKey":"session:terminal"}', + )), + true, + ); + assert.equal(persisted.notificationDedupe?.length, 3); + assert.ok(persisted.completionSummaryDedupe?.length > 0); }); it("dedupes one no-change terminal cycle but allows a later resumed cycle for the same session id", () => { diff --git a/tests/session-store.test.ts b/tests/session-store.test.ts index 0db6b68c..4672194e 100644 --- a/tests/session-store.test.ts +++ b/tests/session-store.test.ts @@ -525,6 +525,50 @@ describe("SessionStore path resolution", () => { assert.deepEqual([...store.persisted.keys()], []); }); + it("preserves completion summary history when a stable session id starts a resumed lifecycle", () => { + const dir = mkdtempSync(join(tmpdir(), "openclaw-store-resumed-summary-")); + const indexPath = join(dir, "sessions.json"); + const recordedAt = new Date().toISOString(); + writeStore(indexPath, [{ + sessionId: "stable-resumed-session", + harnessSessionId: "h-stable-resumed-session", + name: "stable-resumed-session", + prompt: "first turn", + workdir: "/tmp", + status: "completed", + lifecycle: "terminal", + costUsd: 0, + completionSummaryDedupe: [{ + key: "route:cycle-one:outcome", + recordedAt, + label: "completed", + }], + }]); + const store = new SessionStore({ indexPath, env: {} }); + + store.markRunning({ + id: "stable-resumed-session", + name: "stable-resumed-session", + harnessSessionId: "h-stable-resumed-session", + prompt: "second turn", + workdir: "/tmp", + startedAt: Date.now(), + route: DEFAULT_ROUTE, + harnessName: "codex", + } as any); + + assert.deepEqual( + store.getPersistedSession("stable-resumed-session")?.completionSummaryDedupe, + [{ + key: "route:cycle-one:outcome", + linkedKeys: undefined, + recordedAt, + label: "completed", + skipReason: undefined, + }], + ); + }); + it("rebuilds short session ID lookup from persisted index after restart", () => { const dir = mkdtempSync(join(tmpdir(), "openclaw-store-restart-")); const indexPath = join(dir, "sessions.json");