From 6f9129b1fa409314d55b405002ecaaf1e4d7ba81 Mon Sep 17 00:00:00 2001
From: elfcv <cnxpfcv@gmail.com>
Date: Thu, 29 Oct 2026 16:26:30 +0800
Subject: [PATCH 1/3] =?UTF-8?q?fix(scheduler):=20=E6=B4=BE=E5=8F=91?=
 =?UTF-8?q?=E5=89=8D=E6=92=A4=E5=9B=9E=E7=9A=84=E6=8F=90=E7=A4=BA=E7=9B=B4?=
 =?UTF-8?q?=E6=8E=A5=E5=88=A0=E9=99=A4=EF=BC=8C=E5=B9=B6=E5=AE=9A=E6=9C=9F?=
 =?UTF-8?q?=E6=B8=85=E6=8E=89=E5=8E=86=E5=8F=B2=E8=BD=AF=E9=9A=90=E8=97=8F?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

例行任务在送达模型前被撤回时，只写 rewind_at 会让 messages 无界增长。先软隐藏以触发向量清理，再在同一事务删除该行。调度器启动后按批删除已经隐藏的 scheduler 行，不碰可见消息和其他回退墓碑。

Signed-off-by: elfcv <cnxpfcv@gmail.com>
---
 .../localDb/client/WorkerThreadTransport.ts   | 41 +++++++++++++-
 .../src/main/localDb/client/tx/types.ts       |  8 +++
 .../ipc/__tests__/messagesClearRace.test.ts   | 37 ++++++++++++-
 apps/desktop/src/main/localDb/ipc/messages.ts |  9 +++
 .../src/main/localDb/worker/opHandlers/tx.ts  | 51 ++++++++++++++---
 .../__tests__/agent-input-coordinator.test.ts |  2 +-
 .../main/maker-ipc/agent-input-coordinator.ts |  5 +-
 apps/desktop/src/main/maker-ipc/register.ts   |  8 +--
 .../rewoundSchedulerPromptCleanup.test.ts     | 27 +++++++++
 .../__tests__/runnerBackgroundTasks.test.ts   |  2 +
 .../__tests__/runnerModelSelection.test.ts    |  2 +-
 .../__tests__/runnerSilentNotify.test.ts      |  2 +
 apps/desktop/src/main/scheduler-host/index.ts |  3 +
 .../rewoundSchedulerPromptCleanup.ts          | 55 +++++++++++++++++++
 .../desktop/src/main/scheduler-host/runner.ts |  4 +-
 15 files changed, 236 insertions(+), 20 deletions(-)
 create mode 100644 apps/desktop/src/main/scheduler-host/__tests__/rewoundSchedulerPromptCleanup.test.ts
 create mode 100644 apps/desktop/src/main/scheduler-host/rewoundSchedulerPromptCleanup.ts

diff --git a/apps/desktop/src/main/localDb/client/WorkerThreadTransport.ts b/apps/desktop/src/main/localDb/client/WorkerThreadTransport.ts
index f341fb26e26..145dba1a546 100644
--- a/apps/desktop/src/main/localDb/client/WorkerThreadTransport.ts
+++ b/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,34 @@ function messageRewindUserAfterClear(readyDb, args) {
   })();
 }
 
-// ⚠️ 与 worker/opHandlers/tx.ts 的同名事务保持一致。
+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 selectSql = (
+    "SELECT id, session_id AS sessionId FROM messages "
+    + "WHERE role = 'user' AND rewind_at IS NOT NULL AND json_valid(agent_meta) "
+    + "AND json_extract(agent_meta, '$.origin.kind') = 'scheduler' "
+    + "ORDER BY rewind_at LIMIT ?"
+  );
+  return readyDb.transaction(() => {
+    const rows = readyDb.prepare(selectSql).all(limit);
+    if (rows.length === 0) return { deleted: 0, sessionIds: [] };
+    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 };
+  })();
+}
+
 function messageDelete(readyDb, args) {
   const payload = asRecord(args, 'message.delete args');
   const sessionId = expectString(payload.sessionId, 'sessionId');
diff --git a/apps/desktop/src/main/localDb/client/tx/types.ts b/apps/desktop/src/main/localDb/client/tx/types.ts
index 5d4b08acf64..a1124f62526 100644
--- a/apps/desktop/src/main/localDb/client/tx/types.ts
+++ b/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,11 @@ export interface MessageRewindUserAfterClearArgs {
   sessionId: string;
   clientId: string;
   rewoundAt: number;
+  discard?: boolean;
+}
+
+export interface MessagePurgeRewoundSchedulerPromptsArgs {
+  limit: number;
 }
 
 /**
@@ -1405,6 +1411,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 +1513,7 @@ export type DbTxResultByName = {
   'message.updateContent': { changes: number };
   'message.leaseMutate': { changes: number };
   'message.rewindUserAfterClear': { changes: number };
+  'message.purgeRewoundSchedulerPrompts': { deleted: number; sessionIds: string[] };
   'message.delete': MessageDeleteResult;
   'im.deleteBindings': undefined;
   'im.replaceBinding': undefined;
diff --git a/apps/desktop/src/main/localDb/ipc/__tests__/messagesClearRace.test.ts b/apps/desktop/src/main/localDb/ipc/__tests__/messagesClearRace.test.ts
index 950e4143064..d997c8b2e8b 100644
--- a/apps/desktop/src/main/localDb/ipc/__tests__/messagesClearRace.test.ts
+++ b/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,41 @@ 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 already hidden scheduler prompts and leaves other tombstones', async () => {
+    const insert = sqlite.prepare("INSERT INTO messages (id, client_id, session_id, role, content, agent_meta, created_at, rewind_at) VALUES (?, ?, 's1', 'user', ?, ?, ?, ?)");
+    const schedulerMeta = "{\"origin\": {\"kind\": \"scheduler\"}}";
+    insert.run('visible-scheduler', 'sched-live', 'live', schedulerMeta, 1, null);
+    insert.run('hidden-scheduler', 'sched-hidden', 'hidden', schedulerMeta, 2, 50);
+    insert.run('hidden-user', 'user-hidden', 'hello', null, 3, 60);
+    sqlite.prepare('UPDATE sessions SET list_preview = ?, list_message_count = ? WHERE id = ?').run('hidden', 3, 's1');
+    await expect(purgeRewoundSchedulerPrompts(100)).resolves.toEqual({ deleted: 1, sessionIds: ['s1'] });
+    expect(sqlite.prepare('SELECT id FROM messages ORDER BY created_at').all()).toEqual([
+      { id: 'visible-scheduler' },
+      { id: 'hidden-user' },
+    ]);
+    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: [] });
+  });
   it('reconciles session attachment refs for a rewound user row', async () => {
     const { collectCindyMediaHashes } = await import('../../../cindy-media/chatAttachments');
     const { removeSessionAttachmentRefIfUnreferencedByLiveMessage } = await import(
diff --git a/apps/desktop/src/main/localDb/ipc/messages.ts b/apps/desktop/src/main/localDb/ipc/messages.ts
index d76457951cc..0a9b075cd33 100644
--- a/apps/desktop/src/main/localDb/ipc/messages.ts
+++ b/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,13 @@ export async function rewindPersistedUserMessageAfterClear(
   broadcastMessageDeleted({ sessionId, clientId, clientIds: [clientId] }, ownerScope);
 }
 
+export async function purgeRewoundSchedulerPrompts(
+  limit = 100,
+): Promise<{ deleted: number; sessionIds: string[] }> {
+  const ownerScope = captureOwnerBroadcastScope();
+  if (!isOwnerBroadcastScopeCurrent(ownerScope)) return { deleted: 0, sessionIds: [] };
+  return getDbClient().tx('message.purgeRewoundSchedulerPrompts', { limit });
+}
 export async function dismissErrorMessage(
   sessionId: string,
   clientId: string,
diff --git a/apps/desktop/src/main/localDb/worker/opHandlers/tx.ts b/apps/desktop/src/main/localDb/worker/opHandlers/tx.ts
index f87d440bd4d..773b1d2b240 100644
--- a/apps/desktop/src/main/localDb/worker/opHandlers/tx.ts
+++ b/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,38 @@ function messageRewindUserAfterClear(
   return transaction();
 }
 
+function messagePurgeRewoundSchedulerPrompts(
+  db: Database.Database,
+  args: unknown,
+): { deleted: number; sessionIds: string[] } {
+  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 selectSql = (
+    "SELECT id, session_id AS sessionId FROM messages "
+    + "WHERE role = 'user' AND rewind_at IS NOT NULL AND json_valid(agent_meta) "
+    + "AND json_extract(agent_meta, '$.origin.kind') = 'scheduler' "
+    + "ORDER BY rewind_at LIMIT ?"
+  );
+  const transaction = db.transaction(() => {
+    const rows = db.prepare(selectSql).all(limit) as Array<{ id: string; sessionId: string }>;
+    if (rows.length === 0) return { deleted: 0, sessionIds: [] as string[] };
+    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 };
+  });
+  return transaction();
+}
+
 /** 一轮消息内容清除 + 原生上下文失效 + 隐藏重建标记，三者同成同败。 */
 function messageDelete(
   db: Database.Database,
diff --git a/apps/desktop/src/main/maker-ipc/__tests__/agent-input-coordinator.test.ts b/apps/desktop/src/main/maker-ipc/__tests__/agent-input-coordinator.test.ts
index 4f9fd844f51..cc140843878 100644
--- a/apps/desktop/src/main/maker-ipc/__tests__/agent-input-coordinator.test.ts
+++ b/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();
diff --git a/apps/desktop/src/main/maker-ipc/agent-input-coordinator.ts b/apps/desktop/src/main/maker-ipc/agent-input-coordinator.ts
index a49781c8e47..41b58384df0 100644
--- a/apps/desktop/src/main/maker-ipc/agent-input-coordinator.ts
+++ b/apps/desktop/src/main/maker-ipc/agent-input-coordinator.ts
@@ -395,7 +395,7 @@ export interface AgentInputCoordinatorDeps {
    */
   createUserMessage?: typeof createDbMessage;
   /** Hide a user row that was written after `/clear` won the persistence race. */
-  rewindPersistedUserMessageAfterClear?: (sessionId: string, clientId: string) => Promise<void>;
+  rewindPersistedUserMessageAfterClear?: (sessionId: string, clientId: string, options?: { discard?: boolean }) => Promise<void>;
   /**
    * interrupted-turn-resume:判断某条已派发 user 消息之后 agent 是否已产出内容
    * (assistant / tool_use / thinking 持久化行)。retryLastError 用它决定语义:
@@ -5471,6 +5471,7 @@ export class AgentInputCoordinator {
   /** Only proven preparation rejection can erase a scheduled continuation row.
    * The coordinator owns its clientId even when dispatch beats enqueue's return.
    * Unknown vendor outcomes and interactive messages keep their existing history.
+   * The row is deleted, not tombstoned: a deferring routine would otherwise leave one hidden copy per attempt.
    */
   private async rewindSchedulerPreparationFailure(
     sessionId: string,
@@ -5491,7 +5492,7 @@ export class AgentInputCoordinator {
         && (failure?.code === 'PRECONDITION_FAILED' || message.includes('[PRECONDITION_FAILED]')));
     if (!preparationRejected) return false;
     try {
-      await this.deps.rewindPersistedUserMessageAfterClear(sessionId, item.clientId);
+      await this.deps.rewindPersistedUserMessageAfterClear(sessionId, item.clientId, { discard: true });
       return true;
     } catch (err) {
       log.warn('scheduled preparation rollback failed; retaining failure presentation', {
diff --git a/apps/desktop/src/main/maker-ipc/register.ts b/apps/desktop/src/main/maker-ipc/register.ts
index 4066e16ed85..3cfe7783f8d 100644
--- a/apps/desktop/src/main/maker-ipc/register.ts
+++ b/apps/desktop/src/main/maker-ipc/register.ts
@@ -18619,9 +18619,9 @@ export function registerMakerIpc(maker: Maker, options: RegisterMakerIpcOptions)
       } satisfies CindyLearnInvocationGrant;
     },
     createDbMessage: createUserMessageDurably,
-    rewindPersistedUserMessageAfterClear: (sessionId, clientId) =>
+    rewindPersistedUserMessageAfterClear: (sessionId, clientId, options) =>
       enqueueDurableWrite(`user-rewind:${sessionId}:${clientId}`, () =>
-        rewindPersistedUserMessageAfterClear(sessionId, clientId),
+        rewindPersistedUserMessageAfterClear(sessionId, clientId, options),
       ),
     isClearBoundaryCurrent: (sessionId, expected, expectedGeneration) => {
       const coordinator = agentInputCoordinatorHolder;
@@ -20562,9 +20562,9 @@ export function registerMakerIpc(maker: Maker, options: RegisterMakerIpcOptions)
     // auto-resume bookkeeping as direct maker sends.  Keeping one writer also
     // preserves message ordering when a clear-race rewind follows the insert.
     createUserMessage: createUserMessageDurably,
-    rewindPersistedUserMessageAfterClear: (sessionId, clientId) =>
+    rewindPersistedUserMessageAfterClear: (sessionId, clientId, options) =>
       enqueueDurableWrite(`user-rewind:${sessionId}:${clientId}`, () =>
-        rewindPersistedUserMessageAfterClear(sessionId, clientId),
+        rewindPersistedUserMessageAfterClear(sessionId, clientId, options),
       ),
     resolveSessionReferences,
     refreshAgentReferencesBeforeDispatch: async (item) => {
diff --git a/apps/desktop/src/main/scheduler-host/__tests__/rewoundSchedulerPromptCleanup.test.ts b/apps/desktop/src/main/scheduler-host/__tests__/rewoundSchedulerPromptCleanup.test.ts
new file mode 100644
index 00000000000..37dd12e6c13
--- /dev/null
+++ b/apps/desktop/src/main/scheduler-host/__tests__/rewoundSchedulerPromptCleanup.test.ts
@@ -0,0 +1,27 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import {
+  startRewoundSchedulerPromptCleanup,
+  stopRewoundSchedulerPromptCleanup,
+} from '../rewoundSchedulerPromptCleanup.js';
+
+describe('rewound scheduler prompt cleanup', () => {
+  afterEach(() => {
+    stopRewoundSchedulerPromptCleanup();
+    vi.useRealTimers();
+  });
+
+  it('deletes hidden scheduler prompts in bounded batches and then stops', async () => {
+    vi.useFakeTimers();
+    const purge = vi
+      .fn()
+      .mockResolvedValueOnce({ deleted: 100 })
+      .mockResolvedValueOnce({ deleted: 40 })
+      .mockResolvedValue({ deleted: 0 });
+    startRewoundSchedulerPromptCleanup({ purge, firstDelayMs: 30_000, intervalMs: 600_000 });
+    await vi.advanceTimersByTimeAsync(30_000);
+    expect(purge).toHaveBeenCalledTimes(2);
+    expect(purge).toHaveBeenNthCalledWith(1, 100);
+    await vi.advanceTimersByTimeAsync(600_000);
+    expect(purge).toHaveBeenCalledTimes(3);
+  });
+});
diff --git a/apps/desktop/src/main/scheduler-host/__tests__/runnerBackgroundTasks.test.ts b/apps/desktop/src/main/scheduler-host/__tests__/runnerBackgroundTasks.test.ts
index 205f5dd285d..72d9f419b0d 100644
--- a/apps/desktop/src/main/scheduler-host/__tests__/runnerBackgroundTasks.test.ts
+++ b/apps/desktop/src/main/scheduler-host/__tests__/runnerBackgroundTasks.test.ts
@@ -400,6 +400,7 @@ describe('MakerScheduleRunner background subagent task tracking', () => {
       expect(mocks.createMessage).toHaveBeenCalledOnce();
       expect(mocks.rewindPersistedUserMessageAfterClear).toHaveBeenCalledExactlyOnceWith(
         h.session.id, mocks.createMessage.mock.calls[0][1].clientId,
+        { discard: true },
       );
       expect(notifier.notify).not.toHaveBeenCalled();
     },
@@ -454,6 +455,7 @@ describe('MakerScheduleRunner background subagent task tracking', () => {
       } else {
         expect(mocks.rewindPersistedUserMessageAfterClear).toHaveBeenCalledExactlyOnceWith(
           h.session.id, mocks.createMessage.mock.calls[0][1].clientId,
+        { discard: true },
         );
         expect(vendor).not.toHaveBeenCalled();
         expect(notifier.notify).not.toHaveBeenCalled();
diff --git a/apps/desktop/src/main/scheduler-host/__tests__/runnerModelSelection.test.ts b/apps/desktop/src/main/scheduler-host/__tests__/runnerModelSelection.test.ts
index 080dd6c23c8..46423d7ae78 100644
--- a/apps/desktop/src/main/scheduler-host/__tests__/runnerModelSelection.test.ts
+++ b/apps/desktop/src/main/scheduler-host/__tests__/runnerModelSelection.test.ts
@@ -475,7 +475,7 @@ describe('MakerScheduleRunner model selection', () => {
       const acceptedClientId = mocks.createMessage.mock.calls[0][1].clientId;
       expect(acceptedClientId).toEqual(expect.any(String));
       expect(mocks.rewindPersistedUserMessageAfterClear)
-        .toHaveBeenCalledExactlyOnceWith(h.session.id, acceptedClientId);
+        .toHaveBeenCalledExactlyOnceWith(h.session.id, acceptedClientId, { discard: true });
     });
 
     // Merge regression, unit: a legacy remote default must not mask an explicit
diff --git a/apps/desktop/src/main/scheduler-host/__tests__/runnerSilentNotify.test.ts b/apps/desktop/src/main/scheduler-host/__tests__/runnerSilentNotify.test.ts
index 24ba01f7f30..ac4ddede362 100644
--- a/apps/desktop/src/main/scheduler-host/__tests__/runnerSilentNotify.test.ts
+++ b/apps/desktop/src/main/scheduler-host/__tests__/runnerSilentNotify.test.ts
@@ -494,6 +494,7 @@ describe('MakerScheduleRunner silent-run notification skip', () => {
       expect(mocks.rewind).toHaveBeenCalledExactlyOnceWith(
         h.session.id,
         mocks.createMessage.mock.calls[0][1].clientId,
+        { discard: true },
       );
       expect(notifier.notify).not.toHaveBeenCalled();
     },
@@ -521,6 +522,7 @@ describe('MakerScheduleRunner silent-run notification skip', () => {
     expect(mocks.rewind).toHaveBeenCalledExactlyOnceWith(
       h.session.id,
       mocks.createMessage.mock.calls[0][1].clientId,
+        { discard: true },
     );
     expect(notifier.notify).not.toHaveBeenCalled();
   });
diff --git a/apps/desktop/src/main/scheduler-host/index.ts b/apps/desktop/src/main/scheduler-host/index.ts
index 7c0f6382516..868615e33bd 100644
--- a/apps/desktop/src/main/scheduler-host/index.ts
+++ b/apps/desktop/src/main/scheduler-host/index.ts
@@ -26,6 +26,7 @@ import type { FeishuIM } from '@cindy/im';
 import { dialogueWorkspaceRoots } from '../localDb/dialogueWorkspace';
 import { sessions } from '../localDb/schema.js';
 import { getDbClient } from '../localDb/client/current';
+import { startRewoundSchedulerPromptCleanup, stopRewoundSchedulerPromptCleanup } from './rewoundSchedulerPromptCleanup.js';
 import { isReviewSessionSource } from '../../shared/sessionSource.js';
 import { dbToMakerAgentKind } from '../../shared/agentKindConversion.js';
 import { assertScheduledHarnessSupported } from '../maker-ipc/scheduledModelSelection';
@@ -300,6 +301,7 @@ async function startSchedulerInternal(deps: StartSchedulerDeps): Promise<Schedul
   });
   _scheduler = scheduler;
   _promptRunner = promptRunner;
+  startRewoundSchedulerPromptCleanup({ logger: deps.logger });
   setSilencedRunProbe((runId) => scheduler.isRunSilenced(runId));
   void getRoutineEngine().catch((error) => deps.logger.warn?.('routine startup failed', { error: String(error) }));
   _loader = loader;
@@ -364,6 +366,7 @@ export function getProjectAutomationLoader(): ProjectAutomationLoader {
  * 当前 resetMaker（maker-host:131）也不会调本函数。
  */
 export async function resetScheduler(): Promise<void> {
+  stopRewoundSchedulerPromptCleanup();
   await stopRoutines();
   _startupGeneration++;
   const pendingStartup = _startupPromise;
diff --git a/apps/desktop/src/main/scheduler-host/rewoundSchedulerPromptCleanup.ts b/apps/desktop/src/main/scheduler-host/rewoundSchedulerPromptCleanup.ts
new file mode 100644
index 00000000000..8c9ba742a54
--- /dev/null
+++ b/apps/desktop/src/main/scheduler-host/rewoundSchedulerPromptCleanup.ts
@@ -0,0 +1,55 @@
+import { purgeRewoundSchedulerPrompts } from '../localDb/ipc/messages.js';
+
+const BATCH = 100;
+const MAX_BATCHES_PER_SWEEP = 5;
+export const REWOUND_SCHEDULER_PROMPT_CLEANUP_INTERVAL_MS = 10 * 60 * 1000;
+export const REWOUND_SCHEDULER_PROMPT_CLEANUP_FIRST_DELAY_MS = 30 * 1000;
+
+let stopCleanup: (() => void) | null = null;
+
+export function startRewoundSchedulerPromptCleanup(
+  deps: {
+    purge?: (limit: number) => Promise<{ deleted: number }>;
+    logger?: { warn?: (message: string, error?: unknown) => void };
+    firstDelayMs?: number;
+    intervalMs?: number;
+  } = {},
+): void {
+  stopRewoundSchedulerPromptCleanup();
+  const purge = deps.purge ?? ((limit: number) => purgeRewoundSchedulerPrompts(limit));
+  let stopped = false;
+  let running = false;
+  const sweep = async () => {
+    if (stopped || running) return;
+    running = true;
+    try {
+      for (let batch = 0; batch < MAX_BATCHES_PER_SWEEP; batch += 1) {
+        if (stopped) return;
+        const result = await purge(BATCH);
+        if ((result?.deleted ?? 0) < BATCH) return;
+      }
+    } catch (error) {
+      deps.logger?.warn?.('rewound scheduler prompt cleanup failed', error);
+    } finally {
+      running = false;
+    }
+  };
+  const first = setTimeout(() => {
+    void sweep();
+  }, deps.firstDelayMs ?? REWOUND_SCHEDULER_PROMPT_CLEANUP_FIRST_DELAY_MS);
+  first.unref?.();
+  const timer = setInterval(() => {
+    void sweep();
+  }, deps.intervalMs ?? REWOUND_SCHEDULER_PROMPT_CLEANUP_INTERVAL_MS);
+  timer.unref?.();
+  stopCleanup = () => {
+    stopped = true;
+    clearTimeout(first);
+    clearInterval(timer);
+  };
+}
+
+export function stopRewoundSchedulerPromptCleanup(): void {
+  stopCleanup?.();
+  stopCleanup = null;
+}
diff --git a/apps/desktop/src/main/scheduler-host/runner.ts b/apps/desktop/src/main/scheduler-host/runner.ts
index 61464d51f10..54340e9df99 100644
--- a/apps/desktop/src/main/scheduler-host/runner.ts
+++ b/apps/desktop/src/main/scheduler-host/runner.ts
@@ -2258,7 +2258,7 @@ export class MakerScheduleRunner implements ScheduleRunner {
       }
     } catch (err) {
       // A preparation guard proves this turn never reached the vendor. Reuse
-      // the exact-message soft rewind; do not roll back ambiguous send errors.
+      // the exact message. A deferral retries with a new clientId, so no tombstone is kept.
       if (
         (err instanceof RoutineDispatchDeferredError ||
           err instanceof QueuedRouteDisabledError ||
@@ -2266,7 +2266,7 @@ export class MakerScheduleRunner implements ScheduleRunner {
         acceptedMessageClientId
       ) {
         try {
-          await rewindPersistedUserMessageAfterClear(session.id, acceptedMessageClientId);
+          await rewindPersistedUserMessageAfterClear(session.id, acceptedMessageClientId, { discard: true });
         } catch (rollbackError) {
           waiter.stopListening();
           ctx.signal.removeEventListener('abort', onAbort);

From 838e0db6b40d47887298060e702f620ccd396bcf Mon Sep 17 00:00:00 2001
From: elfcv <cnxpfcv@gmail.com>
Date: Fri, 30 Oct 2026 08:28:51 +0800
Subject: [PATCH 2/3] fix(scheduler): keep clear-race rewind on the
 two-argument type

The direct-send rewind callback does not accept discard. Passing options failed typecheck and must not delete a user message.

Signed-off-by: elfcv <cnxpfcv@gmail.com>
---
 apps/desktop/src/main/maker-ipc/register.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/apps/desktop/src/main/maker-ipc/register.ts b/apps/desktop/src/main/maker-ipc/register.ts
index 3cfe7783f8d..eea98e59bd8 100644
--- a/apps/desktop/src/main/maker-ipc/register.ts
+++ b/apps/desktop/src/main/maker-ipc/register.ts
@@ -18619,9 +18619,9 @@ export function registerMakerIpc(maker: Maker, options: RegisterMakerIpcOptions)
       } satisfies CindyLearnInvocationGrant;
     },
     createDbMessage: createUserMessageDurably,
-    rewindPersistedUserMessageAfterClear: (sessionId, clientId, options) =>
+    rewindPersistedUserMessageAfterClear: (sessionId, clientId) =>
       enqueueDurableWrite(`user-rewind:${sessionId}:${clientId}`, () =>
-        rewindPersistedUserMessageAfterClear(sessionId, clientId, options),
+        rewindPersistedUserMessageAfterClear(sessionId, clientId),
       ),
     isClearBoundaryCurrent: (sessionId, expected, expectedGeneration) => {
       const coordinator = agentInputCoordinatorHolder;

From 2966bcdfa7d7d35c4307beb5523a01dc83da75a6 Mon Sep 17 00:00:00 2001
From: elfcv <cnxpfcv@gmail.com>
Date: Mon, 2 Nov 2026 03:28:12 +0800
Subject: [PATCH 3/3] fix(scheduler): only purge provably undelivered prompts,
 in bounded windows

The background purge matched every hidden scheduler prompt. A manual
rewind of a delivered prompt leaves the same shape (rewind_at set,
agent_meta kept), so that history would have been deleted too. Now a row
is only removed when nothing proves delivery: no executionModel was
recorded at dispatch and no other row in the session shares its
rewind_at (rewind.commit stamps the prompt and everything after it with
one timestamp, so a manual rewind never leaves a lone tombstone).

Each purge also walks idx_messages_rewind_at in 500-row windows and
returns a cursor; the sweeper carries it across batches and restarts
from zero on the next sweep. The scan cost per call no longer depends on
how many ordinary tombstones sit in the database.

Signed-off-by: elfcv <cnxpfcv@gmail.com>
---
 .../localDb/client/WorkerThreadTransport.ts   | 27 ++++++++---
 .../src/main/localDb/client/tx/types.ts       | 11 ++++-
 .../ipc/__tests__/messagesClearRace.test.ts   | 45 +++++++++++++++----
 apps/desktop/src/main/localDb/ipc/messages.ts |  7 +--
 .../src/main/localDb/worker/opHandlers/tx.ts  | 34 ++++++++++----
 .../rewoundSchedulerPromptCleanup.test.ts     | 28 +++++++++---
 .../rewoundSchedulerPromptCleanup.ts          | 20 +++++++--
 7 files changed, 134 insertions(+), 38 deletions(-)

diff --git a/apps/desktop/src/main/localDb/client/WorkerThreadTransport.ts b/apps/desktop/src/main/localDb/client/WorkerThreadTransport.ts
index 145dba1a546..f2750cc841c 100644
--- a/apps/desktop/src/main/localDb/client/WorkerThreadTransport.ts
+++ b/apps/desktop/src/main/localDb/client/WorkerThreadTransport.ts
@@ -922,21 +922,34 @@ function messageRewindUserAfterClear(readyDb, args) {
   })();
 }
 
+// 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 id, session_id AS sessionId FROM messages "
-    + "WHERE role = 'user' AND rewind_at IS NOT NULL AND json_valid(agent_meta) "
-    + "AND json_extract(agent_meta, '$.origin.kind') = 'scheduler' "
-    + "ORDER BY rewind_at LIMIT ?"
+    "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 rows = readyDb.prepare(selectSql).all(limit);
-    if (rows.length === 0) return { deleted: 0, sessionIds: [] };
+    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",
     );
@@ -946,7 +959,7 @@ function messagePurgeRewoundSchedulerPrompts(readyDb, args) {
       '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 };
+    return { deleted: rows.length, sessionIds, cursor: nextCursor };
   })();
 }
 
diff --git a/apps/desktop/src/main/localDb/client/tx/types.ts b/apps/desktop/src/main/localDb/client/tx/types.ts
index a1124f62526..ed8508e4d4e 100644
--- a/apps/desktop/src/main/localDb/client/tx/types.ts
+++ b/apps/desktop/src/main/localDb/client/tx/types.ts
@@ -451,6 +451,15 @@ export interface MessageRewindUserAfterClearArgs {
 
 export interface MessagePurgeRewoundSchedulerPromptsArgs {
   limit: number;
+  /** 上一批返回的 cursor（窗口末尾 rewind_at）；省略从头扫。 */
+  cursor?: number;
+}
+
+export interface MessagePurgeRewoundSchedulerPromptsResult {
+  deleted: number;
+  sessionIds: string[];
+  /** 下一批的起点；null 表示所有隐藏行已扫完。 */
+  cursor: number | null;
 }
 
 /**
@@ -1513,7 +1522,7 @@ export type DbTxResultByName = {
   'message.updateContent': { changes: number };
   'message.leaseMutate': { changes: number };
   'message.rewindUserAfterClear': { changes: number };
-  'message.purgeRewoundSchedulerPrompts': { deleted: number; sessionIds: string[] };
+  'message.purgeRewoundSchedulerPrompts': MessagePurgeRewoundSchedulerPromptsResult;
   'message.delete': MessageDeleteResult;
   'im.deleteBindings': undefined;
   'im.replaceBinding': undefined;
diff --git a/apps/desktop/src/main/localDb/ipc/__tests__/messagesClearRace.test.ts b/apps/desktop/src/main/localDb/ipc/__tests__/messagesClearRace.test.ts
index d997c8b2e8b..77c860c4f88 100644
--- a/apps/desktop/src/main/localDb/ipc/__tests__/messagesClearRace.test.ts
+++ b/apps/desktop/src/main/localDb/ipc/__tests__/messagesClearRace.test.ts
@@ -204,20 +204,49 @@ describe('message persistence clear boundary', () => {
     });
   });
 
-  it('purges already hidden scheduler prompts and leaves other tombstones', async () => {
-    const insert = sqlite.prepare("INSERT INTO messages (id, client_id, session_id, role, content, agent_meta, created_at, rewind_at) VALUES (?, ?, 's1', 'user', ?, ?, ?, ?)");
+  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', 'live', schedulerMeta, 1, null);
-    insert.run('hidden-scheduler', 'sched-hidden', 'hidden', schedulerMeta, 2, 50);
-    insert.run('hidden-user', 'user-hidden', 'hello', null, 3, 60);
-    sqlite.prepare('UPDATE sessions SET list_preview = ?, list_message_count = ? WHERE id = ?').run('hidden', 3, 's1');
-    await expect(purgeRewoundSchedulerPrompts(100)).resolves.toEqual({ deleted: 1, sessionIds: ['s1'] });
+    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: [] });
+    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');
diff --git a/apps/desktop/src/main/localDb/ipc/messages.ts b/apps/desktop/src/main/localDb/ipc/messages.ts
index 0a9b075cd33..9c351809b36 100644
--- a/apps/desktop/src/main/localDb/ipc/messages.ts
+++ b/apps/desktop/src/main/localDb/ipc/messages.ts
@@ -1364,10 +1364,11 @@ export async function rewindPersistedUserMessageAfterClear(
 
 export async function purgeRewoundSchedulerPrompts(
   limit = 100,
-): Promise<{ deleted: number; sessionIds: string[] }> {
+  cursor = 0,
+): Promise<{ deleted: number; sessionIds: string[]; cursor: number | null }> {
   const ownerScope = captureOwnerBroadcastScope();
-  if (!isOwnerBroadcastScopeCurrent(ownerScope)) return { deleted: 0, sessionIds: [] };
-  return getDbClient().tx('message.purgeRewoundSchedulerPrompts', { limit });
+  if (!isOwnerBroadcastScopeCurrent(ownerScope)) return { deleted: 0, sessionIds: [], cursor: null };
+  return getDbClient().tx('message.purgeRewoundSchedulerPrompts', { limit, cursor });
 }
 export async function dismissErrorMessage(
   sessionId: string,
diff --git a/apps/desktop/src/main/localDb/worker/opHandlers/tx.ts b/apps/desktop/src/main/localDb/worker/opHandlers/tx.ts
index 773b1d2b240..dce64b6b83c 100644
--- a/apps/desktop/src/main/localDb/worker/opHandlers/tx.ts
+++ b/apps/desktop/src/main/localDb/worker/opHandlers/tx.ts
@@ -1974,24 +1974,42 @@ 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[] } {
+): { 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 id, session_id AS sessionId FROM messages "
-    + "WHERE role = 'user' AND rewind_at IS NOT NULL AND json_valid(agent_meta) "
-    + "AND json_extract(agent_meta, '$.origin.kind') = 'scheduler' "
-    + "ORDER BY rewind_at LIMIT ?"
+    "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 rows = db.prepare(selectSql).all(limit) as Array<{ id: string; sessionId: string }>;
-    if (rows.length === 0) return { deleted: 0, sessionIds: [] as string[] };
+    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",
     );
@@ -2001,7 +2019,7 @@ function messagePurgeRewoundSchedulerPrompts(
       '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 };
+    return { deleted: rows.length, sessionIds, cursor: nextCursor };
   });
   return transaction();
 }
diff --git a/apps/desktop/src/main/scheduler-host/__tests__/rewoundSchedulerPromptCleanup.test.ts b/apps/desktop/src/main/scheduler-host/__tests__/rewoundSchedulerPromptCleanup.test.ts
index 37dd12e6c13..b91b899a6e2 100644
--- a/apps/desktop/src/main/scheduler-host/__tests__/rewoundSchedulerPromptCleanup.test.ts
+++ b/apps/desktop/src/main/scheduler-host/__tests__/rewoundSchedulerPromptCleanup.test.ts
@@ -10,18 +10,32 @@ describe('rewound scheduler prompt cleanup', () => {
     vi.useRealTimers();
   });
 
-  it('deletes hidden scheduler prompts in bounded batches and then stops', async () => {
+  it('walks the cursor across batches, stops when exhausted, and restarts from zero next sweep', async () => {
     vi.useFakeTimers();
     const purge = vi
       .fn()
-      .mockResolvedValueOnce({ deleted: 100 })
-      .mockResolvedValueOnce({ deleted: 40 })
-      .mockResolvedValue({ deleted: 0 });
+      .mockResolvedValueOnce({ deleted: 100, cursor: 0 })
+      .mockResolvedValueOnce({ deleted: 7, cursor: 5_000 })
+      .mockResolvedValueOnce({ deleted: 0, cursor: null })
+      .mockResolvedValue({ deleted: 0, cursor: null });
     startRewoundSchedulerPromptCleanup({ purge, firstDelayMs: 30_000, intervalMs: 600_000 });
     await vi.advanceTimersByTimeAsync(30_000);
-    expect(purge).toHaveBeenCalledTimes(2);
-    expect(purge).toHaveBeenNthCalledWith(1, 100);
+    expect(purge.mock.calls).toEqual([[100, 0], [100, 0], [100, 5_000]]);
     await vi.advanceTimersByTimeAsync(600_000);
-    expect(purge).toHaveBeenCalledTimes(3);
+    expect(purge).toHaveBeenCalledTimes(4);
+    expect(purge).toHaveBeenLastCalledWith(100, 0);
+  });
+
+  it('caps one sweep at five batches even when the index is not exhausted', async () => {
+    vi.useFakeTimers();
+    let next = 0;
+    const purge = vi.fn(async (_limit: number, cursor: number) => {
+      next = cursor + 500;
+      return { deleted: 0, cursor: next };
+    });
+    startRewoundSchedulerPromptCleanup({ purge, firstDelayMs: 1_000, intervalMs: 600_000 });
+    await vi.advanceTimersByTimeAsync(1_000);
+    expect(purge).toHaveBeenCalledTimes(5);
+    expect(purge).toHaveBeenLastCalledWith(100, 2_000);
   });
 });
diff --git a/apps/desktop/src/main/scheduler-host/rewoundSchedulerPromptCleanup.ts b/apps/desktop/src/main/scheduler-host/rewoundSchedulerPromptCleanup.ts
index 8c9ba742a54..1fee813fef2 100644
--- a/apps/desktop/src/main/scheduler-host/rewoundSchedulerPromptCleanup.ts
+++ b/apps/desktop/src/main/scheduler-host/rewoundSchedulerPromptCleanup.ts
@@ -5,28 +5,40 @@ const MAX_BATCHES_PER_SWEEP = 5;
 export const REWOUND_SCHEDULER_PROMPT_CLEANUP_INTERVAL_MS = 10 * 60 * 1000;
 export const REWOUND_SCHEDULER_PROMPT_CLEANUP_FIRST_DELAY_MS = 30 * 1000;
 
+export type RewoundSchedulerPromptPurge = (
+  limit: number,
+  cursor: number,
+) => Promise<{ deleted: number; cursor: number | null }>;
+
 let stopCleanup: (() => void) | null = null;
 
+/**
+ * 每轮从头沿 rewind_at 索引分窗口扫；每批的扫描量由 DB 层限定，与命中多少无关。
+ * cursor 回到 null 说明隐藏行已全部看过，本轮结束；没扫完的留给十分钟后的下一轮。
+ */
 export function startRewoundSchedulerPromptCleanup(
   deps: {
-    purge?: (limit: number) => Promise<{ deleted: number }>;
+    purge?: RewoundSchedulerPromptPurge;
     logger?: { warn?: (message: string, error?: unknown) => void };
     firstDelayMs?: number;
     intervalMs?: number;
   } = {},
 ): void {
   stopRewoundSchedulerPromptCleanup();
-  const purge = deps.purge ?? ((limit: number) => purgeRewoundSchedulerPrompts(limit));
+  const purge: RewoundSchedulerPromptPurge =
+    deps.purge ?? ((limit, cursor) => purgeRewoundSchedulerPrompts(limit, cursor));
   let stopped = false;
   let running = false;
   const sweep = async () => {
     if (stopped || running) return;
     running = true;
     try {
+      let cursor = 0;
       for (let batch = 0; batch < MAX_BATCHES_PER_SWEEP; batch += 1) {
         if (stopped) return;
-        const result = await purge(BATCH);
-        if ((result?.deleted ?? 0) < BATCH) return;
+        const result = await purge(BATCH, cursor);
+        if (result?.cursor == null) return;
+        cursor = result.cursor;
       }
     } catch (error) {
       deps.logger?.warn?.('rewound scheduler prompt cleanup failed', error);