App/cross year matching 1

AGENTS.md

@@ -145,6 +145,7 @@ sequenceDiagram
 
 - `app/api/src/earthbeam/api/earthbeam-api.controller.ts` — HTTP callback endpoints the executor calls
 - `app/api/src/earthbeam/api/earthbeam-api.service.ts` — Job payload assembly, run completion
+- `app/models/src/dtos/earthbeam-api.dto.ts` — Job payload shape
 - `executor/executor/executor.py` — Main executor: S3 operations, HTTP callbacks, earthmover/lightbeam invocation
 
 ### Executor Lifecycle
@@ -154,11 +155,28 @@ sequenceDiagram
 3. **Bundle refresh**: git fetch/checkout/pull the earthmover bundle
 4. **Roster fetch**: `lightbeam fetch` student roster from ODS, upload artifact to S3
 5. **File download**: Download user-uploaded input files from S3
-6. **Transform**: `earthmover run` (with encoding detection + retry)
-7. **Load**: `lightbeam send` to Ed-Fi ODS
-8. **Report**: POST summary, unmatched IDs, errors to app via callback URLs
-9. **Output files**: POST output file path + `sentToOds` flag to `/output-files` callback; app validates path, lists S3, saves `run_output_file_set`
-10. **Done**: POST status `{action: DONE, status: success|failure}`
+6. **Transform**: `earthmover run` against the ODS roster (with encoding detection + retry)
+7. **Cross-year retry** (when `crossYearMatchAvailable` and the first pass produced unmatched students): GET `appUrls.roster` for the cross-year NDJSON roster, write to a `.jsonl` file, and re-run `earthmover` against it using the same ID type.
+8. **Load**: `lightbeam send` to Ed-Fi ODS
+9. **Report**: POST summary, unmatched IDs, errors to app via callback URLs
+10. **Output files**: POST output file path + `sentToOds` flag to `/output-files` callback; app validates path, lists S3, saves `run_output_file_set`
+11. **Done**: POST status `{action: DONE, status: success|failure}`
+
+### Cross-Year Matching Flow
+
+When cross-year matching runs, the executor progresses through each stage of processing for both rosters before moving on, to avoid mixed-status jobs (e.g., a file failing "insufficient matches" against the ODS roster but succeeding against the cross-year roster, when those matches really belonged in the ODS).
+
+```mermaid
+flowchart TD
+    Input[Uploaded input rows] --> T1[earthmover: match + transform<br/>against ODS roster]
+    T1 -->|on success, if crossYearMatchAvailable<br/>and step 7 triggered| T2[earthmover: match + transform<br/>against cross-year EDU roster]
+    T1 -->|on success, otherwise| Load
+    T2 -->|both transforms succeeded| Load[lightbeam send<br/>ODS-matched rows → ODS]
+    Load -->|ODS load succeeded| App[POST results to Runway app<br/>ODS-matched + cross-year-matched rows<br/>exposed via API]
+    App -.fetched by.-> EDU[EDU / external API consumers]
+```
+
+Cross-year-matched rows are never sent to the ODS — they're only made available through the Runway app's API, which EDU and other external consumers query.
 
 ### S3 Path Structure
 

app/api/.env.copyme

@@ -27,5 +27,18 @@ LOCAL_EVENT_EMITTER=log # log | noop (omit for EventBridge)
 AWS_REGION=us-east-2
 BUNDLE_BRANCH=development
 
+# EDU Snowflake cross-year roster fallback for local development only.
+# Private key value should be base64-encoded PEM contents.
+# Example from file:      `base64 -w 0 private_key.pem`
+# Example from clipboard:  `xclip -selection clipboard -o | base64 -w 0`
+EDU_SNOWFLAKE_USERNAME=
+EDU_SNOWFLAKE_ACCOUNT=
+EDU_SNOWFLAKE_DATABASE=
+EDU_SNOWFLAKE_SCHEMA=
+EDU_SNOWFLAKE_PRIVATE_KEY=
+# Optional: leave blank to use the Snowflake user's defaults.
+EDU_SNOWFLAKE_WAREHOUSE=
+EDU_SNOWFLAKE_ROLE=
+
 OAUTH2_ISSUER=http://localhost:8080/realms/example
 OAUTH2_AUDIENCE=runway-local

app/api/integration/factories/partner-user-tenant.ts

@@ -13,6 +13,7 @@ export const makePartnerUserTenantContext = (tag: string) => {
     name: `Partner ${tag}`,
     idpId: idp.id,
     descriptorNamespace: null,
+    crossYearMatchingEnabled: false,
   };
 
   const tenant: WithoutAudit<Tenant> = {

app/api/integration/fixtures/context-fixtures/partner-fixtures.ts

@@ -7,18 +7,21 @@ export const partnerA: WithoutAudit<Partner> = {
   name: 'Partner A',
   idpId: idpA.id,
   descriptorNamespace: 'partner-a',
+  crossYearMatchingEnabled: false,
 };
 
 export const partnerC: WithoutAudit<Partner> = {
   id: 'partner-c',
   name: 'Partner C',
   idpId: idpA.id, // shares idp with partner A
   descriptorNamespace: 'partner-c',
+  crossYearMatchingEnabled: false,
 };
 
 export const partnerX: WithoutAudit<Partner> = {
   id: 'partner-x',
   name: 'Partner X',
   idpId: idpX.id,
   descriptorNamespace: null,
+  crossYearMatchingEnabled: false,
 };

app/api/integration/tests/earthbeam-api.spec.ts

@@ -1,4 +1,7 @@
 import { EarthbeamApiAuthService } from 'api/src/earthbeam/api/auth/earthbeam-api-auth.service';
+import { EduSnowflakePoolService } from 'api/src/earthbeam/api/edu-snowflake-pool.service';
+import { AppConfigService } from 'api/src/config/app-config.service';
+import { Readable } from 'node:stream';
 import request from 'supertest';
 import { seedJob } from '../factories/job-factory';
 import { bundleA, bundleX } from '../fixtures/em-bundle-fixtures';
@@ -103,6 +106,70 @@ describe('Earthbeam API', () => {
       );
     });
 
+    describe('cross-year ID matching', () => {
+      // Default state per test: both gates ON (toggle enabled + creds present)
+      // so the happy path requires no overrides and each negative test reads
+      // as "remove one condition, expect the flag to flip false."
+      let getInfoSpy: jest.SpyInstance;
+
+      beforeEach(async () => {
+        await global.prisma.partner.update({
+          where: { id: partnerA.id },
+          data: { crossYearMatchingEnabled: true },
+        });
+        const configService = app.get(AppConfigService);
+        getInfoSpy = jest.spyOn(configService, 'getEduConnectionInfo').mockResolvedValue({
+          username: 'snowflake-user',
+          account: 'example',
+          database: 'edu_stg',
+          schema: 'public',
+          privateKey: Buffer.from('priv'),
+        });
+      });
+
+      afterEach(() => {
+        getInfoSpy.mockRestore();
+      });
+
+      it('sets crossYearMatchAvailable=true and emits appUrls.roster when toggle on and creds exist', async () => {
+        const res = await request(app.getHttpServer())
+          .get(endpointA)
+          .set('Authorization', `Bearer ${tokenA}`);
+
+        expect(res.status).toBe(200);
+        expect(res.body.crossYearMatchAvailable).toBe(true);
+        expect(res.body.appUrls.roster).toBeDefined();
+        expect(res.body.appUrls.roster).toContain(`/earthbeam/jobs/${runA.id}/roster`);
+      });
+
+      it('sets crossYearMatchAvailable=false and omits appUrls.roster when toggle is off', async () => {
+        await global.prisma.partner.update({
+          where: { id: partnerA.id },
+          data: { crossYearMatchingEnabled: false },
+        });
+
+        const res = await request(app.getHttpServer())
+          .get(endpointA)
+          .set('Authorization', `Bearer ${tokenA}`);
+
+        expect(res.status).toBe(200);
+        expect(res.body.crossYearMatchAvailable).toBe(false);
+        expect(res.body.appUrls.roster).toBeUndefined();
+      });
+
+      it('sets crossYearMatchAvailable=false and omits appUrls.roster when creds are missing', async () => {
+        getInfoSpy.mockResolvedValue(null);
+
+        const res = await request(app.getHttpServer())
+          .get(endpointA)
+          .set('Authorization', `Bearer ${tokenA}`);
+
+        expect(res.status).toBe(200);
+        expect(res.body.crossYearMatchAvailable).toBe(false);
+        expect(res.body.appUrls.roster).toBeUndefined();
+      });
+    });
+
     // TODO: add tests for things other than descriptor mappings
     describe('Authenticated requests: Descriptor Mappings', () => {
       const testDescriptorTypeA = 'testDescriptorTypeA';
@@ -231,6 +298,166 @@ describe('Earthbeam API', () => {
 
   });
 
+  describe('GET /:runId/roster', () => {
+    let runA: Run;
+    let endpointA: string;
+    let tokenA: string;
+
+    beforeEach(async () => {
+      const authService = app.get(EarthbeamApiAuthService);
+      const jobA = await seedJob({
+        odsConfig: odsConfigA2425,
+        bundle: bundleA,
+        tenant: tenantA,
+      });
+      runA = jobA.runs[0];
+      endpointA = `/earthbeam/jobs/${runA.id}/roster`;
+      tokenA = await authService.createAccessToken({ runId: runA.id });
+    });
+
+    it('rejects unauthenticated requests', async () => {
+      const res = await request(app.getHttpServer()).get(endpointA);
+      expect(res.status).toBe(401);
+    });
+
+    it('returns 409 when the partner has cross-year matching disabled', async () => {
+      // partnerA defaults to crossYearMatchingEnabled=false
+      const res = await request(app.getHttpServer())
+        .get(endpointA)
+        .set('Authorization', `Bearer ${tokenA}`);
+      expect(res.status).toBe(409);
+    });
+
+    it('returns 500 when EDU creds are missing for an otherwise-enabled partner', async () => {
+      await global.prisma.partner.update({
+        where: { id: partnerA.id },
+        data: { crossYearMatchingEnabled: true },
+      });
+      // No creds → pool creation will fail before any rows are written;
+      // controller's headersSent check should convert that to a clean 500
+      // rather than tearing the socket.
+      const configService = app.get(AppConfigService);
+      const getInfoSpy = jest
+        .spyOn(configService, 'getEduConnectionInfo')
+        .mockResolvedValue(null);
+
+      const res = await request(app.getHttpServer())
+        .get(endpointA)
+        .set('Authorization', `Bearer ${tokenA}`);
+      expect(res.status).toBe(500);
+
+      getInfoSpy.mockRestore();
+    });
+
+    describe('streaming responses', () => {
+      // Streaming parser for supertest: collects chunks as they arrive and
+      // signals whether the response ended cleanly ('end' fired) or was closed
+      // early ('close'/'error' fired first). Use .buffer(true).parse(streamParser).
+      const streamParser = (
+        response: request.Response,
+        cb: (err: Error | null, body: { chunks: Buffer[]; complete: boolean }) => void
+      ) => {
+        const chunks: Buffer[] = [];
+        let settled = false;
+        const settle = (complete: boolean) => {
+          if (settled) return;
+          settled = true;
+          cb(null, { chunks, complete });
+        };
+        response.on('data', (chunk: Buffer) => chunks.push(chunk));
+        response.on('end', () => settle(true));
+        response.on('close', () => settle(false));
+        response.on('error', () => settle(false));
+      };
+
+      // Stub EduSnowflakePoolService.use with a fake connection that streams
+      // `source` rows. Caller is responsible for `mockRestore()`.
+      const mockEduPoolStream = (source: Iterable<unknown> | AsyncIterable<unknown>) => {
+        const eduPool = app.get(EduSnowflakePoolService);
+        return jest.spyOn(eduPool, 'use').mockImplementation(async (_partnerId, cb) => {
+          return cb({
+            execute: () => ({ streamRows: () => Readable.from(source) }),
+          } as never);
+        });
+      };
+
+      beforeEach(async () => {
+        await global.prisma.partner.update({
+          where: { id: partnerA.id },
+          data: { crossYearMatchingEnabled: true },
+        });
+      });
+
+      it('streams the rows from the EDU pool as NDJSON', async () => {
+        const rows = [
+          { studentUniqueId: '1', priorYear: 2024 },
+          { studentUniqueId: '2', priorYear: 2024 },
+          { studentUniqueId: '3', priorYear: 2024 },
+        ];
+        const spy = mockEduPoolStream(rows);
+
+        const res = await request(app.getHttpServer())
+          .get(endpointA)
+          .set('Authorization', `Bearer ${tokenA}`)
+          .buffer(true)
+          .parse(streamParser);
+
+        expect(res.status).toBe(200);
+        expect(res.headers['content-type']).toContain('application/x-ndjson');
+        expect(res.body.complete).toBe(true);
+        const body = Buffer.concat(res.body.chunks).toString('utf8');
+        expect(body).toBe(rows.map((r) => JSON.stringify(r)).join('\n') + '\n');
+
+        spy.mockRestore();
+      });
+
+      it('closes the response abruptly when the Snowflake row stream errors mid-flight', async () => {
+        const spy = mockEduPoolStream(
+          (async function* () {
+            yield { studentUniqueId: '1' };
+            throw new Error('snowflake exploded mid-stream');
+          })()
+        );
+
+        const res = await request(app.getHttpServer())
+          .get(endpointA)
+          .set('Authorization', `Bearer ${tokenA}`)
+          .buffer(true)
+          .parse(streamParser);
+
+        // Pipeline destroys the response on stream error. Headers were
+        // already sent, so the status is 200, but the socket closes before
+        // 'end' fires — streamParser surfaces that as complete: false.
+        expect(res.status).toBe(200);
+        expect(res.body.complete).toBe(false);
+        const body = Buffer.concat(res.body.chunks).toString('utf8');
+        expect(body).toBe(JSON.stringify({ studentUniqueId: '1' }) + '\n');
+
+        spy.mockRestore();
+      });
+    });
+
+    it('returns 500 when pool acquisition fails before any bytes are streamed', async () => {
+      await global.prisma.partner.update({
+        where: { id: partnerA.id },
+        data: { crossYearMatchingEnabled: true },
+      });
+
+      const eduPool = app.get(EduSnowflakePoolService);
+      const poolUseSpy = jest
+        .spyOn(eduPool, 'use')
+        .mockRejectedValue(new Error('pool acquisition failed'));
+
+      const res = await request(app.getHttpServer())
+        .get(endpointA)
+        .set('Authorization', `Bearer ${tokenA}`);
+
+      expect(res.status).toBe(500);
+
+      poolUseSpy.mockRestore();
+    });
+  });
+
   describe('POST /:runId/status', () => {
     let runA: Run;
     let tokenA: string;