dispatch-batches.ts

// Orchestrate parallel batch sessions for Phase 7c.
// Reads outputs/all-districts.json, filters out the 30 already done,
// splits into batches of BATCH_SIZE, uploads files, creates sessions, fires kickoffs,
// streams events from all in parallel, downloads each batch's result on idle.

import Anthropic from '@anthropic-ai/sdk';
import { Agent, setGlobalDispatcher } from 'undici';
import * as fs from 'fs';
import * as path from 'path';

setGlobalDispatcher(new Agent({ bodyTimeout: 30 * 60_000, headersTimeout: 30 * 60_000 }));

const AGENT_ID = process.env.AGENT_ID;
const ENV_ID = process.env.ENV_ID;
if (!AGENT_ID || !ENV_ID) throw new Error('Set AGENT_ID and ENV_ID');

const BATCH_SIZE = parseInt(process.env.BATCH_SIZE || '15', 10);
const KICKOFF = fs.readFileSync('phase7cbatch.txt', 'utf8');
const OUT_DIR = path.resolve('./outputs/batches');
fs.mkdirSync(OUT_DIR, { recursive: true });

const client = new Anthropic();

interface District {
  name: string;
  hq_city?: string;
  county?: string;
  enrollment?: number | null;
  homepage?: string;
}

async function run() {
  // Use the pre-deduped TODO list from local Python step
  const todo: District[] = JSON.parse(fs.readFileSync('outputs/todo-districts.json', 'utf8'));
  console.log(`To process: ${todo.length} districts in batches of ${BATCH_SIZE}`);

  const batches: District[][] = [];
  for (let i = 0; i < todo.length; i += BATCH_SIZE) batches.push(todo.slice(i, i + BATCH_SIZE));
  console.log(`${batches.length} batches`);

  // Upload vendor-certs.json once
  const certsFile = await client.beta.files.upload({
    file: fs.createReadStream('outputs/vendor-certs.json'),
  });
  console.log(`vendor-certs file: ${certsFile.id}`);

  // For each batch: write batch.json, upload, create session, kickoff, await idle, download
  const results = await Promise.allSettled(
    batches.map((batch, i) => processBatch(batch, i, certsFile.id)),
  );

  let ok = 0, fail = 0;
  for (const [i, r] of results.entries()) {
    if (r.status === 'fulfilled') {
      console.log(`batch ${i}: ✓ ${r.value} districts`);
      ok++;
    } else {
      console.error(`batch ${i}: ✗ ${r.reason}`);
      fail++;
    }
  }
  console.log(`\nDone. ${ok} batches ok, ${fail} failed.`);
}

async function processBatch(batch: District[], idx: number, certsFileId: string): Promise<number> {
  const batchPath = path.join(OUT_DIR, `batch-${idx}.json`);
  fs.writeFileSync(batchPath, JSON.stringify(batch, null, 2));
  const batchFile = await client.beta.files.upload({
    file: fs.createReadStream(batchPath),
  });
  console.log(`[batch ${idx}] uploaded (${batch.length} districts, file ${batchFile.id})`);

  const session = await client.beta.sessions.create({
    agent: AGENT_ID!,
    environment_id: ENV_ID!,
    title: `Phase 7c batch ${idx} — ${batch.length} districts`,
    resources: [
      { type: 'file', file_id: batchFile.id, mount_path: '/workspace/batch.json' },
      { type: 'file', file_id: certsFileId, mount_path: '/workspace/vendor-certs.json' },
    ],
  });
  console.log(`[batch ${idx}] session: ${session.id}`);

  const stream = await client.beta.sessions.events.stream(session.id);
  await client.beta.sessions.events.send(session.id, {
    events: [{ type: 'user.message', content: [{ type: 'text', text: KICKOFF }] }],
  });

  for await (const ev of stream) {
    if (ev.type === 'session.error') {
      console.error(`[batch ${idx}] session.error:`, JSON.stringify(ev));
      throw new Error(`session.error: ${(ev as any).error?.message || 'unknown'}`);
    }
    if (ev.type === 'session.status_terminated') {
      throw new Error('session terminated');
    }
    if (ev.type === 'session.status_idle') {
      const sr = (ev as any).stop_reason;
      if (sr?.type === 'requires_action') {
        const ids: string[] = sr.event_ids ?? [];
        await client.beta.sessions.events.send(session.id, {
          events: ids.map((id) => ({
            type: 'user.tool_confirmation' as const,
            tool_use_id: id,
            result: 'allow' as const,
          })),
        });
        continue;
      }
      console.log(`[batch ${idx}] idle: ${sr?.type ?? 'unknown'}`);
      break;
    }
  }

  // Download batch result
  await new Promise((r) => setTimeout(r, 2500));
  const files = await client.beta.files.list({
    scope_id: session.id,
    betas: ['managed-agents-2026-04-01'],
  });
  let downloaded = 0;
  for (const f of files.data) {
    if (f.filename !== 'batch-results.json') continue;
    try {
      const resp = await client.beta.files.download(f.id);
      const buf = Buffer.from(await resp.arrayBuffer());
      const out = path.join(OUT_DIR, `batch-${idx}-results.json`);
      fs.writeFileSync(out, buf);
      console.log(`[batch ${idx}] → ${out} (${f.size_bytes}b)`);
      downloaded++;
    } catch (err) {
      console.error(`[batch ${idx}] download failed:`, err instanceof Error ? err.message : err);
    }
  }
  if (downloaded === 0) throw new Error('no batch-results.json produced');
  return batch.length;
}

function normalizeName(s: string): string {
  return s.toLowerCase().replace(/[^a-z0-9]/g, '');
}

run().catch((e) => { console.error(e); process.exit(1); });