wangxuming/makelore
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: 对接设计 Agent Gateway WebSocket

Browse Source 
需求：服务端统一 Agent Gateway 将设计任务状态流切换为 WebSocket，客户端需要实时展示生成任务并支持断线恢复。

实现：Electron Main 管理 Session、一次性 Ticket、WebSocket 心跳与游标续传，按关闭码回收会话；Renderer 继续通过本机 Host API 的 SSE 投影接收任务事件，并保留 REST 降级同步。

验证：typecheck、变更文件 ESLint、37 个聚焦测试及 build:vite 通过。
This commit is contained in:
brother7
2026-08-02 20:15:37 +08:00
parent 83e21563dc
commit 518d7ab4cd
17 changed files with 1816 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/lib/image-workspace.ts
View File

@@ -1,5 +1,6 @@
import { AppError } from '@/lib/error-model';
import {
createHostEventSource,
ensureHostApiToken,
getHostApiBase,
hostApiFetch,
@@ -159,6 +160,13 @@ export function fetchImageWorkspaceTasks(
);
}
export async function openImageWorkspaceTaskEvents(workspaceId: string): Promise<EventSource> {
await ensureHostApiToken();
return createHostEventSource(
`${IMAGE_WORKSPACE_API_PATH}/workspaces/${encodeURIComponent(workspaceId)}/events`,
);
}
export async function resolveImageWorkspaceAssetUrl(contentPath: string): Promise<string> {
const token = await ensureHostApiToken();
const separator = contentPath.includes('?') ? '&' : '?';

17
src/pages/ImageCanvas/index.tsx
View File

@@ -194,6 +194,8 @@ export function ImageCanvas() {
const load = useImageWorkspaceStore((state) => state.load);
const refreshWorkspace = useImageWorkspaceStore((state) => state.refreshWorkspace);
const refreshTasks = useImageWorkspaceStore((state) => state.refreshTasks);
const connectTaskStream = useImageWorkspaceStore((state) => state.connectTaskStream);
const disconnectTaskStream = useImageWorkspaceStore((state) => state.disconnectTaskStream);
const sendMessage = useImageWorkspaceStore((state) => state.sendMessage);
const confirmGeneration = useImageWorkspaceStore((state) => state.confirmGeneration);
const [prompt, setPrompt] = useState('');
@@ -206,6 +208,7 @@ export function ImageCanvas() {
() => workspace ? activeQuote(workspace.messages) : null,
[workspace],
);
const taskWorkspaceId = workspace?.workspaceId ?? null;
const hasActiveTasks = tasks.some((task) => ACTIVE_TASK_STATUSES.has(task.status));
useEffect(() => {
@@ -213,12 +216,10 @@ export function ImageCanvas() {
}, [authenticated, load, status]);
useEffect(() => {
if (!hasActiveTasks || !workspace) return;
const timer = window.setInterval(() => {
void refreshTasks().catch(() => undefined);
}, 2_500);
return () => window.clearInterval(timer);
}, [hasActiveTasks, refreshTasks, workspace]);
if (!taskWorkspaceId) return;
connectTaskStream();
return disconnectTaskStream;
}, [connectTaskStream, disconnectTaskStream, taskWorkspaceId]);
useEffect(() => {
if (typeof conversationEndRef.current?.scrollIntoView === 'function') {
@@ -354,7 +355,9 @@ export function ImageCanvas() {
aria-label="刷新设计项目"
className="h-9 w-9 rounded-full"
onClick={() => {
void Promise.all([refreshWorkspace(), refreshTasks()]).catch(() => undefined);
void refreshWorkspace()
.then(() => refreshTasks())
.catch(() => undefined);
}}
>
<RefreshCw className="h-4 w-4" />

298
src/stores/image-workspace.ts
View File

@@ -6,6 +6,7 @@ import {
fetchImageWorkspaceProject,
fetchImageWorkspaceTasks,
ImageWorkspaceApiError,
openImageWorkspaceTaskEvents,
renameImageWorkspaceProject,
sendImageWorkspaceMessage,
} from '@/lib/image-workspace';
@@ -13,6 +14,8 @@ import { useAuthStore } from '@/stores/auth';
import {
IMAGE_WORKSPACE_UNAVAILABLE_CODE,
type DesignGenerationTask,
type DesignGenerationTasksSnapshotEvent,
type DesignGenerationTaskUpdatedEvent,
type DesignWorkspace,
type DesignWorkspaceBootstrap,
type DesignWorkspaceSummary,
@@ -26,12 +29,15 @@ export type ImageWorkspaceLoadStatus =
| 'error'
| 'auth-required';
export type ImageWorkspaceTaskStreamState = 'idle' | 'connecting' | 'connected' | 'degraded';
type ImageWorkspaceState = {
status: ImageWorkspaceLoadStatus;
bootstrap: DesignWorkspaceBootstrap | null;
activeWorkspaceId: string | null;
workspace: DesignWorkspace | null;
tasks: DesignGenerationTask[];
taskStreamState: ImageWorkspaceTaskStreamState;
error: string | null;
load: () => Promise<DesignWorkspaceBootstrap | null>;
createProject: (title: string) => Promise<DesignWorkspace>;
@@ -39,17 +45,88 @@ type ImageWorkspaceState = {
selectProject: (workspaceId: string) => Promise<void>;
refreshWorkspace: () => Promise<DesignWorkspace | null>;
refreshTasks: () => Promise<DesignGenerationTask[]>;
connectTaskStream: () => void;
disconnectTaskStream: () => void;
sendMessage: (message: string) => Promise<DesignWorkspace>;
confirmGeneration: (quoteId: string) => Promise<DesignWorkspace>;
reset: () => void;
};
let inFlightLoad: Promise<DesignWorkspaceBootstrap | null> | null = null;
const GENERATION_TASK_SYNC_ATTEMPTS = 20;
const GENERATION_TASK_SYNC_DELAY_MS = 500;
const TASK_FALLBACK_POLL_INTERVAL_MS = 15_000;
let activeTaskEventSource: EventSource | null = null;
let activeTaskEventWorkspaceId: string | null = null;
let taskStreamGeneration = 0;
let taskFallbackTimer: ReturnType<typeof setInterval> | null = null;
const taskEventRevisions = new Map<string, number>();
function wait(milliseconds: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, milliseconds));
function taskRevisionKey(workspaceId: string, taskId: string): string {
return `${workspaceId}:${taskId}`;
}
function stopFallbackPolling(): void {
if (taskFallbackTimer !== null) clearInterval(taskFallbackTimer);
taskFallbackTimer = null;
}
function closeTaskEventSource(): void {
taskStreamGeneration += 1;
stopFallbackPolling();
if (activeTaskEventSource) {
activeTaskEventSource.onopen = null;
activeTaskEventSource.onerror = null;
activeTaskEventSource.close();
}
activeTaskEventSource = null;
activeTaskEventWorkspaceId = null;
taskEventRevisions.clear();
}
function parseTaskUpdatedEvent(event: Event): DesignGenerationTaskUpdatedEvent | null {
const data = (event as MessageEvent<unknown>).data;
if (typeof data !== 'string') return null;
try {
const payload = JSON.parse(data) as Partial<DesignGenerationTaskUpdatedEvent>;
if (payload.type !== 'design.generation_task.updated'
|| typeof payload.id !== 'string'
|| typeof payload.workspaceId !== 'string'
|| !Number.isInteger(payload.workspaceViewRevision)
|| !payload.generationTask
|| typeof payload.generationTask.taskId !== 'string'
|| payload.generationTask.workspaceId !== payload.workspaceId) {
return null;
}
return payload as DesignGenerationTaskUpdatedEvent;
} catch {
return null;
}
}
function parseTasksSnapshotEvent(event: Event): DesignGenerationTasksSnapshotEvent | null {
const data = (event as MessageEvent<unknown>).data;
if (typeof data !== 'string') return null;
try {
const payload = JSON.parse(data) as Partial<DesignGenerationTasksSnapshotEvent>;
if (payload.type !== 'design.generation_tasks.snapshot'
|| typeof payload.id !== 'string'
|| typeof payload.workspaceId !== 'string'
|| !Number.isInteger(payload.workspaceViewRevision)
|| !Array.isArray(payload.generationTasks)
|| !payload.generationTasks.every((task) => (
typeof task?.taskId === 'string' && task.workspaceId === payload.workspaceId
))) {
return null;
}
return payload as DesignGenerationTasksSnapshotEvent;
} catch {
return null;
}
}
function sortTasks(tasks: DesignGenerationTask[]): DesignGenerationTask[] {
return [...tasks].sort((left, right) => (
right.createdAt.localeCompare(left.createdAt) || right.taskId.localeCompare(left.taskId)
));
}
function unavailable(error: unknown): boolean {
@@ -86,9 +163,147 @@ function upsertSummary(
}
export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) => {
const stopTaskStream = () => {
closeTaskEventSource();
set({ taskStreamState: 'idle' });
};
const startFallbackPolling = () => {
if (taskFallbackTimer !== null) return;
taskFallbackTimer = setInterval(() => {
if (get().taskStreamState !== 'degraded' || !get().activeWorkspaceId) return;
void get().refreshWorkspace()
.catch(() => null)
.then(() => get().refreshTasks().catch(() => []));
}, TASK_FALLBACK_POLL_INTERVAL_MS);
};
const startTaskStream = (workspaceId: string) => {
if (activeTaskEventWorkspaceId === workspaceId
&& get().taskStreamState !== 'idle') return;
closeTaskEventSource();
const generation = taskStreamGeneration;
activeTaskEventWorkspaceId = workspaceId;
set({ taskStreamState: 'connecting' });
void openImageWorkspaceTaskEvents(workspaceId).then((source) => {
if (generation !== taskStreamGeneration || get().activeWorkspaceId !== workspaceId) {
source.close();
return;
}
activeTaskEventSource = source;
source.addEventListener('design.generation_tasks.snapshot', (event) => {
const snapshot = parseTasksSnapshotEvent(event);
if (!snapshot || snapshot.workspaceId !== activeTaskEventWorkspaceId) return;
set((state) => {
if (state.activeWorkspaceId !== snapshot.workspaceId) return state;
const tasksById = new Map(state.tasks.map((task) => [task.taskId, task]));
for (const task of snapshot.generationTasks) {
const key = taskRevisionKey(snapshot.workspaceId, task.taskId);
const knownRevision = taskEventRevisions.get(key) ?? 0;
const existing = tasksById.get(task.taskId);
if (snapshot.workspaceViewRevision >= knownRevision
&& (!existing || existing.updatedAt <= task.updatedAt)) {
tasksById.set(task.taskId, task);
}
taskEventRevisions.set(
key,
Math.max(knownRevision, snapshot.workspaceViewRevision),
);
}
const workspace = state.workspace?.workspaceId === snapshot.workspaceId
? {
...state.workspace,
viewRevision: Math.max(
state.workspace.viewRevision,
snapshot.workspaceViewRevision,
),
}
: state.workspace;
const bootstrap = state.bootstrap
? {
...state.bootstrap,
workspaces: state.bootstrap.workspaces.map((candidate) => (
candidate.workspaceId === snapshot.workspaceId
? {
...candidate,
viewRevision: Math.max(
candidate.viewRevision,
snapshot.workspaceViewRevision,
),
}
: candidate
)),
}
: null;
return { tasks: sortTasks([...tasksById.values()]), workspace, bootstrap };
});
});
source.addEventListener('design.generation_task.updated', (event) => {
const update = parseTaskUpdatedEvent(event);
if (!update || update.workspaceId !== activeTaskEventWorkspaceId) return;
set((state) => {
if (state.activeWorkspaceId !== update.workspaceId) return state;
const key = taskRevisionKey(update.workspaceId, update.generationTask.taskId);
const knownRevision = taskEventRevisions.get(key) ?? 0;
if (update.workspaceViewRevision <= knownRevision) return state;
const existing = state.tasks.find(
(candidate) => candidate.taskId === update.generationTask.taskId,
);
if (existing && existing.updatedAt > update.generationTask.updatedAt) return state;
taskEventRevisions.set(key, update.workspaceViewRevision);
const tasks = sortTasks([
update.generationTask,
...state.tasks.filter((candidate) => candidate.taskId !== update.generationTask.taskId),
]);
const workspace = state.workspace?.workspaceId === update.workspaceId
? {
...state.workspace,
viewRevision: Math.max(
state.workspace.viewRevision,
update.workspaceViewRevision,
),
}
: state.workspace;
const bootstrap = state.bootstrap
? {
...state.bootstrap,
workspaces: state.bootstrap.workspaces.map((candidate) => (
candidate.workspaceId === update.workspaceId
? {
...candidate,
viewRevision: Math.max(
candidate.viewRevision,
update.workspaceViewRevision,
),
}
: candidate
)),
}
: null;
return { tasks, workspace, bootstrap };
});
});
source.onopen = () => {
if (activeTaskEventSource !== source) return;
stopFallbackPolling();
set({ taskStreamState: 'connected' });
};
source.onerror = () => {
if (activeTaskEventSource !== source) return;
set({ taskStreamState: 'degraded' });
startFallbackPolling();
};
}).catch(() => {
if (generation !== taskStreamGeneration || get().activeWorkspaceId !== workspaceId) return;
set({ taskStreamState: 'degraded' });
startFallbackPolling();
});
};
const handleRequestError = (error: unknown): string => {
const message = messageOf(error);
if (authenticationRequired(error)) {
closeTaskEventSource();
useAuthStore.getState().invalidateSession();
set({
status: 'auth-required',
@@ -96,6 +311,7 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
activeWorkspaceId: null,
workspace: null,
tasks: [],
taskStreamState: 'idle',
error: message,
});
}
@@ -113,34 +329,23 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
return workspace;
};
const waitForGenerationTask = async (
workspaceId: string,
quoteId: string,
): Promise<DesignGenerationTask[]> => {
for (let attempt = 0; attempt < GENERATION_TASK_SYNC_ATTEMPTS; attempt += 1) {
if (get().activeWorkspaceId !== workspaceId) return [];
const tasks = await get().refreshTasks().catch(() => []);
if (tasks.some((task) => task.quoteId === quoteId)) return tasks;
if (attempt < GENERATION_TASK_SYNC_ATTEMPTS - 1) {
await wait(GENERATION_TASK_SYNC_DELAY_MS);
}
}
throw new Error('生成请求已确认，但任务列表尚未同步，请点击刷新重试');
};
const loadWorkspace = async (workspaceId: string): Promise<void> => {
const [workspace, tasks] = await Promise.all([
fetchImageWorkspaceProject(workspaceId),
fetchImageWorkspaceTasks(workspaceId),
]);
const workspace = await fetchImageWorkspaceProject(workspaceId);
if (get().activeWorkspaceId !== workspaceId) return;
const tasks = await fetchImageWorkspaceTasks(workspaceId);
if (get().activeWorkspaceId !== workspaceId) return;
set((state) => ({
status: 'ready',
bootstrap: upsertSummary(state.bootstrap, workspace),
activeWorkspaceId: workspaceId,
workspace,
tasks,
tasks: sortTasks(tasks),
error: null,
}));
startTaskStream(workspaceId);
for (const task of tasks) {
taskEventRevisions.set(taskRevisionKey(workspaceId, task.taskId), workspace.viewRevision);
}
};
const recoverRevisionConflict = async (error: unknown): Promise<never> => {
@@ -162,6 +367,7 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
activeWorkspaceId: null,
workspace: null,
tasks: [],
taskStreamState: 'idle',
error: null,
load: () => {
@@ -176,12 +382,14 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
)
? currentId
: bootstrap.workspaces[0]?.workspaceId ?? null;
closeTaskEventSource();
set({
status: 'ready',
bootstrap,
activeWorkspaceId,
workspace: null,
tasks: [],
taskStreamState: 'idle',
error: null,
});
if (activeWorkspaceId) await loadWorkspace(activeWorkspaceId);
@@ -189,12 +397,14 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
} catch (error) {
const message = handleRequestError(error);
if (!authenticationRequired(error)) {
closeTaskEventSource();
set({
status: unavailable(error) ? 'unavailable' : 'error',
bootstrap: null,
activeWorkspaceId: null,
workspace: null,
tasks: [],
taskStreamState: 'idle',
error: message,
});
}
@@ -210,6 +420,7 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
try {
const workspace = applyWorkspace(await createImageWorkspaceProject(title));
set({ tasks: [] });
startTaskStream(workspace.workspaceId);
return workspace;
} catch (error) {
set({ error: handleRequestError(error) });
@@ -228,10 +439,12 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
selectProject: async (workspaceId) => {
if (!get().bootstrap?.workspaces.some((item) => item.workspaceId === workspaceId)) return;
stopTaskStream();
set({
activeWorkspaceId: workspaceId,
workspace: null,
tasks: [],
taskStreamState: 'idle',
error: null,
});
try {
@@ -245,7 +458,14 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
const workspaceId = get().activeWorkspaceId;
if (!workspaceId) return null;
try {
return applyWorkspace(await fetchImageWorkspaceProject(workspaceId));
const refreshed = await fetchImageWorkspaceProject(workspaceId);
if (get().activeWorkspaceId !== workspaceId) return null;
const current = get().workspace;
if (current?.workspaceId === workspaceId
&& current.viewRevision > refreshed.viewRevision) {
return current;
}
return applyWorkspace(refreshed);
} catch (error) {
set({ error: handleRequestError(error) });
throw error;
@@ -258,9 +478,24 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
set({ tasks: [] });
return [];
}
const requestedViewRevision = get().workspace?.workspaceId === workspaceId
? get().workspace?.viewRevision ?? 0
: 0;
try {
const tasks = await fetchImageWorkspaceTasks(workspaceId);
set({ tasks });
const currentWorkspace = get().workspace;
if (get().activeWorkspaceId === workspaceId
&& currentWorkspace?.workspaceId === workspaceId
&& currentWorkspace.viewRevision === requestedViewRevision) {
set({ tasks: sortTasks(tasks) });
for (const task of tasks) {
const key = taskRevisionKey(workspaceId, task.taskId);
taskEventRevisions.set(
key,
Math.max(taskEventRevisions.get(key) ?? 0, requestedViewRevision),
);
}
}
return tasks;
} catch (error) {
set({ error: handleRequestError(error) });
@@ -268,6 +503,13 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
}
},
connectTaskStream: () => {
const workspaceId = get().activeWorkspaceId;
if (workspaceId) startTaskStream(workspaceId);
},
disconnectTaskStream: stopTaskStream,
sendMessage: async (message) => {
const workspace = get().workspace;
if (!workspace) throw new Error('请先选择设计项目');
@@ -294,7 +536,7 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
workspace.turnRevision,
quoteId,
));
await waitForGenerationTask(workspace.workspaceId, quoteId);
await get().refreshTasks().catch(() => []);
return updated;
} catch (error) {
set({ error: handleRequestError(error) });
@@ -304,12 +546,14 @@ export const useImageWorkspaceStore = create<ImageWorkspaceState>((set, get) =>
reset: () => {
inFlightLoad = null;
closeTaskEventSource();
set({
status: 'idle',
bootstrap: null,
activeWorkspaceId: null,
workspace: null,
tasks: [],
taskStreamState: 'idle',
error: null,
});
},