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
57 changes: 49 additions & 8 deletions src/services/triple-writer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@

import { v4 as uuidv4 } from "uuid";
import * as supabase from "./supabase-client.js";
import { hasSupabase } from "./tier.js";
import { hasSupabase, getTableName } from "./tier.js";

// --- Types ---

Expand All @@ -32,7 +32,8 @@ interface TripleCandidate {
predicate: Predicate;
object: string;
source_type: string;
source_id: string;
/** knowledge_triples.source_id is a UUID column: a row id, or null. */
source_id: string | null;
source_linear_issue?: string;
domain?: string[];
project: string;
Expand Down Expand Up @@ -282,11 +283,24 @@ export function extractDecisionTriples(params: DecisionTripleParams): TripleCand
* failure (GIT-104) — callers run this under getEffectTracker().track(), which
* never rethrows.
*/
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;

export async function writeTriples(candidates: TripleCandidate[]): Promise<number> {
if (candidates.length === 0 || !hasSupabase()) {
return 0;
}

// GIT-105: source_id is a UUID column in every schema. A non-UUID value
// (thread ids are "t-xxxxxxxx") failed the whole row with 22P02, so every
// thread triple 400'd and was lost. Never send one; the subject/object
// labels still carry the relation.
for (const c of candidates) {
if (c.source_id !== null && !UUID_RE.test(c.source_id)) {
console.error(`[triple-writer] source_id "${c.source_id}" is not a UUID — writing the triple with source_id null`);
c.source_id = null;
}
}

let written = 0;
const errors: string[] = [];

Expand Down Expand Up @@ -356,6 +370,12 @@ export function writeTriplesForDecision(params: DecisionTripleParams): Promise<n

export interface ThreadCreationTripleParams {
thread_id: string;
/**
* GIT-105: the thread's row id (gitmem_threads.id, a UUID) — what
* source_id references. thread_id ("t-xxxxxxxx") does not fit the column.
* Undefined: looked up by thread_id when writing. Null: not in the store.
*/
thread_row_id?: string | null;
text: string;
linear_issue?: string;
session_id?: string;
Expand All @@ -372,7 +392,7 @@ export function extractThreadCreationTriples(params: ThreadCreationTripleParams)
const threadLabel = buildSubjectLabel("thread", params.text);
const base = {
source_type: "thread",
source_id: params.thread_id,
source_id: params.thread_row_id ?? null,
source_linear_issue: params.linear_issue,
project: params.project,
half_life_days: HALF_LIFE_PROCESS,
Expand Down Expand Up @@ -404,6 +424,8 @@ export function extractThreadCreationTriples(params: ThreadCreationTripleParams)

export interface ThreadResolutionTripleParams {
thread_id: string;
/** GIT-105: see ThreadCreationTripleParams.thread_row_id. */
thread_row_id?: string | null;
text: string;
resolution_note?: string;
session_id?: string;
Expand All @@ -420,7 +442,7 @@ export function extractThreadResolutionTriples(params: ThreadResolutionTriplePar
const threadLabel = buildSubjectLabel("thread", params.text);
const base = {
source_type: "thread",
source_id: params.thread_id,
source_id: params.thread_row_id ?? null,
project: params.project,
half_life_days: HALF_LIFE_PROCESS,
created_by: params.agent,
Expand All @@ -443,16 +465,35 @@ export function extractThreadResolutionTriples(params: ThreadResolutionTriplePar
* Generate and write triples for a newly created thread.
* Fire-and-forget — run under getEffectTracker().track(); rejects if any triple failed.
*/
export function writeTriplesForThreadCreation(params: ThreadCreationTripleParams): Promise<number> {
const triples = extractThreadCreationTriples(params);
export async function writeTriplesForThreadCreation(params: ThreadCreationTripleParams): Promise<number> {
const triples = extractThreadCreationTriples(await withThreadRowId(params));
return writeTriples(triples);
}

/**
* GIT-105: fill thread_row_id from the store when the caller did not have it.
* A failed or empty lookup leaves it null — the triple is still written.
*/
async function withThreadRowId<T extends { thread_id: string; thread_row_id?: string | null }>(params: T): Promise<T> {
if (params.thread_row_id !== undefined || !hasSupabase()) return params;
try {
const rows = await supabase.directQuery<{ id: string }>(getTableName("threads"), {
select: "id",
filters: { thread_id: params.thread_id },
limit: 1,
});
return { ...params, thread_row_id: rows[0]?.id ?? null };
} catch (error) {
console.error(`[triple-writer] Could not look up the row id of ${params.thread_id}:`, error instanceof Error ? error.message : error);
return { ...params, thread_row_id: null };
}
}

/**
* Generate and write triples for a resolved thread.
* Fire-and-forget — run under getEffectTracker().track(); rejects if any triple failed.
*/
export function writeTriplesForThreadResolution(params: ThreadResolutionTripleParams): Promise<number> {
const triples = extractThreadResolutionTriples(params);
export async function writeTriplesForThreadResolution(params: ThreadResolutionTripleParams): Promise<number> {
const triples = extractThreadResolutionTriples(await withThreadRowId(params));
return writeTriples(triples);
}
3 changes: 3 additions & 0 deletions src/tools/create-thread.ts
Original file line number Diff line number Diff line change
Expand Up @@ -296,12 +296,14 @@ export async function createThread(
// Write to Supabase (source of truth) with embedding.
let supabaseSynced = false;
let supabaseError: string | undefined;
let threadRowId: string | null = null; // GIT-105: gitmem_threads.id, for triple source_id
const embeddingJson = newEmbedding ? JSON.stringify(newEmbedding) : null;

if (hasSupabase()) {
try {
const supabaseResult = await createThreadInSupabase(thread, project, embeddingJson);
supabaseSynced = Boolean(supabaseResult);
threadRowId = supabaseResult?.id ?? null;
if (!supabaseSynced) {
supabaseError = "write returned no row";
}
Expand Down Expand Up @@ -347,6 +349,7 @@ export async function createThread(
getEffectTracker().track("triple_write", "thread_creation", () =>
writeTriplesForThreadCreation({
thread_id: thread.id,
thread_row_id: threadRowId,
text: trimmedText,
linear_issue: params.linear_issue,
session_id: sessionId,
Expand Down
11 changes: 10 additions & 1 deletion tests/e2e/blank-supabase.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,6 @@ function isCapabilityProbe(r) {

const EXPECTED_FAILURES = [
{ ticket: "GIT-73", what: "metrics/session_start FK race", method: "POST", path: /^\/rest\/v1\/gitmem_query_metrics$/, status: 409 },
{ ticket: "GIT-105", what: "knowledge-triple thread id into a uuid column", method: "POST", path: /^\/rest\/v1\/knowledge_triples$/, status: 400 },
];
const HERE = dirname(fileURLToPath(import.meta.url));
const REPO = resolve(HERE, "../..");
Expand Down Expand Up @@ -377,6 +376,11 @@ async function flow() {
await waitQuiet(netlog, { minMs: 2000, quietMs: 3000 });
const health = await step("health", { failure_limit: 20 });

// GIT-105: thread triples land, and their source_id is a real thread row id.
const threadTriples = await rest("GET", "knowledge_triples?select=predicate,source_id&source_type=eq.thread", undefined, { Prefer: "" }).then((r) => r.json());
const threadRowIds = new Set((await rest("GET", "gitmem_threads?select=id", undefined, { Prefer: "" }).then((r) => r.json())).map((t) => t.id));
const threadTriplesLinked = threadTriples.length > 0 && threadTriples.every((t) => threadRowIds.has(t.source_id));

// Did the close land? Is relevance readable from the store (GIT-109)?
const closedRows = sessionId
? await rest("GET", `gitmem_sessions?select=id,closing_reflection&id=eq.${sessionId}`, undefined, { Prefer: "" }).then((r) => r.json())
Expand Down Expand Up @@ -408,6 +412,8 @@ 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_triples: threadTriples.length,
thread_triples_linked: threadTriplesLinked,
relevance_readable: relevance.some((r) => scarIds.some((id) => (r.memories_applied || []).includes(id) && r.memory_relevance?.[id])),
relevance,
health_failed_total: failed,
Expand Down Expand Up @@ -520,6 +526,9 @@ 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}`));
if (mode === "flow" && !result.thread_triples_linked) {
failureCheck.unexpected.push(`thread triples missing or not linked to a gitmem_threads row (GIT-105): ${result.thread_triples} found`);
}
if (mode === "flow" && !result.session_close_persisted) {
failureCheck.unexpected.push(`session_close did not persist session ${result.session_id} (closing_reflection missing)`);
}
Expand Down
68 changes: 68 additions & 0 deletions tests/unit/services/thread-triples-write.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
/**
* GIT-105: thread triples reach knowledge_triples on a v1.8.0 store.
*
* source_id is a UUID column; thread ids are "t-xxxxxxxx". Every
* create_thread / resolve_thread triple 400'd (22P02) and the failure was
* swallowed. The writer now stores the thread's row id (gitmem_threads.id),
* looking it up when the caller does not have it, and never sends a non-UUID.
*/
import { describe, it, expect, vi, beforeEach } from "vitest";

const store = vi.hoisted(() => ({
upserts: [] as Array<Record<string, unknown>>,
queries: [] as Array<{ table: string; filters: Record<string, string> }>,
threadRows: {} as Record<string, string>,
queryFails: false,
}));

vi.mock("../../../src/services/supabase-client.js", () => ({
directUpsert: vi.fn(async (_t: string, row: Record<string, unknown>) => { store.upserts.push(row); return row; }),
directQuery: vi.fn(async (table: string, opts: { filters: Record<string, string> }) => {
store.queries.push({ table, filters: opts.filters });
if (store.queryFails) throw new Error("store down");
const id = store.threadRows[opts.filters.thread_id];
return id ? [{ id }] : [];
}),
}));
vi.mock("../../../src/services/tier.js", async (orig) => ({
...(await orig<typeof import("../../../src/services/tier.js")>()),
hasSupabase: () => true,
}));

import { writeTriplesForThreadCreation, writeTriplesForThreadResolution, writeTriples } from "../../../src/services/triple-writer.js";

const ROW = "7d1c9a52-3f4e-4b8a-9c21-5e6f7a8b9c0d";
const SESSION = "550e8400-e29b-41d4-a716-446655440000";
const base = { thread_id: "t-abc12345", text: "Fix auth timeout", session_id: SESSION, project: "p", agent: "cli" };

beforeEach(() => {
store.upserts = []; store.queries = []; store.threadRows = {}; store.queryFails = false;
});

describe("thread triples on a UUID source_id column (GIT-105)", () => {
it("creation: uses the row id create_thread already has — no lookup", async () => {
await writeTriplesForThreadCreation({ ...base, thread_row_id: ROW });
expect(store.queries).toHaveLength(0);
expect(store.upserts.map((r) => r.source_id)).toEqual([ROW]);
});

it("resolution: looks the row id up by thread_id", async () => {
store.threadRows["t-abc12345"] = ROW;
await writeTriplesForThreadResolution(base);
expect(store.queries).toEqual([{ table: "gitmem_threads", filters: { thread_id: "t-abc12345" } }]);
expect(store.upserts.map((r) => r.source_id)).toEqual([ROW]);
});

it("thread not in the store, or the lookup fails: source_id null, triple still written", async () => {
await writeTriplesForThreadResolution(base);
store.queryFails = true;
await writeTriplesForThreadResolution(base);
expect(store.upserts.map((r) => r.source_id)).toEqual([null, null]);
});

it("never sends a non-UUID source_id, whatever the caller passed", async () => {
await writeTriples([{ subject: "s", predicate: "created_thread", object: "o", source_type: "thread",
source_id: "t-abc12345", project: "p", half_life_days: 9999, created_by: "cli" }]);
expect(store.upserts[0].source_id).toBeNull();
});
});
21 changes: 19 additions & 2 deletions tests/unit/services/thread-triples.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,13 @@ import type {

// ---------- Helpers ----------

// GIT-105: source_id is the thread's row id (gitmem_threads.id, a UUID), not
// its "t-" thread_id — knowledge_triples.source_id is a UUID column.
const THREAD_ROW_ID = "7d1c9a52-3f4e-4b8a-9c21-5e6f7a8b9c0d";

const BASE_CREATION_PARAMS: ThreadCreationTripleParams = {
thread_id: "t-abc12345",
thread_row_id: THREAD_ROW_ID,
text: "Fix auth timeout in production",
session_id: "550e8400-e29b-41d4-a716-446655440000",
linear_issue: "PROJ-123",
Expand All @@ -28,6 +33,7 @@ const BASE_CREATION_PARAMS: ThreadCreationTripleParams = {

const BASE_RESOLUTION_PARAMS: ThreadResolutionTripleParams = {
thread_id: "t-abc12345",
thread_row_id: THREAD_ROW_ID,
text: "Fix auth timeout in production",
resolution_note: "Fixed by increasing timeout to 30s",
session_id: "660e8400-e29b-41d4-a716-446655440000",
Expand Down Expand Up @@ -88,7 +94,8 @@ describe("extractThreadCreationTriples", () => {

for (const triple of triples) {
expect(triple.source_type).toBe("thread");
expect(triple.source_id).toBe("t-abc12345");
// GIT-105: was "t-abc12345", which the UUID column rejects (22P02).
expect(triple.source_id).toBe(THREAD_ROW_ID);
expect(triple.half_life_days).toBe(9999);
expect(triple.project).toBe("test-project");
expect(triple.created_by).toBe("CLI");
Expand Down Expand Up @@ -129,7 +136,17 @@ describe("extractThreadResolutionTriples", () => {
expect(triple.subject).toMatch(/^Session: /);
expect(triple.object).toMatch(/^Thread: /);
expect(triple.source_type).toBe("thread");
expect(triple.source_id).toBe("t-abc12345");
// GIT-105: was "t-abc12345", which the UUID column rejects (22P02).
expect(triple.source_id).toBe(THREAD_ROW_ID);
});

it("without a row id, source_id is null — never the thread_id (GIT-105)", () => {
for (const t of [
...extractThreadCreationTriples({ ...BASE_CREATION_PARAMS, thread_row_id: null }),
...extractThreadResolutionTriples({ ...BASE_RESOLUTION_PARAMS, thread_row_id: undefined }),
]) {
expect(t.source_id).toBeNull();
}
});
});

Expand Down
Loading