fix: complete + deterministic dedup candidate set on sync (GIT-70)

src/services/thread-supabase.ts

@@ -71,12 +71,35 @@ export interface ThreadOwnership {
   status: string;
 }
 
+/**
+ * Whether the dedup candidate set for a sync was complete (GIT-70).
+ *
+ * `partial` and `unavailable` do not fail the sync — every thread still syncs
+ * by identity. They mean a duplicate MAY have been created, which the
+ * all_synced gate cannot detect: from the sync's perspective, writing a
+ * duplicate row is a successful write.
+ */
+export type DedupCoverage = "complete" | "partial" | "unavailable";
+
+/** Rows per page when loading dedup candidates. */
+const DEDUP_PAGE_SIZE = 1000;
+
+/** Hard ceiling on the candidate set. Hitting it is reported, never silent. */
+const DEDUP_MAX_ROWS = 10000;
+
 export interface ThreadSyncResult {
   attempted: number;
   synced: string[];
   failed: { id: string; error: string }[];
   skipped: boolean;
   all_synced: boolean;
+  /**
+   * GIT-70: how complete the duplicate check was. Distinct from all_synced,
+   * which reports whether writes landed — not whether they were correct.
+   */
+  dedup_coverage: DedupCoverage;
+  /** Candidate rows actually considered, so the coverage claim is auditable. */
+  dedup_candidates: number;
 }
 
 export interface ActiveThreadsResult {
@@ -560,11 +583,11 @@ export async function syncThreadsToSupabase(
   if (!hasSupabase() || !supabase.isConfigured()) {
     // Not a failure: on free tier Supabase is not the SOT, so there is
     // nothing to sync and nothing to gate on.
-    return { attempted: 0, synced: [], failed: [], skipped: true, all_synced: true };
+    return { attempted: 0, synced: [], failed: [], skipped: true, all_synced: true, dedup_coverage: "complete", dedup_candidates: 0 };
   }
 
   if (threads.length === 0) {
-    return { attempted: 0, synced: [], failed: [], skipped: false, all_synced: true };
+    return { attempted: 0, synced: [], failed: [], skipped: false, all_synced: true, dedup_coverage: "complete", dedup_candidates: 0 };
   }
 
   const synced: string[] = [];
@@ -573,20 +596,49 @@ export async function syncThreadsToSupabase(
   //  Load existing open threads once upfront for text-based dedup.
   // Prevents duplicate creation when closing ceremony generates new thread IDs
   // for threads that already exist with the same (or similar) text.
+  // GIT-70: this was an unordered `limit: 200` single page.
+  //
+  // The window truncated DEDUP COVERAGE, not sync coverage — every thread still
+  // synced via the exact thread_id lookup below. But a text-matching row
+  // outside the loaded 200 meant the map missed it and a DUPLICATE row was
+  // created, silently, with the dedup log never firing. A write that fires
+  // because a read was incomplete.
+  //
+  // Two changes. Pagination, so the candidate set is actually complete
+  // (orchestra_threads is already past 200). And a deterministic total order —
+  // required, not cosmetic: PostgREST Range pagination over an unordered
+  // result set can repeat and skip rows between pages, so without it the
+  // pagination would introduce its own gaps. thread_id is unique, so it is a
+  // total order on its own.
   let existingOpenThreads: { thread_id: string; text: string; status: string }[] = [];
+  let dedupCoverage: DedupCoverage = "complete";
   try {
-    existingOpenThreads = await supabase.directQuery<{ thread_id: string; text: string; status: string }>(
+    existingOpenThreads = await supabase.directQueryAll<{ thread_id: string; text: string; status: string }>(
       getTableName("threads"),
       {
         select: "thread_id,text,status",
         filters: {
           project,
-          status: "not.in.(resolved,archived)",
+          status: `not.in.(${TERMINAL_STATUSES.join(",")})`,
         },
-        limit: 200,
-      }
+        order: "thread_id.asc",
+      },
+      DEDUP_PAGE_SIZE,
+      DEDUP_MAX_ROWS
     );
+
+    // directQueryAll stops at maxRows without saying so. Inheriting that
+    // silence would rebuild the defect one layer up, so the cap is detected
+    // and reported rather than absorbed (R7).
+    if (existingOpenThreads.length >= DEDUP_MAX_ROWS) {
+      dedupCoverage = "partial";
+      console.error(
+        `[thread-supabase] Dedup candidate set hit the ${DEDUP_MAX_ROWS}-row cap — coverage is PARTIAL. ` +
+        `Threads beyond the cap cannot be matched, so a duplicate may be created.`
+      );
+    }
   } catch (err) {
+    dedupCoverage = "unavailable";
     console.error("[thread-supabase] Failed to load existing threads for dedup (proceeding without):", err instanceof Error ? err.message : err);
   }
 
@@ -658,6 +710,8 @@ export async function syncThreadsToSupabase(
     failed,
     skipped: false,
     all_synced: failed.length === 0,
+    dedup_coverage: dedupCoverage,
+    dedup_candidates: existingOpenThreads.length,
   };
 }
 

src/tools/session-close.ts

@@ -1280,6 +1280,7 @@ export async function sessionClose(
   const closeThreads = (sessionData.open_threads || []) as ThreadObject[];
   let threadSync: ThreadSyncResult = {
     attempted: 0, synced: [], failed: [], skipped: true, all_synced: true,
+    dedup_coverage: "complete", dedup_candidates: 0,
   };
 
   if (closeThreads.length > 0) {
@@ -1295,6 +1296,8 @@ export async function sessionClose(
         failed: closeThreads.map((t) => ({ id: t.id, error: message })),
         skipped: false,
         all_synced: false,
+        dedup_coverage: "unavailable",
+        dedup_candidates: 0,
       };
     }
   }
@@ -1546,7 +1549,21 @@ export async function sessionClose(
         ]
       : [];
 
-    const allErrors = [...persistErrors, ...validation.warnings];
+    // GIT-70: coverage is a different axis from success. Every thread synced,
+    // so all_synced is true and the close succeeds — but an incomplete
+    // duplicate check means a duplicate row MAY have been created, and the
+    // sync would report that as a successful write. Disclosed rather than
+    // absorbed (R7): an omitted warning reads as "checked and clean".
+    const coverageWarnings =
+      threadSync.dedup_coverage !== "complete" && threadSync.attempted > 0
+        ? [
+            `Thread duplicate-check coverage was ${threadSync.dedup_coverage} ` +
+            `(${threadSync.dedup_candidates} candidates considered). Threads synced successfully, ` +
+            `but a duplicate may have been created for text matching a thread outside the checked set.`,
+          ]
+        : [];
+
+    const allErrors = [...persistErrors, ...coverageWarnings, ...validation.warnings];
 
     const display = formatCloseDisplay(sessionId, closeCompliance, params, learningsCount, !partialPersist, allErrors.length > 0 ? allErrors : undefined, transcriptStatus, blindspotSnippet);
 

tests/unit/services/thread-sync-result.test.ts

@@ -14,13 +14,15 @@
 import { describe, it, expect, vi, beforeEach } from "vitest";
 
 const mockDirectQuery = vi.fn();
+const mockDirectQueryAll = vi.fn();
 const mockDirectUpsert = vi.fn();
 const mockIsConfigured = vi.fn(() => true);
 const mockHasSupabase = vi.fn(() => true);
 
 vi.mock("../../../src/services/supabase-client.js", () => ({
   isConfigured: (...a: unknown[]) => mockIsConfigured(...a),
   directQuery: (...a: unknown[]) => mockDirectQuery(...a),
+  directQueryAll: (...a: unknown[]) => mockDirectQueryAll(...a),
   directUpsert: (...a: unknown[]) => mockDirectUpsert(...a),
   upsertRecord: vi.fn(),
   listRecords: vi.fn(),
@@ -45,6 +47,10 @@ const thread = (id: string, status = "open") => ({
 
 beforeEach(() => {
   mockDirectQuery.mockReset();
+  mockDirectQueryAll.mockReset();
+  // GIT-70: the dedup candidate set is loaded via paginated directQueryAll.
+  // Empty by default so per-thread directQuery sequencing stays readable.
+  mockDirectQueryAll.mockResolvedValue([]);
   mockDirectUpsert.mockReset();
   mockIsConfigured.mockReturnValue(true);
   mockHasSupabase.mockReturnValue(true);
@@ -97,9 +103,7 @@ describe("outcome reporting", () => {
   });
 
   it("records a failed thread instead of swallowing it", async () => {
-    mockDirectQuery
-      .mockResolvedValueOnce([])                      // candidate load
-      .mockRejectedValueOnce(new Error("boom"));      // t-a lookup
+    mockDirectQuery.mockRejectedValueOnce(new Error("boom")); // t-a lookup
 
     const r = await syncThreadsToSupabase([thread("t-a")], "gitmem", "s-1");
 
@@ -113,7 +117,6 @@ describe("outcome reporting", () => {
     // A partial sync is still worth completing — but the miss must be named,
     // because the local file is the only remaining record of it.
     mockDirectQuery
-      .mockResolvedValueOnce([])                          // candidate load
       .mockRejectedValueOnce(new Error("network"))        // t-a fails
       .mockResolvedValueOnce([]);                         // t-b succeeds
 
@@ -130,9 +133,7 @@ describe("outcome reporting", () => {
   });
 
   it("never reports all_synced while any thread failed", async () => {
-    mockDirectQuery
-      .mockResolvedValueOnce([])
-      .mockRejectedValue(new Error("down"));
+    mockDirectQuery.mockRejectedValue(new Error("down"));
 
     const r = await syncThreadsToSupabase(
       [thread("t-a"), thread("t-b"), thread("t-c")],
@@ -147,9 +148,7 @@ describe("outcome reporting", () => {
   });
 
   it("names every missed id so the caller can report them", async () => {
-    mockDirectQuery
-      .mockResolvedValueOnce([])
-      .mockRejectedValue(new Error("timeout"));
+    mockDirectQuery.mockRejectedValue(new Error("timeout"));
 
     const r = await syncThreadsToSupabase(
       [thread("t-aaa"), thread("t-bbb")],
@@ -165,3 +164,75 @@ describe("outcome reporting", () => {
     }
   });
 });
+
+// ---------------------------------------------------------------------------
+// GIT-70: dedup candidate completeness
+// ---------------------------------------------------------------------------
+
+describe("dedup candidate coverage (GIT-70)", () => {
+  const rows = (n: number) =>
+    Array.from({ length: n }, (_, i) => ({
+      thread_id: `t-${String(i).padStart(8, "0")}`,
+      text: `existing ${i}`,
+      status: "active",
+    }));
+
+  it("loads candidates with a deterministic total order", async () => {
+    // Not cosmetic. PostgREST Range pagination over an unordered result set can
+    // repeat and skip rows between pages, so without a total order the
+    // pagination introduces the very gaps it was added to close. thread_id is
+    // unique, so it orders totally on its own.
+    mockDirectQuery.mockResolvedValue([]);
+    await syncThreadsToSupabase([thread("t-a")], "gitmem", "s-1");
+
+    const [, options] = mockDirectQueryAll.mock.calls[0] as [string, { order?: string }];
+    expect(options.order).toBe("thread_id.asc");
+  });
+
+  it("reports complete coverage past the old 200-row window", async () => {
+    // The defect: an unordered single page of 200 over a table already at 256.
+    // A text match outside that window produced a silent duplicate.
+    mockDirectQueryAll.mockResolvedValue(rows(256));
+    mockDirectQuery.mockResolvedValue([]);
+
+    const r = await syncThreadsToSupabase([thread("t-new")], "gitmem", "s-1");
+
+    expect(r.dedup_coverage).toBe("complete");
+    expect(r.dedup_candidates).toBe(256);
+  });
+
+  it("reports partial coverage rather than absorbing the cap", async () => {
+    // directQueryAll stops at maxRows without saying so. Inheriting that
+    // silence would rebuild the defect one layer up.
+    mockDirectQueryAll.mockResolvedValue(rows(10000));
+    mockDirectQuery.mockResolvedValue([]);
+
+    const r = await syncThreadsToSupabase([thread("t-new")], "gitmem", "s-1");
+
+    expect(r.dedup_coverage).toBe("partial");
+    expect(r.dedup_candidates).toBe(10000);
+  });
+
+  it("reports unavailable when the candidate load fails", async () => {
+    mockDirectQueryAll.mockRejectedValue(new Error("candidate load down"));
+    mockDirectQuery.mockResolvedValue([]);
+
+    const r = await syncThreadsToSupabase([thread("t-new")], "gitmem", "s-1");
+
+    expect(r.dedup_coverage).toBe("unavailable");
+  });
+
+  it("still syncs successfully when coverage is incomplete", async () => {
+    // Coverage and success are different axes. Every thread syncs by identity
+    // regardless; incomplete coverage means a duplicate MAY exist, which
+    // all_synced cannot detect because writing a duplicate IS a successful
+    // write. This is the distinction the whole ticket rests on.
+    mockDirectQueryAll.mockRejectedValue(new Error("down"));
+    mockDirectQuery.mockResolvedValue([]);
+
+    const r = await syncThreadsToSupabase([thread("t-new")], "gitmem", "s-1");
+
+    expect(r.all_synced).toBe(true);
+    expect(r.dedup_coverage).not.toBe("complete");
+  });
+});