Skip to content

Commit c25a576

Browse files
fix(desktop): persist runs and recover question workflows (#428)
## Summary Make generation and question workflows survive renderer refreshes, window switches, and temporary detachment without introducing a new database. The main process now owns durable run/question state, chat projection, and completion snapshots; the renderer reattaches using ordered replay and preserves unanswered form drafts. ### Changes - Persist `runId`/`generationId`, monotonic event sequences, schema version, and explicit `run_settled` outcomes in per-run JSONL journals. Commit events before broadcasting; validate cursors and fail closed on committed corruption while repairing incomplete tails. - Subscribe before replay, buffer overlapping live events, deduplicate cursors, recover gaps, and catch up on focus/visibility changes. Support multiple renderer subscribers without replay-triggered tool execution, auto-polish, or stale workspace overwrites. - Persist questions, answers, cancellation/interruption, and acknowledgement state in `ask-journal.jsonl`. Scope queues and drafts to the owning request/design; retain input on failed persistence, accept identical answer retries, and reject conflicting retries. - Project chat/tool/active-message events idempotently. Committed delivery and tool results outrank inferred terminal fallbacks; repaired rows refresh even when replay contains only already-consumed cursors. - Finalize artifact snapshots in the main process so detached completion does not depend on the original renderer Promise. Reattached renderers refresh the latest snapshot without recreating it. - Preserve the upstream native cancellation and import-ownership protections; keep generation/workspace ownership until the worker actually settles, and reject duplicate run IDs. ## Type of change - [x] Bug fix - [x] New feature - [ ] Breaking change ## Context Follow-up to the continuity work in #425, rebased onto `main` including the v0.2.2 release updates. This PR contains only the recovery implementation, regression tests, and a desktop patch changeset. Behavioral reference: [Open Design run identity, JSONL events, replay, and restart handling](https://github.com/nexu-io/open-design/blob/f5707c8cae7b61014c5e91e114b9aa07826df725/apps/daemon/src/runtimes/runs.ts). Retains CoDesign's existing file/JSONL and session-chat approach rather than importing the separate logical-task database architecture. ## Recovery contract and limits - Renderer refresh/detachment can reattach to a still-running main-process run and pending questions. - A main-process restart preserves history and marks orphaned runs/questions **interrupted**. It does **not** resurrect provider Promises or automatically repeat tools/side effects. - Question answers are acknowledged only after persistence. Invalid submissions leave the question available for correction; losing a window no longer cancels the question by itself. - Additive IPC fields and versioned journals; no SQLite, new production dependencies, or retention/deletion policy. - Very large histories may need future replay pagination and writer backpressure. ## Validation Executed locally on Windows with Node 22 and repository-pinned pnpm: - `pnpm lint` — passed (617 files). - `pnpm typecheck` — passed for the workspace. - `pnpm test` — passed for the workspace; desktop **152 test files, 1937 passed / 1 skipped**. Unchanged workspace packages used valid Turbo cache entries. - `pnpm --filter @open-codesign/desktop build` — main, preload, and renderer bundles passed. - `git diff --check` — passed. Repository commit/push hooks also passed. Regression coverage includes temporary-filesystem reopen/torn-write/corruption cases, failed durable answers, identical/conflicting retries, native cancellation ownership, replay/live overlap and gaps, multiple subscribers, headless completion and snapshot persistence, repaired chat authority, stale status responses, design-scoped drafts, and cancellation/import races. No paid/live-provider manual Electron reload acceptance run was performed. No visual redesign or screenshot recording is included; renderer recovery and import behavior are covered by component tests. ## Checklist - [x] Checked relevant continuity context and upstream implementation - [x] `pnpm lint && pnpm typecheck && pnpm test` passes locally - [x] Added/updated regression tests - [x] Added a user-visible desktop patch changeset - [x] Documented recovery boundaries above - [x] Additive protocol, schema-versioned storage, explicit main-process ownership - [x] No new production dependencies or database layer Signed-off-by: Sun-sunshine06 <Sun-sunshine06@users.noreply.github.com> Co-authored-by: Sun-sunshine06 <Sun-sunshine06@users.noreply.github.com>
1 parent 4284ede commit c25a576

27 files changed

Lines changed: 3938 additions & 349 deletions
Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"@open-codesign/desktop": patch
3+
---
4+
5+
Persist generation run IDs, ordered events and terminal results in local JSONL, and reattach renderer streams after reload or focus without duplicate chat writes. Persist structured questions and answers, recover answer drafts, and make answer retries idempotent across windows. Mark orphaned executions interrupted after a main-process restart instead of rerunning side effects; keep cancellation locks until the original worker settles.

‎apps/desktop/src/main/ask-ipc.test.ts‎

Lines changed: 192 additions & 100 deletions
Large diffs are not rendered by default.

‎apps/desktop/src/main/ask-ipc.ts‎

Lines changed: 170 additions & 133 deletions
Original file line numberDiff line numberDiff line change
@@ -1,206 +1,243 @@
11
import { randomUUID } from 'node:crypto';
2+
import { join } from 'node:path';
23
import type { AskInput, AskResult } from '@open-codesign/core';
34
import { type AskCancelledV1, CodesignError, ERROR_CODES } from '@open-codesign/shared';
4-
import { type BrowserWindow, ipcMain } from 'electron';
5+
import { app, BrowserWindow, ipcMain } from 'electron';
6+
import {
7+
type AskHistoryEntry,
8+
type AskHistoryFilter,
9+
type AskRequestPayload,
10+
AskStore,
11+
} from './ask-store';
512
import { getLogger } from './logger';
613

7-
/**
8-
* Bridge for the core `ask` tool. Mirrors permission-ipc.ts:
9-
* 1. core's ask tool calls `requestAsk(sessionId, input, getMainWindow)`
10-
* 2. `requestAsk` issues a unique requestId, stores a resolver,
11-
* and `webContents.send('ask:request', { requestId, sessionId, input })`
12-
* 3. renderer mounts <AskModal>, user submits or cancels
13-
* 4. renderer invokes `ask:resolve` with the requestId + result
14-
* 5. ipcMain handler resolves the pending promise
15-
*/
14+
export type { AskRequestPayload } from './ask-store';
1615

1716
const log = getLogger('ask-ipc');
18-
1917
interface PendingAsk {
2018
resolve: (result: AskResult) => void;
2119
reject: (reason?: unknown) => void;
22-
sessionId: string;
23-
input: AskInput;
24-
cancel: () => void;
20+
payload: AskRequestPayload;
21+
ready: Promise<AskHistoryEntry>;
22+
cleanup: () => void;
23+
cancelling: boolean;
2524
}
26-
2725
const pending = new Map<string, PendingAsk>();
26+
let store: AskStore | undefined;
2827

29-
export interface AskRequestPayload {
30-
requestId: string;
31-
sessionId: string;
32-
input: AskInput;
28+
function getStore(): AskStore {
29+
store ??= new AskStore(join(app.getPath('userData'), 'ask-journal.jsonl'));
30+
return store;
3331
}
3432

35-
export function registerAskIpc(): void {
33+
export function registerAskIpc(storePath?: string): void {
34+
if (storePath !== undefined) {
35+
if (pending.size > 0) throw new Error('Cannot replace ask store with live questions');
36+
store = new AskStore(storePath);
37+
}
3638
ipcMain.handle('ask:list-pending', () => listPendingAskRequests());
39+
ipcMain.handle('ask:history', (_event, filter: unknown) =>
40+
getStore().history(parseHistoryFilter(filter)),
41+
);
42+
ipcMain.handle('ask:resolve', async (_event, raw: unknown) => {
43+
// Invalid renderer submissions must not consume the runtime's original promise.
44+
const { requestId, ...result } = parseResolveInput(raw);
45+
await pending.get(requestId)?.ready;
46+
const entry = await getStore().settle(requestId, result);
47+
finish(entry);
48+
});
49+
}
3750

38-
ipcMain.handle('ask:resolve', (_event, raw: unknown) => {
39-
const requestId = readRequestId(raw, 'ask:resolve');
40-
const entry = pending.get(requestId);
41-
if (!entry) {
42-
throw new CodesignError(
43-
`ask:resolve called with unknown requestId "${requestId}"`,
44-
ERROR_CODES.IPC_BAD_INPUT,
45-
);
46-
}
47-
let parsed: {
48-
requestId: string;
49-
status: 'answered' | 'cancelled';
50-
answers: AskResult['answers'];
51-
};
51+
function finish(entry: AskHistoryEntry): void {
52+
if (!entry.result) return;
53+
const live = pending.get(entry.requestId);
54+
pending.delete(entry.requestId);
55+
live?.cleanup();
56+
live?.resolve(entry.result);
57+
log.info('ask.resolve', {
58+
sessionId: entry.sessionId,
59+
requestId: entry.requestId,
60+
status: entry.status,
61+
});
62+
const payload = {
63+
requestId: entry.requestId,
64+
sessionId: entry.sessionId,
65+
...(entry.runId ? { runId: entry.runId } : {}),
66+
...(entry.designId ? { designId: entry.designId } : {}),
67+
status: entry.status,
68+
};
69+
for (const win of BrowserWindow.getAllWindows()) {
5270
try {
53-
parsed = parseResolveInput(raw);
54-
} catch (err) {
55-
pending.delete(requestId);
56-
entry.reject(err);
57-
throw err;
71+
if (!win.isDestroyed() && !win.webContents.isDestroyed()) {
72+
win.webContents.send('ask:resolved', payload);
73+
if (entry.status === 'cancelled') {
74+
const cancelled: AskCancelledV1 = {
75+
schemaVersion: 1,
76+
requestId: entry.requestId,
77+
sessionId: entry.sessionId,
78+
};
79+
win.webContents.send('ask:cancelled', cancelled);
80+
}
81+
}
82+
} catch (error) {
83+
log.warn('ask.resolve.delivery_failed', { requestId: entry.requestId, error: String(error) });
5884
}
85+
}
86+
}
87+
88+
async function cancelRequest(requestId: string): Promise<void> {
89+
const live = pending.get(requestId);
90+
if (!live) return;
91+
live.cancelling = true;
92+
try {
93+
await live.ready;
94+
const entry = await getStore().settle(requestId, { status: 'cancelled', answers: [] });
95+
finish(entry);
96+
} catch (error) {
97+
if (pending.get(requestId) !== live) return;
5998
pending.delete(requestId);
60-
log.info('ask.resolve', {
61-
sessionId: entry.sessionId,
62-
requestId,
63-
status: parsed.status,
64-
answers: parsed.answers.length,
65-
});
66-
entry.resolve({ status: parsed.status, answers: parsed.answers });
67-
});
99+
live.cleanup();
100+
live.reject(error);
101+
log.error('ask.cancel.failed', { requestId, error: String(error) });
102+
}
103+
}
104+
105+
export interface RequestAskOptions {
106+
designId?: string;
107+
signal?: AbortSignal;
68108
}
69109

70110
export function requestAsk(
71111
sessionId: string,
72112
input: AskInput,
73113
getMainWindow: () => BrowserWindow | null,
74-
signal?: AbortSignal,
114+
optionsOrSignal: RequestAskOptions | AbortSignal = {},
75115
): Promise<AskResult> {
116+
const options: RequestAskOptions =
117+
'aborted' in optionsOrSignal ? { signal: optionsOrSignal } : optionsOrSignal;
76118
const requestId = `ask-${randomUUID()}`;
119+
const payload: AskRequestPayload = {
120+
requestId,
121+
sessionId,
122+
runId: sessionId,
123+
...(options.designId ? { designId: options.designId } : {}),
124+
input: structuredClone(input),
125+
};
77126
return new Promise<AskResult>((resolve, reject) => {
78-
const win = getMainWindow();
79-
if (signal?.aborted || !win || win.isDestroyed() || win.webContents.isDestroyed()) {
80-
log.warn('ask.request.no_window', { sessionId, requestId });
81-
resolve({ status: 'cancelled', answers: [] });
82-
return;
83-
}
84-
const cleanup = () => {
85-
pending.delete(requestId);
86-
signal?.removeEventListener('abort', cancel);
87-
win.removeListener('closed', cancel);
88-
win.webContents.removeListener('destroyed', cancel);
127+
const onAbort = () => {
128+
void cancelRequest(requestId);
89129
};
90-
const cancel = () => {
91-
if (!pending.has(requestId)) return;
92-
cleanup();
93-
resolve({ status: 'cancelled', answers: [] });
94-
if (!win.isDestroyed() && !win.webContents.isDestroyed()) {
95-
const payload: AskCancelledV1 = { schemaVersion: 1, requestId, sessionId };
130+
const live: PendingAsk = {
131+
resolve,
132+
reject,
133+
payload,
134+
ready: getStore().create(payload),
135+
cleanup: () => options.signal?.removeEventListener('abort', onAbort),
136+
cancelling: false,
137+
};
138+
pending.set(requestId, live);
139+
options.signal?.addEventListener('abort', onAbort, { once: true });
140+
if (options.signal?.aborted) onAbort();
141+
void live.ready
142+
.then(() => {
143+
if (pending.get(requestId) !== live || live.cancelling) return;
144+
// Window closure/reload is not cancellation: pending() replays after remount.
96145
try {
97-
win.webContents.send('ask:cancelled', payload);
146+
const win = getMainWindow();
147+
if (!win || win.isDestroyed() || win.webContents.isDestroyed()) return;
148+
win.webContents.send('ask:request', payload);
98149
} catch (error) {
99-
log.warn('ask.cancel.delivery_failed', {
100-
sessionId,
101-
requestId,
102-
message: error instanceof Error ? error.message : String(error),
103-
});
150+
log.warn('ask.request.delivery_failed', { requestId, error: String(error) });
104151
}
105-
}
106-
};
107-
pending.set(requestId, {
108-
sessionId,
109-
input,
110-
cancel,
111-
resolve: (result) => {
112-
cleanup();
113-
resolve(result);
114-
},
115-
reject: (reason) => {
116-
cleanup();
117-
reject(reason);
118-
},
119-
});
120-
signal?.addEventListener('abort', cancel, { once: true });
121-
win.once('closed', cancel);
122-
win.webContents.once('destroyed', cancel);
123-
const payload: AskRequestPayload = { requestId, sessionId, input };
124-
log.info('ask.request.send', {
125-
sessionId,
126-
requestId,
127-
questions: input.questions.length,
128-
});
129-
try {
130-
win.webContents.send('ask:request', payload);
131-
} catch (error) {
132-
cleanup();
133-
reject(error);
134-
}
152+
})
153+
.catch((error: unknown) => {
154+
if (pending.get(requestId) !== live) return;
155+
pending.delete(requestId);
156+
live.cleanup();
157+
reject(error);
158+
});
135159
});
136160
}
137161

138-
export function listPendingAskRequests(): AskRequestPayload[] {
139-
return Array.from(pending, ([requestId, entry]) => ({
140-
requestId,
141-
sessionId: entry.sessionId,
142-
input: entry.input,
143-
}));
162+
export async function listPendingAskRequests(): Promise<AskRequestPayload[]> {
163+
const history = await getStore().history();
164+
return history
165+
.filter((entry) => entry.status === 'pending' && pending.has(entry.requestId))
166+
.map((entry) => ({
167+
requestId: entry.requestId,
168+
sessionId: entry.sessionId,
169+
...(entry.runId ? { runId: entry.runId } : {}),
170+
...(entry.designId ? { designId: entry.designId } : {}),
171+
input: entry.input,
172+
}));
173+
}
174+
175+
export async function cancelPendingAskRequests(sessionId: string): Promise<void> {
176+
await Promise.all(
177+
[...pending]
178+
.filter(([, entry]) => entry.payload.sessionId === sessionId)
179+
.map(([id]) => cancelRequest(id)),
180+
);
144181
}
145182

146-
export function cancelPendingAskRequests(sessionId: string): void {
147-
for (const entry of pending.values()) {
148-
if (entry.sessionId !== sessionId) continue;
149-
entry.cancel();
183+
function parseHistoryFilter(raw: unknown): AskHistoryFilter {
184+
if (raw === undefined) return {};
185+
if (!raw || typeof raw !== 'object' || Array.isArray(raw)) {
186+
throw new CodesignError('ask:history expects a filter object', ERROR_CODES.IPC_BAD_INPUT);
187+
}
188+
const filter = raw as Record<string, unknown>;
189+
for (const [key, value] of Object.entries(filter)) {
190+
if (!['runId', 'designId'].includes(key) || typeof value !== 'string' || !value.trim()) {
191+
throw new CodesignError('ask:history invalid filter', ERROR_CODES.IPC_BAD_INPUT);
192+
}
150193
}
194+
return filter as AskHistoryFilter;
151195
}
152196

153197
function badResolvePayload(message: string): never {
154198
throw new CodesignError(`ask:resolve ${message}`, ERROR_CODES.IPC_BAD_INPUT);
155199
}
156200

157-
function readRequestId(raw: unknown, channel: string): string {
158-
if (!raw || typeof raw !== 'object') {
159-
throw new CodesignError(`${channel} expects an object payload`, ERROR_CODES.IPC_BAD_INPUT);
160-
}
161-
const obj = raw as Record<string, unknown>;
162-
const requestId = obj['requestId'];
163-
if (typeof requestId !== 'string' || requestId.trim().length === 0) {
164-
throw new CodesignError(`${channel} requires a non-empty requestId`, ERROR_CODES.IPC_BAD_INPUT);
165-
}
201+
function readRequestId(raw: unknown): string {
202+
if (!raw || typeof raw !== 'object' || Array.isArray(raw))
203+
badResolvePayload('expects an object payload');
204+
const requestId = (raw as Record<string, unknown>)['requestId'];
205+
if (typeof requestId !== 'string' || requestId.trim().length === 0)
206+
badResolvePayload('requires a non-empty requestId');
166207
return requestId;
167208
}
168209

169-
function parseResolveInput(raw: unknown): {
170-
requestId: string;
171-
status: 'answered' | 'cancelled';
172-
answers: AskResult['answers'];
173-
} {
174-
const requestId = readRequestId(raw, 'ask:resolve');
210+
function parseResolveInput(raw: unknown): { requestId: string } & AskResult {
211+
const requestId = readRequestId(raw);
175212
const obj = raw as Record<string, unknown>;
176213
assertKnownFields(obj, ['requestId', 'status', 'answers']);
177214
const status = obj['status'];
178215
const answers = obj['answers'];
179-
if (status !== 'answered' && status !== 'cancelled') {
216+
if (status !== 'answered' && status !== 'cancelled')
180217
badResolvePayload('status must be "answered" or "cancelled"');
181-
}
182218
if (!Array.isArray(answers)) badResolvePayload('answers must be an array');
183219
const clean: AskResult['answers'] = [];
184220
for (const a of answers) {
185-
if (!a || typeof a !== 'object') badResolvePayload('answers must contain objects');
221+
if (!a || typeof a !== 'object' || Array.isArray(a))
222+
badResolvePayload('answers must contain objects');
186223
const rec = a as Record<string, unknown>;
187224
assertKnownFields(rec, ['questionId', 'value']);
188225
const questionId = rec['questionId'];
189226
const value = rec['value'];
190-
if (typeof questionId !== 'string') badResolvePayload('answer questionId must be a string');
227+
if (typeof questionId !== 'string' || !questionId.trim())
228+
badResolvePayload('answer questionId must be a non-empty string');
191229
if (
192230
value !== null &&
193231
typeof value !== 'string' &&
194-
typeof value !== 'number' &&
232+
!(typeof value === 'number' && Number.isFinite(value)) &&
195233
!(Array.isArray(value) && value.every((v) => typeof v === 'string'))
196234
) {
197-
badResolvePayload('answer value must be a string, number, string array, or null');
235+
badResolvePayload('answer value must be a string, finite number, string array, or null');
198236
}
199237
clean.push({ questionId, value: value as string | number | string[] | null });
200238
}
201239
return { requestId, status, answers: clean };
202240
}
203-
204241
function assertKnownFields(record: Record<string, unknown>, allowed: readonly string[]): void {
205242
const unsupported = Object.keys(record).find((key) => !allowed.includes(key));
206243
if (unsupported !== undefined) badResolvePayload(`contains unsupported field "${unsupported}"`);

0 commit comments

Comments
 (0)