import { afterAll, beforeAll, expect, test } from "bun:test";
import assert from "node:assert/strict";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp } from "../../src/core/app";
import type { Db } from "../../src/core/db";
import { migrate } from "../../src/core/migrations";
import { createSqliteDb, openDatabase } from "../../src/runtime/bun/sqlite";
import { createVectorizeStore } from "../../src/runtime/cloudflare/vectors";
import { clientVia, fakeVectors, provisionWith } from "./harness";
import { outboxStatement } from "../../src/core/writes";
import {
  RANK_WEIGHTS,
  hasDeadHeat,
  normalizeVectorSimilarity,
  rankCandidates,
  type RankInput,
} from "../../src/core/rank";

/**
 * The optional vector capability, exercised against the shared core.
 *
 * Hybrid retrieval lives entirely in `src/core/context.ts` and `rank.ts`, above
 * the driver boundary, so proving the branch once is sufficient: the two drivers
 * are already covered by the dual-runtime contract suite. What cannot be tested
 * here is a real embedding provider or a real vector index; those are wired in
 * the per-runtime vector adapters and gated on a native extension or a
 * Cloudflare binding.
 */
const directory = mkdtempSync(join(tmpdir(), "titen-vec-"));
const dbPath = join(directory, "titen.db");
const origin = "http://titen.test";

const vectors = fakeVectors();
let db: ReturnType<typeof createSqliteDb>;
let handle: ReturnType<typeof openDatabase>;
let key: string;
let orgId: string;
let subjectId: string;

/** Three claims that all match the query lexically, so only ranking separates them. */
const CLAIMS = [
  "Rollback smoke gates every production release without exception.",
  "Release safety depends on a verified rollback rehearsal.",
  "Rollback ownership for a release sits with the on-call engineer.",
];
const TASK = "rollback release safety";

let claimIds: string[] = [];

const vectorScope = () => ({ org_id: orgId, subject_id: subjectId, project_id: "" });

const rankInput = (id: string, confidence: number, vector_boost?: number): RankInput => ({
  id,
  kind: "procedural",
  statement: `claim ${id}`,
  trust: "verified",
  confidence,
  status: "active",
  created_at: "2026-07-30T00:00:00.000Z",
  bm25: 0,
  disputed: false,
  feedback_positive: 0,
  feedback_negative: 0,
  feedback_total: 0,
  vector_boost,
});

async function seed(client: ReturnType<typeof clientVia>) {
  const observation = await client.call("POST", "/v1/observations", {
    key,
    body: {
      subject_id: subjectId,
      kind: "tool_result",
      content: "Release audit covering rollback rehearsal, gating, and ownership.",
      source: { type: "tool", ref: "audit#1" },
      trust: "verified",
    },
  });
  assert.equal(observation.status, 201);
  const observationId = observation.body.data.observation_id as string;

  const consolidation = await client.call("POST", "/v1/consolidations", {
    key,
    body: {
      subject_id: subjectId,
      claims: CLAIMS.map((statement) => ({
        kind: "procedural",
        statement,
        sources: [{ observation_id: observationId, relation: "supports" }],
      })),
    },
  });
  assert.equal(consolidation.status, 201);
  claimIds = (consolidation.body.data.claims as { claim_id: string }[]).map(
    (claim) => claim.claim_id,
  );
}

/** One app with the capability, one without, over the same seeded database. */
const withVectors = () =>
  clientVia(createApp({ db, revision: "test", runtime: "bun-sqlite", vectors }), origin);
const withoutVectors = () =>
  clientVia(createApp({ db, revision: "test", runtime: "bun-sqlite" }), origin);

beforeAll(async () => {
  handle = openDatabase(dbPath);
  db = createSqliteDb(handle);
  await migrate(db);
  const provisioned = await provisionWith(db, { scopes: ["*"] });
  key = provisioned.key;
  orgId = provisioned.orgId;
  subjectId = "user_vectors";
  await seed(withVectors());
});

afterAll(() => {
  handle.close();
  rmSync(directory, { recursive: true, force: true });
});

test("readiness reports the vector capability only when one is configured", async () => {
  const off = await withoutVectors().call("GET", "/readyz");
  assert.equal(off.body.data.capabilities.vector, "disabled");
  assert.equal(off.body.data.capabilities.model, "disabled");

  const pending = await withVectors().call("GET", "/readyz");
  assert.equal(pending.status, 200);
  assert.equal(pending.body.data.ready, true);
  assert.equal(pending.body.data.checks.semantic_index, "index_projection_pending");
  assert.equal(pending.body.data.capabilities.vector, "enabled");
  const drained = await withVectors().call("POST", "/v1/index/drain?limit=100", { key });
  assert.equal(drained.status, 200);

  const on = await withVectors().call("GET", "/readyz");
  assert.equal(on.status, 200);
  assert.equal(on.body.data.capabilities.vector, "enabled");
  assert.equal(on.body.data.capabilities.model, "enabled");

  // Preserve the pending fixture expected by the outage/recovery cases below.
  await vectors.store.remove(claimIds);
  await db.batch([{
    sql: `UPDATE index_outbox
             SET state = 'pending', attempts = 0,
                 lease_token = NULL, lease_expires_at = NULL
           WHERE record_id IN (${claimIds.map(() => "?").join(", ")})`,
    params: claimIds,
  }]);
});

test("compilation reports vector retrieval as used and calls the embedder", async () => {
  const before = vectors.embedCalls();
  const res = await withVectors().call("POST", "/v1/context/compile", {
    key,
    body: { subject_id: subjectId, task: TASK, max_tokens: 2000 },
  });
  assert.equal(res.status, 200);
  assert.equal(res.body.meta.degraded.vector, "used");
  assert.ok(vectors.embedCalls() > before, "the query must be embedded once");
  assert.ok(res.body.data.items.length >= 1);
  assert.deepEqual(vectors.lastFilter(), {
    org_id: orgId,
    subject_id: subjectId,
    project_id: "",
  });
});

test("vector retrieval treats missing project as unscoped and broad mode as explicit", async () => {
  const client = withVectors();
  const subject = "user_vector_project_scope_141";
  const project = await client.call("POST", "/v1/projects/resolve", {
    key,
    body: { reference: "scope-fixture/vector-project", create: true },
  });
  assert.equal(project.status, 201);
  const projectId = project.body.data.project_id as string;
  const observed = await client.call("POST", "/v1/observations", {
    key,
    body: {
      subject_id: subject,
      project_id: projectId,
      kind: "tool_result",
      content: "Vector-only scoped canary evidence.",
      source: { type: "test", ref: "scope-141" },
      trust: "verified",
      visibility: "organization",
    },
  });
  assert.equal(observed.status, 201);
  const consolidated = await client.call("POST", "/v1/consolidations", {
    key,
    body: {
      subject_id: subject,
      project_id: projectId,
      claims: [{
        kind: "procedural",
        statement: "Vector-only scoped canary claim.",
        visibility: "organization",
        sources: [{ observation_id: observed.body.data.observation_id, relation: "supports" }],
      }],
    },
  });
  assert.equal(consolidated.status, 201);
  const claimId = consolidated.body.data.claims[0].claim_id as string;
  vectors.setScore(claimId, 1, {
    org_id: orgId,
    subject_id: subject,
    project_id: projectId,
  });

  const request = (body: Record<string, unknown>) =>
    client.call("POST", "/v1/context/compile", {
      key,
      body: { subject_id: subject, task: "semantically adjacent phrase", max_tokens: 900, ...body },
    });
  const omitted = await request({});
  assert.equal(omitted.status, 200);
  assert.equal(omitted.body.data.items.length, 0);
  assert.deepEqual(vectors.lastFilter(), {
    org_id: orgId,
    subject_id: subject,
    project_id: "",
  });

  const exact = await request({ project_id: projectId });
  assert.equal(exact.status, 200);
  assert.equal(exact.body.data.items[0].claim_id, claimId);
  assert.deepEqual(vectors.lastFilter(), {
    org_id: orgId,
    subject_id: subject,
    project_id: projectId,
  });

  const broad = await request({ cross_project: true });
  assert.equal(broad.status, 200);
  assert.equal(broad.body.data.items[0].claim_id, claimId);
  assert.equal(broad.body.data.scope.project_mode, "cross_project");
  assert.deepEqual(vectors.lastFilter(), { org_id: orgId, subject_id: subject });
});

test("Vectorize receives the same canonical metadata and query filter", async () => {
  let upserted: unknown;
  let queried: unknown;
  let lookedUp: unknown;
  const store = createVectorizeStore({
    async upsert(records) {
      upserted = records;
    },
    async query(_vector, options) {
      queried = options;
      return { matches: [] };
    },
    async getByIds(ids) {
      lookedUp = ids;
      return [{ id: "claim" }];
    },
    async deleteByIds() {},
  });
  const scope = { org_id: "org", subject_id: "subject", project_id: "project" };
  await store.upsert([{ id: "claim", vector: new Float32Array([1, 0]), metadata: scope }]);
  await store.query(new Float32Array([1, 0]), { topK: 2, filter: scope });
  const present = await store.present!(["claim", "missing"]);

  assert.deepEqual(upserted, [
    { id: "claim", values: [1, 0], metadata: scope },
  ]);
  assert.deepEqual(queried, { topK: 2, filter: scope });
  assert.deepEqual(lookedUp, ["claim", "missing"]);
  assert.deepEqual([...present], ["claim"]);
});

test("unchanged claims skip embeddings while verification repairs a missing vector", async () => {
  const localHandle = openDatabase(join(directory, "index-verification.db"));
  const localDb = createSqliteDb(localHandle);
  try {
    await migrate(localDb);
    const auth = await provisionWith(localDb, { scopes: ["*"] });
    const capability = fakeVectors();
    const client = clientVia(createApp({
      db: localDb,
      revision: "test",
      runtime: "bun-sqlite",
      vectors: capability,
    }), origin);
    const observed = await client.call("POST", "/v1/observations", {
      key: auth.key,
      body: {
        subject_id: "index-verification",
        kind: "tool_result",
        content: "Canonical evidence for bounded vector verification.",
        source: { type: "test", id: "index-verification-evidence", ref: "verify#1" },
        trust: "verified",
      },
    });
    assert.equal(observed.status, 201);
    const consolidated = await client.call("POST", "/v1/consolidations", {
      key: auth.key,
      body: {
        subject_id: "index-verification",
        claims: [{
          kind: "procedural",
          statement: "Bounded verification repairs a missing semantic projection.",
          sources: [{ observation_id: observed.body.data.observation_id, relation: "supports" }],
        }],
      },
    });
    assert.equal(consolidated.status, 201);
    const claimId = consolidated.body.data.claims[0].claim_id as string;

    const first = await client.call("POST", "/v1/index/drain?limit=100", { key: auth.key });
    assert.equal(first.status, 200);
    assert.equal(first.body.data.indexed, 1);
    const calls = capability.embedCalls();
    await localDb.batch([outboxStatement(
      auth.orgId,
      "claim",
      claimId,
      "upsert",
      new Date().toISOString(),
    )]);
    const duplicate = await client.call("POST", "/v1/index/drain?limit=100", { key: auth.key });
    assert.equal(duplicate.status, 200);
    assert.equal(duplicate.body.data.indexed, 0);
    assert.equal(duplicate.body.data.skipped, 1);
    assert.equal(capability.embedCalls(), calls);

    capability.drop(claimId);
    const verified = await client.call("POST", "/v1/index/verify?limit=100", { key: auth.key });
    assert.equal(verified.status, 200);
    assert.deepEqual(verified.body.data, {
      checked: 1,
      present: 0,
      missing: 1,
      repairs_queued: 1,
      next_after: null,
    });
    const repaired = await client.call("POST", "/v1/index/drain?limit=100", { key: auth.key });
    assert.equal(repaired.status, 200);
    assert.equal(repaired.body.data.indexed, 1);
    assert.equal(capability.embedCalls(), calls + 1);
    assert.deepEqual(await capability.store.present!([claimId]), new Set([claimId]));
  } finally {
    localHandle.close();
  }
});

test("Vectorize caps native queries without changing scope filters", async () => {
  let queried: unknown;
  const store = createVectorizeStore({
    async upsert() {},
    async query(_vector, options) {
      if (options.topK > 100) throw new Error("maximum topK is 100");
      queried = options;
      return { matches: [] };
    },
    async getByIds() { return []; },
    async deleteByIds() {},
  });
  const filter = { org_id: "org", subject_id: "subject", project_id: "project" };
  await store.query(new Float32Array([1, 0]), { topK: 200, filter });
  assert.deepEqual(queried, { topK: 100, filter });
});

test("a semantic hit lifts a lexically weak claim up the ranking", async () => {
  // With no vector signal, record the order lexical scoring alone produces.
  const lexical = await withoutVectors().call("POST", "/v1/context/compile", {
    key,
    body: { subject_id: subjectId, task: TASK, max_tokens: 2000 },
  });
  assert.equal(lexical.status, 200);
  const lexicalOrder = lexical.body.data.items.map((item: any) => item.claim_id) as string[];
  assert.equal(lexical.body.meta.degraded.vector, "disabled");
  assert.ok(lexicalOrder.length >= 3, "the fixture needs three ranked claims");

  // Pin the weakest lexical match as semantically near: the case FTS cannot see.
  const weakest = lexicalOrder[lexicalOrder.length - 1]!;
  const weakestBefore = lexical.body.data.items.at(-1);
  assert.ok(
    weakestBefore.score_components.relevance < 1,
    "the chosen claim must start below the top relevance",
  );
  vectors.setScore(weakest, 1, vectorScope());

  const hybrid = await withVectors().call("POST", "/v1/context/compile", {
    key,
    body: { subject_id: subjectId, task: TASK, max_tokens: 2000 },
  });
  assert.equal(hybrid.status, 200);
  assert.equal(hybrid.body.meta.degraded.vector, "used");
  const hybridOrder = hybrid.body.data.items.map((item: any) => item.claim_id) as string[];

  // Position must improve, and it now improves all the way. Within-set
  // normalization scores the top lexical hit and the top semantic hit at
  // relevance 1 alike, so the two tie on the weighted score; the cosine
  // tie-break decides that tie on evidence instead of on a freshly minted uuid.
  assert.ok(
    hybridOrder.indexOf(weakest) < lexicalOrder.indexOf(weakest),
    `a semantic match must rank higher: was ${lexicalOrder.indexOf(weakest)}, now ${hybridOrder.indexOf(weakest)}`,
  );
  assert.equal(
    hybridOrder[0],
    weakest,
    "a tie with the top lexical hit must be decided by the cosine, not by the uuid",
  );
  const boosted = hybrid.body.data.items.find((item: any) => item.claim_id === weakest);
  assert.equal(boosted.score_components.relevance, 1, "similarity feeds relevance");
  assert.ok(
    boosted.score > weakestBefore.score,
    "the boosted claim must score higher than it did without the vector signal",
  );

  // Clearing the score restores the original order: the boost is not persisted.
  vectors.setScore(weakest, 0, vectorScope());
  const cleared = await withVectors().call("POST", "/v1/context/compile", {
    key,
    body: { subject_id: subjectId, task: TASK, max_tokens: 2000 },
  });
  assert.deepEqual(
    cleared.body.data.items.map((item: any) => item.claim_id),
    lexicalOrder,
    "with no similarity the ranking must match lexical-only",
  );
});

test("the absolute cosine floor rejects a best bad neighbor before hydration", async () => {
  const subject = "user_absolute_gate_144";
  const observation = await withVectors().call("POST", "/v1/observations", {
    key,
    body: {
      subject_id: subject,
      kind: "tool_result",
      content: "The unrelated archive stores violet telescope calibration notes.",
      source: { type: "test", ref: "absolute-gate-144" },
      trust: "verified",
    },
  });
  assert.equal(observation.status, 201);
  const consolidation = await withVectors().call("POST", "/v1/consolidations", {
    key,
    body: {
      subject_id: subject,
      claims: [{
        kind: "semantic_fact",
        statement: "Violet telescope calibration belongs in the archive.",
        sources: [{
          observation_id: observation.body.data.observation_id,
          relation: "supports",
        }],
      }],
    },
  });
  assert.equal(consolidation.status, 201);
  const claimId = consolidation.body.data.claims[0].claim_id as string;
  vectors.setScore(claimId, 0.099, {
    org_id: orgId,
    subject_id: subject,
    project_id: "",
  });

  let vectorHydrations = 0;
  const trackedDb: Db = {
    all: (sql, params) => {
      if (sql.includes("WHERE c.id IN")) vectorHydrations += 1;
      return db.all(sql, params);
    },
    batch: (statements) => db.batch(statements),
    exec: (sql) => db.exec(sql),
  };
  const client = clientVia(
    createApp({ db: trackedDb, revision: "test", runtime: "bun-sqlite", vectors }),
    origin,
  );
  const compile = () => client.call("POST", "/v1/context/compile", {
    key,
    body: { subject_id: subject, task: "...\u200d...", max_tokens: 900 },
  });

  const rejected = await compile();
  assert.equal(rejected.status, 200);
  assert.deepEqual(rejected.body.data.items, []);
  assert.equal(rejected.body.meta.candidates, 0);
  assert.equal(rejected.body.meta.degraded.vector, "used");
  assert.equal(vectorHydrations, 0, "sub-threshold ids must not reach canonical SQL");
  assert.doesNotMatch(
    JSON.stringify(rejected.body),
    /min_cosine|raw-unit-v1|0\.099/,
  );

  vectors.setScore(claimId, 0.1, {
    org_id: orgId,
    subject_id: subject,
    project_id: "",
  });
  const accepted = await compile();
  assert.equal(accepted.status, 200);
  assert.equal(accepted.body.data.items[0].claim_id, claimId);
  assert.equal(vectorHydrations, 1, "an eligible vector id is hydrated once");
});

test("narrow-band vector similarity passes through and still orders the set", () => {
  // Cosine used to be min-max normalized inside the set, which stretched a
  // 0.002 gap to the full 0..1 and pinned the best to exactly 1 however poor it
  // was. #227: it is an absolute 0..1 already, so it is used as it stands.
  const candidates = [rankInput("near", 0.8, 0.991), rankInput("exact", 0.8, 0.993)];
  const normalized = normalizeVectorSimilarity(candidates);
  assert.equal(normalized.get("near"), 0.991);
  assert.equal(normalized.get("exact"), 0.993);

  // The narrow band still decides the order — that was the property worth
  // keeping, and it never needed the stretch to hold.
  const ranked = rankCandidates(candidates, new Date("2026-07-30T00:00:00.000Z"));
  assert.equal(ranked[0]!.candidate.id, "exact");
  assert.equal(ranked[0]!.components.relevance, 0.993);
  assert.ok(ranked[0]!.score > ranked[1]!.score, "the best no longer ties the rest at a ceiling");

  const tied = normalizeVectorSimilarity([rankInput("a", 1, 0.99), rankInput("b", 1, 0.99)]);
  assert.equal(tied.get("a"), 0.99);
  assert.equal(tied.get("b"), 0.99);

  const absent = normalizeVectorSimilarity([rankInput("none", 1, 0)]);
  assert.equal(absent.has("none"), false);
});

test("a score tie between the lexical best and the semantic best breaks on cosine", () => {
  const now = new Date("2026-07-30T00:00:00.000Z");
  // Both arms used to be min-max normalized, so each arm's own best scored
  // relevance exactly 1 and any hybrid pair tied by construction. After #227
  // both are absolute, so a tie has to be built rather than assumed: -9.5474
  // per term saturates to 0.7207, the cosine the other candidate carries.
  // The case is still the ordinary hybrid one — on the release fixture it
  // decided rank 1 of `id_temporal_endpoint` and `jv_in_id_preference` with a
  // uuid, and measured recall@1 flipped between repeats because of it.
  const lexicalBest = { ...rankInput("aaa", 1), bm25: -9.547404 };
  const semanticBest = { ...rankInput("zzz", 1, 0.7207) };
  const ranked = rankCandidates([lexicalBest, semanticBest], now);
  assert.equal(ranked[0]!.score, ranked[1]!.score, "the fixture must actually tie on score");
  assert.equal(ranked[0]!.components.relevance, ranked[1]!.components.relevance);
  // Fails on the id-only fallback, which returns "aaa".
  assert.equal(ranked[0]!.candidate.id, "zzz", "semantic evidence outranks a lower uuid");

  // Two cosines are compared only against each other, so no constant and no
  // cross-scale comparison enters: the larger raw cosine wins a tied score even
  // when within-set normalization has already flattened both to relevance 1.
  const near = { ...rankInput("aaa", 1, 0.7026), bm25: -9.547404 };
  const nearer = { ...rankInput("zzz", 1, 0.7207), bm25: -9.547404 };
  const bothSemantic = rankCandidates([near, nearer], now);
  assert.equal(bothSemantic[0]!.score, bothSemantic[1]!.score);
  assert.equal(bothSemantic[0]!.candidate.id, "zzz");

  // A genuine dead heat — equal score and equal evidence — must stay total and
  // repeatable, which is the property the cosine term must not have broken.
  // Ids are freshly minted uuids in production, so the order has to come from
  // content: here the lower id carries the higher statement, and the statement
  // decides. With an id-only fallback this returns ["aaa", "zzz"].
  const twins = rankCandidates(
    [
      { ...rankInput("zzz", 1), statement: "Alpha rollback rehearsal.", bm25: -4 },
      { ...rankInput("aaa", 1), statement: "Zulu rollback rehearsal.", bm25: -4 },
    ],
    now,
  );
  assert.deepEqual(twins.map((entry) => entry.candidate.id), ["zzz", "aaa"]);

  // Byte-identical statements still fall through to the id, so the order stays
  // total when content cannot separate two candidates.
  const identical = rankCandidates(
    [
      { ...rankInput("zzz", 1), statement: "Same rollback rehearsal.", bm25: -4 },
      { ...rankInput("aaa", 1), statement: "Same rollback rehearsal.", bm25: -4 },
    ],
    now,
  );
  assert.deepEqual(identical.map((entry) => entry.candidate.id), ["aaa", "zzz"]);

  // Lexical-only ranking carries no cosine at all, so the new term is inert
  // there: a stronger bm25 still wins on score, never on the tie-break.
  const lexicalOnly = rankCandidates(
    [{ ...rankInput("zzz", 1), bm25: -1.73 }, { ...rankInput("aaa", 1), bm25: -8.08 }],
    now,
  );
  assert.equal(lexicalOnly[0]!.candidate.id, "aaa");
  assert.ok(lexicalOnly[0]!.score > lexicalOnly[1]!.score);
});

test("confidence is an explicit weighted and auditable ranking factor", () => {
  const weightSum = Object.values(RANK_WEIGHTS).reduce((sum, weight) => sum + weight, 0);
  assert.ok(Math.abs(weightSum - 1) < Number.EPSILON * 2);
  const ranked = rankCandidates(
    [rankInput("low", 0.2), rankInput("high", 0.9)],
    new Date("2026-07-30T00:00:00.000Z"),
  );
  assert.equal(ranked[0]!.candidate.id, "high");
  assert.equal(ranked[0]!.components.confidence, 0.9);
  assert.equal(ranked[1]!.components.confidence, 0.2);
  assert.equal(ranked[0]!.score - ranked[1]!.score, 0.07);
});

test("corroboration decides a dead heat and never overrides score or cosine", () => {
  const now = new Date("2026-07-30T00:00:00.000Z");
  // Equal bm25 and no semantic signal: the FTS-only dead heat that the
  // statement key used to decide alphabetically. Measured on LongMemEval-S at
  // n=500 this never happens at rank 1, which is why the signal ships on this
  // contract rather than on a benchmark win.
  const shallow = { ...rankInput("zzz", 0.8), statement: "Alpha rehearsal.", bm25: -4 };
  const deep = {
    ...rankInput("aaa", 0.8),
    statement: "Zulu rehearsal.",
    bm25: -4,
    evidence_depth: 3,
  };
  const ranked = rankCandidates([shallow, deep], now);
  assert.equal(ranked[0]!.score, ranked[1]!.score, "the fixture must actually tie on score");
  // Without the corroboration key this returns "zzz", whose statement sorts first.
  assert.equal(ranked[0]!.candidate.id, "aaa");

  // Corroboration is a tie-break, not a seventh weighted term: no score and no
  // component moves, so no fitted weight enters the ranker.
  const withoutDepth = rankCandidates([{ ...deep, evidence_depth: undefined }], now);
  assert.equal(ranked.find((entry) => entry.candidate.id === "aaa")!.score, withoutDepth[0]!.score);
  assert.deepEqual(
    ranked.find((entry) => entry.candidate.id === "aaa")!.components,
    withoutDepth[0]!.components,
  );

  // Uniform depth leaves the pre-change order exactly as it was.
  assert.deepEqual(
    rankCandidates([{ ...deep, evidence_depth: 3 }, { ...shallow, evidence_depth: 3 }], now)
      .map((entry) => entry.candidate.id),
    ["zzz", "aaa"],
  );

  // A stronger lexical match still wins on score; corroboration cannot rescue a
  // worse match.
  const scored = rankCandidates(
    [
      { ...rankInput("weak", 0.8), bm25: -1, evidence_depth: 9 },
      { ...rankInput("strong", 0.8), bm25: -8 },
    ],
    now,
  );
  assert.equal(scored[0]!.candidate.id, "strong");
  assert.ok(scored[0]!.score > scored[1]!.score);

  // A stronger cosine also still wins: an unmeasured signal does not override a
  // measured one.
  const semantic = rankCandidates(
    [
      { ...rankInput("cosine", 0.8, 0.91), bm25: -4 },
      // -37.4111 per term saturates to 0.91, so both reach relevance 0.91 —
      // one through cosine, one through lexical — and genuinely tie on score.
      { ...rankInput("corroborated", 0.8, 0.72), bm25: -37.411111, evidence_depth: 9 },
    ],
    now,
  );
  assert.equal(semantic[0]!.score, semantic[1]!.score);
  assert.equal(semantic[0]!.candidate.id, "cosine");
});

test("a dead heat is detected only inside the window that is returned", () => {
  const now = new Date("2026-07-30T00:00:00.000Z");
  // Three tie on score; the fourth is a strictly worse lexical match. Only the
  // returned window can be changed by a tie-break, so a tie below it must not
  // charge the request for an evidence lookup it cannot use.
  const ranked = rankCandidates(
    [
      { ...rankInput("a", 0.9), bm25: -8 },
      { ...rankInput("b", 0.8), bm25: -4, statement: "Bravo." },
      { ...rankInput("c", 0.8), bm25: -4, statement: "Charlie." },
      { ...rankInput("d", 0.8), bm25: -4, statement: "Delta." },
    ],
    now,
  );
  assert.deepEqual(ranked.map((entry) => entry.candidate.id), ["a", "b", "c", "d"]);
  assert.equal(hasDeadHeat(ranked, 1), false, "rank 1 is unique and rank 2 does not tie it");
  assert.equal(hasDeadHeat(ranked, 2), true, "the pair straddling the boundary can move");
  assert.equal(hasDeadHeat(ranked), true);
  assert.equal(hasDeadHeat([]), false);
  assert.equal(hasDeadHeat(ranked.slice(0, 1)), false);

  // Equal score but different cosine is not a dead heat: the semantic key
  // already decides it, so no lookup is needed.
  const semantic = rankCandidates(
    [
      { ...rankInput("x", 0.8, 0.91), bm25: -4 },
      { ...rankInput("y", 0.8, 0.72), bm25: -37.411111 },
    ],
    now,
  );
  assert.equal(semantic[0]!.score, semantic[1]!.score);
  assert.equal(hasDeadHeat(semantic), false);
});

test("compressing relevance lets the other components decide between two strong matches", () => {
  const now = new Date("2026-07-30T00:00:00.000Z");
  // The one behavioural consequence of #227 that the benchmark could not
  // measure. Both bench stores were ingested in a single pass, so trust,
  // confidence, recency, conflict and feedback are constant across every
  // candidate -- exactly one distinct non-relevance tuple across 89,467 packed
  // items. With those held constant, score is monotone in relevance and both
  // the old and the new transform are monotone in -bm25, so the order is
  // identical by arithmetic and recall could not have moved. It did not.
  //
  // Here they are not constant, which is the case the corpus never contained.
  // The old transform was s / best -- proportional, so a 25% gap in bm25 stayed
  // a 25% gap in relevance and outweighed a full trust level. The new one
  // saturates: at this strength both matches are already near the ceiling, the
  // gap compresses from 0.250 to 0.018, and a verified claim beats a marginally
  // better-matching asserted one.
  //
  // That is the intended direction. When two candidates both match well, which
  // of them matches 25% better is a weaker signal than which of them a human
  // verified. Asserting it here is what keeps it from being silent.
  const candidates = [
    { ...rankInput("asserted-better", 0.8), bm25: -60, trust: "asserted" as const },
    { ...rankInput("verified-worse", 0.8), bm25: -45, trust: "verified" as const },
  ];
  const ranked = rankCandidates(candidates, now);
  assert.equal(ranked[0]!.candidate.id, "verified-worse");
  assert.ok(
    ranked[0]!.components.relevance < ranked[1]!.components.relevance,
    "the winner is the worse lexical match; trust is what carried it",
  );
  assert.ok(
    ranked[1]!.components.relevance - ranked[0]!.components.relevance < 0.05,
    "and it only carries because saturation compressed the relevance gap",
  );

  // The compression is bounded, not unlimited: a genuinely weak match still
  // loses to a strong one however trusted it is.
  const wide = rankCandidates(
    [
      { ...rankInput("strong-asserted", 0.8), bm25: -60, trust: "asserted" as const },
      { ...rankInput("weak-verified", 0.8), bm25: -0.5, trust: "verified" as const },
    ],
    now,
  );
  assert.equal(wide[0]!.candidate.id, "strong-asserted", "trust cannot rescue a poor match");
});

test("identical corpus content ranks identically across ingests that mint fresh ids", () => {
  const now = new Date("2026-07-30T00:00:00.000Z");
  // #226: two stores holding the same claims must answer in the same order even
  // though every uuid differs. Corroboration is content-derived, so it has to
  // keep that property rather than reintroduce the coin flip it replaces.
  const corpus = (ids: [string, string, string]) => [
    { ...rankInput(ids[0], 0.8), statement: "Alpha rehearsal.", bm25: -4, evidence_depth: 1 },
    { ...rankInput(ids[1], 0.8), statement: "Zulu rehearsal.", bm25: -4, evidence_depth: 2 },
    { ...rankInput(ids[2], 0.8), statement: "Mike rehearsal.", bm25: -4, evidence_depth: 1 },
  ];
  const order = (ids: [string, string, string]) =>
    rankCandidates(corpus(ids), now).map((entry) => entry.candidate.statement);
  assert.deepEqual(order(["c1", "c2", "c3"]), order(["z9", "a0", "m5"]));
  assert.equal(order(["c1", "c2", "c3"])[0], "Zulu rehearsal.");
});

test("dispute is an explicit ranking penalty, never a bonus", () => {
  const clean = rankInput("clean", 0.8);
  const disputed = { ...rankInput("disputed", 0.8), disputed: true };
  const ranked = rankCandidates(
    [disputed, clean],
    new Date("2026-07-30T00:00:00.000Z"),
  );
  assert.equal(ranked[0]!.candidate.id, "clean");
  assert.equal(ranked[0]!.components.conflict, 1);
  assert.equal(ranked[1]!.components.conflict, 0);
  assert.ok(Math.abs(ranked[0]!.score - ranked[1]!.score - 0.05) < 1e-9);
});

async function pendingCount() {
  const rows = await db.all<{ count: number }>(
    `SELECT COUNT(*) AS count FROM index_outbox WHERE state = 'pending'`,
  );
  return Number(rows[0]!.count);
}

test("index drain reports embedder outages as retryable and preserves pending rows", async () => {
  const broken = fakeVectors();
  broken.breakEmbedder();
  const app = clientVia(createApp({ db, revision: "test", runtime: "bun-sqlite", vectors: broken }), origin);
  const before = await pendingCount();
  const failed = await app.call("POST", "/v1/index/drain", { key });
  assert.equal(failed.status, 503);
  assert.equal(failed.body.error.code, "UNAVAILABLE");
  assert.equal(failed.body.meta.dependency, "embedder");
  assert.equal(failed.body.meta.retryable, true);
  assert.equal(failed.body.meta.pending, before);
  const afterFailure = await pendingCount();
  assert.ok(afterFailure > 0 && afterFailure <= before);
  const unavailable = await app.call("GET", "/readyz");
  assert.equal(unavailable.status, 503);
  assert.equal(unavailable.body.meta.capabilities.embedding, "configured_error");
  assert.equal(unavailable.body.meta.capabilities.vector, "enabled");
  assert.equal(unavailable.body.meta.checks.semantic_index, "embedding_dependency_unavailable");

  const other = await provisionWith(db, { scopes: ["*"] });
  const otherObservation = await app.call("POST", "/v1/observations", {
    key: other.key,
    body: {
      subject_id: "other_org_recovery",
      kind: "tool_result",
      content: "One tenant cannot clear another tenant's dependency outage.",
      source: { type: "tool", ref: "cross-org-recovery" },
      trust: "verified",
    },
  });
  assert.equal(otherObservation.status, 201);
  assert.equal(
    (
      await app.call("POST", "/v1/consolidations", {
        key: other.key,
        body: {
          subject_id: "other_org_recovery",
          claims: [{
            kind: "procedural",
            statement: "Cross-organization recovery remains isolated.",
            sources: [{
              observation_id: otherObservation.body.data.observation_id,
              relation: "supports",
            }],
          }],
        },
      })
    ).status,
    201,
  );
  broken.restoreEmbedder();
  const otherRecovery = await app.call("POST", "/v1/index/drain", { key: other.key });
  assert.equal(otherRecovery.status, 200);
  assert.ok(otherRecovery.body.data.indexed > 0);
  assert.equal((await app.call("GET", "/readyz")).status, 503);

  const retiring = await db.all<{ id: string; record_id: string; status: string }>(
    `SELECT queued.id, queued.record_id, c.status
       FROM (
         SELECT id, record_id FROM index_outbox
          WHERE org_id = ? AND state = 'pending'
          ORDER BY created_at, id LIMIT 50
       ) queued
       JOIN claims c ON c.id = queued.record_id
      WHERE c.status IN ('active', 'disputed')`,
    [orgId],
  );
  assert.ok(retiring.length > 0);
  await db.batch(
    [...new Set(retiring.map(({ record_id }) => record_id))].map((id) => ({
      sql: `UPDATE claims SET status = 'expired' WHERE id = ?`,
      params: [id],
    })),
  );
  const deleteOnly = await app.call("POST", "/v1/index/drain", { key });
  assert.equal(deleteOnly.status, 200);
  assert.equal(deleteOnly.body.data.indexed, 0);
  assert.equal((await app.call("GET", "/readyz")).status, 503);

  const originalStatus = new Map(retiring.map(({ record_id, status }) => [record_id, status]));
  await db.batch([
    ...[...originalStatus].map(([id, status]) => ({
      sql: `UPDATE claims SET status = ? WHERE id = ?`,
      params: [status, id],
    })),
    ...retiring.map(({ id }) => ({
      sql: `UPDATE index_outbox SET state = 'pending' WHERE id = ?`,
      params: [id],
    })),
  ]);
  broken.restoreEmbedder();
  const recovered = await app.call("POST", "/v1/index/drain", { key });
  assert.equal(recovered.status, 200);
  assert.ok(recovered.body.data.indexed > 0);
  assert.ok((await pendingCount()) < before);
  assert.equal((await app.call("GET", "/readyz")).status, 200);
  assert.deepEqual(broken.metadataFor(claimIds[0]!), {
    org_id: orgId,
    subject_id: subjectId,
    project_id: "",
  });
});

test("index drain reports vector-store outages as retryable and preserves pending rows", async () => {
  await seed(withVectors());
  const broken = fakeVectors();
  const upsert = broken.store.upsert.bind(broken.store);
  let mutationCalls = 0;
  broken.store.upsert = async (records) => {
    mutationCalls += 1;
    await upsert(records);
  };
  broken.breakStore();
  const app = clientVia(createApp({ db, revision: "test", runtime: "bun-sqlite", vectors: broken }), origin);
  const before = await pendingCount();
  const repairsBefore = Number((await db.all<{ count: number }>(
    `SELECT COUNT(*) AS count FROM index_outbox
      WHERE state = 'pending' AND operation = 'reconcile'`,
  ))[0]?.count ?? 0);
  const failed = await app.call("POST", "/v1/index/drain", { key });
  assert.equal(failed.status, 503);
  assert.equal(failed.body.error.code, "UNAVAILABLE");
  assert.equal(failed.body.meta.dependency, "vector_store");
  assert.equal(failed.body.meta.retryable, true);
  assert.equal(failed.body.meta.pending, before);
  const after = await pendingCount();
  const repairsAfter = Number((await db.all<{ count: number }>(
    `SELECT COUNT(*) AS count FROM index_outbox
      WHERE state = 'pending' AND operation = 'reconcile'`,
  ))[0]?.count ?? 0);
  assert.ok(repairsAfter > repairsBefore, "ambiguous external writes retain canonical repair");
  assert.equal(after - before, repairsAfter - repairsBefore);
  for (let attempt = 2; attempt <= 6; attempt += 1) {
    const repeated = await app.call("POST", "/v1/index/drain", { key });
    assert.equal(repeated.status, 503);
    assert.equal(await pendingCount(), after);
    assert.equal(Number((await db.all<{ count: number }>(
      `SELECT COUNT(*) AS count FROM index_outbox
        WHERE state = 'pending' AND operation = 'reconcile'`,
    ))[0]?.count ?? 0), repairsAfter);
  }
  assert.equal(mutationCalls, 6, "one vector mutation is attempted per bounded retry");
  const unavailable = await app.call("GET", "/readyz");
  assert.equal(unavailable.status, 503);
  assert.equal(unavailable.body.meta.capabilities.embedding, "enabled");
  assert.equal(unavailable.body.meta.capabilities.vector, "configured_error");
  assert.equal(unavailable.body.meta.checks.semantic_index, "vector_dependency_unavailable");

  broken.restoreStore();
  const recovered = await app.call("POST", "/v1/index/drain", { key });
  assert.equal(recovered.status, 200);
  assert.ok(recovered.body.data.indexed > 0);
  assert.ok((await pendingCount()) < before);
  assert.equal((await app.call("GET", "/readyz")).status, 200);
});

test("purge drain removes an already indexed claim", async () => {
  const client = withVectors();
  const observation = await client.call("POST", "/v1/observations", {
    key,
    body: {
      subject_id: "purge-vector",
      kind: "tool_result",
      content: "Vector purge canary evidence.",
      source: { type: "tool", ref: "vector-purge" },
      trust: "verified",
    },
  });
  assert.equal(observation.status, 201);
  const consolidated = await client.call("POST", "/v1/consolidations", {
    key,
    body: {
      subject_id: "purge-vector",
      claims: [{
        kind: "procedural",
        statement: "Vector purge canary claim.",
        sources: [{ observation_id: observation.body.data.observation_id, relation: "supports" }],
      }],
    },
  });
  assert.equal(consolidated.status, 201);
  const claimId = consolidated.body.data.claims[0].claim_id as string;

  const indexed = await client.call("POST", "/v1/index/drain?limit=100", { key });
  assert.equal(indexed.status, 200);
  assert.ok(vectors.metadataFor(claimId), "fixture must prove the vector existed first");

  const purged = await client.call("DELETE", `/v1/observations/${observation.body.data.observation_id}`, { key });
  assert.equal(purged.status, 200);
  const repeated = await client.call("DELETE", `/v1/observations/${observation.body.data.observation_id}`, { key });
  assert.equal(repeated.status, 200);
  assert.equal(repeated.body.data.already_purged, true);
  const remove = vectors.store.remove.bind(vectors.store);
  let removalUnavailable = true;
  vectors.store.remove = async (ids) => {
    if (removalUnavailable) throw new Error("controlled vector removal outage");
    await remove(ids);
  };
  const failed = await client.call("POST", "/v1/index/drain?limit=100", { key });
  assert.equal(failed.status, 503);
  assert.deepEqual(await db.all(
    `SELECT embedder_failure_at IS NOT NULL AS embedder_failed,
            vector_store_failure_at IS NOT NULL AS vector_failed
       FROM semantic_index_metadata WHERE id = 'claims'`,
  ), [{ embedder_failed: 0, vector_failed: 1 }]);
  await db.batch([{
    sql: `UPDATE semantic_index_metadata SET embedder_failure_at = ?
           WHERE id = 'claims'`,
    params: ["2026-07-31T00:00:00.000Z"],
  }]);
  removalUnavailable = false;
  const removed = await client.call("POST", "/v1/index/drain?limit=100", { key });
  vectors.store.remove = remove;
  assert.equal(removed.status, 200);
  assert.ok(removed.body.data.removed >= 1);
  assert.equal(vectors.metadataFor(claimId), undefined);
  assert.deepEqual(await db.all(
    `SELECT embedder_failure_at, vector_store_failure_at
       FROM semantic_index_metadata WHERE id = 'claims'`,
  ), [{
    embedder_failure_at: "2026-07-31T00:00:00.000Z",
    vector_store_failure_at: null,
  }]);
  assert.equal((await client.call("GET", "/readyz")).status, 503);
  await db.batch([{
    sql: `UPDATE semantic_index_metadata SET embedder_failure_at = NULL
           WHERE id = 'claims'`,
  }]);
  assert.equal((await client.call("GET", "/readyz")).status, 200);
});

test("an unavailable embedder degrades to lexical retrieval instead of failing", async () => {
  const lexicalOnly = await withoutVectors().call("POST", "/v1/context/compile", {
    key,
    body: { subject_id: subjectId, task: TASK, max_tokens: 2000 },
  });

  const broken = fakeVectors();
  broken.breakEmbedder();
  const app = clientVia(
    createApp({ db, revision: "test", runtime: "bun-sqlite", vectors: broken }),
    origin,
  );
  const res = await app.call("POST", "/v1/context/compile", {
    key,
    body: { subject_id: subjectId, task: TASK, max_tokens: 2000 },
  });

  assert.equal(res.status, 200, "a vector outage must not fail the request");
  assert.equal(res.body.meta.degraded.vector, "error");
  assert.deepEqual(
    res.body.data.items.map((item: any) => item.claim_id),
    lexicalOnly.body.data.items.map((item: any) => item.claim_id),
    "the pack must match lexical-only retrieval",
  );
});

test("scoping still runs before retrieval when vectors are enabled", async () => {
  // The vector index is deliberately told every claim is a perfect match. Scope
  // is enforced in SQL before ranking, so a foreign reader must still see none.
  for (const id of claimIds) vectors.setScore(id, 1, vectorScope());

  const intruder = await provisionWith(db, { scopes: ["*"] });
  const res = await withVectors().call("POST", "/v1/context/compile", {
    key: intruder.key,
    body: { subject_id: subjectId, task: TASK, max_tokens: 2000 },
  });
  assert.equal(res.status, 200);
  assert.equal(res.body.data.items.length, 0, "a vector hit cannot bypass scope");
  assert.equal(res.body.meta.candidates, 0);

  for (const id of claimIds) vectors.setScore(id, 0, vectorScope());
});