fix: batch streamed events and simplify project exports (#6915) (#6918)

* fix: batch streamed message event persistence

* fix: derive export scope from project id

(cherry picked from commit cf094aa)

Co-authored-by: lefarcen <935902669@qq.com>

Author: open-design-release-bot[bot]

apps/daemon/src/cli.ts

@@ -469,6 +469,8 @@ async function runAgent(args) {
 
 const EXPORT_STRING_FLAGS = new Set([
   'daemon-url', 'project', 'format', 'out', 'output', 'image-format', 'title', 'file',
+  // Backwards-compatible no-ops. Older scripts may still pass these, but
+  // export authority is derived from the project id by the daemon.
   'workspace', 'workspace-member',
 ]);
 const EXPORT_BOOLEAN_FLAGS = new Set(['help', 'h', 'json', 'deck', 'page', 'no-deck']);
@@ -493,8 +495,6 @@ Options:
   --deck                   Treat the artifact as a multi-slide deck
   --page, --no-deck        Treat the artifact as a normal scrollable page
   --title <title>          Title used for metadata / default filename
-  --workspace <id>        Explicit Workspace id for a bound project
-  --workspace-member <id> Explicit Workspace member id for a bound project
   --json                   Print a machine-readable result envelope
   --daemon-url <url>       Override daemon URL
 
@@ -541,7 +541,7 @@ async function runExport(args) {
   const token = process.env.OD_TOOL_TOKEN;
   const requestHeaders = token
     ? { authorization: `Bearer ${token}` }
-    : workspaceHeadersFromExplicitFlags(flags) ?? {};
+    : {};
   // Visual formats rasterize through the desktop screenshot renderer so the
   // CLI matches the UI exactly. In particular `pdf` uses `/export/pdf-image`
   // (one raster page per deck slide / per viewport for a page) — NOT the generic

apps/daemon/src/db.ts

@@ -2584,6 +2584,9 @@ export function conversationTurnIndexForRun(
 }
 
 export function upsertMessage(db: SqliteDb, conversationId: string, m: DbRow) {
+  const persistedEvents = Array.isArray(m.events)
+    ? compactAdjacentMessageAgentEvents(m.events)
+    : m.events;
   const existing = db
     .prepare(`SELECT position FROM messages WHERE id = ?`)
     .get(m.id) as DbRow | undefined;
@@ -2613,7 +2616,7 @@ export function upsertMessage(db: SqliteDb, conversationId: string, m: DbRow) {
       m.runStatus ?? null,
       normalizeResultDeliveryStateForStorage(m.resultDeliveryState),
       m.lastRunEventId ?? null,
-      m.events ? JSON.stringify(m.events) : null,
+      persistedEvents ? JSON.stringify(persistedEvents) : null,
       m.attachments ? JSON.stringify(m.attachments) : null,
       m.commentAttachments ? JSON.stringify(m.commentAttachments) : null,
       m.producedFiles ? JSON.stringify(m.producedFiles) : null,
@@ -2667,7 +2670,7 @@ export function upsertMessage(db: SqliteDb, conversationId: string, m: DbRow) {
       m.runStatus ?? null,
       normalizeResultDeliveryStateForStorage(m.resultDeliveryState),
       m.lastRunEventId ?? null,
-      m.events ? JSON.stringify(m.events) : null,
+      persistedEvents ? JSON.stringify(persistedEvents) : null,
       m.attachments ? JSON.stringify(m.attachments) : null,
       m.commentAttachments ? JSON.stringify(m.commentAttachments) : null,
       m.producedFiles ? JSON.stringify(m.producedFiles) : null,
@@ -2759,25 +2762,73 @@ export function appendMessageStatusEvent(db: SqliteDb, messageId: string, event:
   return next;
 }
 
-export function appendMessageAgentEvent(db: SqliteDb, messageId: string, event: DbRow) {
-  if (!event || typeof event !== 'object') return null;
-  const kind = typeof event.kind === 'string' ? event.kind : '';
-  if (!kind) return null;
+export function compactAdjacentMessageAgentEvents(
+  incomingEvents: readonly DbRow[],
+): DbRow[] {
+  const events: DbRow[] = [];
+  for (const event of incomingEvents) {
+    const kind = typeof event?.kind === 'string' ? event.kind : '';
+    const last = events[events.length - 1];
+    const isMergeableDelta =
+      (kind === 'text' || kind === 'thinking') && typeof event?.text === 'string';
+    if (isMergeableDelta && last?.kind === kind && typeof last.text === 'string') {
+      events[events.length - 1] = { ...last, text: last.text + event.text };
+    } else {
+      events.push(event);
+    }
+  }
+  return events;
+}
+
+export function appendMessageAgentEvents(
+  db: SqliteDb,
+  messageId: string,
+  incomingEvents: readonly DbRow[],
+): DbRow[] | null {
+  if (incomingEvents.length === 0) return null;
   const row = db
     .prepare(`SELECT content, events_json AS eventsJson FROM messages WHERE id = ?`)
     .get(messageId) as DbRow | undefined;
   if (!row) return null;
   const parsed = parseJsonOrUndef(row.eventsJson);
-  const events = Array.isArray(parsed) ? parsed : [];
-  const last = events[events.length - 1];
-  if (last && JSON.stringify(last) === JSON.stringify(event)) {
-    return events;
+  const parsedEvents = Array.isArray(parsed) ? parsed : [];
+  const events = compactAdjacentMessageAgentEvents(parsedEvents);
+  let textDelta = '';
+  let changed = events.length !== parsedEvents.length;
+
+  for (const event of incomingEvents) {
+    if (!event || typeof event !== 'object') continue;
+    const kind = typeof event.kind === 'string' ? event.kind : '';
+    if (!kind) continue;
+    const last = events[events.length - 1];
+    const isMergeableDelta =
+      (kind === 'text' || kind === 'thinking') && typeof event.text === 'string';
+    if (isMergeableDelta && last?.kind === kind && typeof last.text === 'string') {
+      last.text += event.text;
+      if (kind === 'text') textDelta += event.text;
+      changed = changed || event.text.length > 0;
+      continue;
+    }
+    if (!isMergeableDelta && last && JSON.stringify(last) === JSON.stringify(event)) {
+      continue;
+    }
+    events.push(event);
+    if (kind === 'text' && typeof event.text === 'string') textDelta += event.text;
+    changed = true;
   }
-  const next = [...events, event];
-  const textDelta = kind === 'text' && typeof event.text === 'string' ? event.text : '';
+
+  if (!changed) return events;
   db.prepare(`UPDATE messages SET content = COALESCE(content, '') || ?, events_json = ? WHERE id = ?`)
-    .run(textDelta, JSON.stringify(next), messageId);
-  return next;
+    .run(textDelta, JSON.stringify(events), messageId);
+  return events;
+}
+
+export function appendMessageAgentEvent(
+  db: SqliteDb,
+  messageId: string,
+  event: DbRow,
+): DbRow[] | null {
+  return appendMessageAgentEvents(db, messageId, [event]);
 }
 
 export function deleteMessage(db: SqliteDb, id: string) {

apps/daemon/src/import-export-routes.ts

@@ -602,7 +602,11 @@ export function registerProjectExportRoutes(app: Express, ctx: RegisterProjectEx
   async function authorizeExportRead(
     req: any,
     res: any,
-    options: { allowNavigationQuery?: boolean; toolEndpoint?: string } = {},
+    options: {
+      allowNavigationQuery?: boolean;
+      deriveWorkspaceFromProject?: boolean;
+      toolEndpoint?: string;
+    } = {},
   ): Promise<AuthorizedExportRead | null> {
     const authorization = req.get('authorization');
     if (
@@ -627,6 +631,14 @@ export function registerProjectExportRoutes(app: Express, ctx: RegisterProjectEx
       );
       return authority ? { previewWorkspace: authority.workspace } : null;
     }
+    if (options.deriveWorkspaceFromProject) {
+      const authority = await ctx.authorizeProjectToolRequest(
+        res,
+        req.params.id,
+        { mode: 'read' },
+      );
+      return authority ? { previewWorkspace: authority.workspace } : null;
+    }
     const authorized = await ctx.authorizeProjectRequest(
       req,
       res,
@@ -1392,7 +1404,10 @@ export function registerProjectExportRoutes(app: Express, ctx: RegisterProjectEx
   // PNG and assemble a one-image-per-slide .pptx. Replaces the old "send a prompt
   // to the agent and hope it runs python-pptx" path with a deterministic export.
   app.post('/api/projects/:id/export/pptx', async (req, res) => {
-    const authority = await authorizeExportRead(req, res, { toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT });
+    const authority = await authorizeExportRead(req, res, {
+      deriveWorkspaceFromProject: true,
+      toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT,
+    });
     if (!authority) return;
     await handleScreenshotExport(res, 'pptx', req.params.id, { authority, body: req.body });
   });
@@ -1401,7 +1416,10 @@ export function registerProjectExportRoutes(app: Express, ctx: RegisterProjectEx
   // The print-ready vector PDF stays on POST /export/pdf; this is the "exactly
   // what you see" counterpart that shares the slide renderer with PPTX.
   app.post('/api/projects/:id/export/pdf-image', async (req, res) => {
-    const authority = await authorizeExportRead(req, res, { toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT });
+    const authority = await authorizeExportRead(req, res, {
+      deriveWorkspaceFromProject: true,
+      toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT,
+    });
     if (!authority) return;
     await handleScreenshotExport(res, 'pdf', req.params.id, { authority, body: req.body });
   });
@@ -1411,7 +1429,10 @@ export function registerProjectExportRoutes(app: Express, ctx: RegisterProjectEx
   // the whole document at natural size. Viewport-independent — unlike the
   // host-compositor snapshot, the size never depends on the preview pane.
   app.post('/api/projects/:id/export/image', async (req, res) => {
-    const authority = await authorizeExportRead(req, res, { toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT });
+    const authority = await authorizeExportRead(req, res, {
+      deriveWorkspaceFromProject: true,
+      toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT,
+    });
     if (!authority) return;
     await handleScreenshotExport(res, 'image', req.params.id, { authority, body: req.body });
   });
@@ -1420,7 +1441,10 @@ export function registerProjectExportRoutes(app: Express, ctx: RegisterProjectEx
   // embedded by the daemon. Remote HTTP(S) dependencies remain external and
   // are listed in a machine-readable manifest inside the output.
   app.post('/api/projects/:id/export/html', async (req, res) => {
-    const authority = await authorizeExportRead(req, res, { toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT });
+    const authority = await authorizeExportRead(req, res, {
+      deriveWorkspaceFromProject: true,
+      toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT,
+    });
     if (!authority) return;
     await handleStandaloneHtmlExport(res, req.params.id, req.body);
   });
@@ -1438,7 +1462,10 @@ export function registerProjectExportRoutes(app: Express, ctx: RegisterProjectEx
     if (!isExportFormat(format)) {
       return sendApiError(res, 400, 'BAD_REQUEST', 'invalid export format');
     }
-    const authority = await authorizeExportRead(req, res, { toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT });
+    const authority = await authorizeExportRead(req, res, {
+      deriveWorkspaceFromProject: true,
+      toolEndpoint: PROJECT_EXPORT_TOOL_ENDPOINT,
+    });
     if (!authority) return;
     if (format === 'html') {
       return handleStandaloneHtmlExport(res, req.params.id, {

apps/daemon/src/routes/project/conversations.ts

@@ -10,6 +10,7 @@ import { TERMINAL_RUN_STATUSES } from '../../runtimes/runs.js';
 import { registerProjectCommentRoutes } from './comments.js';
 import { cancelRunsOwnedBy } from './cancel-owned-runs.js';
 import {
+  compactAdjacentMessageAgentEvents,
   deleteConversationAndRepairTeamCommentAnchor,
   isProjectCommentAnchorConversationId,
 } from '../../db.js';
@@ -548,8 +549,11 @@ export function registerProjectConversationRoutes(app: Express, ctx: RegisterPro
     if (existing === null && getMessage(db, req.params.mid) !== null) {
       return res.status(404).json({ error: 'message not found' });
     }
+    const normalizedMessage = Array.isArray(m.events)
+      ? { ...m, events: compactAdjacentMessageAgentEvents(m.events) }
+      : m;
     const saved = upsertMessage(db, req.params.cid, {
-      ...mergeMessageWriteForDaemonBacked(existing, m),
+      ...mergeMessageWriteForDaemonBacked(existing, normalizedMessage),
       id: req.params.mid,
     });
     // Bump the parent project's updatedAt so the project list re-orders.

apps/daemon/src/runtimes/chat-run-messages.ts

@@ -1,7 +1,7 @@
 import type Database from 'better-sqlite3';
 import type { PersistedAgentEvent } from '@open-design/contracts';
 import {
-  appendMessageAgentEvent,
+  appendMessageAgentEvents,
   upsertMessage,
 } from '../db.js';
 
@@ -22,6 +22,18 @@ type ChatRunMessageState = {
   failureDetail?: string | null;
 };
 
+type PendingMessageEvents = {
+  db: SqliteDb;
+  messageId: string;
+  events: PersistedAgentEvent[];
+  bytes: number;
+  timer: ReturnType<typeof setTimeout> | null;
+};
+
+export const RUN_MESSAGE_EVENT_FLUSH_INTERVAL_MS = 250;
+const RUN_MESSAGE_EVENT_FLUSH_BYTES = 64 * 1024;
+const pendingMessageEvents = new WeakMap<ChatRunMessageState, PendingMessageEvents>();
+
 function isRecord(value: unknown): value is Record<string, unknown> {
   return Boolean(value && typeof value === 'object' && !Array.isArray(value));
 }
@@ -34,9 +46,67 @@ export function persistRunEventToAssistantMessage(
 ): void {
   if (!run.assistantMessageId) return;
   const persisted = runSseEventToPersistedAgentEvent(event, data);
-  if (!persisted) return;
+  if (!persisted) {
+    if (event === 'end' || event === 'close') flushRunMessageEvents(run);
+    return;
+  }
+
+  let pending = pendingMessageEvents.get(run);
+  if (pending && (pending.db !== db || pending.messageId !== run.assistantMessageId)) {
+    flushRunMessageEvents(run);
+    pending = undefined;
+  }
+  if (!pending) {
+    pending = {
+      db,
+      messageId: run.assistantMessageId,
+      events: [],
+      bytes: 0,
+      timer: null,
+    };
+    pendingMessageEvents.set(run, pending);
+  }
+  appendPendingMessageEvent(pending, persisted);
+
+  const isDelta = persisted.kind === 'text' || persisted.kind === 'thinking';
+  if (!isDelta || pending.bytes >= RUN_MESSAGE_EVENT_FLUSH_BYTES) {
+    flushRunMessageEvents(run);
+    return;
+  }
+  if (!pending.timer) {
+    pending.timer = setTimeout(() => {
+      flushRunMessageEvents(run);
+    }, RUN_MESSAGE_EVENT_FLUSH_INTERVAL_MS);
+    pending.timer.unref?.();
+  }
+}
+
+function appendPendingMessageEvent(
+  pending: PendingMessageEvents,
+  event: PersistedAgentEvent,
+): void {
+  const last = pending.events[pending.events.length - 1];
+  if (
+    (event.kind === 'text' || event.kind === 'thinking') &&
+    last?.kind === event.kind
+  ) {
+    last.text += event.text;
+  } else {
+    pending.events.push(event);
+  }
+  pending.bytes += event.kind === 'text' || event.kind === 'thinking'
+    ? event.text.length
+    : JSON.stringify(event).length;
+}
+
+export function flushRunMessageEvents(run: ChatRunMessageState): void {
+  const pending = pendingMessageEvents.get(run);
+  if (!pending) return;
+  pendingMessageEvents.delete(run);
+  if (pending.timer) clearTimeout(pending.timer);
+  if (pending.events.length === 0) return;
   try {
-    appendMessageAgentEvent(db, run.assistantMessageId, persisted);
+    appendMessageAgentEvents(pending.db, pending.messageId, pending.events);
   } catch (err) {
     console.warn('[runs] message event persistence failed', err);
   }