import assert from "node:assert/strict";
import { createApp } from "../../src/core/app";
import type { Db } from "../../src/core/db";
import { runMaintenance } from "../../src/core/maintenance";
import {
  claimSemanticIndexWork,
  completeSemanticIndexWork,
  embeddingPolicyFingerprint,
  recordSemanticDependencyFailure,
  SEMANTIC_INDEX_LEASE_MS,
} from "../../src/core/vectors";
import { clientVia, fakeVectors, provisionWith } from "./harness";

const readyWith = (db: Db, runtime: string, vectors?: ReturnType<typeof fakeVectors>) =>
  clientVia(
    createApp({ db, runtime, revision: "semantic-readiness", vectors }),
    "http://titen.test",
  ).call("GET", "/readyz");

/** Same persisted fingerprint/readiness contract against D1 and bun:sqlite. */
export async function assertSemanticReadiness(db: Db, runtime: string) {
  const diagnosticFree = await clientVia(
    createApp({
      db,
      runtime,
      semanticReadiness: {
        embedding: "configured_error",
        vector: "configured_error",
      },
    }),
    "http://titen.test",
  ).call("GET", "/readyz");
  assert.equal(diagnosticFree.status, 503);

  await db.batch([
    { sql: `DELETE FROM semantic_index_metadata` },
    { sql: `DELETE FROM index_outbox WHERE id = 'idx_semantic_legacy'` },
    {
      sql: `INSERT INTO index_outbox
              (id, org_id, record_type, record_id, operation, state, attempts, created_at)
            VALUES ('idx_semantic_legacy', 'org_test', 'claim', 'clm_test',
                    'upsert', 'done', 1, '2026-07-31T00:00:00.000Z')`,
    },
  ]);

  const legacyVectors = fakeVectors();
  const missing = await readyWith(db, runtime, legacyVectors);
  assert.equal(missing.status, 503);
  assert.equal(missing.body.meta.ready, false);
  assert.equal(missing.body.meta.capabilities.embedding, "enabled");
  assert.equal(missing.body.meta.capabilities.vector, "configured_error");
  assert.equal(missing.body.meta.checks.semantic_index, "index_fingerprint_missing");
  assert.equal(legacyVectors.embedCalls(), 0, "readiness must not call the provider");

  const dishonest = fakeVectors();
  dishonest.embedder.model = "different-runtime-model";
  dishonest.embedder.dimensions = 8;
  const invalid = await readyWith(db, runtime, dishonest);
  assert.equal(invalid.status, 503);
  assert.equal(invalid.body.meta.checks.semantic_index, "embedding_configuration_invalid");
  assert.equal(dishonest.embedCalls(), 0);

  await db.batch([
    { sql: `DELETE FROM index_outbox WHERE id = 'idx_semantic_legacy'` },
    {
      sql: `INSERT OR IGNORE INTO organizations (id, name, created_at)
            VALUES ('org_semantic_backfill', 'Semantic Backfill', '2026-07-31T00:00:00.000Z')`,
    },
    {
      sql: `INSERT INTO claims
              (id, org_id, subject_id, project_id, workspace_id, observer_id,
               actor_id, kind, statement, confidence, trust, visibility, status,
               version, valid_from, valid_to, created_at)
            VALUES ('clm_semantic_backfill', 'org_semantic_backfill', 'subject',
                    NULL, NULL, NULL, 'agent', 'semantic_fact', 'Backfill me',
                    1, 'asserted', 'private', 'active', 1,
                    '2026-07-31T00:00:00.000Z', NULL, '2026-07-31T00:00:00.000Z')`,
    },
  ]);
  const vectors = fakeVectors();
  const backfillRequired = await readyWith(db, runtime, vectors);
  assert.equal(backfillRequired.status, 503);
  assert.equal(
    backfillRequired.body.meta.checks.semantic_index,
    "index_backfill_required",
  );
  assert.equal(vectors.embedCalls(), 0);
  await db.batch([
    {
      sql: `INSERT INTO index_outbox
              (id, org_id, record_type, record_id, operation, state, attempts, created_at)
            VALUES ('idx_semantic_backfill', 'org_semantic_backfill', 'claim',
                    'clm_semantic_backfill', 'upsert', 'pending', 0,
                    '2026-07-31T00:00:00.000Z')`,
    },
  ]);
  const persistent = clientVia(
    createApp({ db, runtime, revision: "semantic-readiness", vectors }),
    "http://titen.test",
  );
  const pendingProjection = await persistent.call("GET", "/readyz");
  assert.equal(pendingProjection.status, 200);
  assert.equal(pendingProjection.body.data.ready, true);
  assert.equal(
    pendingProjection.body.data.checks.semantic_index,
    "index_projection_pending",
  );
  assert.equal(pendingProjection.body.data.capabilities.vector, "enabled");
  await db.batch([{
    sql: `UPDATE index_outbox SET state = 'done'
           WHERE id = 'idx_semantic_backfill'`,
  }]);
  const healthy = await persistent.call("GET", "/readyz");
  assert.equal(healthy.status, 200);
  assert.equal(healthy.body.data.capabilities.version, 1);
  assert.equal(healthy.body.data.capabilities.embedding, "enabled");
  assert.equal(healthy.body.data.capabilities.vector, "enabled");
  assert.equal(healthy.body.data.capabilities.extraction, "disabled");
  assert.equal(healthy.body.data.capabilities.background_enrichment, "disabled");
  assert.equal(healthy.body.data.capabilities.model, "enabled");
  assert.equal(vectors.embedCalls(), 0, "readiness must remain local");
  await db.batch([
    {
      sql: `INSERT INTO claims
              (id, org_id, subject_id, project_id, workspace_id, observer_id,
               actor_id, kind, statement, confidence, trust, visibility, status,
               version, valid_from, valid_to, created_at)
            VALUES ('clm_semantic_repeated_write', 'org_semantic_backfill', 'subject',
                    NULL, NULL, NULL, 'agent', 'semantic_fact', 'Repeated write',
                    1, 'asserted', 'private', 'active', 1,
                    '2026-07-31T00:00:00.000Z', NULL, '2026-07-31T00:00:00.000Z')`,
    },
    {
      sql: `INSERT INTO index_outbox
              (id, org_id, record_type, record_id, operation, state, attempts, created_at)
            VALUES ('idx_semantic_repeated_write', 'org_semantic_backfill', 'claim',
                    'clm_semantic_repeated_write', 'upsert', 'pending', 0,
                    '2026-07-31T00:00:00.000Z')`,
    },
  ]);
  const repeatedWrite = await persistent.call("GET", "/readyz");
  assert.equal(repeatedWrite.status, 200);
  assert.equal(repeatedWrite.body.data.checks.semantic_index, "index_projection_pending");
  await db.batch([
    { sql: `DELETE FROM index_outbox WHERE id = 'idx_semantic_repeated_write'` },
    { sql: `DELETE FROM claims WHERE id = 'clm_semantic_repeated_write'` },
  ]);
  assert.equal((await persistent.call("GET", "/readyz")).status, 200);
  const persisted = await db.all<Record<string, unknown>>(
    `SELECT provider, model, revision, dimensions, metric, preprocessing, index_schema
       FROM semantic_index_metadata WHERE id = 'claims'`,
  );
  assert.deepEqual(persisted, [{ ...vectors.fingerprint }]);

  // #250. The model-convention guard stays: a bare raw profile on an
  // EmbeddingGemma model is a configuration error, not a slow quality decay.
  const refusedProfile = fakeVectors();
  refusedProfile.embedder.model = "tuf/embeddinggemma";
  refusedProfile.fingerprint.model = "tuf/embeddinggemma";
  const refused = await readyWith(db, runtime, refusedProfile);
  assert.equal(refused.status, 503);
  assert.equal(
    refused.body.meta.checks.semantic_index,
    "embedding_configuration_invalid",
  );

  // The deliberate opt-out clears that check and lands on a rebuild instead,
  // because the profile is part of the persisted fingerprint. Every other
  // fingerprint field is byte-identical to what is stored, so the profile is
  // the only thing that can be demanding the rebuild.
  const acknowledged = fakeVectors();
  acknowledged.fingerprint.preprocessing = embeddingPolicyFingerprint(
    "raw-unit-v1-model-mismatch-acknowledged",
    0.1,
  );
  const rebuild = await readyWith(db, runtime, acknowledged);
  assert.equal(rebuild.status, 503);
  assert.equal(
    rebuild.body.meta.checks.semantic_index,
    "index_fingerprint_mismatch",
    "switching preprocessing convention must force a rebuild, never mix silently",
  );
  assert.equal(acknowledged.embedCalls(), 0, "the opt-out must not add provider I/O");

  await db.batch([{ sql: `DELETE FROM semantic_index_metadata WHERE id = 'claims'` }]);
  const lostMetadata = await persistent.call("GET", "/readyz");
  assert.equal(lostMetadata.status, 503);
  assert.equal(lostMetadata.body.meta.checks.semantic_index, "index_metadata_unavailable");
  assert.equal(vectors.embedCalls(), 0, "metadata loss readiness stays local");
  const missingFingerprint = await readyWith(db, runtime, vectors);
  assert.equal(missingFingerprint.status, 503);
  assert.equal(
    missingFingerprint.body.meta.checks.semantic_index,
    "index_fingerprint_missing",
  );
  await db.batch([{
    sql: `INSERT INTO semantic_index_metadata
            (id, provider, model, revision, dimensions, metric,
             preprocessing, index_schema, created_at)
          VALUES ('claims', ?, ?, ?, ?, ?, ?, ?, ?)`,
    params: [
      vectors.fingerprint.provider,
      vectors.fingerprint.model,
      vectors.fingerprint.revision,
      vectors.fingerprint.dimensions,
      vectors.fingerprint.metric,
      vectors.fingerprint.preprocessing,
      vectors.fingerprint.index_schema,
      "2026-07-31T00:00:00.000Z",
    ],
  }]);

  for (const dependency of ["embedder", "vector_store"] as const) {
    await db.batch([
      {
        sql: `UPDATE semantic_index_metadata
                 SET ${dependency === "embedder" ? "embedder_failure_at" : "vector_store_failure_at"} = ?
               WHERE id = 'claims'`,
        params: ["2026-07-31T00:00:01.000Z"],
      },
    ]);
    const observedFailure = await readyWith(db, runtime, vectors);
    assert.equal(observedFailure.status, 503);
    assert.equal(
      observedFailure.body.meta.checks.semantic_index,
      dependency === "embedder"
        ? "embedding_dependency_unavailable"
        : "vector_dependency_unavailable",
    );
    assert.equal(vectors.embedCalls(), 0, "observed failure readiness stays local");
    await db.batch([
      {
        sql: `UPDATE semantic_index_metadata
                 SET ${dependency === "embedder" ? "embedder_failure_at" : "vector_store_failure_at"} = NULL
               WHERE id = 'claims'`,
      },
    ]);
    assert.equal((await readyWith(db, runtime, vectors)).status, 200);
  }
  await db.batch([
    {
      sql: `UPDATE semantic_index_metadata
               SET embedder_failure_at = '2026-07-31T00:00:02.000Z',
                   vector_store_failure_at = '2026-07-31T00:00:03.000Z'
             WHERE id = 'claims'`,
    },
  ]);
  const dualFailure = await readyWith(db, runtime, vectors);
  assert.equal(dualFailure.status, 503);
  assert.equal(
    dualFailure.body.meta.checks.semantic_index,
    "semantic_dependencies_unavailable",
  );
  assert.equal(dualFailure.body.meta.capabilities.embedding, "configured_error");
  assert.equal(dualFailure.body.meta.capabilities.vector, "configured_error");
  await db.batch([
    {
      sql: `UPDATE semantic_index_metadata
               SET embedder_failure_at = NULL, vector_store_failure_at = NULL
             WHERE id = 'claims'`,
    },
  ]);

  await db.batch([
    {
      sql: `INSERT INTO organizations (id, name, created_at)
            VALUES ('org_semantic_recovery_other', 'Other recovery org',
                    '2026-07-31T00:00:00.000Z')`,
    },
    {
      sql: `INSERT INTO index_outbox
              (id, org_id, record_type, record_id, operation, state, attempts, created_at)
            VALUES ('idx_semantic_recovery_other', 'org_semantic_recovery_other',
                    'claim', 'clm_semantic_recovery_other', 'upsert', 'pending', 0,
                    '2026-07-31T00:00:00.000Z')`,
    },
  ]);
  await db.batch([{
    sql: `UPDATE index_outbox SET state = 'pending'
           WHERE id = 'idx_semantic_backfill'`,
  }]);
  const failedLease = await claimSemanticIndexWork(
    db,
    ["idx_semantic_backfill"],
  );
  await recordSemanticDependencyFailure(
    db,
    "embedder",
    "2026-07-31T00:00:04.000Z",
    ["idx_semantic_backfill"],
    failedLease.token,
  );
  assert.deepEqual(
    await db.all(`SELECT attempts FROM index_outbox WHERE id = 'idx_semantic_backfill'`),
    [{ attempts: 1 }],
  );
  assert.equal((await readyWith(db, runtime, vectors)).status, 503);
  const otherLease = await claimSemanticIndexWork(
    db,
    ["idx_semantic_recovery_other"],
  );
  await completeSemanticIndexWork(
    db,
    ["idx_semantic_recovery_other"],
    true,
    otherLease.token,
  );
  assert.equal(
    (await readyWith(db, runtime, vectors)).status,
    503,
    "another organization cannot clear unresolved failure evidence",
  );
  const recoveryLease = await claimSemanticIndexWork(
    db,
    ["idx_semantic_backfill"],
  );
  await completeSemanticIndexWork(
    db,
    ["idx_semantic_backfill"],
    true,
    recoveryLease.token,
  );
  assert.equal((await readyWith(db, runtime, vectors)).status, 200);
  await db.batch([
    { sql: `DELETE FROM index_outbox WHERE id = 'idx_semantic_recovery_other'` },
    { sql: `DELETE FROM organizations WHERE id = 'org_semantic_recovery_other'` },
  ]);

  await db.batch([
    {
      sql: `UPDATE semantic_index_metadata
               SET embedder_failure_at = '2026-07-31T00:00:05.000Z'
             WHERE id = 'claims'`,
    },
    {
      sql: `INSERT INTO organizations (id, name, created_at)
            VALUES ('org_semantic_unowned_recovery', 'Unowned recovery org',
                    '2026-07-31T00:00:00.000Z')`,
    },
    {
      sql: `INSERT INTO index_outbox
              (id, org_id, record_type, record_id, operation, state, attempts, created_at)
            VALUES
              ('idx_semantic_superseded_failure', 'org_semantic_backfill',
               'claim', 'clm_semantic_superseded_failure', 'upsert', 'done', 1,
               '2026-07-31T00:00:00.000Z'),
              ('idx_semantic_delete_only', 'org_semantic_backfill',
               'claim', 'clm_semantic_superseded_failure', 'delete', 'pending', 0,
               '2026-07-31T00:00:01.000Z'),
              ('idx_semantic_unowned_success', 'org_semantic_unowned_recovery',
               'claim', 'clm_semantic_unowned_success', 'upsert', 'pending', 0,
               '2026-07-31T00:00:02.000Z')`,
    },
  ]);
  const deleteOnlyLease = await claimSemanticIndexWork(
    db,
    ["idx_semantic_delete_only"],
  );
  await completeSemanticIndexWork(
    db,
    ["idx_semantic_delete_only"],
    false,
    deleteOnlyLease.token,
  );
  assert.equal((await readyWith(db, runtime, vectors)).status, 503,
    "delete-only completion is not dependency recovery");
  const unownedSuccessLease = await claimSemanticIndexWork(
    db,
    ["idx_semantic_unowned_success"],
  );
  await completeSemanticIndexWork(
    db,
    ["idx_semantic_unowned_success"],
    true,
    unownedSuccessLease.token,
  );
  assert.equal((await readyWith(db, runtime, vectors)).status, 503,
    "unowned success without a failed attempt cannot erase outage evidence");
  await db.batch([{
    sql: `INSERT INTO index_outbox
            (id, org_id, record_type, record_id, operation, state, attempts, created_at)
          VALUES ('idx_semantic_owned_retry', 'org_semantic_backfill', 'claim',
                  'clm_semantic_owned_retry', 'upsert', 'pending', 1,
                  '2026-07-31T00:00:03.000Z')`,
  }]);
  const ownedRetryLease = await claimSemanticIndexWork(
    db,
    ["idx_semantic_owned_retry"],
  );
  await completeSemanticIndexWork(
    db,
    ["idx_semantic_owned_retry"],
    true,
    ownedRetryLease.token,
  );
  assert.equal((await readyWith(db, runtime, vectors)).status, 200,
    "a successful retry of previously failed work proves recovery");
  await db.batch([
    {
      sql: `DELETE FROM index_outbox
             WHERE id IN ('idx_semantic_superseded_failure',
                          'idx_semantic_delete_only',
                          'idx_semantic_unowned_success',
                          'idx_semantic_owned_retry')`,
    },
    { sql: `DELETE FROM organizations WHERE id = 'org_semantic_unowned_recovery'` },
  ]);

  await db.batch([
    {
      sql: `INSERT INTO claims
              (id, org_id, subject_id, project_id, workspace_id, observer_id,
               actor_id, kind, statement, confidence, trust, visibility, status,
               version, valid_from, valid_to, created_at)
            VALUES ('clm_semantic_interim', 'org_semantic_backfill', 'subject',
                    NULL, NULL, NULL, 'agent', 'semantic_fact', 'Interim FTS claim',
                    1, 'asserted', 'private', 'active', 1,
                    '2026-07-31T00:00:00.000Z', NULL, '2026-07-31T00:00:00.000Z')`,
    },
  ]);
  const interimGuard = await readyWith(db, runtime, vectors);
  assert.equal(interimGuard.status, 503);
  assert.equal(interimGuard.body.meta.checks.semantic_index, "index_backfill_required");
  await db.batch([
    {
      sql: `INSERT INTO index_outbox
              (id, org_id, record_type, record_id, operation, state, attempts, created_at)
            VALUES ('idx_semantic_interim', 'org_semantic_backfill', 'claim',
                    'clm_semantic_interim', 'upsert', 'pending', 0,
                    '2026-07-31T00:00:00.000Z')`,
    },
  ]);
  const queuedProjection = await readyWith(db, runtime, vectors);
  assert.equal(queuedProjection.status, 200);
  assert.equal(
    queuedProjection.body.data.checks.semantic_index,
    "index_projection_pending",
  );
  assert.equal(queuedProjection.body.data.capabilities.vector, "enabled");

  await db.batch([
    {
      sql: `UPDATE index_outbox SET state = 'done'
             WHERE id IN ('idx_semantic_backfill', 'idx_semantic_interim')`,
    },
  ]);
  const restoredWithoutProjection = fakeVectors();
  restoredWithoutProjection.indexEmpty = true;
  for (let restart = 0; restart < 2; restart += 1) {
    const restoreGuard = await readyWith(db, runtime, restoredWithoutProjection);
    assert.equal(restoreGuard.status, 503);
    assert.equal(restoreGuard.body.meta.checks.semantic_index, "index_backfill_required");
  }
  assert.equal(restoredWithoutProjection.embedCalls(), 0);

  const changed = fakeVectors();
  changed.fingerprint = { ...changed.fingerprint, model: "contract-stub-v2" };
  changed.embedder.model = "contract-stub-v2";
  const mismatch = await readyWith(db, runtime, changed);
  assert.equal(mismatch.status, 503);
  assert.equal(mismatch.body.meta.capabilities.embedding, "enabled");
  assert.equal(mismatch.body.meta.capabilities.vector, "configured_error");
  assert.equal(mismatch.body.meta.checks.semantic_index, "index_fingerprint_mismatch");
  assert.doesNotMatch(JSON.stringify(mismatch.body), /contract-stub-v2/);
  assert.equal(changed.embedCalls(), 0);

  const lexicalOnly = await readyWith(db, runtime);
  assert.equal(lexicalOnly.status, 200);
  assert.equal(lexicalOnly.body.data.capabilities.embedding, "disabled");
  assert.equal(lexicalOnly.body.data.capabilities.vector, "disabled");

  // Explicit reindex resets metadata and requeues derived work before the new
  // fingerprint can become ready. The vector backend reset is runtime-specific.
  await db.batch([
    { sql: `DELETE FROM semantic_index_metadata WHERE id = 'claims'` },
    {
      sql: `UPDATE index_outbox SET state = 'pending', attempts = 0
             WHERE record_type = 'claim'`,
    },
  ]);
  const recovered = await readyWith(db, runtime, changed);
  assert.equal(recovered.status, 200);
  assert.equal(recovered.body.data.checks.semantic_index, "index_projection_pending");
  assert.equal(recovered.body.data.capabilities.vector, "enabled");
  assert.equal(changed.embedCalls(), 0);

  await db.batch([
    { sql: `DELETE FROM semantic_index_metadata` },
    {
      sql: `DELETE FROM index_outbox WHERE org_id = 'org_semantic_backfill'`,
    },
    {
      sql: `DELETE FROM claims
             WHERE id IN ('clm_semantic_backfill', 'clm_semantic_interim')`,
    },
    { sql: `DELETE FROM organizations WHERE id = 'org_semantic_backfill'` },
  ]);
}

/** Same owner-fence race against bun:sqlite and D1. */
export async function assertSemanticIndexOwnership(db: Db, runtime: string) {
  for (const winner of ["manual", "background"] as const) {
    const provisioned = await provisionWith(db, { scopes: ["*"] });
    const claimId = `clm_index_owner_${winner}`;
    const outboxId = `idx_index_owner_${winner}`;
    const base = new Date("2026-07-31T01:00:00.000Z");
    const takeover = new Date(base.getTime() + SEMANTIC_INDEX_LEASE_MS + 1);
    const vectors = fakeVectors();
    const embed = vectors.embedder.embed.bind(vectors.embedder);
    let providerCalls = 0;
    let releaseFirst!: () => void;
    let markStarted!: () => void;
    const firstStarted = new Promise<void>((resolve) => {
      markStarted = resolve;
    });
    const firstRelease = new Promise<void>((resolve) => {
      releaseFirst = resolve;
    });
    vectors.embedder.embed = async (texts) => {
      providerCalls += 1;
      if (providerCalls === 1) {
        markStarted();
        await firstRelease;
        throw new Error("controlled late embedder failure");
      }
      return embed(texts);
    };

    await db.batch([
      {
        sql: `INSERT OR IGNORE INTO semantic_index_metadata
                (id, provider, model, revision, dimensions, metric,
                 preprocessing, index_schema, created_at)
              VALUES ('claims', ?, ?, ?, ?, ?, ?, ?, ?)`,
        params: [
          vectors.fingerprint.provider,
          vectors.fingerprint.model,
          vectors.fingerprint.revision,
          vectors.fingerprint.dimensions,
          vectors.fingerprint.metric,
          vectors.fingerprint.preprocessing,
          vectors.fingerprint.index_schema,
          base.toISOString(),
        ],
      },
      {
        sql: `UPDATE semantic_index_metadata
                 SET embedder_failure_at = NULL, vector_store_failure_at = NULL
               WHERE id = 'claims'`,
      },
      {
        sql: `INSERT INTO claims
                (id, org_id, subject_id, project_id, workspace_id, observer_id,
                 actor_id, kind, statement, confidence, trust, visibility, status,
                 version, valid_from, valid_to, created_at)
              VALUES (?, ?, 'subject_index_owner', NULL, NULL, NULL, ?,
                      'semantic_fact', 'Owner fencing keeps readiness truthful.',
                      1, 'verified', 'private', 'active', 1, ?, NULL, ?)`,
        params: [
          claimId,
          provisioned.orgId,
          provisioned.principalId,
          base.toISOString(),
          base.toISOString(),
        ],
      },
      {
        sql: `INSERT INTO index_outbox
                (id, org_id, record_type, record_id, operation, state, attempts, created_at)
              VALUES (?, ?, 'claim', ?, 'upsert', 'pending', 0, ?)`,
        params: [outboxId, provisioned.orgId, claimId, base.toISOString()],
      },
    ]);

    const manualAt = (at: Date) => clientVia(
      createApp({
        db,
        runtime,
        revision: "index-owner",
        vectors,
        semanticPrepared: true,
        now: () => at,
      }),
      "http://titen.test",
    );
    const late = manualAt(base).call("POST", "/v1/index/drain?limit=1", {
      key: provisioned.key,
    });
    await firstStarted;
    assert.deepEqual(
      await db.all(
        `SELECT state, attempts, lease_token IS NOT NULL AS owned
           FROM index_outbox WHERE id = ?`,
        [outboxId],
      ),
      [{ state: "pending", attempts: 0, owned: 1 }],
    );

    const contend = async (at: Date): Promise<number> => {
      if (winner === "manual") {
        const result = await manualAt(at).call("POST", "/v1/index/drain?limit=1", {
          key: provisioned.key,
        });
        assert.equal(result.status, 200);
        return result.body.data.indexed;
      }
      const result = await runMaintenance({
        db,
        vectors,
        limit: 1,
        now: at,
        deliverWebhooks: false,
      });
      assert.deepEqual(result.errors, []);
      return result.indexed;
    };
    assert.equal(await contend(new Date(base.getTime() + 1)), 0);
    assert.equal(providerCalls, 1, "an active owner excludes an overlapping drain");
    await db.batch([{
      sql: `UPDATE index_outbox
               SET lease_expires_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-1 second')
             WHERE id = ?`,
      params: [outboxId],
    }]);
    assert.equal(await contend(takeover), 1);

    releaseFirst();
    const stale = await late;
    assert.equal(stale.status, 503);
    assert.equal(stale.body.meta.dependency, "embedder");
    assert.deepEqual(
      await db.all(
        `SELECT state, attempts, lease_token, lease_expires_at
           FROM index_outbox WHERE id = ?`,
        [outboxId],
      ),
      [{ state: "done", attempts: 1, lease_token: null, lease_expires_at: null }],
    );
    assert.deepEqual(
      await db.all(
        `SELECT embedder_failure_at, vector_store_failure_at
           FROM semantic_index_metadata WHERE id = 'claims'`,
      ),
      [{ embedder_failure_at: null, vector_store_failure_at: null }],
    );

    const callsBeforeReadiness = providerCalls;
    const ready = await manualAt(takeover).call("GET", "/readyz");
    assert.equal(ready.status, 200);
    assert.equal(providerCalls, callsBeforeReadiness);
    const idle = await manualAt(takeover).call("POST", "/v1/index/drain?limit=1", {
      key: provisioned.key,
    });
    assert.equal(idle.status, 200);
    assert.equal(idle.body.data.drained, 0);
    assert.equal(idle.body.data.remaining, 0);

    await db.batch([
      { sql: `DELETE FROM index_outbox WHERE id = ?`, params: [outboxId] },
      { sql: `DELETE FROM claims WHERE id = ?`, params: [claimId] },
      { sql: `DELETE FROM api_keys WHERE org_id = ?`, params: [provisioned.orgId] },
      { sql: `DELETE FROM organizations WHERE id = ?`, params: [provisioned.orgId] },
    ]);
  }
}

/** Same stale external-write repair contract against bun:sqlite and D1. */
export async function assertSemanticIndexWriteRepair(db: Db, runtime: string) {
  const fingerprint = fakeVectors().fingerprint;
  await db.batch([
    {
      sql: `INSERT OR IGNORE INTO semantic_index_metadata
              (id, provider, model, revision, dimensions, metric,
               preprocessing, index_schema, created_at)
            VALUES ('claims', ?, ?, ?, ?, ?, ?, ?, ?)`,
      params: [
        fingerprint.provider,
        fingerprint.model,
        fingerprint.revision,
        fingerprint.dimensions,
        fingerprint.metric,
        fingerprint.preprocessing,
        fingerprint.index_schema,
        "2026-07-31T00:00:00.000Z",
      ],
    },
    {
      sql: `UPDATE semantic_index_metadata
               SET embedder_failure_at = NULL, vector_store_failure_at = NULL
             WHERE id = 'claims'`,
    },
  ]);
  const run = async (
    winner: "manual" | "background",
    purge: boolean,
    crashAfterWrite = false,
    applyThenThrow = false,
  ) => {
    const provisioned = await provisionWith(db, { scopes: ["*"] });
    const caseId = `${winner}_${purge}_${crashAfterWrite}_${applyThenThrow}`;
    const base = new Date(
      `2026-07-31T0${winner === "manual" ? 2 : 3}:00:00.000Z`,
    );
    let contenderAt = purge
      ? new Date(base.getTime() + 1)
      : new Date(base.getTime() + SEMANTIC_INDEX_LEASE_MS + 1);
    const vectors = fakeVectors();
    const upsert = vectors.store.upsert.bind(vectors.store);
    const remove = vectors.store.remove.bind(vectors.store);
    let release!: () => void;
    let started!: () => void;
    const blocked = new Promise<void>((resolve) => { started = resolve; });
    const resume = new Promise<void>((resolve) => { release = resolve; });
    let upsertCalls = 0;
    let visibleGeneration = 0;
    vectors.store.upsert = async (records) => {
      upsertCalls += 1;
      const generation = upsertCalls;
      if (upsertCalls === 1) {
        started();
        await resume;
      }
      await upsert(records);
      visibleGeneration = generation;
      if (generation === 1 && applyThenThrow)
        throw new Error("controlled apply-then-throw vector failure");
    };
    let releaseRemoval!: () => void;
    let removalStarted!: () => void;
    const removalBlocked = new Promise<void>((resolve) => { removalStarted = resolve; });
    const removalResume = new Promise<void>((resolve) => { releaseRemoval = resolve; });
    let firstRemoval = !purge;
    vectors.store.remove = async (ids) => {
      if (firstRemoval) {
        firstRemoval = false;
        removalStarted();
        await removalResume;
      }
      await remove(ids);
    };

    let stopAfterWrite = crashAfterWrite;
    const ownerDb: Db = {
      all: (sql, params) => db.all(sql, params),
      async batch(statements) {
        if (
          stopAfterWrite && visibleGeneration > 0 &&
          statements.some(({ sql }) => sql.includes("SET state = 'done'"))
        ) {
          stopAfterWrite = false;
          throw new Error("controlled process stop after external vector write");
        }
        await db.batch(statements);
      },
      exec: (sql) => db.exec(sql),
    };
    const clientAt = (at: Date | (() => Date), targetDb: Db = db) => clientVia(
      createApp({
        db: targetDb,
        runtime,
        revision: "index-write-repair",
        vectors,
        semanticPrepared: true,
        now: typeof at === "function" ? at : () => at,
      }),
      "http://titen.test",
    );
    const seed = clientAt(base);
    const observation = await seed.call("POST", "/v1/observations", {
      key: provisioned.key,
      body: {
        subject_id: `subject_write_repair_${winner}_${purge}`,
        kind: "tool_result",
        content: "Synthetic stale write repair evidence.",
        source: { type: "tool", ref: `write-repair-${winner}-${purge}` },
        trust: "verified",
      },
    });
    assert.equal(observation.status, 201);
    const observationId = observation.body.data.observation_id as string;
    const consolidated = await seed.call("POST", "/v1/consolidations", {
      key: provisioned.key,
      body: {
        subject_id: `subject_write_repair_${winner}_${purge}`,
        claims: [{
          kind: "procedural",
          statement: "A stale vector write must retain durable repair.",
          sources: [{ observation_id: observationId, relation: "supports" }],
        }],
      },
    });
    assert.equal(consolidated.status, 201);
    const claimId = consolidated.body.data.claims[0].claim_id as string;

    if (!purge) await db.batch([{
      sql: `INSERT INTO index_outbox
              (id, org_id, record_type, record_id, operation, state, attempts, created_at)
            VALUES (?, ?, 'claim', ?, 'delete', 'pending', 0, ?)`,
      params: [
        `idx_slow_delete_${caseId}`,
        provisioned.orgId,
        `clm_slow_delete_${caseId}`,
        new Date(base.getTime() - 1).toISOString(),
      ],
    }]);

    let ownerAt = base;
    const staleCall = clientAt(() => ownerAt, ownerDb).call("POST", "/v1/index/drain?limit=100", {
      key: provisioned.key,
    });
    if (!purge) {
      await removalBlocked;
      ownerAt = contenderAt;
      releaseRemoval();
    }
    await blocked;
    const writeAhead = await db.all<{ operation: string; state: string; org_id: string }>(
      `SELECT operation, state, org_id FROM index_outbox
        WHERE record_type = 'claim' AND record_id = ? AND operation = 'reconcile'
          AND state = 'pending' AND lease_token IS NOT NULL`,
      [claimId],
    );
    assert.deepEqual(writeAhead, [{
      operation: "reconcile",
      state: "pending",
      org_id: provisioned.orgId,
    }]);

    if (purge) {
      const purged = await clientAt(contenderAt).call(
        "DELETE",
        `/v1/observations/${observationId}`,
        { key: provisioned.key },
      );
      assert.equal(purged.status, 200);
    }
    const contend = async (at: Date): Promise<number> => {
      if (winner === "manual") {
        const result = await clientAt(at).call(
          "POST",
          "/v1/index/drain?limit=100",
          { key: provisioned.key },
        );
        assert.equal(result.status, 200);
        return result.body.data.indexed;
      }
      const result = await runMaintenance({
        db,
        vectors,
        limit: 100,
        now: at,
        deliverWebhooks: false,
      });
      assert.deepEqual(result.errors, []);
      return result.indexed;
    };
    if (purge) {
      await contend(contenderAt);
    } else {
      const lease = (await db.all<{ lease_expires_at: string; inspected_at: string }>(
        `SELECT lease_expires_at,
                strftime('%Y-%m-%dT%H:%M:%fZ', 'now') AS inspected_at
           FROM index_outbox
          WHERE record_type = 'claim' AND record_id = ? AND operation = 'upsert'`,
        [claimId],
      ))[0];
      assert.ok(lease?.lease_expires_at);
      assert.ok(
        Date.parse(lease.lease_expires_at) - Date.parse(lease.inspected_at) >=
          SEMANTIC_INDEX_LEASE_MS - 1_000,
        "later work receives a full database-clock lease",
      );
      assert.equal(await contend(new Date("2099-01-01T00:00:00.000Z")), 0);
      assert.equal(await contend(new Date("2000-01-01T00:00:00.000Z")), 0);
      assert.equal(upsertCalls, 1, "an immediate contender performs no provider write");
      await db.batch([{
        sql: `UPDATE index_outbox
                 SET lease_expires_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-1 second')
               WHERE record_type = 'claim' AND record_id = ? AND state = 'pending'`,
        params: [claimId],
      }]);
      assert.ok(await contend(contenderAt) >= 1);
    }
    if (purge) assert.equal(vectors.metadataFor(claimId), undefined);
    else assert.ok(vectors.metadataFor(claimId));

    release();
    const stale = await staleCall;
    assert.equal(stale.status, crashAfterWrite ? 500 : applyThenThrow ? 503 : 200);
    if (applyThenThrow) {
      assert.equal(visibleGeneration, 1, "the stale provider result must be externally visible");
      assert.ok(vectors.metadataFor(claimId));
    } else if (!crashAfterWrite) {
      assert.equal(stale.body.data.indexed, 0);
      assert.ok(stale.body.data.remaining > 0);
      assert.equal(vectors.metadataFor(claimId), undefined);
    } else {
      assert.ok(vectors.metadataFor(claimId), "the crash fixture must leave the external write visible");
    }
    const canonical = (await db.all<{ status: string }>(
      `SELECT status FROM claims WHERE id = ? AND org_id = ?`,
      [claimId, provisioned.orgId],
    ))[0];
    assert.equal(canonical?.status, purge ? "revoked" : "active");
    const queued = await db.all<{ org_id: string; operation: string }>(
      `SELECT org_id, operation FROM index_outbox
        WHERE record_type = 'claim' AND record_id = ? AND state = 'pending'`,
      [claimId],
    );
    assert.ok(queued.length > 0);
    assert.ok(queued.every(({ org_id }) => org_id === provisioned.orgId));

    // A fresh app instance against the same SQL and vector stores converges.
    await db.batch([{
      sql: `UPDATE index_outbox
               SET lease_expires_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-1 second')
             WHERE record_type = 'claim' AND record_id = ? AND state = 'pending'
               AND lease_token IS NOT NULL`,
      params: [claimId],
    }]);
    const repairAt = new Date(contenderAt.getTime() + 1);
    const repaired = await clientAt(repairAt).call(
      "POST",
      "/v1/index/drain?limit=100",
      { key: provisioned.key },
    );
    assert.equal(repaired.status, 200);
    assert.equal(repaired.body.data.remaining, 0);
    if (purge) assert.equal(vectors.metadataFor(claimId), undefined);
    else {
      assert.ok(vectors.metadataFor(claimId));
      if (applyThenThrow)
        assert.ok(visibleGeneration > 1, "fresh repair must overwrite the ambiguous stale generation");
    }
  };

  assert.deepEqual((await claimSemanticIndexWork(db, [])).ids, []);

  await run("manual", true, true);
  await run("background", true);
  await run("manual", false);
  await run("background", false);
  await run("manual", false, false, true);

  const runStaleRemoval = async (
    winner: "manual" | "background",
    applyThenThrow: boolean,
  ) => {
    const provisioned = await provisionWith(db, { scopes: ["*"] });
    const suffix = `${winner}_${applyThenThrow ? "ambiguous" : "success"}`;
    const vectors = fakeVectors();
    const at = (value: Date) => clientVia(createApp({
      db,
      runtime,
      revision: "index-remove-repair",
      vectors,
      semanticPrepared: true,
      now: () => value,
    }), "http://titen.test");
    const base = new Date("2026-07-31T05:00:00.000Z");
    const seed = at(base);
    const observation = await seed.call("POST", "/v1/observations", {
      key: provisioned.key,
      body: {
        subject_id: `subject_remove_repair_${suffix}`,
        kind: "tool_result",
        content: "Synthetic stale removal repair evidence.",
        source: { type: "tool", ref: `remove-repair-${suffix}` },
        trust: "verified",
      },
    });
    assert.equal(observation.status, 201);
    const consolidated = await seed.call("POST", "/v1/consolidations", {
      key: provisioned.key,
      body: {
        subject_id: `subject_remove_repair_${suffix}`,
        claims: [{
          kind: "procedural",
          statement: "Canonical reconciliation repairs a stale vector removal.",
          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 seed.call("POST", "/v1/index/drain?limit=100", {
      key: provisioned.key,
    });
    assert.equal(indexed.status, 200);
    assert.ok(vectors.metadataFor(claimId));

    await db.batch([{
      sql: `INSERT INTO index_outbox
              (id, org_id, record_type, record_id, operation, state, attempts, created_at)
            VALUES (?, ?, 'claim', ?, 'delete', 'pending', 0, ?)`,
      params: [`idx_stale_remove_${suffix}`, provisioned.orgId, claimId, base.toISOString()],
    }]);
    const remove = vectors.store.remove.bind(vectors.store);
    let release!: () => void;
    let started!: () => void;
    const blocked = new Promise<void>((resolve) => { started = resolve; });
    const resume = new Promise<void>((resolve) => { release = resolve; });
    let removeCalls = 0;
    vectors.store.remove = async (ids) => {
      removeCalls += 1;
      const call = removeCalls;
      if (call === 1) {
        started();
        await resume;
      }
      await remove(ids);
      if (call === 1 && applyThenThrow)
        throw new Error("controlled apply-then-throw remove failure");
    };

    const staleCall = at(base).call("POST", "/v1/index/drain?limit=1", {
      key: provisioned.key,
    });
    await blocked;
    assert.equal(Number((await db.all<{ count: number }>(
      `SELECT COUNT(*) AS count FROM index_outbox
        WHERE org_id = ? AND record_id = ? AND operation = 'reconcile'
          AND state = 'pending' AND lease_token IS NOT NULL`,
      [provisioned.orgId, claimId],
    ))[0]?.count ?? 0), 1);
    await db.batch([{
      sql: `UPDATE index_outbox
               SET lease_expires_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-1 second')
             WHERE org_id = ? AND record_id = ? AND state = 'pending'`,
      params: [provisioned.orgId, claimId],
    }]);

    if (winner === "manual") {
      const takeover = await at(new Date("2099-01-01T00:00:00.000Z")).call(
        "POST",
        "/v1/index/drain?limit=100",
        { key: provisioned.key },
      );
      assert.equal(takeover.status, 200);
      assert.ok(takeover.body.data.indexed >= 1);
    } else {
      const takeover = await runMaintenance({
        db,
        vectors,
        limit: 100,
        now: new Date("2000-01-01T00:00:00.000Z"),
        deliverWebhooks: false,
      });
      assert.deepEqual(takeover.errors, []);
      assert.ok(takeover.indexed >= 1);
    }
    assert.ok(vectors.metadataFor(claimId), "takeover restores the current claim");

    release();
    const stale = await staleCall;
    assert.equal(stale.status, applyThenThrow ? 503 : 200);
    if (!applyThenThrow) assert.equal(stale.body.data.removed, 0);
    assert.equal(vectors.metadataFor(claimId), undefined);
    assert.ok(Number((await db.all<{ count: number }>(
      `SELECT COUNT(*) AS count FROM index_outbox
        WHERE org_id = ? AND record_id = ? AND operation = 'reconcile'
          AND state = 'pending' AND lease_token IS NULL`,
      [provisioned.orgId, claimId],
    ))[0]?.count ?? 0) >= 1);
    assert.deepEqual(await db.all(
      `SELECT embedder_failure_at, vector_store_failure_at
         FROM semantic_index_metadata WHERE id = 'claims'`,
    ), [{ embedder_failure_at: null, vector_store_failure_at: null }]);

    const repaired = await at(base).call("POST", "/v1/index/drain?limit=100", {
      key: provisioned.key,
    });
    assert.equal(repaired.status, 200);
    assert.equal(repaired.body.data.remaining, 0);
    assert.ok(vectors.metadataFor(claimId));
  };

  await runStaleRemoval("manual", false);
  await runStaleRemoval("background", true);

  const early = await provisionWith(db, { scopes: ["*"] });
  const later = await provisionWith(db, { scopes: ["*"] });
  const base = new Date("2026-07-31T04:00:00.000Z");
  const vectors = fakeVectors();
  const seed = clientAt(base);
  function clientAt(at: Date) {
    return clientVia(createApp({
      db,
      runtime,
      revision: "index-cross-org-lease",
      vectors,
      semanticPrepared: true,
      now: () => at,
    }), "http://titen.test");
  }
  const observation = await seed.call("POST", "/v1/observations", {
    key: later.key,
    body: {
      subject_id: "subject_cross_org_lease",
      kind: "tool_result",
      content: "Synthetic later organization lease evidence.",
      source: { type: "tool", ref: "cross-org-lease" },
      trust: "verified",
    },
  });
  assert.equal(observation.status, 201);
  const consolidated = await seed.call("POST", "/v1/consolidations", {
    key: later.key,
    body: {
      subject_id: "subject_cross_org_lease",
      claims: [{
        kind: "procedural",
        statement: "Later organizations receive fresh index leases.",
        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;
  await db.batch([{
    sql: `INSERT INTO index_outbox
            (id, org_id, record_type, record_id, operation, state, attempts, created_at)
          VALUES ('idx_early_org_delete', ?, 'claim', 'clm_early_org_delete',
                  'delete', 'pending', 0, ?)`,
    params: [early.orgId, new Date(base.getTime() - 1).toISOString()],
  }]);

  const remove = vectors.store.remove.bind(vectors.store);
  let releaseRemoval!: () => void;
  let removalStarted!: () => void;
  const removalBlocked = new Promise<void>((resolve) => { removalStarted = resolve; });
  const removalResume = new Promise<void>((resolve) => { releaseRemoval = resolve; });
  vectors.store.remove = async (ids) => {
    removalStarted();
    await removalResume;
    await remove(ids);
  };
  const upsert = vectors.store.upsert.bind(vectors.store);
  let releaseUpsert!: () => void;
  let upsertStarted!: () => void;
  const upsertBlocked = new Promise<void>((resolve) => { upsertStarted = resolve; });
  const upsertResume = new Promise<void>((resolve) => { releaseUpsert = resolve; });
  let upserts = 0;
  vectors.store.upsert = async (records) => {
    upserts += 1;
    upsertStarted();
    await upsertResume;
    await upsert(records);
  };

  let ownerAt = base;
  const owner = runMaintenance({
    db,
    vectors,
    limit: 100,
    now: () => ownerAt,
    deliverWebhooks: false,
  });
  await removalBlocked;
  ownerAt = new Date(base.getTime() + 86_400_000);
  releaseRemoval();
  await upsertBlocked;
  const lease = (await db.all<{ lease_expires_at: string; inspected_at: string }>(
    `SELECT lease_expires_at,
            strftime('%Y-%m-%dT%H:%M:%fZ', 'now') AS inspected_at
       FROM index_outbox
      WHERE org_id = ? AND record_id = ? AND operation = 'upsert'`,
    [later.orgId, claimId],
  ))[0];
  assert.ok(lease?.lease_expires_at);
  assert.ok(
    Date.parse(lease.lease_expires_at) - Date.parse(lease.inspected_at) >=
      SEMANTIC_INDEX_LEASE_MS - 1_000,
    "a later organization receives a full database-clock lease",
  );
  const providerCalls = vectors.embedCalls();
  const contender = await clientAt(new Date("2000-01-01T00:00:00.000Z")).call("POST", "/v1/index/drain?limit=100", {
    key: later.key,
  });
  assert.equal(contender.status, 200);
  assert.equal(contender.body.data.indexed, 0);
  assert.equal(vectors.embedCalls(), providerCalls);
  assert.equal(upserts, 1);
  releaseUpsert();
  const result = await owner;
  assert.deepEqual(result.errors, []);
  assert.equal(result.indexed, 1);
}