From 21559593e9acfc5f368519f2f25968bba5b20079 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 23:33:54 -0400 Subject: [PATCH] fix: complete + deterministic dedup candidate set on sync (GIT-70) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The last item before any tag conversation exists. session_close now awaits this sync (GIT-69 item 3), so the window runs synchronously on the close path. The defect: an unordered `limit: 200` single page feeding the text-dedup map, against a table already at 256 rows. Every thread still synced — the per-thread lookup is an exact thread_id match, unbounded by the window — so what truncated was DEDUP COVERAGE. 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: 1. Pagination via directQueryAll, which already existed — checked before building, same as the harness. The candidate set is now actually complete rather than an arbitrary slice. 2. A deterministic total order (thread_id.asc). 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 the very gaps it was added to close. thread_id is unique, so it orders totally on its own. directQueryAll stops at its maxRows cap WITHOUT saying so. Inheriting that silence would rebuild the defect one layer up, so the cap is detected and reported: dedup_coverage is "complete" | "partial" | "unavailable", with dedup_candidates so the claim is auditable. Coverage is deliberately a separate axis from all_synced, and session_close surfaces it as a warning rather than a failure. Every thread syncs by identity regardless; incomplete coverage means a duplicate MAY exist — which all_synced structurally cannot detect, because writing a duplicate IS a successful write. That distinction is the whole ticket: the outcome contract makes the path honest about failures, not about correctness. Identity matching at the exact-thread_id lookup is untouched, per R8a. Tests: +5 (deterministic order asserted at the call; complete coverage past the old 200 window; cap reported as partial not absorbed; load failure reported as unavailable; sync still succeeds while coverage is incomplete). Also updated the 4 existing sync tests whose mocks assumed directQuery for the candidate load — they caught the change correctly. 1093 unit + 12 e2e green, typecheck and build clean. Refs GIT-70, GIT-69, GIT-63 Co-Authored-By: Claude Opus 5 --- src/services/thread-supabase.ts | 66 ++++++++++++-- src/tools/session-close.ts | 19 +++- .../unit/services/thread-sync-result.test.ts | 91 +++++++++++++++++-- 3 files changed, 159 insertions(+), 17 deletions(-) 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"); + }); +});