Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 49 additions & 6 deletions src/services/startup.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,9 @@

import * as fs from "fs";
import * as path from "path";
import { isConfigured, loadScarsWithEmbeddings, getLearningsFingerprint, getSupabaseUrl } from "./supabase-client.js";
import { computeStoreKey, readVectorCache, writeVectorCache, tryAcquireDownloadLock, waitForLeaderCache } from "./vector-disk-cache.js";
import { isConfigured, loadScarsWithEmbeddings, loadLearningsManifest, loadLearningsByIds, getLearningsFingerprint, getSupabaseUrl } from "./supabase-client.js";
import { computeStoreKey, readVectorCache, readVectorCacheBase, planDelta, assembleDelta, writeVectorCache, tryAcquireDownloadLock, waitForLeaderCache } from "./vector-disk-cache.js";
import type { StoreFingerprint } from "./vector-disk-cache.js";
import { getTableName } from "./tier.js";
import { getGitmemDir } from "./gitmem-dir.js";
import {
Expand Down Expand Up @@ -103,6 +104,38 @@ interface ScarWithEmbedding {
[key: string]: unknown;
}

/** Rows the bulk load and the delta manifest both cap at. */
const INDEX_ROW_LIMIT = 500;

/**
* GIT-98 per-row delta: bring a cache from this store up to date by fetching
* only the rows whose updated_at moved. Returns null — and the caller does the
* full download — when there is no base to reuse or anything goes wrong.
*/
async function deltaSyncFromCache(storeKey: string, fingerprint: StoreFingerprint): Promise<ScarWithEmbedding[] | null> {
const base = readVectorCacheBase<ScarWithEmbedding>(storeKey);
if (!base || base.length === 0) return null;
const startTime = Date.now();
try {
const manifest = await loadLearningsManifest(INDEX_ROW_LIMIT);
// An empty manifest is more likely a transient failure than an emptied
// store; let the full download decide.
if (manifest.length === 0) return null;
const plan = planDelta(base, manifest);
const fetched = plan.fetchIds.length > 0 ? await loadLearningsByIds<ScarWithEmbedding>(plan.fetchIds) : [];
const rows = assembleDelta(manifest, plan, fetched);
console.error(
`[startup] Vector cache DELTA: ${rows.length} learnings in ${Date.now() - startTime}ms — ` +
`${plan.reuse.size} reused from disk, ${fetched.length} fetched, ${plan.dropped} dropped (bulk download skipped)`
);
if (rows.length > 0) writeVectorCache(storeKey, fingerprint, rows);
return rows;
} catch (error) {
console.error("[startup] Delta sync failed, falling back to full download:", error instanceof Error ? error.message : error);
return null;
}
}

/**
* Load all learnings with embeddings from Supabase
*
Expand All @@ -112,7 +145,8 @@ interface ScarWithEmbedding {
* NOTE: Now loads all learning types (scars, patterns, wins, anti-patterns),
* not just scars. This fixes the issue where ~64 patterns were being ignored.
*/
async function loadScarsFromSupabase(options: { bypassDiskCache?: boolean } = {}): Promise<{
/** @internal exported for tests (GIT-98 delta sync). */
export async function loadScarsFromSupabase(options: { bypassDiskCache?: boolean } = {}): Promise<{
scars: ScarWithEmbedding[];
latestUpdatedAt: string | null;
source: "disk" | "network";
Expand Down Expand Up @@ -155,11 +189,20 @@ async function loadScarsFromSupabase(options: { bypassDiskCache?: boolean } = {}
}
releaseLock = releaseLock ?? (() => {});

console.error(`[startup] Loading ALL learnings with embeddings from Supabase (cross-project)`);

try {
// GIT-98: a miss is usually a handful of changed rows, not a new store.
// cache-flush (bypassDiskCache) always takes the full download.
if (!options.bypassDiskCache && fingerprint.count >= 0) {
const synced = await deltaSyncFromCache(storeKey, fingerprint);
if (synced) {
return { scars: synced, latestUpdatedAt: synced[0]?.updated_at || null, source: "network" };
}
}

console.error(`[startup] Loading ALL learnings with embeddings from Supabase (cross-project)`);

// Load ALL learnings across projects — semantic similarity handles relevance
const learnings = await loadScarsWithEmbeddings<ScarWithEmbedding>(undefined, 500);
const learnings = await loadScarsWithEmbeddings<ScarWithEmbedding>(undefined, INDEX_ROW_LIMIT);

const elapsed = Date.now() - startTime;
console.error(`[startup] Loaded ${learnings.length} learnings in ${elapsed}ms`);
Expand Down
56 changes: 49 additions & 7 deletions src/services/supabase-client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -518,6 +518,18 @@ export async function directPatch<T = unknown>(
*
* NOTE: Changed from "scars only" to "all learning types"
*/
/**
* The learnings the local index is built from. The bulk load, the fingerprint,
* the delta manifest and the delta fetch MUST all use these, or they describe
* different sets (GIT-98).
*/
const INDEXED_LEARNING_FILTERS: Record<string, string> = {
learning_type: "in.(scar,pattern,win,anti_pattern)",
is_active: "eq.true",
};
const INDEXED_LEARNING_COLUMNS =
"id,title,description,severity,counter_arguments,applies_when,source_linear_issue,project,embedding,updated_at,learning_type,decay_multiplier";

export async function loadScarsWithEmbeddings<T = unknown>(
project?: string,
limit = 500
Expand All @@ -526,16 +538,13 @@ export async function loadScarsWithEmbeddings<T = unknown>(
const startTime = Date.now();

try {
const filters: Record<string, string> = {
learning_type: "in.(scar,pattern,win,anti_pattern)",
is_active: "eq.true",
};
const filters: Record<string, string> = { ...INDEXED_LEARNING_FILTERS };
if (project) {
filters.project = project;
}

const learnings = await directQuery<T>(getTableName("learnings"), {
select: "id,title,description,severity,counter_arguments,applies_when,source_linear_issue,project,embedding,updated_at,learning_type,decay_multiplier",
select: INDEXED_LEARNING_COLUMNS,
filters,
order: "updated_at.desc",
limit,
Expand All @@ -551,6 +560,40 @@ export async function loadScarsWithEmbeddings<T = unknown>(
}
}

/**
* Delta manifest (GIT-98): the id and updated_at of every row the bulk load
* would return — same filters, order and limit — and nothing else. About 70
* bytes a row, versus ~20 KB a row with its 1536-dim embedding.
*/
export async function loadLearningsManifest(limit = 500): Promise<Array<{ id: string; updated_at: string | null }>> {
return directQuery<{ id: string; updated_at: string | null }>(getTableName("learnings"), {
select: "id,updated_at",
filters: { ...INDEXED_LEARNING_FILTERS },
order: "updated_at.desc",
limit,
});
}

/** Ids per delta request. A UUID is 36 characters, so this keeps URLs near 4 KB. */
const DELTA_FETCH_CHUNK = 100;

/**
* Delta fetch (GIT-98): full rows, embedding included, for the given ids only.
* The indexed-learning filters still apply, so a row deactivated since the
* manifest was read is simply absent.
*/
export async function loadLearningsByIds<T = unknown>(ids: string[]): Promise<T[]> {
const rows: T[] = [];
for (let i = 0; i < ids.length; i += DELTA_FETCH_CHUNK) {
const chunk = ids.slice(i, i + DELTA_FETCH_CHUNK);
rows.push(...await directQuery<T>(getTableName("learnings"), {
select: INDEXED_LEARNING_COLUMNS,
filters: { ...INDEXED_LEARNING_FILTERS, id: `in.(${chunk.join(",")})` },
}));
}
return rows;
}

/**
* Cheap fingerprint of the learnings the local index is built from (GIT-98).
*
Expand All @@ -569,8 +612,7 @@ export async function getLearningsFingerprint(): Promise<{ count: number; latest
try {
const url = new URL(`${SUPABASE_REST_URL}/${getTableName("learnings")}`);
url.searchParams.set("select", "updated_at");
url.searchParams.set("learning_type", "in.(scar,pattern,win,anti_pattern)");
url.searchParams.set("is_active", "eq.true");
for (const [key, value] of Object.entries(INDEXED_LEARNING_FILTERS)) url.searchParams.set(key, value);
url.searchParams.set("order", "updated_at.desc");
url.searchParams.set("limit", "1");

Expand Down
111 changes: 99 additions & 12 deletions src/services/vector-disk-cache.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,21 @@
* 1. Ask the store for a FINGERPRINT — row count + newest updated_at. One
* request, one row, a few hundred bytes (PostgREST `Prefer: count=exact`).
* 2. If a cache file with the same fingerprint exists, load vectors from disk.
* 3. Otherwise download as before and rewrite the cache.
* 3. Otherwise sync the difference (below) and rewrite the cache.
*
* ## Per-row delta sync
*
* A fingerprint miss used to mean a full download, and on a Pro store it
* misses on nearly every session_start: refresh_scar_behavioral_scores()
* rewrites every scar with enough recent usage and bumps its updated_at. So
* a handful of changed rows cost the whole index.
*
* On a miss with a usable cache from this store, only the manifest (id,
* updated_at; no vectors) is downloaded. A row whose id and updated_at match
* the cache is reused from disk; only changed and new ids are fetched in full;
* rows no longer in the manifest are dropped. The result is in manifest order,
* exactly what a full download would have returned, given that updated_at
* moves on every edit (the same assumption the fingerprint already makes).
*
* ## What the fingerprint can and cannot see
*
Expand Down Expand Up @@ -105,31 +119,104 @@ export function readVectorCache<T>(storeKey: string, expected: StoreFingerprint)
// failures would otherwise "agree" and serve whatever is on disk.
if (expected.count < 0) return { hit: false, reason: "store fingerprint unavailable" };

const file = readCacheFile<T>(storeKey);
if (!file.ok) return { hit: false, reason: file.reason };
if (!file.parsed.fingerprint || !fingerprintsMatch(file.parsed.fingerprint, expected)) {
return { hit: false, reason: "store changed since cache was written" };
}
return { hit: true, rows: file.parsed.rows, bytesOnDisk: file.bytesOnDisk };
}

/**
* The cached rows of this store, whatever fingerprint they were written under:
* the base a delta sync reuses unchanged rows from. Never served as-is — every
* row is checked against the store's manifest first.
*/
export function readVectorCacheBase<T>(storeKey: string): T[] | null {
if (!isVectorDiskCacheEnabled()) return null;
const file = readCacheFile<T>(storeKey);
return file.ok ? file.parsed.rows : null;
}

export interface ManifestRow {
id: string;
updated_at: string | null;
}

export interface DeltaPlan<T> {
/** Cached rows still current in the store, by id. */
reuse: Map<string, T>;
/** Ids to fetch in full: changed since the cache was written, or new. */
fetchIds: string[];
/** Cached rows the store no longer returns. */
dropped: number;
}

/**
* Compare the store's manifest with the cached rows. A row is reused only when
* its id AND updated_at match; a cached row without updated_at is refetched.
*/
export function planDelta<T extends { id: string; updated_at?: string | null }>(
cached: T[],
manifest: ManifestRow[]
): DeltaPlan<T> {
const cachedById = new Map(cached.map((r) => [r.id, r]));
const reuse = new Map<string, T>();
const fetchIds: string[] = [];
for (const m of manifest) {
const c = cachedById.get(m.id);
if (c && m.updated_at != null && c.updated_at === m.updated_at) reuse.set(m.id, c);
else fetchIds.push(m.id);
}
const inManifest = new Set(manifest.map((m) => m.id));
const dropped = cached.filter((r) => !inManifest.has(r.id)).length;
return { reuse, fetchIds, dropped };
}

/**
* Rows in manifest order: reused where current, fetched otherwise. A manifest
* id that is in neither (deactivated between the manifest and the fetch) is
* left out, as a full download at that moment would have.
*/
export function assembleDelta<T extends { id: string }>(
manifest: ManifestRow[],
plan: DeltaPlan<T>,
fetched: T[]
): T[] {
const fetchedById = new Map(fetched.map((r) => [r.id, r]));
const rows: T[] = [];
for (const m of manifest) {
const row = fetchedById.get(m.id) ?? plan.reuse.get(m.id);
if (row) rows.push(row);
}
return rows;
}

function readCacheFile<T>(storeKey: string):
| { ok: true; parsed: VectorCacheFile<T>; bytesOnDisk: number }
| { ok: false; reason: string } {
const file = getVectorCachePath(storeKey);
let raw: string;
try {
if (!fs.existsSync(file)) return { hit: false, reason: "no cache file" };
if (!fs.existsSync(file)) return { ok: false, reason: "no cache file" };
raw = fs.readFileSync(file, "utf-8");
} catch (error) {
return { hit: false, reason: `unreadable: ${error instanceof Error ? error.message : String(error)}` };
return { ok: false, reason: `unreadable: ${error instanceof Error ? error.message : String(error)}` };
}

let parsed: VectorCacheFile<T>;
try {
parsed = JSON.parse(raw) as VectorCacheFile<T>;
} catch {
return { hit: false, reason: "corrupt JSON" };
return { ok: false, reason: "corrupt JSON" };
}

if (!parsed || typeof parsed !== "object") return { hit: false, reason: "corrupt shape" };
if (parsed.format !== VECTOR_CACHE_FORMAT) return { hit: false, reason: `format ${parsed.format} != ${VECTOR_CACHE_FORMAT}` };
if (parsed.storeKey !== storeKey) return { hit: false, reason: "written for a different store" };
if (!Array.isArray(parsed.rows)) return { hit: false, reason: "corrupt rows" };
if (!parsed.fingerprint || !fingerprintsMatch(parsed.fingerprint, expected)) {
return { hit: false, reason: "store changed since cache was written" };
}
if (!parsed || typeof parsed !== "object") return { ok: false, reason: "corrupt shape" };
if (parsed.format !== VECTOR_CACHE_FORMAT) return { ok: false, reason: `format ${parsed.format} != ${VECTOR_CACHE_FORMAT}` };
if (parsed.storeKey !== storeKey) return { ok: false, reason: "written for a different store" };
if (!Array.isArray(parsed.rows)) return { ok: false, reason: "corrupt rows" };

return { hit: true, rows: parsed.rows, bytesOnDisk: Buffer.byteLength(raw) };
return { ok: true, parsed, bytesOnDisk: Buffer.byteLength(raw) };
}

/**
Expand Down
11 changes: 11 additions & 0 deletions tests/e2e/blank-supabase.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
* egress bytes received from the venue on a cold start, a warm start, and around
* two session_starts (GIT-98 disk cache). --usage N seeds N scars with
* >=3 usage rows so refresh_scar_behavioral_scores() has work to do.
* Fails if any start after the cold one downloads the whole index again
* (GIT-98 per-row delta).
*
* VENUE_ENV is a KEY=VALUE file OUTSIDE the repo defining SUPABASE_URL,
* SUPABASE_SERVICE_ROLE_KEY and VENUE_REF. Credentials are never committed.
Expand Down Expand Up @@ -520,6 +522,15 @@ const failureCheck = checkFailures(allVenueRequests);
const edgeCalls = allVenueRequests.filter((r) => r.path.startsWith("/functions/v1/"))
.map((r) => `${r.method} ${r.path} -> ${r.status}`);
if (edgeCalls.length) failureCheck.unexpected.push(...edgeCalls.map((c) => `edge function called (GIT-97): ${c}`));
// GIT-98 per-row delta: after the cold start, a changed store costs the changed
// rows, never the whole index again.
if (mode === "egress") {
for (const st of result.starts.slice(1)) {
if (st.cache_log.some((l) => /Loading ALL learnings/.test(l))) {
failureCheck.unexpected.push(`full index download after the cold start (GIT-98): ${st.start}, ${st.total_bytes} B`);
}
}
}
if (mode === "flow" && !result.session_close_persisted) {
failureCheck.unexpected.push(`session_close did not persist session ${result.session_id} (closing_reflection missing)`);
}
Expand Down
Loading
Loading