diff --git a/src/services/thread-supabase.ts b/src/services/thread-supabase.ts index 88d0da3..1494521 100644 --- a/src/services/thread-supabase.ts +++ b/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, }; } diff --git a/src/tools/session-close.ts b/src/tools/session-close.ts index edbef8c..fb2ab92 100644 --- a/src/tools/session-close.ts +++ b/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); diff --git a/tests/unit/services/thread-sync-result.test.ts b/tests/unit/services/thread-sync-result.test.ts index 99e0d6a..ba5e976 100644 --- a/tests/unit/services/thread-sync-result.test.ts +++ b/tests/unit/services/thread-sync-result.test.ts @@ -14,6 +14,7 @@ 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); @@ -21,6 +22,7 @@ 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"); + }); +});