feat(schedule): sync occurrence overrides to the client

backend/src/timeflow/gateway/websocket/agent_ports.py

@@ -153,6 +153,11 @@ def schedules(self) -> list[dict[str, Any]] | None:
         """Matches a query found, when the command was a query."""
         ...
 
+    @property
+    def occurrence_overrides(self) -> list[dict[str, Any]] | None:
+        """Per-occurrence exceptions (cancel/replace) a recurring-schedule mutation produced."""
+        ...
+
 
 class AudioReplyInfo(Protocol):
     """Format and purpose of a spoken reply, announced before its audio."""

backend/src/timeflow/gateway/websocket/handlers/agent_result.py

@@ -91,6 +91,7 @@ async def deliver_result(self, result: CommandOutcome, stream: StreamIdentity) -
                 status=result.status,
                 schedule=result.schedule,
                 schedules=result.schedules,
+                occurrence_overrides=result.occurrence_overrides,
             ),
         )
         await self._send(stream.session_id, message.type, message.model_dump(exclude_none=True))

backend/src/timeflow/gateway/websocket/messages/agent.py

@@ -29,6 +29,7 @@ class VoiceCommandResultPayload(BaseModel):
     status: str
     schedule: dict[str, Any] | None = None
     schedules: list[dict[str, Any]] | None = None
+    occurrence_overrides: list[dict[str, Any]] | None = None
 
 
 class VoiceCommandResult(BaseModel):

backend/src/timeflow/intelligence/ports.py

@@ -108,6 +108,7 @@ class CommandResult:
     status: str
     schedule: dict[str, Any] | None = None
     schedules: list[dict[str, Any]] | None = None
+    occurrence_overrides: list[dict[str, Any]] | None = None
 
 
 @dataclass(frozen=True, slots=True)

backend/src/timeflow/intelligence/realtime/agent.py

@@ -399,6 +399,7 @@ async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any
                     status=str(outcome["status"]),
                     schedule=outcome.get("schedule"),
                     schedules=outcome.get("schedules"),
+                    occurrence_overrides=outcome.get("occurrence_overrides"),
                 ),
                 self._stream,
             )

backend/src/timeflow/intelligence/realtime/schedule_tools.py

@@ -463,6 +463,8 @@ def _business_error(error: ScheduleBusinessError) -> ToolResult:
 def _mutation_result(result: ScheduleMutationResult, operation: str, tz: ZoneInfo) -> ToolResult:
     """Convert a mutation result into model output and client outcome."""
     snapshot = result.schedules[0] if result.schedules else None
+    schedules = [_snapshot_for_client(s) for s in result.schedules]
+    overrides = [_override_for_client(o) for o in result.occurrence_overrides]
     return ToolResult(
         output=json.dumps(
             {"status": "applied", "schedule": _for_model_dict(snapshot, tz)}, ensure_ascii=False
@@ -471,6 +473,8 @@ def _mutation_result(result: ScheduleMutationResult, operation: str, tz: ZoneInf
             "operation": operation,
             "status": "applied",
             "schedule": _snapshot_for_client(snapshot) if snapshot else None,
+            "schedules": schedules or None,
+            "occurrence_overrides": overrides or None,
         },
     )
 
@@ -500,6 +504,12 @@ def _snapshot_for_client(snapshot: Any) -> dict[str, Any]:
     }
 
 
+def _override_for_client(override: Any) -> dict[str, Any]:
+    """Convert ScheduleOccurrenceOverrideSnapshot to client dict, filtering out audit fields."""
+    d = asdict(override)
+    return {k: _json_value(v) for k, v in d.items() if k not in {"created_at", "updated_at"}}
+
+
 def _local_text(instant: datetime | None, tz: ZoneInfo) -> str:
     """Render a stored instant as local wall-clock text, empty for a schedule without one."""
     if instant is None:

backend/tests/intelligence/realtime/test_realtime_toolbox.py

@@ -11,11 +11,13 @@
 import pytest
 
 from timeflow.business.calendar import (
+    OccurrenceOverrideAction,
     ScheduleAgentService,
     ScheduleBusinessError,
     ScheduleErrorCode,
     ScheduleKind,
     ScheduleMutationResult,
+    ScheduleOccurrenceOverrideSnapshot,
     ScheduleSearchResult,
     ScheduleSnapshot,
     ScheduleStatus,
@@ -305,6 +307,78 @@ def test_a_delete_reaches_the_call_that_matches_the_kind(
     assert service.calls == [expected]
 
 
+OVERRIDE = ScheduleOccurrenceOverrideSnapshot(
+    id="ovr_1",
+    schedule_id="sch_1",
+    occurrence_start=datetime(2026, 9, 8, 7, 0, tzinfo=UTC),
+    action=OccurrenceOverrideAction.CANCEL,
+    created_at=datetime(2026, 9, 7, 1, 0, tzinfo=UTC),
+    updated_at=datetime(2026, 9, 7, 1, 0, tzinfo=UTC),
+)
+
+
+class OverrideProducingService(RecordingService):
+    """A this_occurrence delete that cancels rather than replaces: no new schedule, one override."""
+
+    def delete_recurring_schedule(self, *, account_id: str, command: Any) -> ScheduleMutationResult:
+        self.calls.append("delete_recurring")
+        return ScheduleMutationResult(schedules=(), occurrence_overrides=(OVERRIDE,))
+
+
+def test_a_this_occurrence_delete_reports_the_override_it_produced() -> None:
+    result = run(
+        "schedule_delete",
+        {
+            "schedule_id": "sch_1",
+            "expected_revision": 1,
+            "schedule_kind": "recurring",
+            "scope": "this_occurrence",
+        },
+        ToolBox("acc_test", OverrideProducingService()),
+    )
+    assert result.outcome is not None
+    assert result.outcome["schedule"] is None
+    assert result.outcome["occurrence_overrides"] == [
+        {
+            "id": "ovr_1",
+            "schedule_id": "sch_1",
+            "occurrence_start": "2026-09-08T07:00:00+00:00",
+            "action": "cancel",
+            "replacement_schedule_id": None,
+        }
+    ]
+
+
+class MultiScheduleProducingService(RecordingService):
+    """A this_occurrence delete hitting an existing replace override: no new override,
+    but two schedules — the untouched parent and the soft-deleted replacement."""
+
+    def delete_recurring_schedule(self, *, account_id: str, command: Any) -> ScheduleMutationResult:
+        self.calls.append("delete_recurring")
+        replacement = replace(SNAPSHOT, id="sch_replacement", status=ScheduleStatus.DELETED)
+        return ScheduleMutationResult(schedules=(SNAPSHOT, replacement))
+
+
+def test_a_this_occurrence_delete_with_existing_replacement_reports_both_schedules() -> None:
+    result = run(
+        "schedule_delete",
+        {
+            "schedule_id": "sch_1",
+            "expected_revision": 1,
+            "schedule_kind": "recurring",
+            "scope": "this_occurrence",
+        },
+        ToolBox("acc_test", MultiScheduleProducingService()),
+    )
+    assert result.outcome is not None
+    # schedule（单数）保持只给第一条，向后兼容
+    assert result.outcome["schedule"]["id"] == "sch_1"
+    # schedules（复数）必须两条都在，这是这次要修的东西——之前只有 schedules[0] 会下发，
+    # 软删除的 replacement 永远到不了客户端
+    ids = [s["id"] for s in result.outcome["schedules"]]
+    assert ids == ["sch_1", "sch_replacement"]
+
+
 def test_a_delete_with_nothing_left_to_report_still_says_it_applied() -> None:
     result = run(
         "schedule_delete",

frontend/src/contracts/conversation.ts

@@ -64,6 +64,14 @@ export interface VoiceAsrCompletedMessage {
 /** 日程快照的形状由后端 `ScheduleSnapshot` 决定；前端只透传给本地落库，不在此处强约束字段。 */
 export type ScheduleSnapshotPayload = Record<string, unknown>;
 
+export interface OccurrenceOverridePayload {
+  id: string;
+  schedule_id: string;
+  occurrence_start: string;
+  action: 'cancel' | 'replace';
+  replacement_schedule_id: string | null;
+}
+
 export interface VoiceCommandResultMessage {
   type: 'voice.command.result';
   message_id: string;
@@ -74,6 +82,7 @@ export interface VoiceCommandResultMessage {
     status: string;
     schedule?: ScheduleSnapshotPayload;
     schedules?: ScheduleSnapshotPayload[];
+    occurrence_overrides?: OccurrenceOverridePayload[];
   };
 }
 

frontend/src/features/assistant/application/AssistantContinuousConversationService.ts

@@ -342,6 +342,7 @@ export class AssistantContinuousConversationService implements AssistantApplicat
           operation: message.payload.operation,
           schedule: message.payload.schedule,
           schedules: message.payload.schedules,
+          occurrence_overrides: message.payload.occurrence_overrides,
           status: message.payload.status,
         };
         // 状态立刻回到 listening（麦克风还开着），不等写库；message.ack 必须等

frontend/src/features/assistant/application/AssistantConversationService.ts

@@ -224,6 +224,7 @@ export class AssistantConversationService implements AssistantApplicationPort {
           operation: message.payload.operation,
           schedule: message.payload.schedule,
           schedules: message.payload.schedules,
+          occurrence_overrides: message.payload.occurrence_overrides,
           status: message.payload.status,
         };
         // 状态立刻回到 idle，不等写库；message.ack 必须等写库成功才发（AGENTS.md §6）。

frontend/src/features/assistant/data/local/LocalScheduleWriter.ts

@@ -1,35 +1,64 @@
-import type { CloudScheduleRow, ScheduleLocalRepository } from '../../../schedule/data';
+import type {
+  CloudScheduleRow,
+  LocalScheduleOccurrenceOverrideRow,
+  ScheduleLocalRepository,
+} from '../../../schedule/data';
 import type { LocalScheduleWriterPort } from '../../application/interfaces/LocalScheduleWriterPort';
-import type { AppliedCommand } from '../../domain/ConversationTurn';
+import type { AppliedCommand, AppliedOccurrenceOverride } from '../../domain/ConversationTurn';
 
 /**
  * `applyCloudSchedule` 是仓储里现成的 upsert，专门给"服务端权威数据落到本地"用的
  * （固定写 sync_status='synced'），voice.command.result 正好是这种场景——服务端
  * 已经提交过了，这里只是把它同步进本地日历读服务能看到的地方。
  *
- * 不按 operation 名字判断该不该写：后端 create_schedule/update_schedule/
+ * 用 `operation` 显式排除 list_schedules：后端 create_schedule/update_schedule/
  * delete_schedule 三个操作共用同一条 `_mutation_result` 路径
- * （backend/src/timeflow/intelligence/realtime/schedule_tools.py:384），
- * `schedule` 字段结构完全一样——删除是软删除，`status` 会变成 'deleted'，
+ * （backend/src/timeflow/intelligence/realtime/schedule_tools.py:463），
+ * `list_schedules`（查询，见同文件 `_find`）复用同一个 `schedules` 复数字段填
+ * outcome——不能靠"字段是否存在"这种隐式判断排除它，query 结果不是已提交的写入，
+ * 不能当成本地权威数据 upsert。删除是软删除，`status` 会变成 'deleted'，
  * `applyCloudSchedule` 照样按 id upsert，本地这行的 status 也跟着变成
- * 'deleted'，日历读服务只认 status==='active' 的行，自然就不显示了。只要
- * `schedule` 字段存在就同步，这样天然排除 list_schedules（只有 `schedules`
- * 复数字段，没有 `schedule`）。
+ * 'deleted'，日历读服务只认 status==='active' 的行，自然就不显示了。
  *
- * 已知没覆盖的情况：重复日程"仅删除这一次"且不需要截断 RRULE 时
- * （service.py 的 _delete_recurring_range 里创建 occurrence_override 那条
- * 分支），后端只把 override 记到 ScheduleMutationResult.occurrence_overrides
- * 里，`voice.command.result` 完全不下发这个字段——这种情况本地日历目前没法
- * 同步，得等后端把 override 数据也传下来。
+ * 重复日程"仅删除这一次"命中已存在的 replace override 时（service.py 的
+ * _delete_recurring_range），不产生新 override，而是让 `schedules` 里带上被
+ * 连带软删的 replacement——`schedules` 复数字段是"这次命令落地的全部 schedule
+ * 快照"，不是只服务查询场景；`occurrence_overrides` 跟 `schedules` 各自独立，
+ * 一条 command.result 里两者都可能有、也可能都没有。
+ *
+ * 一条命令可能同时产生多条 schedule 写入和多条 override 写入，合起来才代表这条
+ * 语音指令完整落地，所以全部包在一个事务里：任何一次写入失败（仓储返回 false，
+ * 比如父日程缺失、账号不匹配）都会抛错并回滚整个事务，调用方据此不发
+ * message.ack，不会向服务端谎报已落库、也不会在本地留下半吊子状态。
  */
 export class LocalScheduleWriter implements LocalScheduleWriterPort {
   public constructor(private readonly repository: ScheduleLocalRepository) {}
 
   public async applyCommandResult(accountId: string, command: AppliedCommand): Promise<void> {
-    if (command.status !== 'applied' || !command.schedule) {
+    if (command.status !== 'applied' || command.operation === 'list_schedules') {
+      return;
+    }
+    const schedules = command.schedules ?? (command.schedule ? [command.schedule] : []);
+    if (schedules.length === 0 && !command.occurrence_overrides?.length) {
       return;
     }
-    await this.repository.applyCloudSchedule(toCloudScheduleRow(accountId, command.schedule));
+    await this.repository.withTransaction(async (repository) => {
+      for (const raw of schedules) {
+        const applied = await repository.applyCloudSchedule(toCloudScheduleRow(accountId, raw));
+        if (!applied) {
+          throw new Error(`Could not apply cloud schedule ${String(raw.id)}`);
+        }
+      }
+      for (const override of command.occurrence_overrides ?? []) {
+        const applied = await repository.upsertOccurrenceOverride(
+          accountId,
+          toOverrideRow(override),
+        );
+        if (!applied) {
+          throw new Error(`Could not apply occurrence override ${override.id}`);
+        }
+      }
+    });
   }
 }
 
@@ -70,6 +99,16 @@ function toCloudScheduleRow(accountId: string, raw: Record<string, unknown>): Cl
   };
 }
 
+function toOverrideRow(override: AppliedOccurrenceOverride): LocalScheduleOccurrenceOverrideRow {
+  return {
+    id: override.id,
+    schedule_id: override.schedule_id,
+    occurrence_start: override.occurrence_start,
+    action: override.action,
+    replacement_schedule_id: override.replacement_schedule_id,
+  };
+}
+
 function requireString(value: unknown, field: string): string {
   if (typeof value !== 'string' || value.length === 0) {
     throw new TypeError(`command.result.schedule.${field} must be a non-empty string`);

frontend/src/features/assistant/domain/ConversationTurn.ts

@@ -18,9 +18,18 @@ export type ConversationTurnState =
   | { phase: 'paused'; conversationId: string | null }
   | { phase: 'error'; message: string };
 
+export interface AppliedOccurrenceOverride {
+  id: string;
+  schedule_id: string;
+  occurrence_start: string;
+  action: 'cancel' | 'replace';
+  replacement_schedule_id: string | null;
+}
+
 export interface AppliedCommand {
   operation: string;
   status: string;
   schedule?: Record<string, unknown>;
   schedules?: Record<string, unknown>[];
+  occurrence_overrides?: AppliedOccurrenceOverride[];
 }

frontend/src/features/schedule/data/local/scheduleLocalRepository.ts

@@ -89,6 +89,20 @@ export class ScheduleLocalRepository {
     );
   }
 
+  /**
+   * 一批相关写入要么全部生效、要么全部不生效——调用方在 task 里用参数给的 repository
+   * （绑定在事务连接上）做写入，抛错会让整个事务回滚。
+   */
+  public async withTransaction<T>(
+    task: (repository: ScheduleLocalRepository) => Promise<T>,
+  ): Promise<T> {
+    let result!: T;
+    await this.database.withExclusiveTransactionAsync(async (transaction) => {
+      result = await task(new ScheduleLocalRepository(transaction));
+    });
+    return result;
+  }
+
   /** Apply cloud-owned fields while preserving existing device runtime state. */
   public async applyCloudSchedule(row: CloudScheduleRow): Promise<boolean> {
     const result = await this.database.runAsync(