fix(adv): bound the ingest folder and keep the newest filing per CRD

src/task/adv/IngestAdvSnapshotTask.test.ts

@@ -6,7 +6,7 @@
 
 import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
-import { join } from "node:path";
+import { join, relative } from "node:path";
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
 import { globalServiceRegistry } from "workglow";
 import { SEC_RAW_DATA_FOLDER } from "../../config/tokens";
@@ -144,6 +144,94 @@ describe("IngestAdvSnapshotTask", () => {
     expect((await rows.query({ snapshot: "2026-06" })) ?? []).toHaveLength(3);
   });
 
+  it("keeps the newest base filing per CRD, not whichever the file listed last", async () => {
+    // One adviser, two of its annual amendments, newest FIRST — so a pass that
+    // takes the last row it sees keeps the 2013 description of a firm that has
+    // filed every year since. The cumulative archive stamps thirteen years of
+    // filings with one snapshot label, so every one of them collides here.
+    extract("2011-2024", {
+      "IA_ADV_Base_A.csv": [
+        "FilingID,DateSubmitted,1A,1B1,1D,1E1,1F1-City,1F1-State,1F1-Country,5F2c",
+        '900002,6/30/2024,"Acme Capital Management, LP",Acme Capital,801-12345,110001,Boston,MA,United States,"3,000,000,000"',
+        '900001,4/1/2013,Acme Capital LLC,Acme,801-12345,110001,Providence,RI,United States,"90,000,000"',
+      ].join("\n"),
+    });
+
+    const out = await new IngestAdvSnapshotTask().run({ snapshot: "2011-2024" });
+
+    const advisers = globalServiceRegistry.get(ADV_ADVISER_REPOSITORY_TOKEN);
+    const landed = (await advisers.query({ snapshot: "2011-2024" })) ?? [];
+    expect(landed).toHaveLength(1);
+    // The reported count is rows that landed, not rows pushed at the key.
+    expect(out.advisers).toBe(1);
+
+    const acme = landed[0]!;
+    expect(acme.date_submitted).toBe("2024-06-30");
+    expect(acme.filing_id).toBe("900002");
+    expect(acme.regulatory_aum).toBe(3_000_000_000);
+    expect(acme.main_office_state).toBe("MA");
+  });
+
+  it("orders undated filings behind dated ones rather than by position", async () => {
+    extract("2011-2024", {
+      "IA_ADV_Base_A.csv": [
+        "FilingID,DateSubmitted,1A,1B1,1D,1E1,1F1-City,1F1-State,1F1-Country,5F2c",
+        '900010,5/1/2020,Gamma Advisers LLC,Gamma,801-22222,110003,Denver,CO,United States,"5,000,000"',
+        "900011,,Gamma Advisers LLC,Gamma,801-22222,110003,Nowhere,ZZ,United States,",
+      ].join("\n"),
+    });
+
+    await new IngestAdvSnapshotTask().run({ snapshot: "2011-2024" });
+
+    const advisers = globalServiceRegistry.get(ADV_ADVISER_REPOSITORY_TOKEN);
+    const gamma = await advisers.get({ snapshot: "2011-2024", crd_number: "110003" });
+    expect(gamma?.date_submitted).toBe("2020-05-01");
+    expect(gamma?.main_office_state).toBe("CO");
+  });
+
+  it("refuses a folder that resolves outside SEC_RAW_DATA_FOLDER", async () => {
+    const outside = mkdtempSync(join(tmpdir(), "sec-adv-outside-"));
+    writeFileSync(join(outside, "Private.csv"), "secret\nvalue\n");
+    try {
+      // The same shape `BootstrapDownloadTask` guards: the task is registered,
+      // so this input arrives from `sec-base task run`, the console form and
+      // the MCP tool surface, not only from the sync leaf.
+      await expect(
+        new IngestAdvSnapshotTask().run({
+          snapshot: "2026-06",
+          folder: relative(dir, outside),
+        })
+      ).rejects.toThrow(/SEC_RAW_DATA_FOLDER/);
+
+      const rows = globalServiceRegistry.get(ADV_ROW_REPOSITORY_TOKEN);
+      expect((await rows.getAll()) ?? []).toHaveLength(0);
+    } finally {
+      rmSync(outside, { recursive: true, force: true });
+    }
+  });
+
+  it("still validates the snapshot label when folder is given", async () => {
+    // `snapshot` is stamped on every row AND names the rows dropped before the
+    // read, so a label that never passed `advArchiveFolder` can wipe a real one.
+    await expect(
+      new IngestAdvSnapshotTask().run({
+        snapshot: "not-a-period",
+        folder: advArchiveFolder("2026-06"),
+      })
+    ).rejects.toThrow(/YYYY-MM/);
+  });
+
+  it("does not drop another snapshot's rows when the read is refused", async () => {
+    await new IngestAdvSnapshotTask().run({ snapshot: "2026-06" });
+
+    await expect(
+      new IngestAdvSnapshotTask().run({ snapshot: "2026-06", folder: "../.." })
+    ).rejects.toThrow(/SEC_RAW_DATA_FOLDER/);
+
+    const rows = globalServiceRegistry.get(ADV_ROW_REPOSITORY_TOKEN);
+    expect((await rows.query({ snapshot: "2026-06" })) ?? []).toHaveLength(3);
+  });
+
   it("says what to run when the archive was never downloaded", async () => {
     await expect(new IngestAdvSnapshotTask().run({ snapshot: "2026-07" })).rejects.toThrow(
       /sec update adv --period 2026-07/

src/task/adv/IngestAdvSnapshotTask.ts

@@ -5,7 +5,7 @@
  */
 
 import { readdir, readFile } from "node:fs/promises";
-import { extname, join } from "node:path";
+import { extname, join, resolve, sep } from "node:path";
 import { Type } from "typebox";
 import type { IExecuteContext } from "workglow";
 import { globalServiceRegistry, Task, TaskAbortedError } from "workglow";
@@ -89,6 +89,40 @@ function toNumber(value: string | undefined): number | null {
   return Number.isFinite(parsed) ? parsed : null;
 }
 
+/**
+ * Orders two `filing_id`s, ascending, with a missing one lowest.
+ *
+ * Numerically where both parse, because ADV writes the id unpadded and a
+ * lexical comparison puts `900010` below `900002`.
+ */
+function compareFilingId(a: string | null, b: string | null): number {
+  if (a === b) return 0;
+  if (a === null) return -1;
+  if (b === null) return 1;
+  const na = Number(a);
+  const nb = Number(b);
+  if (Number.isFinite(na) && Number.isFinite(nb) && na !== nb) return na < nb ? -1 : 1;
+  return a < b ? -1 : 1;
+}
+
+/**
+ * Whether `candidate` is the later of two base filings for one CRD.
+ *
+ * `date_submitted` decides, and `filing_id` breaks a tie. An undated filing
+ * loses to a dated one: it cannot be ordered against it, and a header the
+ * archive left undated must not displace one that says when it was filed.
+ */
+function isLaterFiling(candidate: AdvAdviser, current: AdvAdviser): boolean {
+  const a = candidate.date_submitted;
+  const b = current.date_submitted;
+  if (a !== b) {
+    if (a === null) return false;
+    if (b === null) return true;
+    return a > b;
+  }
+  return compareFilingId(candidate.filing_id, current.filing_id) > 0;
+}
+
 /** ISO date, or null — ADV dates arrive in several spellings across archives. */
 function toIsoDate(value: string | undefined): string | null {
   if (value === undefined) return null;
@@ -102,7 +136,9 @@ function toIsoDate(value: string | undefined): string | null {
 
 /**
  * Lands one extracted Form ADV archive: every member as `adv_row`, and the
- * base-filing member additionally as typed `adv_adviser` rows.
+ * base-filing member additionally as typed `adv_adviser` rows — one per CRD,
+ * carrying that adviser's latest filing in the snapshot. Per-filing history
+ * stays in `adv_row`, which keeps every base row verbatim.
  *
  * Nothing here knows the SEC's column set. The two headline tables people
  * filter on get columns; everything else stays queryable as JSON. A member the
@@ -145,8 +181,17 @@ export class IngestAdvSnapshotTask extends Task<
     context: IExecuteContext
   ): Promise<TaskPorts<IngestAdvSnapshotTaskOutput>> {
     const root = globalServiceRegistry.get(SEC_RAW_DATA_FOLDER);
-    const folder = input.folder ?? advArchiveFolder(input.snapshot);
-    const dir = join(root, folder);
+    // Unconditionally, so `folder` cannot smuggle past it: `snapshot` is
+    // stamped on every row and names the rows dropped before the read.
+    const snapshotFolder = advArchiveFolder(input.snapshot);
+    const folder = input.folder ?? snapshotFolder;
+    const dir = resolve(root, folder);
+    const safeBase = resolve(root) + sep;
+    if (!dir.startsWith(safeBase)) {
+      throw new Error(
+        `Invalid folder "${folder}": must resolve to a subdirectory of SEC_RAW_DATA_FOLDER`
+      );
+    }
     const members = (await readMembers(dir)).sort();
 
     if (members.length === 0) {
@@ -168,7 +213,11 @@ export class IngestAdvSnapshotTask extends Task<
     }
 
     let rowTotal = 0;
-    let adviserTotal = 0;
+    // Keyed by CRD because `adv_adviser` is, and the base member is one row per
+    // FILING: the cumulative archive carries every amendment an adviser filed
+    // between 2011 and 2024 under one snapshot label. Resolved here rather than
+    // left to the write, whose last-write-wins is CSV position.
+    const advisers = new Map<string, AdvAdviser>();
 
     for (const [position, member] of members.entries()) {
       if (context.signal?.aborted) throw new TaskAbortedError();
@@ -197,11 +246,10 @@ export class IngestAdvSnapshotTask extends Task<
 
       if (!isBaseFilingMember(table)) continue;
       const era = isEraMember(table);
-      const advisers: AdvAdviser[] = [];
       for (const row of rows) {
         const crd = advField(row, "1E1", "CRD Number", "crd_number");
         if (crd === undefined) continue;
-        advisers.push({
+        const adviser: AdvAdviser = {
           snapshot: input.snapshot,
           crd_number: crd,
           sec_file_number: advField(row, "1D", "SEC File Number") ?? null,
@@ -214,16 +262,26 @@ export class IngestAdvSnapshotTask extends Task<
           regulatory_aum: toNumber(advField(row, "5F2c", "5F(2)(c)")),
           filing_id: advField(row, "FilingID") ?? null,
           date_submitted: toIsoDate(advField(row, "DateSubmitted", "Date Submitted")),
-        });
+        };
+        const held = advisers.get(crd);
+        if (held === undefined || isLaterFiling(adviser, held)) advisers.set(crd, adviser);
       }
-      if (!dryRun) {
-        for (let i = 0; i < advisers.length; i += WRITE_BATCH) {
-          await adviserRepo.putBulk(advisers.slice(i, i + WRITE_BATCH));
-        }
+    }
+
+    // After every member, because the IA and ERA base members are one adviser
+    // set split in two and either can hold the later filing for a CRD.
+    const landedAdvisers = [...advisers.values()];
+    if (!dryRun) {
+      for (let i = 0; i < landedAdvisers.length; i += WRITE_BATCH) {
+        await adviserRepo.putBulk(landedAdvisers.slice(i, i + WRITE_BATCH));
       }
-      adviserTotal += advisers.length;
     }
 
-    return { success: true, tables: members.length, rows: rowTotal, advisers: adviserTotal };
+    return {
+      success: true,
+      tables: members.length,
+      rows: rowTotal,
+      advisers: landedAdvisers.length,
+    };
   }
 }