fix: close the two bugs deferred from the engine hunt

.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.

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({

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",

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<boolean> {
+    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",

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<unknown> {
   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 } : {}),

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