/**
 * Applying an inbox deposit is a DISPATCH on what the deposit declares itself to be, and
 * this suite is about the two things a `match` owes: every arm does real work, and the
 * arm that does not exist SAYS SO.
 *
 * ── What each test would have looked like before ──────────────────────────
 * Before 2026-08-21 `processInbox` iterated the caps a read had observed. A deposit of any
 * other shape was read, counted, logged — and produced nothing: no effect, and no reported
 * failure. So a reference deposit vanished, a malformed Link vanished, and a kind nobody
 * had written an arm for vanished, all three indistinguishable from an empty inbox. Every
 * test below fails against that code, and fails again if the arm it exercises is removed.
 *
 * ── Nobody is handed an inbox address ─────────────────────────────────────
 * Each depositor names the DOCUMENT (`inbox.postToDocument`), which is all an application
 * has; the address is resolved inside. The one address the TEST resolves for itself is
 * used only to assert what landed, never given to an actor.
 */
import { test, expect, mock, afterEach } from "bun:test";
import { configure } from "../src/index";
import {
  configureStoreRegistry,
  resetCaps,
  resetConfig,
  resetStoreRegistry,
  setCurrentUser,
} from "../src/shared-wallet/bootstrap";
import {
  createEntityDoc,
  resetRegistryCache,
  resolveWriteGraph,
} from "../src/shared-wallet/account-registry";
import type { RegistrySession } from "../src/shared-wallet/account-registry";
import { openDocumentInbox } from "../src/emulated-verifier/branch-registers";
import { connectedUser } from "../src/emulated-verifier/connect";
import {
  cancelScheduledInboxProcessing,
  runScheduledInboxProcessingNow,
} from "../src/emulated-verifier/inbox-processor";
import { stopObservingInboxes } from "../src/emulated-verifier/inbox-observer";
import { resetOpenedRepos, setOpenTimeoutForTests } from "../src/emulated-verifier/open-repo";
import { resetPublicStoreFetches } from "../src/emulated-verifier/public-store";
import { ENTRY_VALUE, INDEX_FIELD } from "../src/emulated-verifier/index-deposit";
import { postToDocument } from "../src/surface/inbox";
import { readUnion } from "../src/surface/read-model";
import { sparqlUpdate } from "../src/surface/docs";
import type { Nuri } from "../src/model/types";

const SESSION: RegistrySession = { sessionId: "sid-dispatch", privateStoreId: "PRIV-DISPATCH" };
const SHIM = "urn:ng-eventually:shim";
const INBOX = "urn:ng-eventually:inbox";
/** The field the index documents below declare they index by. */
const WHEN = "urn:test:when";

interface Quad { g: string; s: string; p: string; o: string }

function unescapeLiteral(s: string): string {
  let out = "";
  for (let i = 0; i < s.length; i++) {
    if (s[i] === "\\" && i + 1 < s.length) {
      const next = s[++i];
      out += next === "n" ? "\n" : next === "r" ? "\r" : next === "t" ? "\t" : next!;
    } else out += s[i];
  }
  return out;
}

/** A stateful fake `ng`: the shim SPARQL, the inbox SPARQL, and the anchored per-doc read. */
function makeFakeNg() {
  const quads: Quad[] = [];
  let docCounter = 0;

  const doc_create = mock(async () => `did:ng:o:doc${++docCounter}`);

  const sparql_update = mock(async (...a: unknown[]) => {
    const query = a[1] as string;
    const anchor = a[2] as string | undefined;
    if (!anchor) return undefined;
    const del = query.match(/^\s*DELETE\s+WHERE\s*\{\s*<([^>]+)>\s+<([^>]+)>\s+\?/);
    if (del) {
      const [s0, p0] = [del[1]!, del[2]!];
      for (let i = quads.length - 1; i >= 0; i--) {
        const q = quads[i]!;
        if (q.g === anchor && q.s === s0 && q.p === p0) quads.splice(i, 1);
      }
      return undefined;
    }
    // Retiring a DEPOSIT: every triple of one subject, whatever predicates it carries
    // (`emulated-verifier/inbox-consume.removeDeposit`). Modelled here rather than ignored
    // — a fake that dropped this delete would leave every deposit in place and let this
    // whole suite pass on a queue that never shrinks.
    const delAll = query.match(/^\s*DELETE\s+WHERE\s*\{\s*<([^>]+)>\s+\?\w+\s+\?\w+\s*\}/);
    if (delAll) {
      const s0 = delAll[1]!;
      for (let i = quads.length - 1; i >= 0; i--) {
        const q = quads[i]!;
        if (q.g === anchor && q.s === s0) quads.splice(i, 1);
      }
      return undefined;
    }
    const body = query.replace(/^\s*INSERT DATA\s*\{/, "").replace(/\}\s*$/, "");
    const sm = body.match(/<([^>]+)>/);
    if (!sm) return undefined;
    const s = sm[1]!;
    const pairRe = /(?:a|<([^>]+)>)\s+(?:"((?:[^"\\]|\\.)*)"|<([^>]+)>)/g;
    let m: RegExpExecArray | null;
    const after = body.slice(body.indexOf(sm[0]) + sm[0].length);
    while ((m = pairRe.exec(after)) !== null) {
      const p = m[1] ?? (query.includes(`${INBOX}:Deposit`) ? `${INBOX}:Deposit` : `${SHIM}:Account`);
      const o = m[2] !== undefined ? unescapeLiteral(m[2]) : (m[3] ?? "");
      quads.push({ g: anchor, s, p, o });
    }
    return undefined;
  });

  const sparql_query = mock(async (...a: unknown[]) => {
    const query = a[1] as string;
    const anchor = a[3] as string | undefined;
    const byPred = (pred: string, v: string) => ({
      results: {
        bindings: quads
          .filter((q) => q.g === anchor && q.p === pred)
          .map((q) => ({ [v]: { value: q.o } })),
      },
    });
    if (query.includes(`<${SHIM}:shimDoc>`)) return byPred(`${SHIM}:shimDoc`, "shimDoc");
    if (query.includes(`<${SHIM}:id>`)) {
      const subjM = query.match(/<([^>]+)>\s+a\s+<urn:ng-eventually:shim:Account>/);
      const only = subjM ? subjM[1]! : null;
      const bySubject = new Map<string, Record<string, string>>();
      for (const q of quads) {
        if (q.g !== anchor) continue;
        if (only !== null && q.s !== only) continue;
        const rec = bySubject.get(q.s) ?? {};
        if (q.p === `${SHIM}:id`) rec.id = q.o;
        if (q.p === `${SHIM}:docPublic`) rec.docPublic = q.o;
        if (q.p === `${SHIM}:docProtected`) rec.docProtected = q.o;
        if (q.p === `${SHIM}:docPrivate`) rec.docPrivate = q.o;
        bySubject.set(q.s, rec);
      }
      return {
        results: {
          bindings: [...bySubject.values()].filter((r) => r.id).map((r) => ({
            id: { value: r.id! },
            docPublic: { value: r.docPublic ?? "" },
            docProtected: { value: r.docProtected ?? "" },
            docPrivate: { value: r.docPrivate ?? "" },
          })),
        },
      };
    }
    if (query.includes(`<${INBOX}:payload>`)) {
      const bySubject = new Map<string, Record<string, string>>();
      for (const q of quads) {
        if (q.g !== anchor) continue;
        const rec = bySubject.get(q.s) ?? {};
        // The deposit's own subject — the key a drain names to retire it.
        rec.d = q.s;
        if (q.p === `${INBOX}:payload`) rec.payload = q.o;
        if (q.p === `${INBOX}:ts`) rec.ts = q.o;
        if (q.p === `${INBOX}:from`) rec.from = q.o;
        bySubject.set(q.s, rec);
      }
      return {
        results: {
          bindings: [...bySubject.values()]
            .filter((r) => r.payload !== undefined && r.ts !== undefined)
            .map((r) => {
              const row: Record<string, { value: string }> = {
                d: { value: r.d! },
                payload: { value: r.payload! },
                ts: { value: r.ts! },
              };
              if (r.from !== undefined) row.from = { value: r.from };
              return row;
            }),
        },
      };
    }
    if (query.includes(`<${SHIM}:inboxCap>`)) return byPred(`${SHIM}:inboxCap`, "c");
    if (query.includes(`<${SHIM}:inboxAddress>`)) return byPred(`${SHIM}:inboxAddress`, "a");
    if (query.includes(`<${SHIM}:readCap>`)) return byPred(`${SHIM}:readCap`, "c");
    if (query.includes(`<${SHIM}:link>`)) return byPred(`${SHIM}:link`, "c");
    if (query.includes(`${SHIM}:isInbox`)) return byPred(`${SHIM}:isInbox`, "i");
    if (query.includes(`<${SHIM}:inboxOwner>`)) {
      const sm = query.match(/<([^>]+)>\s+<urn:ng-eventually:shim:inboxOwner>/);
      const subj = sm ? sm[1]! : null;
      return {
        results: {
          bindings: quads
            .filter((q) => q.g === anchor && q.p === `${SHIM}:inboxOwner` && q.s === subj)
            .map((q) => ({ u: { value: q.o } })),
        },
      };
    }
    if (query.includes(`${SHIM}:docInbox`)) {
      const pm = query.match(/<(urn:ng-eventually:shim:docInbox:[a-z]+)>/);
      const pred = pm ? pm[1]! : "";
      const sm = query.match(/<([^>]+)>\s+<urn:ng-eventually:shim:docInbox/);
      const subj = sm ? sm[1]! : null;
      return {
        results: {
          bindings: quads
            .filter((q) => q.g === anchor && q.p === pred && (subj === null || q.s === subj))
            .map((q) => ({ d: { value: q.o } })),
        },
      };
    }
    if (query.includes(`<${SHIM}:exposedReadCap>`)) return byPred(`${SHIM}:exposedReadCap`, "c");
    if (query.includes(`<${SHIM}:contains>`)) return byPred(`${SHIM}:contains`, "e");
    return {
      results: {
        bindings: quads
          .filter((q) => q.g === anchor)
          .map((q) => ({ s: { value: q.s }, p: { value: q.p }, o: { value: q.o } })),
      },
    };
  });

  // `doc_subscribe` is declared and left undefined: absent, `ensureRepoOpen` is the
  // documented no-op of the unit-fake path. One test installs a subscription that never
  // pushes `State`, to reach the `timed-out` repo state a cold start really produces.
  const fakeNg: {
    doc_create: typeof doc_create;
    sparql_update: typeof sparql_update;
    sparql_query: typeof sparql_query;
    quads: Quad[];
    doc_subscribe?: (...a: unknown[]) => () => void;
  } = { doc_create, sparql_update, sparql_query, quads };
  return fakeNg;
}

let fake: ReturnType<typeof makeFakeNg>;

function inject() {
  fake = makeFakeNg();
  configure({ ng: fake as never, useShape: (() => {}) as never });
  configureStoreRegistry({ getSession: async () => SESSION });
  resetRegistryCache();
  resetOpenedRepos();
  resetPublicStoreFetches();
  resetCaps();
  setCurrentUser(null);
}

afterEach(() => {
  cancelScheduledInboxProcessing();
  stopObservingInboxes();
  resetConfig();
  resetStoreRegistry();
  resetCaps();
  resetRegistryCache();
  resetOpenedRepos();
  resetPublicStoreFetches();
  setCurrentUser(null);
});

/** Write one triple into `doc`, as the consumer's own write path would. */
async function write(doc: Nuri, p: string, o: string): Promise<void> {
  await sparqlUpdate(SESSION.sessionId, `INSERT DATA { <${doc}> <${p}> "${o}" }`, doc, "test");
}

/** Every value the index holds for `object` — one per entry written. */
async function entryValuesFor(index: Nuri, object: Nuri): Promise<string[]> {
  const subjects = await readUnion([index]);
  return subjects.filter((s) => s.subject === object).flatMap((s) => s.props[ENTRY_VALUE] ?? []);
}

/** What the package reported, for the span of `run`. Restores `console.error` whatever
 *  happens — a suite that leaks a stub takes the next file down with it. */
async function reportedDuring(run: () => Promise<void>): Promise<string[]> {
  const lines: string[] = [];
  const real = console.error;
  console.error = ((...args: unknown[]) => {
    lines.push(args.map((a) => String(a)).join(" "));
  }) as typeof console.error;
  try {
    await run();
  } finally {
    console.error = real;
  }
  return lines;
}

/**
 * Alice's index: a public document declaring the field it indexes by, with an inbox open
 * on it so that anyone can hand it a reference. Returns the index and one PUBLIC object
 * carrying a value for that field.
 */
async function aliceOpensAnIndex(fields: string[]): Promise<{ index: Nuri; object: Nuri }> {
  setCurrentUser("alice");
  const index = await createEntityDoc("alice", "public");
  for (const field of fields) await write(index, INDEX_FIELD, field);
  await openDocumentInbox(index);
  const object = await createEntityDoc("alice", "public");
  await write(object, WHEN, "2026-01-01");
  setCurrentUser(null);
  return { index, object };
}

/** Bob has been in the page once, so he is someone the wallet knows. */
async function bobSignsInOnce(): Promise<void> {
  setCurrentUser("bob");
  await resolveWriteGraph("bob", "protected");
  setCurrentUser(null);
}

/** Bob hands `payload` to the owner of `doc`, naming the document and nothing else. */
async function bobDepositsInto(doc: Nuri, payload: unknown, ts: number): Promise<void> {
  setCurrentUser("bob");
  await postToDocument(doc, { payload, ts });
  setCurrentUser(null);
}

/** Alice connects — which restores her registers and drains every inbox she holds. */
async function aliceConnects(): Promise<void> {
  setCurrentUser("alice");
  await connectedUser();
}

// --- the reference arm -------------------------------------------------------

test("an entry appears in the index from a bare reference deposited by a stranger", async () => {
  inject();
  await bobSignsInOnce();
  const { index, object } = await aliceOpensAnIndex([WHEN]);

  // The whole payload is the reference: no operation, no claim, and no copy of the value.
  // What lands in the index is what the OBJECT says, which is why a stranger depositing
  // achieves exactly what the owner would have.
  await bobDepositsInto(index, object, 1);
  expect(await entryValuesFor(index, object)).toEqual([]);

  await aliceConnects();

  expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
});

test("a reference nobody can resolve is reported, and costs the index nothing", async () => {
  inject();
  await bobSignsInOnce();
  const { index, object } = await aliceOpensAnIndex([WHEN]);

  // A reference this holder cannot even attempt — no cap, so `readUnion` drops it before
  // issuing a query — and a resolvable one BEHIND it, which is what makes the question
  // answerable at all.
  await bobDepositsInto(index, "did:ng:o:nowhere", 1);
  await bobDepositsInto(index, object, 2);

  const reported = await reportedDuring(aliceConnects);

  // NOT called an absence, and this is the correction that cost the most to find. "No cap"
  // reads as deterministic and is not: for a stranger's public document possession comes
  // from a network fetch that swallows its errors and memoises the failure, so one outage
  // presents exactly like a document that never existed. Classing it absent destroyed the
  // deposit and every re-deposit after it. It stops the drain instead.
  expect(reported.some((l) => /did:ng:o:nowhere/.test(l))).toBe(true);
  expect(reported.some((l) => /was consumed and could not be applied/.test(l))).toBe(true);
  expect(await entryValuesFor(index, "did:ng:o:nowhere")).toEqual([]);

  // The price is LATENCY, not loss: the unreadable one was popped, so the next connection
  // resumes past it and the resolvable one lands. Nothing wedges, and nothing is destroyed
  // that a working broker would have delivered.
  expect(await entryValuesFor(index, object)).toEqual([]);
  await aliceConnects();
  expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
});

test("an object already in the index is passed over — the same reference twice makes ONE entry", async () => {
  inject();
  await bobSignsInOnce();
  const { index, object } = await aliceOpensAnIndex([WHEN]);

  // TWO deposits, each consumed when it is applied — not one deposit replayed. The check
  // that a reference already in the index is passed over is what makes this land on ONE
  // entry: without it the second deposit writes a second value for the same object, and an
  // entry carrying two values is exactly what once made the index SHRINK through additions.
  // That is why the check outlived consumption: it never answered the replay.
  await bobDepositsInto(index, object, 1);
  await bobDepositsInto(index, object, 2);

  await aliceConnects();

  expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
});

test("an index declaring NO field refuses, loudly, rather than writing something wrong", async () => {
  inject();
  await bobSignsInOnce();
  // A document with an inbox and no declaration: either it is not an index, or it could
  // not be read — and here those are the same empty result.
  const { index, object } = await aliceOpensAnIndex([]);

  await bobDepositsInto(index, object, 1);
  const reported = await reportedDuring(aliceConnects);

  expect(reported.some((l) => /declares no index field/.test(l))).toBe(true);
  expect(await entryValuesFor(index, object)).toEqual([]);
});

test("an index declaring SEVERAL fields refuses rather than picking one", async () => {
  inject();
  await bobSignsInOnce();
  // Entries already written are never re-read, so curating under a second field would
  // leave one list ordered by two different properties — a quiet wrong answer.
  const { index, object } = await aliceOpensAnIndex([WHEN, "urn:test:other"]);

  await bobDepositsInto(index, object, 1);
  const reported = await reportedDuring(aliceConnects);

  expect(reported.some((l) => /declares 2 index fields/.test(l))).toBe(true);
  expect(await entryValuesFor(index, object)).toEqual([]);
});

test("a reference deposit is left for its owner's own session, not applied from a stranger's", async () => {
  inject();
  await bobSignsInOnce();
  const { index, object } = await aliceOpensAnIndex([WHEN]);

  await bobDepositsInto(index, object, 1);
  // The deferred stand-in for an absent owner runs under BOB's session. It files the
  // verifier's own registers for an absent owner and stops there: writing application data
  // into Alice's documents from Bob's page is a different act, and it is not taken.
  setCurrentUser("bob");
  const reported = await reportedDuring(async () => {
    await runScheduledInboxProcessingNow();
  });
  setCurrentUser(null);
  expect(await entryValuesFor(index, object)).toEqual([]);
  // Waiting is not failing, and it must not be reported as one: the arm did not try and
  // fall short, it is not this session's to run. Running it anyway would report a refusal
  // here every twenty seconds — Bob's User branch pairs no document with Alice's inbox.
  expect(reported.filter((l) => /could not apply a deposit/.test(l))).toEqual([]);

  // Nothing was consumed, so Alice's own next connection applies it — the real path.
  await aliceConnects();
  expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
});

// --- the cap arm -------------------------------------------------------------

test("a deposit declaring itself a Link and carrying no readable cap is reported, not passed over", async () => {
  inject();
  await bobSignsInOnce();
  const { index } = await aliceOpensAnIndex([WHEN]);

  // A bare reference where a ReadCap belongs — the one type confusion that would invert
  // the model if it were filed, and the shape a version skew produces. Upstream an arm
  // opens by validating its envelope and a malformed one is a hard failure, not a skip.
  await bobDepositsInto(index, { kind: "urn:ng-eventually:inbox:link", cap: "did:ng:o:bare" }, 1);

  const reported = await reportedDuring(aliceConnects);

  expect(reported.some((l) => /declares itself a Link and carries no readable cap/.test(l))).toBe(true);
});

// --- the arm that does not exist ---------------------------------------------

test("a kind no arm answers says so — the dispatch's `NotImplemented`", async () => {
  inject();
  await bobSignsInOnce();
  const { index } = await aliceOpensAnIndex([WHEN]);

  // A deposit declaring itself in THIS package's reserved namespace, for a variant nothing
  // here applies. Upstream that is `_ => Err(NotImplemented)`; here it used to be silence.
  await bobDepositsInto(index, { kind: "urn:ng-eventually:inbox:not-a-thing" }, 1);

  const reported = await reportedDuring(aliceConnects);

  expect(
    reported.some((l) => /no arm applies deposits of kind "urn:ng-eventually:inbox:not-a-thing"/.test(l)),
  ).toBe(true);
});

test("a consumer payload is not a variant: no arm claims it and nothing is reported", async () => {
  inject();
  await bobSignsInOnce();
  const { index } = await aliceOpensAnIndex([WHEN]);

  // `kind` is an ordinary word an application may use for its own messages. Keying the
  // dispatch on the mere presence of the field would have turned every one of them into an
  // unapplied message reported at each drain.
  await bobDepositsInto(index, { kind: "join", who: "bob" }, 1);

  const reported = await reportedDuring(aliceConnects);

  expect(reported.filter((l) => /could not apply a deposit/.test(l))).toEqual([]);
});

// --- a consumed deposit is never discarded in silence ------------------------

test("a reference to an object carrying nothing for the field is REPORTED, not silently dropped", async () => {
  inject();
  await bobSignsInOnce();
  const { index } = await aliceOpensAnIndex([WHEN]);

  // A draft: it resolves, and it says nothing for the field the index declares. While
  // deposits were re-read for ever this was a benign WAIT — the object might gain the field
  // and a later drain would index it — so it was traced behind `debugAccessLog` and nothing
  // more. Consumption ends the wait: this deposit is gone once it has been looked at, so the
  // discard has to be audible without any flag turned on.
  setCurrentUser("alice");
  const draft = await createEntityDoc("alice", "public");
  // It has to SAY something, or the read comes back empty and this is the `refused` path
  // instead — a different outcome with a different report. What is under test is the
  // object that resolves fine and simply has no value for the indexed field.
  await write(draft, "urn:test:title", "a draft with no date yet");
  setCurrentUser(null);
  await bobDepositsInto(index, draft, 1);

  const reported = await reportedDuring(aliceConnects);

  expect(reported.some((l) => /was discarded/.test(l) && new RegExp(`carries no ${WHEN}`).test(l)))
    .toBe(true);
  expect(await entryValuesFor(index, draft)).toEqual([]);
});

// --- a broker that goes quiet must not walk the whole queue -----------------

test("a broker that cannot read the referenced objects costs ONE deposit, not the queue", async () => {
  inject();
  await bobSignsInOnce();
  const { index } = await aliceOpensAnIndex([WHEN]);

  // Three references waiting in the index's inbox.
  setCurrentUser("alice");
  const objects: Nuri[] = [];
  for (let i = 0; i < 3; i++) {
    const o = await createEntityDoc("alice", "public");
    await write(o, WHEN, `2026-01-0${i + 1}`);
    objects.push(o);
  }
  setCurrentUser(null);
  for (let i = 0; i < 3; i++) await bobDepositsInto(index, objects[i]!, i + 1);

  const queued = () =>
    fake.quads.filter((q) => q.p === `${INBOX}:payload` && q.o.includes("did:ng:o:")).length;
  expect(queued()).toBe(3);

  // The broker goes quiet on the referenced objects. `read-model.readDoc` SWALLOWS a failed
  // read and answers `[]` — so the arm is TOLD, through the `onUnreadable` reporter, rather
  // than left to guess from an empty answer. Until 2026-08-26 this was classed as an
  // ordinary refusal: the drain reported it and carried on, popping and destroying all three
  // while printing "will not be retried" for each.
  const realQuery = fake.sparql_query;
  fake.sparql_query = (async (...a: unknown[]) => {
    if (objects.includes(a[3] as Nuri)) throw new Error("the broker went quiet");
    return realQuery(...a);
  }) as typeof fake.sparql_query;

  const reported = await reportedDuring(aliceConnects);

  expect(reported.some((l) => /was consumed and could not be applied/.test(l))).toBe(true);
  expect(queued()).toBe(2);
});

test("an EMPTY but readable object is absent — reported, and the queue behind it carries on", async () => {
  inject();
  await bobSignsInOnce();
  const { index, object } = await aliceOpensAnIndex([WHEN]);

  // A document Alice really holds and that really reads — it just has no triples. This is
  // the genuine clean-empty read, and until 2026-08-26 NOTHING exercised it: the only test
  // of this branch pointed at `did:ng:o:nowhere`, which `readUnion`'s possession gate drops
  // before any query is issued. So the branch that decides "absent vs unreadable" on a real
  // read was uncovered, which is how two opposite misclassifications shipped through it.
  setCurrentUser("alice");
  const empty = await createEntityDoc("alice", "public");
  setCurrentUser(null);

  // The empty one FIRST, so "does it stall what is behind it" is a question with an answer.
  await bobDepositsInto(index, empty, 1);
  await bobDepositsInto(index, object, 2);

  const reported = await reportedDuring(aliceConnects);

  expect(reported.some((l) => /read cleanly from a synced repo and found nothing/.test(l)))
    .toBe(true);
  expect(await entryValuesFor(index, empty)).toEqual([]);
  expect(await entryValuesFor(index, object)).toEqual(["2026-01-01"]);
});

test("a rejection valued `null` is still a failure, not an absence", async () => {
  inject();
  await bobSignsInOnce();
  const { index, object } = await aliceOpensAnIndex([WHEN]);
  await bobDepositsInto(index, object, 1);

  // `throw null` is legal, and the first version of this classification tracked the error
  // VALUE against a `null` sentinel — so a broker that rejected with `null` was recorded as
  // "no failure happened" and its deposit was destroyed as an absence. Tracked by a boolean
  // now, which is the only thing that cannot collide with a payload.
  const realQuery = fake.sparql_query;
  fake.sparql_query = (async (...a: unknown[]) => {
    if (a[3] === object) throw null;
    return realQuery(...a);
  }) as typeof fake.sparql_query;

  const reported = await reportedDuring(aliceConnects);

  expect(reported.some((l) => /was consumed and could not be applied/.test(l))).toBe(true);
  expect(reported.some((l) => /came back clean and empty/.test(l))).toBe(false);
  expect(await entryValuesFor(index, object)).toEqual([]);
});

test("an empty read whose repo is NOT SYNCED is not an absence — it stops the drain", async () => {
  inject();
  await bobSignsInOnce();
  const { index, object } = await aliceOpensAnIndex([WHEN]);

  setCurrentUser("alice");
  const unsynced = await createEntityDoc("alice", "public");
  setCurrentUser(null);
  await bobDepositsInto(index, unsynced, 1);
  await bobDepositsInto(index, object, 2);

  // A broker that ACCEPTS the subscription and never pushes `State`. The bounded fallback
  // fires, the repo is `timed-out` rather than `synced`, and the anchored read then answers
  // 0 rows WITH NO ERROR — which is the shape a cold start takes, not an exotic one:
  // `Verifier::load` repopulates `self.repos` from storage, so the repo is present and its
  // commits simply have not arrived. Reading that as "the object is absent" consumed and
  // destroyed the whole queue while reporting each deposit as absent. `open-repo` keeps
  // `synced` and `timed-out` apart precisely so this arm can tell, and now it asks.
  setOpenTimeoutForTests(20);
  fake.doc_subscribe = () => () => {};

  const reported = await reportedDuring(aliceConnects);

  expect(reported.some((l) => /is not synced in this session/.test(l))).toBe(true);
  expect(reported.some((l) => /came back clean and empty/.test(l))).toBe(false);
  // Stopped rather than carried on: the deposit behind it is untouched.
  expect(await entryValuesFor(index, object)).toEqual([]);
});