workglow-dev · sroussey · Sep 8, 2026 · Sep 8, 2026
diff --git a/src/task/kb/IndexFilingSectionsTask.test.ts b/src/task/kb/IndexFilingSectionsTask.test.ts
@@ -5,15 +5,17 @@
  */
 
 import { afterEach, describe, expect, it, vi } from "vitest";
-import { globalServiceRegistry } from "workglow";
+import type { DocumentNode } from "workglow";
+import { Document, globalServiceRegistry, NodeKind } from "workglow";
 import { withSqliteDb } from "../../config/testing/withSqliteDb";
-import { resetSecKnowledgeBaseForTesting } from "../../kb/secKnowledgeBase";
+import { getSecKnowledgeBase, resetSecKnowledgeBaseForTesting } from "../../kb/secKnowledgeBase";
 import {
   FILING_DOCUMENT_REPOSITORY_TOKEN,
   type FilingDocument,
 } from "../../storage/document/FilingDocumentSchema";
 import { FILING_SECTION_REPOSITORY_TOKEN } from "../../storage/document/FilingSectionSchema";
 import { IndexFilingSectionsTask } from "./IndexFilingSectionsTask";
+import { kbDocIdFor } from "./selectDocumentsToIndex";
 
 const header = (index: number, filingDate: string): FilingDocument => ({
   cik: 320193,
@@ -54,30 +56,55 @@ describe("IndexFilingSectionsTask selection", () => {
     }
   };
 
-  it("streams the converted filings rather than loading every header first", async () => {
+  /** Mark a filing as already in the knowledge base. */
+  const markIndexed = async (index: number): Promise<void> => {
+    const kb = await getSecKnowledgeBase();
+    const title = `Filing ${index}`;
+    const root = { kind: NodeKind.DOCUMENT, title, children: [] } as unknown as DocumentNode;
+    const docId = kbDocIdFor(`0000320193-26-${String(index).padStart(6, "0")}`, "primary.htm");
+    await kb.upsertDocument(new Document(root, { title } as never, [], docId));
+  };
+
+  it("picks the work in the database rather than reading headers to sift them", async () => {
     await seed(5);
     const repo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN);
-    const getAll = vi.spyOn(repo, "getAll");
+    const reads = (["getAll", "records", "query", "queryPage"] as const).map((method) =>
+      vi.spyOn(repo, method)
+    );
 
-    const out = await new IndexFilingSectionsTask().run({ limit: 0 });
+    const out = await new IndexFilingSectionsTask().run({ cik: 320193, limit: 0 });
 
-    // `getAll()` on the unscoped path is the whole converted corpus in memory
-    // — hundreds of thousands of rows to take the first few of.
-    expect(getAll).not.toHaveBeenCalled();
-    expect(out).toMatchObject({ indexed: 0, truncated: true });
+    // Reading headers to decide is the whole converted corpus crossing the
+    // process boundary to take the first few of it.
+    for (const read of reads) expect(read).not.toHaveBeenCalled();
+    expect(out.truncated).toBe(true);
   });
 
-  it("streams a scoped selection by page too", async () => {
+  it("still reports what is already indexed once nothing is left to do", async () => {
+    // The count `sec index` prints, and what makes it suggest asking a question
+    // rather than converting filings there are none of.
     await seed(3);
-    const repo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN);
-    const query = vi.spyOn(repo, "query");
-    const queryPage = vi.spyOn(repo, "queryPage");
+    for (const index of [0, 1, 2]) await markIndexed(index);
 
-    const out = await new IndexFilingSectionsTask().run({ cik: 320193, limit: 0 });
+    const out = await new IndexFilingSectionsTask().run({ limit: 5 });
 
-    expect(query).not.toHaveBeenCalled();
-    expect(queryPage).toHaveBeenCalled();
-    expect(out.truncated).toBe(true);
+    expect(out).toMatchObject({ indexed: 0, sections: 0, skipped: 3, truncated: false });
+  });
+
+  it("spends its limit on filings that need work, not on ones already indexed", async () => {
+    // The defect: `limit` counted filings INDEXED, and an already-indexed one
+    // was skipped without counting toward it — so a small limit on an indexed
+    // corpus read every row and probed the knowledge base once per row.
+    await seed(4);
+    await markIndexed(0);
+    await markIndexed(1);
+    const kb = await getSecKnowledgeBase();
+    const getDocument = vi.spyOn(kb, "getDocument");
+
+    const out = await new IndexFilingSectionsTask().run({ limit: 2 });
+
+    expect(getDocument).not.toHaveBeenCalled();
+    expect(out).toMatchObject({ skipped: 2, truncated: false });
   });
 
   it("reports no truncation when the scope leaves nothing to index", async () => {

diff --git a/src/task/kb/IndexFilingSectionsTask.ts b/src/task/kb/IndexFilingSectionsTask.ts
@@ -5,7 +5,7 @@
  */
 
 import { Type } from "typebox";
-import type { DocumentNode, IExecuteContext, PageCursor } from "workglow";
+import type { DocumentNode, IExecuteContext } from "workglow";
 import {
   Document,
   globalServiceRegistry,
@@ -15,17 +15,14 @@ import {
   TaskAbortedError,
 } from "workglow";
 import { getSecKnowledgeBase } from "../../kb/secKnowledgeBase";
-import {
-  FILING_DOCUMENT_REPOSITORY_TOKEN,
-  type FilingDocument,
-  type FilingDocumentRepositoryStorage,
-} from "../../storage/document/FilingDocumentSchema";
+import type { FilingDocument } from "../../storage/document/FilingDocumentSchema";
 import {
   FILING_SECTION_REPOSITORY_TOKEN,
   type FilingSection,
 } from "../../storage/document/FilingSectionSchema";
 import { accessionWithoutDashes } from "../../util/accession";
 import type { TaskPorts } from "../taskPorts";
+import { countAlreadyIndexed, kbDocIdFor, selectDocumentsToIndex } from "./selectDocumentsToIndex";
 
 export interface IndexFilingSectionsTaskInput {
   /** Only this issuer's filings. */
@@ -63,9 +60,6 @@ export interface IndexFilingSectionsTaskOutput {
  */
 export const DEFAULT_ASK_INDEX_LIMIT = 25;
 
-/** Rows per read while walking the converted filings. */
-const HEADER_PAGE_SIZE = 500;
-
 /** The EDGAR URL a citation points at. */
 function filingUrl(cik: number, accession: string, docFile: string): string {
   return (
@@ -117,32 +111,6 @@ function toDocument(header: FilingDocument, sections: readonly FilingSection[]):
   } as never);
 }
 
-/**
- * The converted filings matching `criteria`, a page at a time.
- *
- * Streamed rather than collected: the unscoped case is every converted filing
- * in the database, and materializing that array to take the first few of it
- * costs the whole corpus in memory before any work starts.
- */
-async function* streamHeaders(
-  repo: FilingDocumentRepositoryStorage,
-  criteria: Record<string, unknown>
-): AsyncGenerator<FilingDocument> {
-  if (Object.keys(criteria).length === 0) {
-    yield* repo.records(HEADER_PAGE_SIZE);
-    return;
-  }
-  let cursor: PageCursor | undefined;
-  for (;;) {
-    const page = await repo.queryPage(criteria as never, { limit: HEADER_PAGE_SIZE, cursor });
-    yield* page.items;
-    // Both conditions: a cursor can be handed back for a page that concurrent
-    // deletes have since emptied, and looping on it alone would not terminate.
-    if (page.nextCursor === undefined || page.items.length === 0) return;
-    cursor = page.nextCursor;
-  }
-}
-
 /**
  * Embeds converted filing sections into the knowledge base `sec ask` reads.
  *
@@ -187,43 +155,33 @@ export class IndexFilingSectionsTask extends Task<
     context: IExecuteContext
   ): Promise<TaskPorts<IndexFilingSectionsTaskOutput>> {
     const kb = await getSecKnowledgeBase();
-    const documentRepo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN);
     const sectionRepo = globalServiceRegistry.get(FILING_SECTION_REPOSITORY_TOKEN);
 
-    const criteria: Record<string, unknown> = {};
-    if (input.cik !== undefined) criteria.cik = input.cik;
-    if (input.form !== undefined) criteria.form = input.form;
-    if (input.accession !== undefined) criteria.accession_number = input.accession;
+    const scope = {
+      cik: input.cik,
+      form: input.form,
+      since: input.since,
+      accession: input.accession,
+    };
 
-    // Only a denominator for the progress line, so an over-count from the
-    // `since` filter (which no backend expresses as a criterion) is harmless.
-    const candidates = await documentRepo.count(
-      Object.keys(criteria).length === 0 ? undefined : (criteria as never)
-    );
+    // One document more than asked for, so "there are more" is something this
+    // run observed rather than inferred from having filled the limit exactly.
     const limit = input.limit;
-    const denominator = Math.max(1, limit === undefined ? candidates : Math.min(limit, candidates));
+    const candidates = await selectDocumentsToIndex({
+      ...scope,
+      limit: limit === undefined ? undefined : limit + 1,
+      force: input.force,
+    });
+    const truncated = limit !== undefined && candidates.length > limit;
+    const work = truncated ? candidates.slice(0, limit) : candidates;
+    const skipped = input.force === true ? 0 : await countAlreadyIndexed(scope);
+    const denominator = Math.max(1, work.length);
 
     let indexed = 0;
     let sectionTotal = 0;
-    let skipped = 0;
-    let truncated = false;
-    for await (const header of streamHeaders(documentRepo, criteria)) {
+    for (const header of work) {
       if (context.signal?.aborted) throw new TaskAbortedError();
-      // Scope first, then the limit, so a run that stops has genuinely left
-      // candidates behind rather than rows the scope excludes.
-      if (input.since !== undefined && (header.filing_date ?? "") < input.since) continue;
-      if (limit !== undefined && indexed >= limit) {
-        truncated = true;
-        break;
-      }
-
-      // The document id is derived from the filing, so "already indexed" is a
-      // lookup rather than a second table to keep in step with this one.
-      const docId = `${header.accession_number}:${header.doc_file}`;
-      if (input.force !== true && (await kb.getDocument(docId)) !== undefined) {
-        skipped += 1;
-        continue;
-      }
+      const docId = kbDocIdFor(header.accession_number, header.doc_file);
 
       const sections = (
         (await sectionRepo.query({

diff --git a/src/task/kb/selectDocumentsToIndex.sqlite.test.ts b/src/task/kb/selectDocumentsToIndex.sqlite.test.ts
@@ -0,0 +1,151 @@
+/**
+ * @license
+ * Copyright 2026 Steven Roussey <sroussey@gmail.com>
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import type { DocumentNode } from "workglow";
+import { Document, globalServiceRegistry, NodeKind } from "workglow";
+import { withSqliteDb } from "../../config/testing/withSqliteDb";
+import { getSecKnowledgeBase, resetSecKnowledgeBaseForTesting } from "../../kb/secKnowledgeBase";
+import {
+  FILING_DOCUMENT_REPOSITORY_TOKEN,
+  type FilingDocument,
+} from "../../storage/document/FilingDocumentSchema";
+import { countAlreadyIndexed, kbDocIdFor, selectDocumentsToIndex } from "./selectDocumentsToIndex";
+
+const doc = (index: number, over: Partial<FilingDocument> = {}): FilingDocument => ({
+  cik: 320193,
+  accession_number: `0000320193-26-${String(index).padStart(6, "0")}`,
+  doc_file: "primary.htm",
+  doc_type: "10-K",
+  description: null,
+  sequence: 1,
+  is_primary: true,
+  form: "10-K",
+  filing_date: "2026-03-01",
+  title: `Filing ${index}`,
+  section_count: 1,
+  char_count: 100,
+  converter_version: "1",
+  converted_at: "2026-01-01T00:00:00.000Z",
+  ...over,
+});
+
+/**
+ * `limit` used to bound the number of documents INDEXED, and an already-indexed
+ * one was skipped without counting toward it. So a small `--limit` on a corpus
+ * that is already indexed read every row and probed the knowledge base once per
+ * row before concluding there was nothing to do — and `sec ask` builds the
+ * index implicitly, so that ran on every question.
+ *
+ * These cases are about which rows come back and how many, not about embedding.
+ */
+describe("selectDocumentsToIndex (sqlite)", () => {
+  withSqliteDb("select_documents_to_index", [FILING_DOCUMENT_REPOSITORY_TOKEN]);
+
+  beforeEach(async () => {
+    await resetSecKnowledgeBaseForTesting();
+    delete process.env.SEC_EMBEDDING_MODEL;
+  });
+
+  afterEach(async () => {
+    await resetSecKnowledgeBaseForTesting();
+  });
+
+  async function seed(count: number, over: (i: number) => Partial<FilingDocument> = () => ({})) {
+    const repo = globalServiceRegistry.get(FILING_DOCUMENT_REPOSITORY_TOKEN);
+    for (let i = 1; i <= count; i += 1) await repo.put(doc(i, over(i)) as never);
+  }
+
+  /** Mark a document as already in the knowledge base. */
+  async function markIndexed(index: number) {
+    const kb = await getSecKnowledgeBase();
+    const title = `Filing ${index}`;
+    const root = { kind: NodeKind.DOCUMENT, title, children: [] } as unknown as DocumentNode;
+    const docId = kbDocIdFor(`0000320193-26-${String(index).padStart(6, "0")}`, "primary.htm");
+    await kb.upsertDocument(new Document(root, { title } as never, [], docId));
+  }
+
+  it("returns nothing once every document is indexed", async () => {
+    await seed(3);
+    for (const i of [1, 2, 3]) await markIndexed(i);
+
+    expect(await selectDocumentsToIndex({ limit: 5 })).toEqual([]);
+  });
+
+  it("counts the limit in documents that need work, not documents examined", async () => {
+    // The defect: two of five are already indexed, so a limit of 2 must return
+    // the two that are NOT — never fewer because the indexed ones consumed it.
+    await seed(5);
+    await markIndexed(1);
+    await markIndexed(2);
+
+    const picked = await selectDocumentsToIndex({ limit: 2 });
+
+    expect(picked).toHaveLength(2);
+    expect(picked.map((d) => d.accession_number)).not.toContain("0000320193-26-000001");
+    expect(picked.map((d) => d.accession_number)).not.toContain("0000320193-26-000002");
+  });
+
+  it("re-selects everything under force", async () => {
+    await seed(3);
+    for (const i of [1, 2, 3]) await markIndexed(i);
+
+    expect(await selectDocumentsToIndex({ limit: 5, force: true })).toHaveLength(3);
+  });
+
+  it("applies `since` in the query rather than after it", async () => {
+    await seed(4, (i) => ({ filing_date: i <= 2 ? "2020-01-01" : "2026-06-01" }));
+
+    const picked = await selectDocumentsToIndex({ limit: 10, since: "2026-01-01" });
+
+    expect(picked).toHaveLength(2);
+    expect(picked.every((d) => (d.filing_date ?? "") >= "2026-01-01")).toBe(true);
+  });
+
+  it("narrows by cik, form and accession", async () => {
+    await seed(3, (i) => ({ form: i === 2 ? "8-K" : "10-K" }));
+
+    expect(await selectDocumentsToIndex({ limit: 10, form: "8-K" })).toHaveLength(1);
+    expect(await selectDocumentsToIndex({ limit: 10, cik: 999 })).toHaveLength(0);
+    expect(
+      await selectDocumentsToIndex({ limit: 10, accession: "0000320193-26-000003" })
+    ).toHaveLength(1);
+  });
+
+  it("returns newest first, so an interrupted backfill covers what people read", async () => {
+    await seed(3, (i) => ({ filing_date: `2026-0${i}-01` }));
+
+    const picked = await selectDocumentsToIndex({ limit: 3 });
+
+    expect(picked.map((d) => d.filing_date)).toEqual(["2026-03-01", "2026-02-01", "2026-01-01"]);
+  });
+
+  it("leaves out documents with no sections, which can never be indexed", async () => {
+    // They embed nothing, so they never enter the knowledge base — selecting
+    // them spends the limit on work that cannot happen, on every run.
+    await seed(3, (i) => ({ section_count: i === 2 ? 0 : 1 }));
+
+    const picked = await selectDocumentsToIndex({ limit: 10 });
+
+    expect(picked.map((d) => d.accession_number)).not.toContain("0000320193-26-000002");
+    expect(picked).toHaveLength(2);
+  });
+
+  it("counts nothing as already indexed before the index exists", async () => {
+    // The knowledge-base tables are built by the first command that opens the
+    // index, and this runs before that — a database with converted filings and
+    // no index has no table to join against.
+    await seed(2);
+
+    expect(await selectDocumentsToIndex({ limit: 10 })).toHaveLength(2);
+    expect(await countAlreadyIndexed({})).toBe(0);
+  });
+
+  it("returns nothing for a non-positive limit rather than everything", async () => {
+    await seed(2);
+    expect(await selectDocumentsToIndex({ limit: 0 })).toEqual([]);
+  });
+});