diff --git a/src/services/display-protocol.ts b/src/services/display-protocol.ts index ec712e8..5e0cd2c 100644 --- a/src/services/display-protocol.ts +++ b/src/services/display-protocol.ts @@ -156,6 +156,7 @@ export const STATUS = { rejected: `${red}REJECTED${reset}`, complete: `${green}COMPLETE${reset}`, failed: `${red}FAILED${reset}`, + partial: `${yellow}PARTIAL${reset}`, pass: `${green}+${reset}`, miss: `${red}-${reset}`, } as const; diff --git a/src/services/thread-supabase.ts b/src/services/thread-supabase.ts index 1494521..9674ae5 100644 --- a/src/services/thread-supabase.ts +++ b/src/services/thread-supabase.ts @@ -205,19 +205,24 @@ export async function createThreadInSupabase( } try { - const row = threadObjectToRow(thread, project, embedding); - const result = await supabase.directUpsert( - getTableName("threads"), - row - ); - console.error(`[thread-supabase] Created thread ${thread.id} in Supabase`); - return result; + return await writeThreadRow(thread, project, embedding); } catch (error) { console.error("[thread-supabase] Failed to create thread:", error instanceof Error ? error.message : error); return null; } } +/** + * GIT-117: create a thread row, or throw. directUpsert asks for the stored + * representation and throws on an empty one, so a return is a confirmed row. + */ +async function writeThreadRow(thread: ThreadObject, project: Project, embedding?: string | null): Promise { + const row = threadObjectToRow(thread, project, embedding); + const result = await supabase.directUpsert(getTableName("threads"), row); + console.error(`[thread-supabase] Created thread ${thread.id} in Supabase`); + return result; +} + /** * Resolve a thread in Supabase by thread_id. * Updates status, resolved_at, resolution_note, resolved_by_session. @@ -236,33 +241,7 @@ export async function resolveThreadInSupabase( } try { - // First, find the UUID primary key for this thread_id - const rows = await supabase.directQuery(getTableName("threads"), { - select: "id,thread_id", - filters: { thread_id: threadId }, - limit: 1, - }); - - if (rows.length === 0) { - console.error(`[thread-supabase] Thread ${threadId} not found in Supabase (will proceed with local-only)`); - return false; - } - - const uuid = rows[0].id; - const patchData: Record = { - status: "resolved", - resolved_at: options.resolvedAt || new Date().toISOString(), - }; - - if (options.resolutionNote) { - patchData.resolution_note = options.resolutionNote; - } - if (options.resolvedBySession) { - patchData.resolved_by_session = options.resolvedBySession; - } - - await supabase.directPatch(getTableName("threads"), { id: uuid }, patchData); - console.error(`[thread-supabase] Resolved thread ${threadId} in Supabase`); + await resolveThreadRow(threadId, options); return true; } catch (error) { console.error("[thread-supabase] Failed to resolve thread:", error instanceof Error ? error.message : error); @@ -270,6 +249,33 @@ export async function resolveThreadInSupabase( } } +/** + * GIT-117: resolve a thread row, or throw. Confirmed only when the PATCH + * returns the row: an empty result means nothing was updated. + */ +async function resolveThreadRow( + threadId: string, + options: { resolvedAt?: string; resolutionNote?: string; resolvedBySession?: string } +): Promise { + const rows = await supabase.directQuery(getTableName("threads"), { + select: "id,thread_id", + filters: { thread_id: threadId }, + limit: 1, + }); + if (rows.length === 0) throw new Error(`thread ${threadId} not found in the store`); + + const patchData: Record = { + status: "resolved", + resolved_at: options.resolvedAt || new Date().toISOString(), + }; + if (options.resolutionNote) patchData.resolution_note = options.resolutionNote; + if (options.resolvedBySession) patchData.resolved_by_session = options.resolvedBySession; + + const patched = await supabase.directPatch(getTableName("threads"), { id: rows[0].id }, patchData); + if (!Array.isArray(patched) || patched.length === 0) throw new Error(`resolve of ${threadId} updated no row`); + console.error(`[thread-supabase] Resolved thread ${threadId} in Supabase`); +} + /** * Fetch a single thread from Supabase by its thread_id ("t-XXXX"). * @@ -508,9 +514,10 @@ export async function loadActiveThreadsFromSupabase( */ export async function touchThreadsInSupabase( threadIds: string[] -): Promise { +): Promise<{ touched: string[]; failed: { id: string; error: string }[] }> { + const outcome = { touched: [] as string[], failed: [] as { id: string; error: string }[] }; if (!hasSupabase() || !supabase.isConfigured() || threadIds.length === 0) { - return; + return outcome; } for (const threadId of threadIds) { @@ -522,12 +529,17 @@ export async function touchThreadsInSupabase( limit: 1, }); - if (rows.length === 0) continue; + // GIT-117: a thread we believe exists but the store does not have is a + // failed write, not a no-op. + if (rows.length === 0) throw new Error(`thread ${threadId} not found in the store`); const row = rows[0]; // Skip resolved/archived threads — no point recomputing vitality - if (row.status === "resolved" || row.status === "archived") continue; + if (row.status === "resolved" || row.status === "archived") { + outcome.touched.push(threadId); + continue; + } const now = new Date(); const newTouchCount = (row.touch_count || 0) + 1; @@ -551,18 +563,23 @@ export async function touchThreadsInSupabase( delete metadata.dormant_since; } - await supabase.directPatch(getTableName("threads"), { id: row.id }, { + const patched = await supabase.directPatch(getTableName("threads"), { id: row.id }, { touch_count: newTouchCount, last_touched_at: nowIso, vitality_score: vitality.vitality_score, status: lifecycle_status, metadata, }); + if (!Array.isArray(patched) || patched.length === 0) throw new Error(`touch of ${threadId} updated no row`); + outcome.touched.push(threadId); } catch (error) { - console.error(`[thread-supabase] Failed to touch thread ${threadId}:`, error instanceof Error ? error.message : error); + const message = error instanceof Error ? error.message : String(error); + console.error(`[thread-supabase] Failed to touch thread ${threadId}:`, message); + outcome.failed.push({ id: threadId, error: message }); // Continue with other threads } } + return outcome; } /** @@ -665,13 +682,18 @@ export async function syncThreadsToSupabase( const normalizedNewText = normalizeText(thread.text || ""); const matchedThreadId = normalizedNewText ? textToExistingId.get(normalizedNewText) : undefined; + // GIT-117: every write below throws unless the store confirmed a row. + // The helpers this loop called before swallowed their failures and + // returned null / false / void, so a thread was counted as synced — + // and pruned from threads.json — when nothing had been written. if (matchedThreadId) { // Duplicate text found — touch existing instead of creating console.error(`[thread-supabase] Dedup: "${thread.id}" matches existing "${matchedThreadId}" by text — touching instead of creating`); - await touchThreadsInSupabase([matchedThreadId]); + const touched = await touchThreadsInSupabase([matchedThreadId]); + if (touched.failed.length > 0) throw new Error(touched.failed[0].error); } else { // Genuinely new thread — create it - await createThreadInSupabase(thread, project); + await writeThreadRow(thread, project); // Register in lookup so subsequent threads in this batch also dedup if (normalizedNewText) { textToExistingId.set(normalizedNewText, thread.id); @@ -679,14 +701,15 @@ export async function syncThreadsToSupabase( } } else if (thread.status === "resolved" && existing[0].status !== "resolved") { // Thread was resolved during this session - await resolveThreadInSupabase(thread.id, { + await resolveThreadRow(thread.id, { resolvedAt: thread.resolved_at, resolutionNote: thread.resolution_note, resolvedBySession: thread.resolved_by_session || sessionId, }); } else { // Existing thread, just touch it - await touchThreadsInSupabase([thread.id]); + const touched = await touchThreadsInSupabase([thread.id]); + if (touched.failed.length > 0) throw new Error(touched.failed[0].error); } synced.push(thread.id); } catch (error) { diff --git a/src/tools/session-close.ts b/src/tools/session-close.ts index f398efc..ff0ee76 100644 --- a/src/tools/session-close.ts +++ b/src/tools/session-close.ts @@ -16,7 +16,7 @@ import { getStorage } from "../services/storage.js"; import { filterToStoreSessionColumns } from "../services/session-columns.js"; import { storeHasTable } from "../services/store-columns.js"; import { clearCurrentSession, resolveCurrentSession, getSurfacedScars, getConfirmations, getReflections, getObservations, getChildren, getThreads, getSessionActivity, isRecallCalled } from "../services/session-state.js"; -import { normalizeThreads, mergeThreadStates, migrateStringThread, saveThreadsFile } from "../services/thread-manager.js"; // +import { normalizeThreads, mergeThreadStates, migrateStringThread, saveThreadsFile, loadThreadsFile } from "../services/thread-manager.js"; // import { deduplicateThreadList } from "../services/thread-dedup.js"; import { syncThreadsToSupabase, loadOpenThreadEmbeddings } from "../services/thread-supabase.js"; import type { ThreadSyncResult } from "../services/thread-supabase.js"; @@ -369,8 +369,9 @@ function formatCloseDisplay( ): string { const lines: string[] = []; - // Header: branded product line - const status = success ? STATUS.complete : STATUS.failed; + // Header: branded product line. GIT-117: the session saved but some threads + // did not — PARTIAL, not FAILED (the session is there) and not COMPLETE. + const status = success ? STATUS.complete : sessionStored ? STATUS.partial : STATUS.failed; lines.push(productLine("close", status)); // Stats line: compact one-liner with key counts @@ -1445,8 +1446,19 @@ export async function sessionClose( `[session_close] Local thread prune SKIPPED — ${threadSync.failed.length} thread(s) did not reach Supabase (${threadSync.failed.map((f) => f.id).join(", ")}). ` + `threads.json retained in full; it is the only record of the unsynced threads.` ); + // GIT-117: mark what did not land so the next session_start keeps it + // (whatever its age) and the next close writes it again. + const failedIds = new Set(threadSync.failed.map((f) => f.id)); + const fileThreads = loadThreadsFile(); + const byId = new Map(fileThreads.map((t) => [t.id, t])); + for (const t of closeThreads) { + if (failedIds.has(t.id)) byId.set(t.id, { ...(byId.get(t.id) ?? {}), ...t, sync_pending: true }); + } + saveThreadsFile([...byId.values()]); } else { - const openThreadsOnly = closeThreads.filter(t => t.status === "open" || !t.status); + const openThreadsOnly = closeThreads + .filter(t => t.status === "open" || !t.status) + .map(({ sync_pending: _written, ...t }) => t); // GIT-117: written now saveThreadsFile(openThreadsOnly); console.error(`[session_close] Pruned threads.json: ${openThreadsOnly.length} open threads (removed ${closeThreads.length - openThreadsOnly.length} resolved/archived)`); } @@ -1702,8 +1714,8 @@ export async function sessionClose( const persistErrors = partialPersist ? [ - `Thread sync incomplete: ${threadSync.failed.length} of ${threadSync.attempted} thread(s) did not reach Supabase — ${threadSync.failed.map((f) => `${f.id} (${f.error})`).join("; ")}. ` + - `threads.json was NOT pruned and session files were retained; re-run session_close to retry.`, + `PARTIAL: the session saved, but ${threadSync.failed.length} of ${threadSync.attempted} thread(s) did not reach Supabase — ${threadSync.failed.map((f) => `${f.id} (${f.error})`).join("; ")}. ` + + `They are kept in threads.json and carried into the next session, whose close writes them again; session files were retained.`, ] : []; diff --git a/src/tools/session-start.ts b/src/tools/session-start.ts index b177c3c..0d8ce51 100644 --- a/src/tools/session-start.ts +++ b/src/tools/session-start.ts @@ -870,9 +870,17 @@ function writeSessionFiles( // because the NOT-IN query excludes them, making them look "local-only". const supabaseIds = new Set(threads.map(t => t.id)); const cutoff = Date.now() - 24 * 60 * 60 * 1000; + // GIT-117: a thread a close could not write is kept whatever its age, and + // its local state (e.g. a resolution) wins over the store's copy until a + // later close writes it. + const pending = new Map(existingFileThreads.filter(t => t.sync_pending && t.id).map(t => [t.id, t])); + if (pending.size > 0) { + console.error(`[session_start] Carrying forward ${pending.size} thread(s) a previous close could not write: ${[...pending.keys()].join(", ")}`); + } const localOnlyThreads = existingFileThreads.filter(t => { if (supabaseIds.has(t.id)) return false; // exists in Supabase active set if (!t.id) return false; // no ID = malformed, drop + if (t.sync_pending) return true; const created = t.created_at ? new Date(t.created_at).getTime() : 0; return created > cutoff; // only keep if created within last 24h }); @@ -883,7 +891,7 @@ function writeSessionFiles( if (dropped > 0) { console.error(`[session_start] Dropped ${dropped} stale local-only threads (resolved/archived in Supabase)`); } - merged = deduplicateThreadList([...threads, ...localOnlyThreads]); + merged = deduplicateThreadList([...threads.map(t => pending.get(t.id) ?? t), ...localOnlyThreads]); } else { // Fallback (free tier / Supabase offline): merge with existing file merged = existingFileThreads.length > 0 diff --git a/src/types/index.ts b/src/types/index.ts index 6de5c9b..383344a 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -86,6 +86,12 @@ export interface ThreadObject { source_session?: string; /** Session ID that resolved this thread */ resolved_by_session?: string; + /** + * GIT-117: a close could not write this thread (or its resolution) to the + * durable store. Kept locally, whatever its age, and preferred over the + * store's copy until a later close writes it. + */ + sync_pending?: boolean; /** Brief resolution note */ resolution_note?: string; } diff --git a/tests/e2e/blank-supabase.mjs b/tests/e2e/blank-supabase.mjs index 0a16761..5b61093 100644 --- a/tests/e2e/blank-supabase.mjs +++ b/tests/e2e/blank-supabase.mjs @@ -422,6 +422,56 @@ async function flow() { const relevance = relevanceRows.map((m) => ({ tool: m.tool_name, memories_applied: m.metadata.memories_applied, memory_relevance: m.metadata.memory_relevance })); await waitQuiet(netlog, { minMs: 1000, quietMs: 2000 }); await srv.close(); + + // GIT-117: a close whose thread write the store rejects. A thread naming a + // session the store has never seen violates the source_session FK (23503). + // The close must say PARTIAL and name it, keep it in threads.json marked + // sync_pending, and the next session_start must carry it forward. + const threadSyncProbe = async () => { + const h = isolatedHome(`${label}-threadsync`); + const probeLog = join(outDir, `netlog-${label}-threadsync.jsonl`); + writeFileSync(probeLog, ""); + const badId = `t-fk${randomUUID().slice(0, 6)}`; + const errs1 = []; + const p1 = await startServer({ home: h.home, gitmemDir: h.gitmemDir, ollamaUrl: ollama.url, netlog: probeLog, stderrSink: errs1 }); + const ss1 = await p1.call("session_start", { project: PROJECT, agent_identity: "cli", force: true }); + const sid1 = (ss1.match(UUID) || [])[0]; + await waitQuiet(probeLog, { minMs: 1500, quietMs: 2000 }); + const close = await p1.call("session_close", { + session_id: sid1, close_type: "standard", human_corrections: "none", + open_threads: [{ id: badId, text: `Thread whose write the store rejects (${label})`, status: "open", + created_at: new Date().toISOString(), source_session: randomUUID() }], + closing_reflection: { + what_broke: "n/a", what_took_longer: "n/a", do_differently: "n/a", what_worked: "n/a", + wrong_assumption: "n/a", scars_applied: [], institutional_memory_items: "n/a", + collaborative_dynamic: "n/a", rapport_notes: "n/a", + }, + }); + await p1.close(); + const threadsFile = join(h.gitmemDir, "threads.json"); + const local = existsSync(threadsFile) ? JSON.parse(readFileSync(threadsFile, "utf8")) : []; + const localEntry = local.find((t) => t.id === badId); + const inStore = await rest("GET", `gitmem_threads?select=id&thread_id=eq.${badId}`, undefined, { Prefer: "" }).then((r) => r.json()); + const errs2 = []; + const p2 = await startServer({ home: h.home, gitmemDir: h.gitmemDir, ollamaUrl: ollama.url, netlog: probeLog, stderrSink: errs2 }); + await p2.call("session_start", { project: PROJECT, agent_identity: "cli" }); + await p2.close(); + allVenueRequests.push(...venueReqs(readNet(probeLog)).filter((r) => !(r.method === "POST" && /gitmem_threads$/.test(r.path) && r.status === 409))); + const after = existsSync(threadsFile) ? JSON.parse(readFileSync(threadsFile, "utf8")) : []; + writeFileSync(join(outDir, `stderr-${label}-threadsync-restart.log`), errs2.join("")); + writeFileSync(join(outDir, `threads-${label}-threadsync-after.json`), JSON.stringify(after, null, 2)); + const plain = close.replace(/\x1b\[[0-9;]*m/g, ""); + return { + bad_thread: badId, + store_rejected_it: inStore.length === 0, + close_says_partial: /close · PARTIAL/.test(plain) && plain.includes(badId), + kept_locally_pending: !!localEntry && localEntry.sync_pending === true, + carried_into_next_session: after.some((t) => t.id === badId) && + errs2.join("").split("\n").some((l) => l.includes("Carrying forward") && l.includes(badId)), + close_head: plain.split("\n").slice(0, 2).join(" | "), + }; + }; + const threadSync = await threadSyncProbe(); ollama.close(); const reqs = readNet(netlog); @@ -441,6 +491,7 @@ async function flow() { remote_sessions_after: await count("gitmem_sessions"), remote_scar_usage_after: await count("gitmem_scar_usage"), session_close_persisted: closedRows.length === 1 && closedRows[0].closing_reflection != null, + thread_sync: threadSync, missing_payload_named: missingPayloadNamed, resolve_thread_durable: resolveDurable, close_warns: closeWarns, @@ -648,6 +699,13 @@ if (mode === "flow" && !result.thread_triples_linked) { if (mode === "flow" && result.false_stranded_notice) { failureCheck.unexpected.push("session_start reported the store it reads as 'NOT being read' (GIT-107)"); } +if (mode === "flow") { + const t = result.thread_sync; + if (!t.store_rejected_it) failureCheck.unexpected.push(`thread-sync probe: the store accepted ${t.bad_thread}; the probe did not exercise a failed write (GIT-117)`); + else if (!(t.close_says_partial && t.kept_locally_pending && t.carried_into_next_session)) { + failureCheck.unexpected.push(`a rejected thread write was not reported PARTIAL / kept pending / carried forward (GIT-117): ${JSON.stringify(t)}`); + } +} if (mode === "flow" && !result.session_close_persisted) { failureCheck.unexpected.push(`session_close did not persist session ${result.session_id} (closing_reflection missing)`); } diff --git a/tests/unit/services/thread-sync-result.test.ts b/tests/unit/services/thread-sync-result.test.ts index ba5e976..338ed91 100644 --- a/tests/unit/services/thread-sync-result.test.ts +++ b/tests/unit/services/thread-sync-result.test.ts @@ -16,6 +16,7 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; const mockDirectQuery = vi.fn(); const mockDirectQueryAll = vi.fn(); const mockDirectUpsert = vi.fn(); +const mockDirectPatch = vi.fn(); const mockIsConfigured = vi.fn(() => true); const mockHasSupabase = vi.fn(() => true); @@ -24,6 +25,7 @@ vi.mock("../../../src/services/supabase-client.js", () => ({ directQuery: (...a: unknown[]) => mockDirectQuery(...a), directQueryAll: (...a: unknown[]) => mockDirectQueryAll(...a), directUpsert: (...a: unknown[]) => mockDirectUpsert(...a), + directPatch: (...a: unknown[]) => mockDirectPatch(...a), upsertRecord: vi.fn(), listRecords: vi.fn(), })); @@ -55,6 +57,8 @@ beforeEach(() => { mockIsConfigured.mockReturnValue(true); mockHasSupabase.mockReturnValue(true); mockDirectUpsert.mockResolvedValue([{ id: "row" }]); + mockDirectPatch.mockReset(); + mockDirectPatch.mockResolvedValue([{ id: "row" }]); }); describe("tier awareness (R3)", () => { @@ -236,3 +240,81 @@ describe("dedup candidate coverage (GIT-70)", () => { expect(r.dedup_coverage).not.toBe("complete"); }); }); + +// --------------------------------------------------------------------------- +// GIT-117: the WRITE fails, not the lookup +// --------------------------------------------------------------------------- +// The tests above fail only directQuery, the one call whose error propagated. +// The create / resolve / touch helpers caught their own failures and returned +// null / false / void, which the loop never read — so a thread whose write +// failed was counted as synced, and session_close pruned it from threads.json. + +describe("GIT-117: a thread counts as synced only when the store confirms a row", () => { + it("create rejected by the store -> failed, not synced", async () => { + mockDirectQuery.mockResolvedValue([]); // not in the store yet -> create + mockDirectUpsert.mockRejectedValue(new Error("Supabase upsert error: 409 - violates foreign key constraint")); + + const r = await syncThreadsToSupabase([thread("t-new")], "gitmem", "s-1"); + + expect(r.synced).toEqual([]); + expect(r.failed.map((f) => f.id)).toEqual(["t-new"]); + expect(r.failed[0].error).toMatch(/foreign key/); + expect(r.all_synced).toBe(false); + }); + + it("resolve whose PATCH updates no row -> failed", async () => { + mockDirectQuery + .mockResolvedValueOnce([{ id: "u1", thread_id: "t-r", status: "open" }]) // existence + .mockResolvedValueOnce([{ id: "u1", thread_id: "t-r" }]); // resolve lookup + mockDirectPatch.mockResolvedValue([]); + + const r = await syncThreadsToSupabase([thread("t-r", "resolved")], "gitmem", "s-1"); + + expect(r.failed.map((f) => f.id)).toEqual(["t-r"]); + expect(r.failed[0].error).toMatch(/updated no row/); + expect(r.all_synced).toBe(false); + }); + + it("resolve rejected by the store -> failed", async () => { + mockDirectQuery + .mockResolvedValueOnce([{ id: "u1", thread_id: "t-r", status: "open" }]) + .mockResolvedValueOnce([{ id: "u1", thread_id: "t-r" }]); + mockDirectPatch.mockRejectedValue(new Error("Supabase patch error: 503")); + + const r = await syncThreadsToSupabase([thread("t-r", "resolved")], "gitmem", "s-1"); + expect(r.failed.map((f) => f.id)).toEqual(["t-r"]); + }); + + it("touch of an existing thread whose PATCH fails -> failed", async () => { + mockDirectQuery + .mockResolvedValueOnce([{ id: "u1", thread_id: "t-o", status: "active" }]) // existence + .mockResolvedValueOnce([{ id: "u1", touch_count: 1, created_at: "2026-08-01T00:00:00Z", thread_class: "backlog", status: "active" }]); + mockDirectPatch.mockRejectedValue(new Error("Supabase patch error: 500")); + + const r = await syncThreadsToSupabase([thread("t-o")], "gitmem", "s-1"); + expect(r.failed.map((f) => f.id)).toEqual(["t-o"]); + expect(r.all_synced).toBe(false); + }); + + it("one failure among several: the others are synced, only the failure is reported", async () => { + mockDirectQuery.mockResolvedValue([]); + mockDirectUpsert + .mockResolvedValueOnce([{ id: "a" }]) + .mockRejectedValueOnce(new Error("boom")) + .mockResolvedValueOnce([{ id: "c" }]); + + const r = await syncThreadsToSupabase([thread("t-a"), thread("t-b"), thread("t-c")], "gitmem", "s-1"); + expect(r.synced).toEqual(["t-a", "t-c"]); + expect(r.failed.map((f) => f.id)).toEqual(["t-b"]); + }); + + it("every write confirmed -> all synced (the contract did not get stricter than the store)", async () => { + mockDirectQuery + .mockResolvedValueOnce([]) // t-new: create + .mockResolvedValueOnce([{ id: "u1", thread_id: "t-r", status: "open" }]) // t-r: exists + .mockResolvedValueOnce([{ id: "u1", thread_id: "t-r" }]); // t-r: resolve lookup + const r = await syncThreadsToSupabase([thread("t-new"), thread("t-r", "resolved")], "gitmem", "s-1"); + expect(r.failed).toEqual([]); + expect(r.all_synced).toBe(true); + }); +});