deploy: c2c7937

space/src/consumer-coverage-update.ts

@@ -1,9 +1,14 @@
 import type Database from "better-sqlite3";
 import { canonicalJson } from "@xtap-pool/shared";
 import type { ResolvedConsumerContext } from "./consumer-context.js";
+import { ConsumerBootstrapRequired, ConsumerIndexState } from "./consumer-index-state.js";
 import { historicalSelection } from "./consumer-changes.js";
 import { selectedCompleteThrough } from "./enrich-store.js";
-import { selectedObservationThrough } from "./consumer-coverage.js";
+import {
+  selectedCurrentCoverageUnitIds,
+  selectedCurrentObservationThrough,
+  selectedObservationThrough,
+} from "./consumer-coverage.js";
 import { ConsumerCoverageEffects } from "./consumer-coverage-effects.js";
 
 export type ConsumerCoverage = {
@@ -63,6 +68,27 @@ function metricCoverage(
 function recalculate(
   database: Database.Database,
   target: ResolvedConsumerContext,
+): ConsumerCoverage {
+  try {
+    new ConsumerIndexState(database, target.context.contract, "read").require(
+      target.context.source,
+    );
+  } catch (error) {
+    if (error instanceof ConsumerBootstrapRequired) return historicalCoverage(database, target);
+    throw error;
+  }
+  const unitIds = selectedCurrentCoverageUnitIds(database, target);
+  return {
+    kind: "coverage",
+    mode: "semantic",
+    completeThrough: selectedCompleteThrough(database, { unitIds }) ?? null,
+    observationsThrough: selectedCurrentObservationThrough(database, target, unitIds),
+  };
+}
+
+function historicalCoverage(
+  database: Database.Database,
+  target: ResolvedConsumerContext,
 ): ConsumerCoverage {
   return {
     kind: "coverage",

space/src/consumer-coverage.ts

@@ -4,6 +4,36 @@ import { eligibleUnits } from "./enrich-store.js";
 import { historicalSelection } from "./consumer-changes.js";
 import type { ResolvedConsumerContext } from "./consumer-context.js";
 
+const unitRow = z.object({ unit_id: z.string() });
+
+/** Historical membership is an indexed superset of current selected units.
+ * Start there, then apply the exact current author and publication rules once
+ * for both coverage clocks. */
+export function selectedCurrentCoverageUnitIds(
+  database: Database.Database,
+  target: ResolvedConsumerContext,
+): string[] {
+  const context = target.context;
+  const authors = JSON.stringify(context.selection.author_ids);
+  return database
+    .prepare(
+      `WITH candidates AS MATERIALIZED (
+      SELECT DISTINCT unit_id FROM consumer_post_units
+      WHERE author_id IN (SELECT value FROM json_each(@authors))
+    ) SELECT c.unit_id FROM candidates c WHERE
+      NOT EXISTS (SELECT 1 FROM unit_members m JOIN tweets t ON t.id = m.tweet_id
+        WHERE m.unit_id = c.unit_id AND (
+          json_extract(t.json, '$.author.id') IS NULL
+          OR json_extract(t.json, '$.author.id') NOT IN (SELECT value FROM json_each(@authors))
+          OR json_extract(t.json, '$.is_subscriber_only') = 1))
+      AND EXISTS (SELECT 1 FROM unit_members m JOIN tweets t ON t.id = m.tweet_id
+        WHERE m.unit_id = c.unit_id AND COALESCE(json_extract(t.json, '$.is_retweet'), 0) != 1)
+      ORDER BY c.unit_id`,
+    )
+    .all({ authors })
+    .map((row) => unitRow.parse(row).unit_id);
+}
+
 /** Current selected, completed public content only. Counters have their own clock;
  * pending, unselected, subscriber-only and retweet samples cannot advance it. */
 export function selectedObservationThrough(
@@ -11,8 +41,6 @@ export function selectedObservationThrough(
   target: ResolvedConsumerContext,
   postIds?: readonly string[],
 ): string | null {
-  const context = target.context;
-  const selection = historicalSelection(target);
   const unitIds =
     postIds === undefined
       ? undefined
@@ -21,13 +49,40 @@ export function selectedObservationThrough(
             "SELECT DISTINCT unit_id FROM unit_members WHERE tweet_id IN (SELECT value FROM json_each(?))",
           )
           .all(JSON.stringify(postIds))
-          .map((row) => z.object({ unit_id: z.string() }).parse(row).unit_id);
+          .map((row) => unitRow.parse(row).unit_id);
+  return observationThrough(database, target, unitIds, postIds, false);
+}
+
+/** The caller proved that the database is exactly the target source and applied
+ * current author and publication rules to unitIds. Do not repeat either scan. */
+export function selectedCurrentObservationThrough(
+  database: Database.Database,
+  target: ResolvedConsumerContext,
+  unitIds: readonly string[],
+): string | null {
+  return observationThrough(database, target, unitIds, undefined, true);
+}
+
+function observationThrough(
+  database: Database.Database,
+  target: ResolvedConsumerContext,
+  unitIds: readonly string[] | undefined,
+  postIds: readonly string[] | undefined,
+  currentSource: boolean,
+): string | null {
+  const context = target.context;
+  const selection = historicalSelection(target);
   const eligible = eligibleUnits({
-    ...selection,
-    ...(unitIds === undefined ? {} : { unitIds }),
+    ...(currentSource ? { labels: selection.labels, labelMode: selection.labelMode } : selection),
+    unitIds,
     taxonomyVersion: context.taxonomy.version,
     contractHash: context.contract,
   });
+  const sourceSql = currentSource
+    ? ""
+    : `AND EXISTS (SELECT 1 FROM observation_sources s INDEXED BY idx_observation_source_id
+          WHERE s.observation_id = o.observation_id
+            AND s.segment_key IN (SELECT json_extract(value, '$.key') FROM json_each(?)))`;
   // Read each selected post once and stop at its newest permitted observation.
   // Scanning every historical counter sample repeats source and visibility checks.
   const row = database
@@ -51,8 +106,7 @@ export function selectedObservationThrough(
         AND (json_type(c.payload_json, '$.is_subscriber_only') IS NULL OR json_type(c.payload_json, '$.is_subscriber_only') = 'false')
         AND (json_type(c.payload_json, '$.is_retweet') IS NULL OR json_type(c.payload_json, '$.is_retweet') = 'false')
         AND json_extract(c.payload_json, '$.author.id') IN (SELECT value FROM json_each(?))
-        AND EXISTS (SELECT 1 FROM observation_sources s INDEXED BY idx_observation_source_id WHERE s.observation_id = o.observation_id
-          AND s.segment_key IN (SELECT json_extract(value, '$.key') FROM json_each(?)))
+        ${sourceSql}
       ORDER BY o.observed_at DESC, o.observation_id DESC LIMIT 1
     )) AS latest FROM selected_posts p
     `,
@@ -65,7 +119,7 @@ export function selectedObservationThrough(
       postIds === undefined ? null : JSON.stringify(postIds),
       postIds === undefined ? null : JSON.stringify(postIds),
       JSON.stringify(selection.authorIds),
-      JSON.stringify(target.snapshot.files),
+      ...(currentSource ? [] : [JSON.stringify(target.snapshot.files)]),
     );
   return z.object({ latest: z.string().nullable() }).parse(row).latest;
 }

space/src/enrich-store.ts

@@ -1536,6 +1536,7 @@ function nullable<T>(value: T | undefined): T | null {
 }
 
 type UnitSelection = {
+  unitIds?: readonly string[] | undefined;
   authorIds?: readonly string[] | undefined;
   labels?: readonly string[] | undefined;
   labelMode?: "any" | "all" | undefined;
@@ -1544,7 +1545,6 @@ type UnitSelection = {
 };
 
 type EligibleUnitOptions = UnitSelection & {
-  unitIds?: readonly string[];
   taxonomyVersion: number;
   contractHash: string;
 };
@@ -1626,10 +1626,16 @@ function selectedUnits(options: UnitSelection): { sql: string; params: unknown[]
   const cutoffParams = options.cutoff === undefined ? [] : [options.cutoff];
   const authorParams = options.authorIds ?? [];
   return {
-    sql: `WITH unit_ids AS MATERIALIZED (SELECT DISTINCT unit_id FROM unit_members)
+    sql: `WITH unit_ids AS MATERIALIZED (SELECT DISTINCT unit_id FROM unit_members${
+      options.unitIds === undefined ? "" : " WHERE unit_id IN (SELECT value FROM json_each(?))"
+    })
           SELECT um.unit_id FROM unit_ids um
           WHERE 1 = 1${cutoffSql}${authorSql}${publicationSql}`,
-    params: [...cutoffParams, ...authorParams],
+    params: [
+      ...(options.unitIds === undefined ? [] : [JSON.stringify(options.unitIds)]),
+      ...cutoffParams,
+      ...authorParams,
+    ],
   };
 }
 

space/tests/consumer-coverage-update.test.ts

@@ -156,6 +156,36 @@ describe("source coverage maintenance", () => {
     expect(plan.some((line) => line.includes("segment_key=? AND observation_id=?"))).toBe(false);
   });
 
+  it("uses the indexed current selection for both semantic coverage clocks", async () => {
+    await f.postMany([
+      consumerTweet("100"),
+      consumerTweet("200", {
+        conversation_id: "100",
+        captured_at: "2026-09-06T05:00:00.000Z",
+        is_subscriber_only: "yes",
+      }),
+      consumerTweet("300", {
+        conversation_id: "100",
+        captured_at: "2026-09-06T04:00:00.000Z",
+      }),
+      consumerTweet("400", { captured_at: "2026-09-06T01:00:00.000Z" }),
+      consumerTweet("500", { author: { id: "22", username: "other" } }),
+    ]);
+    const source = await finish(BOOTSTRAP);
+    expect(source.observations_through).toBe("2026-09-06T01:00:00.000Z");
+    const pinned = await context(source.cursor);
+    const database = f.index.store.database;
+    const plan = consumerQueryPlan(
+      database,
+      /WITH candidates AS MATERIALIZED|AS latest FROM selected_posts/u,
+    );
+    const coverage = updateConsumerCoverage(database, pinned);
+    expect(coverage.completeThrough).toBe(source.complete_through);
+    expect(coverage.observationsThrough).toBe(source.observations_through);
+    expect(plan.some((line) => line.includes("idx_consumer_post_units_author"))).toBe(true);
+    expect(plan.some((line) => line.includes("observation_sources"))).toBe(false);
+  });
+
   it("does zero body reads for twenty no-op polls and twenty harmless revisions, including restart and replay", async () => {
     await f.postMany(Array.from({ length: 250 }, (_, n) => consumerTweet(String(100 + n))));
     let source = await finish(BOOTSTRAP);