fix(scheduler): 派发前撤回的提示直接删除，并定期清掉历史软隐藏

apps/desktop/src/main/localDb/client/WorkerThreadTransport.ts

@@ -499,6 +499,8 @@ function dispatchTransaction(readyDb, payload) {
       return messageLeaseMutate(readyDb, request.args);
     case 'message.rewindUserAfterClear':
       return messageRewindUserAfterClear(readyDb, request.args);
+    case 'message.purgeRewoundSchedulerPrompts':
+      return messagePurgeRewoundSchedulerPrompts(readyDb, request.args);
     case 'message.delete':
       return messageDelete(readyDb, request.args);
     case 'im.deleteBindings':
@@ -899,11 +901,19 @@ function messageRewindUserAfterClear(readyDb, args) {
   const sessionId = expectString(payload.sessionId, 'sessionId');
   const clientId = expectString(payload.clientId, 'clientId');
   const rewoundAt = expectNumber(payload.rewoundAt, 'rewoundAt');
+  const discard = payload.discard === true;
   return readyDb.transaction(() => {
     const changes = readyDb.prepare(
       "UPDATE messages SET rewind_at = ? WHERE session_id = ? AND client_id = ? AND role = 'user' AND rewind_at IS NULL",
     ).run(rewoundAt, sessionId, clientId).changes;
-    if (changes > 0) {
+    if (changes > 0 && discard) {
+      readyDb.prepare(
+        "DELETE FROM messages WHERE session_id = ? AND client_id = ? AND role = 'user' AND rewind_at = ?",
+      ).run(sessionId, clientId, rewoundAt);
+      readyDb.prepare(
+        'UPDATE sessions SET list_preview = NULL, list_preview_role = NULL, list_message_count = NULL WHERE id = ?',
+      ).run(sessionId);
+    } else if (changes > 0) {
       readyDb.prepare(
         'UPDATE sessions SET list_preview = NULL, list_preview_role = NULL WHERE id = ?',
       ).run(sessionId);
@@ -912,7 +922,47 @@ function messageRewindUserAfterClear(readyDb, args) {
   })();
 }
 
-// ⚠️ 与 worker/opHandlers/tx.ts 的同名事务保持一致。
+// Mirrors opHandlers/tx.ts: only provably undelivered scheduler prompts (no executionModel,
+// lone rewind_at in the session) are deleted, scanning idx_messages_rewind_at in bounded windows.
+const PURGE_REWOUND_SCAN_WINDOW = 500;
+
+function messagePurgeRewoundSchedulerPrompts(readyDb, args) {
+  const payload = asRecord(args, 'message.purgeRewoundSchedulerPrompts args');
+  const limit = expectNumber(payload.limit, 'limit');
+  if (!Number.isSafeInteger(limit) || limit < 1 || limit > 200) {
+    throw Object.assign(new Error('limit must be an integer from 1 to 200'), { code: 'INVALID_ARGS' });
+  }
+  const cursor = payload.cursor === undefined ? 0 : expectNumber(payload.cursor, 'cursor');
+  const windowSql = 'SELECT rewind_at AS rewindAt FROM messages WHERE rewind_at > ? ORDER BY rewind_at LIMIT ?';
+  const selectSql = (
+    "SELECT m.id, m.session_id AS sessionId FROM messages m "
+    + "WHERE m.rewind_at > ? AND m.rewind_at <= ? AND m.role = 'user' AND json_valid(m.agent_meta) "
+    + "AND json_extract(m.agent_meta, '$.origin.kind') = 'scheduler' "
+    + "AND json_extract(m.agent_meta, '$.executionModel') IS NULL "
+    + 'AND NOT EXISTS (SELECT 1 FROM messages o WHERE o.rewind_at = m.rewind_at AND o.session_id = m.session_id AND o.id <> m.id) '
+    + 'ORDER BY m.rewind_at LIMIT ?'
+  );
+  return readyDb.transaction(() => {
+    const window = readyDb.prepare(windowSql).all(cursor, PURGE_REWOUND_SCAN_WINDOW);
+    if (window.length === 0) return { deleted: 0, sessionIds: [], cursor: null };
+    const windowEnd = window[window.length - 1].rewindAt;
+    const exhausted = window.length < PURGE_REWOUND_SCAN_WINDOW;
+    const rows = readyDb.prepare(selectSql).all(cursor, windowEnd, limit);
+    const nextCursor = rows.length === limit ? cursor : exhausted ? null : windowEnd;
+    if (rows.length === 0) return { deleted: 0, sessionIds: [], cursor: nextCursor };
+    const remove = readyDb.prepare(
+      "DELETE FROM messages WHERE id = ? AND role = 'user' AND rewind_at IS NOT NULL",
+    );
+    for (const row of rows) remove.run(row.id);
+    const sessionIds = [...new Set(rows.map((row) => row.sessionId))];
+    const clearPreview = readyDb.prepare(
+      'UPDATE sessions SET list_preview = NULL, list_preview_role = NULL, list_message_count = NULL WHERE id = ?',
+    );
+    for (const sessionId of sessionIds) clearPreview.run(sessionId);
+    return { deleted: rows.length, sessionIds, cursor: nextCursor };
+  })();
+}
+
 function messageDelete(readyDb, args) {
   const payload = asRecord(args, 'message.delete args');
   const sessionId = expectString(payload.sessionId, 'sessionId');

apps/desktop/src/main/localDb/client/tx/types.ts

@@ -38,6 +38,7 @@ export type DbTxName =
   | 'message.updateContent'
   | 'message.leaseMutate'
   | 'message.rewindUserAfterClear'
+  | 'message.purgeRewoundSchedulerPrompts'
   | 'message.delete'
   | 'im.deleteBindings'
   | 'im.replaceBinding'
@@ -445,6 +446,20 @@ export interface MessageRewindUserAfterClearArgs {
   sessionId: string;
   clientId: string;
   rewoundAt: number;
+  discard?: boolean;
+}
+
+export interface MessagePurgeRewoundSchedulerPromptsArgs {
+  limit: number;
+  /** 上一批返回的 cursor（窗口末尾 rewind_at）；省略从头扫。 */
+  cursor?: number;
+}
+
+export interface MessagePurgeRewoundSchedulerPromptsResult {
+  deleted: number;
+  sessionIds: string[];
+  /** 下一批的起点；null 表示所有隐藏行已扫完。 */
+  cursor: number | null;
 }
 
 /**
@@ -1405,6 +1420,7 @@ export type DbTxArgsByName = {
   'message.updateContent': MessageUpdateContentArgs;
   'message.leaseMutate': MessageLeaseMutateArgs;
   'message.rewindUserAfterClear': MessageRewindUserAfterClearArgs;
+  'message.purgeRewoundSchedulerPrompts': MessagePurgeRewoundSchedulerPromptsArgs;
   'message.delete': MessageDeleteArgs;
   'im.deleteBindings': ImDeleteBindingsArgs;
   'im.replaceBinding': ImReplaceBindingArgs;
@@ -1506,6 +1522,7 @@ export type DbTxResultByName = {
   'message.updateContent': { changes: number };
   'message.leaseMutate': { changes: number };
   'message.rewindUserAfterClear': { changes: number };
+  'message.purgeRewoundSchedulerPrompts': MessagePurgeRewoundSchedulerPromptsResult;
   'message.delete': MessageDeleteResult;
   'im.deleteBindings': undefined;
   'im.replaceBinding': undefined;

apps/desktop/src/main/localDb/ipc/__tests__/messagesClearRace.test.ts

@@ -54,7 +54,7 @@ vi.mock('../../client/current', () => ({
   getDbClient: () => h.client,
 }));
 
-import { createMessage, rewindPersistedUserMessageAfterClear } from '../messages';
+import { createMessage, purgeRewoundSchedulerPrompts, rewindPersistedUserMessageAfterClear } from '../messages';
 
 function createDb(): Database.Database {
   const sqlite = new Database(':memory:');
@@ -184,6 +184,70 @@ describe('message persistence clear boundary', () => {
     expect(h.broadcast).not.toHaveBeenCalled();
   });
 
+  it('discards an undispatched scheduled row instead of keeping a tombstone', async () => {
+    sqlite.exec("CREATE TABLE embedding_jobs (source TEXT NOT NULL, source_id TEXT NOT NULL); CREATE TABLE chat_messages_vec_v1 (embedding BLOB); CREATE TRIGGER trg_chat_rewind_clean_vec AFTER UPDATE OF rewind_at ON messages WHEN NEW.rewind_at IS NOT NULL AND OLD.rewind_at IS NULL BEGIN DELETE FROM chat_messages_vec_v1 WHERE rowid IN (SELECT rowid FROM embedding_jobs WHERE source = 'chat' AND source_id = NEW.id); DELETE FROM embedding_jobs WHERE source = 'chat' AND source_id = NEW.id; END;");
+    const insert = sqlite.prepare("INSERT INTO messages (id, client_id, session_id, role, content, created_at, rewind_at) VALUES (?, ?, ?, 'user', ?, ?, NULL)");
+    insert.run('row-owner', 'owner-1', 's1', 'owner message', 100);
+    insert.run('row-scheduled', 'scheduled-1', 's1', 'check tasks', 200);
+    sqlite.prepare("INSERT INTO embedding_jobs (source, source_id) VALUES ('chat', ?)").run('row-owner');
+    sqlite.prepare("INSERT INTO embedding_jobs (source, source_id) VALUES ('chat', ?)").run('row-scheduled');
+    sqlite.prepare("INSERT INTO chat_messages_vec_v1 (rowid, embedding) VALUES (?, X'00')").run(1);
+    sqlite.prepare("INSERT INTO chat_messages_vec_v1 (rowid, embedding) VALUES (?, X'00')").run(2);
+    sqlite.prepare('UPDATE sessions SET list_preview = ?, list_preview_role = ?, list_message_count = ? WHERE id = ?').run('check tasks', 'user', 2, 's1');
+    await rewindPersistedUserMessageAfterClear('s1', 'scheduled-1', { discard: true });
+    expect(sqlite.prepare('SELECT client_id AS clientId FROM messages').all()).toEqual([{ clientId: 'owner-1' }]);
+    expect(sqlite.prepare('SELECT source_id AS sourceId FROM embedding_jobs').all()).toEqual([{ sourceId: 'row-owner' }]);
+    expect(sqlite.prepare('SELECT rowid FROM chat_messages_vec_v1').all()).toEqual([{ rowid: 1 }]);
+    expect(sqlite.prepare('SELECT list_preview AS preview, list_message_count AS count FROM sessions WHERE id = ?').get('s1')).toEqual({ preview: null, count: null });
+    expect(h.broadcast).toHaveBeenCalledWith('local-db:messages:deleted', {
+      sessionId: 's1', clientId: 'scheduled-1', clientIds: ['scheduled-1'],
+    });
+  });
+
+  it('purges only undelivered hidden scheduler prompts and keeps manually rewound history', async () => {
+    const insert = sqlite.prepare("INSERT INTO messages (id, client_id, session_id, role, content, agent_meta, created_at, rewind_at) VALUES (?, ?, 's1', ?, ?, ?, ?, ?)");
+    const schedulerMeta = "{\"origin\": {\"kind\": \"scheduler\"}}";
+    insert.run('visible-scheduler', 'sched-live', 'user', 'live', schedulerMeta, 1, null);
+    // Never reached the vendor: the only row stamped with rewind_at=50.
+    insert.run('hidden-scheduler', 'sched-hidden', 'user', 'hidden', schedulerMeta, 2, 50);
+    insert.run('hidden-user', 'user-hidden', 'user', 'hello', null, 3, 60);
+    // Delivered, then rewound by the user: rewind.commit stamps the prompt and its reply alike.
+    insert.run('manual-prompt', 'sched-manual', 'user', 'manual', schedulerMeta, 4, 70);
+    insert.run('manual-reply', 'reply-manual', 'assistant', 'done', null, 5, 70);
+    // Delivered: dispatch recorded the execution model, even though the row is now alone.
+    insert.run('dispatched-prompt', 'sched-dispatched', 'user', 'dispatched', "{\"origin\": {\"kind\": \"scheduler\"}, \"executionModel\": {\"model\": \"m\"}}", 6, 80);
+    sqlite.prepare('UPDATE sessions SET list_preview = ?, list_message_count = ? WHERE id = ?').run('hidden', 6, 's1');
+    await expect(purgeRewoundSchedulerPrompts(100)).resolves.toEqual({ deleted: 1, sessionIds: ['s1'], cursor: null });
+    expect(sqlite.prepare('SELECT id FROM messages ORDER BY created_at').all()).toEqual([
+      { id: 'visible-scheduler' },
+      { id: 'hidden-user' },
+      { id: 'manual-prompt' },
+      { id: 'manual-reply' },
+      { id: 'dispatched-prompt' },
+    ]);
+    expect(sqlite.prepare('SELECT list_preview AS preview, list_message_count AS count FROM sessions WHERE id = ?').get('s1')).toEqual({ preview: null, count: null });
+    await expect(purgeRewoundSchedulerPrompts(100)).resolves.toEqual({ deleted: 0, sessionIds: [], cursor: null });
+  });
+
+  it('bounds each purge to a window of the rewind index and hands back a cursor', async () => {
+    const insert = sqlite.prepare("INSERT INTO messages (id, client_id, session_id, role, content, agent_meta, created_at, rewind_at) VALUES (?, ?, 's1', 'user', 'x', ?, ?, ?)");
+    const schedulerMeta = "{\"origin\": {\"kind\": \"scheduler\"}}";
+    const total = 620;
+    for (let i = 0; i < total; i += 1) {
+      // Mostly ordinary tombstones; two scheduler prompts sit past the first 500-row window.
+      const meta = i === 510 || i === 600 ? schedulerMeta : null;
+      insert.run(`t-${i}`, `c-${i}`, meta, i, 1_000 + i);
+    }
+    const first = await purgeRewoundSchedulerPrompts(100);
+    expect(first).toEqual({ deleted: 0, sessionIds: [], cursor: 1_499 });
+    const second = await purgeRewoundSchedulerPrompts(100, first.cursor!);
+    expect(second).toEqual({ deleted: 2, sessionIds: ['s1'], cursor: null });
+    expect(sqlite.prepare('SELECT COUNT(*) AS n FROM messages').get()).toEqual({ n: total - 2 });
+    // A page filled to `limit` re-reads the same window before moving on.
+    for (let i = 0; i < 3; i += 1) insert.run(`s-${i}`, `sc-${i}`, schedulerMeta, 2_000 + i, 3_000 + i);
+    await expect(purgeRewoundSchedulerPrompts(2, 2_000)).resolves.toEqual({ deleted: 2, sessionIds: ['s1'], cursor: 2_000 });
+    await expect(purgeRewoundSchedulerPrompts(2, 2_000)).resolves.toEqual({ deleted: 1, sessionIds: ['s1'], cursor: null });
+  });
   it('reconciles session attachment refs for a rewound user row', async () => {
     const { collectCindyMediaHashes } = await import('../../../cindy-media/chatAttachments');
     const { removeSessionAttachmentRefIfUnreferencedByLiveMessage } = await import(

apps/desktop/src/main/localDb/ipc/messages.ts

@@ -1299,6 +1299,7 @@ export function broadcastMessageDeleted(
 export async function rewindPersistedUserMessageAfterClear(
   sessionId: string,
   clientId: string,
+  options?: { discard?: boolean },
 ): Promise<void> {
   const ownerScope = captureOwnerBroadcastScope();
   if (!isOwnerBroadcastScopeCurrent(ownerScope)) return;
@@ -1324,6 +1325,7 @@ export async function rewindPersistedUserMessageAfterClear(
     sessionId,
     clientId,
     rewoundAt,
+    ...(options?.discard ? { discard: true } : {}),
   });
   if (!isOwnerBroadcastScopeCurrent(ownerScope)) return;
   if (updated.changes === 0) return;
@@ -1360,6 +1362,14 @@ export async function rewindPersistedUserMessageAfterClear(
   broadcastMessageDeleted({ sessionId, clientId, clientIds: [clientId] }, ownerScope);
 }
 
+export async function purgeRewoundSchedulerPrompts(
+  limit = 100,
+  cursor = 0,
+): Promise<{ deleted: number; sessionIds: string[]; cursor: number | null }> {
+  const ownerScope = captureOwnerBroadcastScope();
+  if (!isOwnerBroadcastScopeCurrent(ownerScope)) return { deleted: 0, sessionIds: [], cursor: null };
+  return getDbClient().tx('message.purgeRewoundSchedulerPrompts', { limit, cursor });
+}
 export async function dismissErrorMessage(
   sessionId: string,
   clientId: string,

apps/desktop/src/main/localDb/worker/opHandlers/tx.ts

@@ -139,6 +139,8 @@ function dispatchTransaction(db: Database.Database, args: unknown): unknown {
       return messageLeaseMutate(db, txArgs);
     case 'message.rewindUserAfterClear':
       return messageRewindUserAfterClear(db, txArgs);
+    case 'message.purgeRewoundSchedulerPrompts':
+      return messagePurgeRewoundSchedulerPrompts(db, txArgs);
     case 'message.delete':
       return messageDelete(db, txArgs);
     case 'im.deleteBindings':
@@ -1948,18 +1950,21 @@ function messageRewindUserAfterClear(
   const sessionId = expectString(payload.sessionId, 'sessionId');
   const clientId = expectString(payload.clientId, 'clientId');
   const rewoundAt = expectNumber(payload.rewoundAt, 'rewoundAt');
+  const discard = payload.discard === true;
   const transaction = db.transaction(() => {
     const changes = db
       .prepare(
-        `UPDATE messages
-            SET rewind_at = ?
-          WHERE session_id = ?
-            AND client_id = ?
-            AND role = 'user'
-            AND rewind_at IS NULL`,
+        "UPDATE messages SET rewind_at = ? WHERE session_id = ? AND client_id = ? AND role = 'user' AND rewind_at IS NULL",
       )
       .run(rewoundAt, sessionId, clientId).changes;
-    if (changes > 0) {
+    if (changes > 0 && discard) {
+      db.prepare(
+        "DELETE FROM messages WHERE session_id = ? AND client_id = ? AND role = 'user' AND rewind_at = ?",
+      ).run(sessionId, clientId, rewoundAt);
+      db.prepare(
+        'UPDATE sessions SET list_preview = NULL, list_preview_role = NULL, list_message_count = NULL WHERE id = ?',
+      ).run(sessionId);
+    } else if (changes > 0) {
       db.prepare(
         'UPDATE sessions SET list_preview = NULL, list_preview_role = NULL WHERE id = ?',
       ).run(sessionId);
@@ -1969,6 +1974,56 @@ function messageRewindUserAfterClear(
   return transaction();
 }
 
+/** 每次最多沿 idx_messages_rewind_at 扫这么多条隐藏行，扫描量与命中数无关。 */
+const PURGE_REWOUND_SCAN_WINDOW = 500;
+
+/**
+ * 只删能证明从未送达模型的调度提示：没有派发时写入的 executionModel，且同会话里
+ * 没有别的行共用同一个 rewind_at。用户手动回退会给提示和其后所有消息盖同一个时间戳，
+ * 所以那种墓碑不会落单。cursor 是上一窗口的末尾 rewind_at，null 表示扫完。
+ */
+function messagePurgeRewoundSchedulerPrompts(
+  db: Database.Database,
+  args: unknown,
+): { deleted: number; sessionIds: string[]; cursor: number | null } {
+  const payload = asRecord(args, 'message.purgeRewoundSchedulerPrompts args');
+  const limit = expectNumber(payload.limit, 'limit');
+  if (!Number.isSafeInteger(limit) || limit < 1 || limit > 200) {
+    throw invalidArgs('limit must be an integer from 1 to 200');
+  }
+  const cursor = payload.cursor === undefined ? 0 : expectNumber(payload.cursor, 'cursor');
+  const windowSql = 'SELECT rewind_at AS rewindAt FROM messages WHERE rewind_at > ? ORDER BY rewind_at LIMIT ?';
+  const selectSql = (
+    "SELECT m.id, m.session_id AS sessionId FROM messages m "
+    + "WHERE m.rewind_at > ? AND m.rewind_at <= ? AND m.role = 'user' AND json_valid(m.agent_meta) "
+    + "AND json_extract(m.agent_meta, '$.origin.kind') = 'scheduler' "
+    + "AND json_extract(m.agent_meta, '$.executionModel') IS NULL "
+    + 'AND NOT EXISTS (SELECT 1 FROM messages o WHERE o.rewind_at = m.rewind_at AND o.session_id = m.session_id AND o.id <> m.id) '
+    + 'ORDER BY m.rewind_at LIMIT ?'
+  );
+  const transaction = db.transaction(() => {
+    const window = db.prepare(windowSql).all(cursor, PURGE_REWOUND_SCAN_WINDOW) as Array<{ rewindAt: number }>;
+    if (window.length === 0) return { deleted: 0, sessionIds: [] as string[], cursor: null };
+    const windowEnd = window[window.length - 1].rewindAt;
+    const exhausted = window.length < PURGE_REWOUND_SCAN_WINDOW;
+    const rows = db.prepare(selectSql).all(cursor, windowEnd, limit) as Array<{ id: string; sessionId: string }>;
+    // A full page may hide more matches in this window; rescan it before moving on.
+    const nextCursor = rows.length === limit ? cursor : exhausted ? null : windowEnd;
+    if (rows.length === 0) return { deleted: 0, sessionIds: [] as string[], cursor: nextCursor };
+    const remove = db.prepare(
+      "DELETE FROM messages WHERE id = ? AND role = 'user' AND rewind_at IS NOT NULL",
+    );
+    for (const row of rows) remove.run(row.id);
+    const sessionIds = [...new Set(rows.map((row) => row.sessionId))];
+    const clearPreview = db.prepare(
+      'UPDATE sessions SET list_preview = NULL, list_preview_role = NULL, list_message_count = NULL WHERE id = ?',
+    );
+    for (const sessionId of sessionIds) clearPreview.run(sessionId);
+    return { deleted: rows.length, sessionIds, cursor: nextCursor };
+  });
+  return transaction();
+}
+
 /** 一轮消息内容清除 + 原生上下文失效 + 隐藏重建标记，三者同成同败。 */
 function messageDelete(
   db: Database.Database,

apps/desktop/src/main/maker-ipc/__tests__/agent-input-coordinator.test.ts

@@ -12180,7 +12180,7 @@ describe('AgentInputCoordinator scheduler 排队心跳(review 反馈回归)', ()
       });
       h.coordinator.enqueue(sid, makeItem('scheduled-row', 'heartbeat', { origin: schedOrigin }));
       await flush();
-      expect(h.rewindPersistedUserMessageAfterClear).toHaveBeenCalledExactlyOnceWith(sid, 'scheduled-row');
+      expect(h.rewindPersistedUserMessageAfterClear).toHaveBeenCalledExactlyOnceWith(sid, 'scheduled-row', { discard: true });
       expect(h.persistTerminalSendError).not.toHaveBeenCalled();
       expect(latestProjection(h.projections).error).toBeNull();
       expect(latestProjection(h.projections).recovery).toBeNull();