diff --git a/.changeset/deferred-hunt-fixes.md b/.changeset/deferred-hunt-fixes.md new file mode 100644 index 0000000..5310831 --- /dev/null +++ b/.changeset/deferred-hunt-fixes.md @@ -0,0 +1,17 @@ +--- +"@marrowhq/core": patch +"@marrowhq/web": patch +--- + +Close the two bugs deferred from the engine hunt. + +- Distill embedding reconcile: a node insert and its embedding write are separate + transactions, so a transient embedding failure could leave a committed node + with no vector, and the idempotent re-distill skip made that permanent, hiding + a real fact from semantic search. Distill now re-embeds any existing node + missing its vector at the start of the pass (a no-op once everything is + embedded), via a new `store.hasEmbedding`. +- API timestamp validation: the `since` / `until` / `before` query params are + validated as ISO-8601 at the boundary and rejected with a fixed 400, so a + malformed value never reaches Postgres as an uncastable timestamp whose raw + error the error classifier would otherwise reflect to the client. diff --git a/packages/core/src/distill.test.ts b/packages/core/src/distill.test.ts index 1406a9d..6df1013 100644 --- a/packages/core/src/distill.test.ts +++ b/packages/core/src/distill.test.ts @@ -258,6 +258,23 @@ describe("distillation", () => { await expect(core.traceToSource("dec_missing")).rejects.toThrow(/not found/); }); + it("re-embeds an existing node that has no vector on the next distill (reconcile)", async () => { + const evId = await core.ingest({ text: gdyniaTranscript, source: "interviews/reembed.md" }); + // a node committed with no embedding, exactly as a transient embed failure + // after the row's own commit would leave it. insertEntity does not embed. + const orphan = await store.insertEntity({ + name: "reembed target", + status: "open", + confidence: { value: 0.6, source: "model" }, + provenance: [{ evidenceId: evId, start: 0, end: 5 }], + }); + expect(await store.hasEmbedding(orphan.id, "entity")).toBe(false); + + // distilling the same evidence runs the reconcile pass over existing nodes. + await core.distill(evId); + expect(await store.hasEmbedding(orphan.id, "entity")).toBe(true); + }); + it("traceToSource surfaces the skeptic's latest verdict, advisory and status-neutral", async () => { const ev = await store.insertEvidence({ text: "auth notes here", source: "room/skeptic.md" }); const dec = await store.insertDecision({ diff --git a/packages/core/src/marrow.ts b/packages/core/src/marrow.ts index bf41080..0ed2155 100644 --- a/packages/core/src/marrow.ts +++ b/packages/core/src/marrow.ts @@ -600,6 +600,19 @@ export class Marrow { traced(this.store, { kind: "distill", label: evidence.source }, async (report) => { const existing = await this.store.getNodesForEvidence(evidenceId); const seen = new Set(existing.map((node) => nodeKey(node, evidenceId))); + // reconcile: a node insert and its embedding write are separate + // transactions, so an earlier run could commit a node but fail to embed it + // (a transient embedding error). the dedupe skip below would then make that + // miss permanent, leaving a real fact invisible to semantic search. re-embed + // any existing node missing its vector so a re-distill of the same source + // repairs it. cheap: a no-op once every node is embedded. + if (this.embedding) { + for (const node of existing) { + if (!(await this.store.hasEmbedding(node.id, node.kind))) { + await this.embedNode(node.id, node.kind, this.embedTextForNode(node)); + } + } + } const created: Distilled[] = []; let tokensIn = 0; let tokensOut = 0; @@ -2749,6 +2762,21 @@ export class Marrow { return new SyncEngine({ store: this.store }).runAll(); } + /** The text a node is embedded from, matching what each create path passes to + * embedNode, so a reconcile re-embed produces the same vector. */ + private embedTextForNode(node: Distilled): string { + switch (node.kind) { + case "entity": + return node.name; + case "decision": + return `${node.title} ${node.rationale}`; + case "goal": + return `${node.title} ${node.description ?? ""}`; + case "question": + return node.prompt; + } + } + private async embedNode( nodeId: string, nodeKind: "entity" | "decision" | "question" | "goal", diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index a356a58..2bf1de6 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -1322,6 +1322,17 @@ export class Store { ); } + /** Whether a node already has an embedding row. Used by distill to reconcile a + * node that was committed but whose embedding write failed on an earlier run, + * so a transient embedding failure is not permanent. */ + async hasEmbedding(nodeId: string, nodeKind: string): Promise { + const res = await this.pool.query( + "select 1 from embedding where node_id = $1 and node_kind = $2 limit 1", + [nodeId, nodeKind], + ); + return res.rows.length > 0; + } + async embeddingProfile(): Promise<{ model: string; dim: number } | undefined> { const res = await this.pool.query<{ embedding_model: string; dim: number }>( "select embedding_model, dim from embedding limit 1", diff --git a/packages/web/src/api.ts b/packages/web/src/api.ts index 0af1854..aec4fe7 100644 --- a/packages/web/src/api.ts +++ b/packages/web/src/api.ts @@ -353,6 +353,17 @@ export class ApiError extends Error { } } +/** Validate an optional timestamp query param at the boundary, returning a + * canonical ISO string. A bad value is rejected with a fixed 400 here, so it + * never reaches Postgres as an uncastable timestamp whose raw error (engine, + * column type, the reflected input) the classifier would otherwise echo. */ +function parseIsoParam(raw: string | null, name: string): string | undefined { + if (raw === null || raw === "") return undefined; + const t = Date.parse(raw); + if (Number.isNaN(t)) throw new ApiError(400, `${name} must be an ISO-8601 timestamp`); + return new Date(t).toISOString(); +} + async function readBody(req: IncomingMessage): Promise { const chunks: Buffer[] = []; let size = 0; @@ -521,8 +532,8 @@ async function handle( // --- observability: the run trace ---------------------------------------- if (path === "/api/metrics" && req.method === "GET") { const store = requireStore(resolved.store); - const since = url.searchParams.get("since") ?? undefined; - const until = url.searchParams.get("until") ?? undefined; + const since = parseIsoParam(url.searchParams.get("since"), "since"); + const until = parseIsoParam(url.searchParams.get("until"), "until"); const metrics: RunMetrics = await store.runMetrics({ ...(since ? { since } : {}), ...(until ? { until } : {}), @@ -534,7 +545,7 @@ async function handle( const kind = url.searchParams.get("kind") as RunKind | null; const status = url.searchParams.get("status") as RunStatus | null; const limitRaw = url.searchParams.get("limit"); - const before = url.searchParams.get("before"); + const before = parseIsoParam(url.searchParams.get("before"), "before"); const limit = limitRaw ? Number(limitRaw) : undefined; const runs: RunRecord[] = await store.listRuns({ ...(kind ? { kind } : {}), diff --git a/packages/web/src/server.test.ts b/packages/web/src/server.test.ts index ffdf520..cf2cc43 100644 --- a/packages/web/src/server.test.ts +++ b/packages/web/src/server.test.ts @@ -190,6 +190,15 @@ describe("web api server", () => { expect(res.status).toBe(404); }); + it("rejects a malformed timestamp param with a fixed 400, never reflecting a DB error", async () => { + const res = await fetch(`${base}/api/metrics?since=not-a-date`); + expect(res.status).toBe(400); + const body = (await res.json()) as { error: string }; + expect(body.error).toBe("since must be an ISO-8601 timestamp"); + // the raw Postgres cast error (engine, column type, reflected input) never leaks. + expect(JSON.stringify(body)).not.toMatch(/invalid input syntax|for type timestamp/i); + }); + it("GET /api/runs with a non-positive or fractional limit falls back to the default, not a 500", async () => { for (const bad of ["-5", "0", "2.5"]) { const res = await fetch(`${base}/api/runs?limit=${bad}`);