fix: a thread counts as synced only when the store confirms its row (GIT-117)

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;

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<ThreadRow>(
-      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<ThreadRow> {
+  const row = threadObjectToRow(thread, project, embedding);
+  const result = await supabase.directUpsert<ThreadRow>(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,40 +241,41 @@ export async function resolveThreadInSupabase(
   }
 
   try {
-    // First, find the UUID primary key for this thread_id
-    const rows = await supabase.directQuery<ThreadRow>(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<string, unknown> = {
-      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);
     return false;
   }
 }
 
+/**
+ * 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<void> {
+  const rows = await supabase.directQuery<ThreadRow>(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<string, unknown> = {
+    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<void> {
+): 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,28 +682,34 @@ 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);
           }
         }
       } 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) {

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.`,
         ]
       : [];
 

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

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;
 }

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)`);
 }