Accept duplicate replay uploads without a read-back

src/api/services/notification.test.ts

@@ -154,11 +154,7 @@ describe("notification service", () => {
     });
   });
 
-  it("preserves navigation metadata and disposes native action listeners", async () => {
-    const unregister = vi.fn();
-    const handler = vi.fn();
-    mocks.onAction.mockResolvedValueOnce({ unregister });
-
+  it("preserves navigation metadata on the sent notification", async () => {
     await sendSystemNotification("Assigned", "Review it", {
       orgiiTarget: "team-inbox",
       teamInboxItemKey: "assigned_work_item:WI-1",
@@ -174,45 +170,18 @@ describe("notification service", () => {
       actionTypeId: undefined,
       autoCancel: true,
     });
-
-    const dispose = await listenForSystemNotificationActions(handler);
-    const nativeHandler = mocks.onAction.mock.calls[0]?.[0] as
-      | ((notification: { extra?: Record<string, unknown> }) => void)
-      | undefined;
-    nativeHandler?.({
-      extra: {
-        orgiiTarget: "team-inbox",
-        teamInboxItemKey: "assigned_work_item:WI-1",
-      },
-    });
-    expect(handler).toHaveBeenCalledWith({
-      extra: {
-        orgiiTarget: "team-inbox",
-        teamInboxItemKey: "assigned_work_item:WI-1",
-      },
-    });
-
-    dispose();
-    expect(unregister).toHaveBeenCalledOnce();
   });
 
-  it("registers a foreground View action for Team Inbox notifications", async () => {
-    mocks.registerActionTypes.mockResolvedValueOnce(undefined);
+  it("keeps the mobile-only action entry points inert on desktop", async () => {
+    const handler = vi.fn();
 
     await registerTeamInboxNotificationActionType("View");
+    expect(mocks.registerActionTypes).not.toHaveBeenCalled();
 
-    expect(mocks.registerActionTypes).toHaveBeenCalledWith([
-      {
-        id: "orgii-team-inbox",
-        actions: [
-          {
-            id: "view-team-inbox",
-            title: "View",
-            foreground: true,
-          },
-        ],
-      },
-    ]);
+    const dispose = await listenForSystemNotificationActions(handler);
+    expect(mocks.onAction).not.toHaveBeenCalled();
+    expect(handler).not.toHaveBeenCalled();
+    expect(() => dispose()).not.toThrow();
   });
 
   it("projects positive and cleared dock badge values", async () => {

src/api/services/notification.ts

@@ -2,8 +2,6 @@ import { invoke, isTauri } from "@tauri-apps/api/core";
 import { getCurrentWindow } from "@tauri-apps/api/window";
 import {
   isPermissionGranted,
-  onAction,
-  registerActionTypes,
   requestPermission,
   sendNotification,
 } from "@tauri-apps/plugin-notification";
@@ -168,35 +166,22 @@ export const sendSystemNotification = async (
   }
 };
 
-/** Project the authoritative Team Inbox unread count into the dock badge. */
-export const registerTeamInboxNotificationActionType = async (
-  viewLabel: string
-): Promise<void> => {
-  await registerActionTypes([
-    {
-      id: TEAM_INBOX_NOTIFICATION_ACTION_TYPE_ID,
-      actions: [
-        {
-          id: "view-team-inbox",
-          title: viewLabel,
-          foreground: true,
-        },
-      ],
-    },
-  ]);
-};
-
 /**
- * Listen for native notification activation while the application process is
- * alive. The returned disposer is safe to call during React effect cleanup.
+ * Notification action buttons are a mobile-only concept in the notification
+ * plugin: its desktop `invoke_handler` registers just notify/permission
+ * commands, so `registerActionTypes` and the action listener can only fail
+ * with "Command not found" on every desktop launch. Both entry points are
+ * kept as inert seams for a future mobile target.
  */
+export const registerTeamInboxNotificationActionType = async (
+  _viewLabel: string
+): Promise<void> => {};
+
+/** See `registerTeamInboxNotificationActionType` — inert on desktop. */
 export const listenForSystemNotificationActions = async (
-  handler: (action: SystemNotificationAction) => void
+  _handler: (action: SystemNotificationAction) => void
 ): Promise<() => void> => {
-  const listener = await onAction((notification) => {
-    handler({ extra: notification.extra ?? {} });
-  });
-  return () => listener.unregister();
+  return () => {};
 };
 
 /**

src/features/Org2Cloud/org2CloudStorageClient.test.ts

@@ -125,14 +125,32 @@ describe("uploadReplayObject", () => {
     expect((error as Org2CloudStorageError).status).toBe(400);
   });
 
-  it("accepts a plain 409 duplicate when the object is readable", async () => {
-    fetchMock
-      .mockResolvedValueOnce(new Response(null, { status: 409 }))
-      .mockResolvedValueOnce(new Response(null, { status: 200 }));
+  it("accepts a plain 409 duplicate without a read-back probe", async () => {
+    fetchMock.mockResolvedValueOnce(new Response(null, { status: 409 }));
+
+    await expect(
+      uploadReplayObject("jwt-1", "org-1/s-1/1/1-h.gz", new Uint8Array([1]))
+    ).resolves.toBeUndefined();
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("accepts a 400-wrapped KeyAlreadyExists even when the object is unreadable", async () => {
+    fetchMock.mockResolvedValueOnce(
+      new Response(
+        JSON.stringify({
+          statusCode: "409",
+          error: "Duplicate",
+          message: "The resource already exists",
+          code: "KeyAlreadyExists",
+        }),
+        { status: 400 }
+      )
+    );
 
     await expect(
       uploadReplayObject("jwt-1", "org-1/s-1/1/1-h.gz", new Uint8Array([1]))
     ).resolves.toBeUndefined();
+    expect(fetchMock).toHaveBeenCalledTimes(1);
   });
 });
 

src/features/Org2Cloud/org2CloudStorageClient.ts

@@ -70,15 +70,17 @@ export async function uploadReplayObject(
   });
   if (response.ok) return;
   const body = await response.text().catch(() => "");
-  // Replay objects are content-addressed (segment hash in the key) and the
-  // storage policies grant INSERT but never UPDATE, so re-uploading an
-  // existing name is rejected rather than applied — the normal retry/resume
-  // path. Supabase reports it as 409, or as a 400 envelope wrapping a 403
-  // RLS denial when the policy blocks the implied update, which is
-  // indistinguishable by status from a genuine authorization failure.
-  // Confirm the object is actually readable before treating it as done, so
-  // a real denial still surfaces.
-  if (mayMeanReplayObjectExists(response.status, body)) {
+  // Replay objects are content-addressed (segment hash in the key), so a
+  // duplicate-name rejection means the exact bytes are already stored —
+  // resume treats it as success without a read-back: the member read path
+  // walks the session read ladder, which denies replay reads on
+  // metadata-only shares, so an existence probe cannot confirm objects the
+  // uploader is not allowed to read and would fail this path forever.
+  if (isDuplicateObjectRejection(response.status, body)) return;
+  // A 400/403 RLS denial is ambiguous: the policy blocks the implied
+  // update on an existing name with the same text as a genuine
+  // authorization failure. Only here does the read-back decide.
+  if (mayBeMaskedDuplicate(response.status, body)) {
     const exists = await replayObjectExists(
       accessToken,
       path,
@@ -94,17 +96,25 @@ export async function uploadReplayObject(
   );
 }
 
-/** Statuses/bodies that can mean "this object name is already stored". */
-function mayMeanReplayObjectExists(status: number, body: string): boolean {
+/** Rejections that unambiguously mean "this object name is already stored". */
+function isDuplicateObjectRejection(status: number, body: string): boolean {
   if (status === 409) return true;
   if (status !== 400 && status !== 403) return false;
   return (
-    body.includes("row-level security policy") ||
     body.includes("Duplicate") ||
+    body.includes("KeyAlreadyExists") ||
     body.includes("already exists")
   );
 }
 
+/** RLS denials that may be masking an insert onto an existing name. */
+function mayBeMaskedDuplicate(status: number, body: string): boolean {
+  return (
+    (status === 400 || status === 403) &&
+    body.includes("row-level security policy")
+  );
+}
+
 /** HEAD probe used only to confirm an upload rejection was a duplicate. */
 async function replayObjectExists(
   accessToken: string,

src/features/Org2Cloud/org2CloudSyncEngine.retentionPark.test.ts

@@ -0,0 +1,65 @@
+/**
+ * Retention parking on the push loop.
+ *
+ * A session past the org's retention window fails its push with
+ * ORG2_RETENTION_EXPIRED on every pass; retention only recedes further
+ * within a signed-in run, so the engine must stop re-walking the doomed
+ * upload chain instead of retrying it each pass.
+ */
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+
+import { Org2CloudSyncError } from "./org2CloudSyncClient";
+import {
+  cleanupEngineFixture,
+  createEngineFixture,
+} from "./org2CloudSyncEngine.testUtils";
+import type { EngineFixture } from "./org2CloudSyncEngine.testUtils";
+import {
+  getSyncJournalSnapshot,
+  resetSyncJournalForTests,
+} from "./org2CloudSyncJournal";
+
+describe("Org2CloudSyncEngine retention parking", () => {
+  let fixture: EngineFixture;
+  let engine: EngineFixture["engine"];
+
+  beforeEach(() => {
+    resetSyncJournalForTests();
+    fixture = createEngineFixture();
+    ({ engine } = fixture);
+  });
+
+  afterEach(() => {
+    cleanupEngineFixture(engine);
+    resetSyncJournalForTests();
+  });
+
+  it("parks a retention-expired session instead of retrying every pass", async () => {
+    fixture.client.upsertSessionMetadata.mockRejectedValue(
+      new Org2CloudSyncError("ORG2_RETENTION_EXPIRED", 400)
+    );
+
+    await engine.runSyncPass();
+    expect(fixture.client.upsertSessionMetadata).toHaveBeenCalledTimes(1);
+    expect(
+      getSyncJournalSnapshot().some(
+        (event) =>
+          event.kind === "session_retention_parked" &&
+          event.code === "ORG2_RETENTION_EXPIRED"
+      )
+    ).toBe(true);
+
+    await engine.runSyncPass();
+    expect(fixture.client.upsertSessionMetadata).toHaveBeenCalledTimes(1);
+  });
+
+  it("keeps retrying pushes that fail with other codes", async () => {
+    fixture.client.upsertSessionMetadata.mockRejectedValue(
+      new Org2CloudSyncError("ORG2_VALIDATION", 400)
+    );
+
+    await engine.runSyncPass();
+    await engine.runSyncPass();
+    expect(fixture.client.upsertSessionMetadata).toHaveBeenCalledTimes(2);
+  });
+});

src/features/Org2Cloud/org2CloudSyncEngine.ts

@@ -147,6 +147,7 @@ import {
   resolveContinuationStatusesViaCache,
   resolveLocalSessionIdsViaAggregateList,
 } from "./org2CloudSyncEngine.vanishedSessions";
+import { recordSyncEvent } from "./org2CloudSyncJournal";
 import {
   type CloudStore,
   Org2CloudSyncLifecycle,
@@ -187,6 +188,12 @@ export class Org2CloudSyncEngine extends Org2CloudSyncLifecycle {
   private readonly orgBackoff: Org2CloudOrgBackoffTracker;
   /** Generation whose background-org retract reconcile already ran (P2). */
   private reconciledGeneration = -1;
+  /** "orgId|sessionId" keys whose push failed with ORG2_RETENTION_EXPIRED.
+   * Retention only recedes further within a signed-in run, so the push is
+   * doomed until the org's entitlement changes — parked until the next
+   * resetSyncState() (sign-in cycle / endpoint switch / app restart)
+   * instead of re-walking the full upload chain every pass. */
+  private readonly retentionParked = new Set<string>();
   /** TTL-gated `org2CloudRepoScopesAtom` mirror hydration, split out to
    * `Org2CloudRepoScopeSync`. */
   private readonly repoScopeSync: Org2CloudRepoScopeSync;
@@ -324,6 +331,7 @@ export class Org2CloudSyncEngine extends Org2CloudSyncLifecycle {
   }
 
   protected override resetSyncState(): void {
+    this.retentionParked.clear();
     this.orgBackoff.reset();
     this.sessionSync.reset();
     this.repoScopeSync.reset();
@@ -483,6 +491,9 @@ export class Org2CloudSyncEngine extends Org2CloudSyncLifecycle {
       for (const session of store.get(sessionsAtom)) {
         if (this.generation !== generation) return;
         if (!isCloudPushCandidate(session)) continue;
+        if (this.retentionParked.has(`${org.orgId}|${session.session_id}`)) {
+          continue;
+        }
         // A fork is a continuation inside the source collaboration boundary,
         // not a new ordinary repo session. Repo scopes may overlap across a
         // team org and the forker's personal org, so scope matching alone
@@ -740,6 +751,25 @@ export class Org2CloudSyncEngine extends Org2CloudSyncLifecycle {
             this.orgBackoff.backOffOrg(org.orgId, error);
             break; // Stop touching this org for the rest of the run.
           }
+          if (
+            org2CloudSyncClient.isOrg2SyncErrorCode(
+              error,
+              "ORG2_RETENTION_EXPIRED"
+            )
+          ) {
+            this.retentionParked.add(`${org.orgId}|${session.session_id}`);
+            recordSyncEvent({
+              level: "warn",
+              kind: "session_retention_parked",
+              orgId: org.orgId,
+              message: `Push parked for session ${session.session_id}: past the org's retention window`,
+              code: "ORG2_RETENTION_EXPIRED",
+            });
+            log.warn(
+              `cloud push parked for retention-expired session ${session.session_id}`
+            );
+            continue;
+          }
           log.warn(
             `cloud push failed for session ${session.session_id}:`,
             error