diff --git a/CHANGELOG.md b/CHANGELOG.md index 83e6b4d..3f36a98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All notable changes to this project are documented here. The format is based on ### Added +- **Broad-Side: lens batches poll concurrently** (#136). Collect previously polled one lens at a time to completion, so the slowest lens serialized the wall clock for lenses that had already finished server-side. In-flight batches now poll in parallel against one shared deadline via the new `pollBatchesConcurrently` helper, with progress callbacks tagged per lens; results still save in deterministic lens order. The poll interval is now injectable, which the new peak-concurrency regression tests use to prove the parallelism without timing flakiness. - **Broad-Side: batch reconnaissance over the OpenRouter Batch API** (#144). New `codecarto_broadside` MCP tool (actions: `submit`, `collect`, `status`) fires six single-turn analysis lenses — architecture, API surface, security, mechanical defect scan, convention extraction, porting — at any git repository as asynchronous batch jobs on a cheap batch model (~50% of sync pricing, unattended, 24h window), slices large modules by top-level directory, saves JSON plus rendered markdown to `.codecarto/broadside//`, and optionally synthesizes a cross-lens executive report. Works without an initialized workspace; needs an OpenRouter key via the `api_key` parameter, `OPENROUTER_API_KEY`, or `.codecarto/broadside/config.yaml`. Broad-Side findings are explicitly unverified scouting signals — file:line leads for the interactive pipeline to confirm, never evidence themselves. `codecarto_init` tolerates a `.codecarto/` that holds only `broadside/` (no force/backup needed), and scaffold refresh never touches broadside state, config, or results. - **Broad-Side: expense guardrails and live per-model pricing** (#144). `config.yaml` now accepts `model`, `max_cost`, and `pricing.input_per_m`/`output_per_m` overrides, and the MCP tool accepts `max_cost` and `force` parameters. Before submitting, Broad-Side estimates the run cost from collected file sizes (≈4 chars/token) against the configured model's per-token pricing — looked up live from OpenRouter's model catalog (cached 24h), so models like `openai/gpt-5.2-pro:batch` at ~$84/M output are priced correctly, not at the default model's rates. A submit whose estimate exceeds `max_cost` refuses with a per-lens breakdown and creates no run entry unless `force: true`. The submit response now reports the pricing used and its source (built-in/config/live/cache). - **Broad-Side: model catalog action and capability pre-flight** (#144). New `models` action lists every `:batch` model on OpenRouter — pricing per million tokens, context window, completion ceiling, structured-output support, and optional Artificial Analysis coding indices (via `GET /api/v1/benchmarks`, attribution preserved) — cheapest first with the configured model marked. Submits now pre-flight the chosen model against that catalog: lens `max_tokens` clamps to the provider's completion ceiling, deprecated models are flagged, and models that do not advertise structured-output support are refused outright, since every lens depends on `json_schema` response_format. The catalog cache is shared between the `models` action and submit-time pricing resolution. diff --git a/ROADMAP.md b/ROADMAP.md index 509a757..b76e014 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -35,7 +35,7 @@ file only moves when a tier completes. |---|---|---| | **Triage lens** — prioritized fix queue (impact × difficulty, grouped by module) | [#135](https://github.com/HuginnIndustries/CodeCartographer/issues/135) | **Shipped**: triage pass runs on collect alongside synthesis (`include_triage` to skip) | | **Truncation repair** — detect max_tokens-cutoff JSON, resubmit slices, report truncation in summaries | [#133](https://github.com/HuginnIndustries/CodeCartographer/issues/133) | Partially shipped: fence-tolerant parsing + `truncated` flagging in collect, meta, and synthesis. Remaining: automatic resubmit of truncated slices | -| **Concurrent polling** — poll all in-flight batches round-robin against one deadline | [#136](https://github.com/HuginnIndustries/CodeCartographer/issues/136) | Submissions already parallel; polling is sequential today | +| **Concurrent polling** — poll all in-flight batches round-robin against one deadline | [#136](https://github.com/HuginnIndustries/CodeCartographer/issues/136) | **Shipped**: `pollBatchesConcurrently` polls in parallel with per-lens progress tags | | **Per-language prompts** — Go/Python/Rust/TS lens prompts; globs already adapt | [#137](https://github.com/HuginnIndustries/CodeCartographer/issues/137) | Schemas stay shared so synthesis is unaffected | ## Tier 2 — integration depth diff --git a/core/broadside.ts b/core/broadside.ts index 92abb0d..225533d 100644 --- a/core/broadside.ts +++ b/core/broadside.ts @@ -1540,9 +1540,11 @@ export async function pollBatchUntilTerminal( deadlineMs?: number; onStatus?: (status: string, counts: Record) => void; fetcher?: FetchLike; + pollIntervalMs?: number; } = {}, ): Promise> { const deadline = Date.now() + (opts.deadlineMs ?? BROADSIDE_DEFAULT_POLL_BUDGET_MS); + const intervalMs = opts.pollIntervalMs ?? BROADSIDE_POLL_INTERVAL_MS; const fetcher = opts.fetcher ?? (fetch as FetchLike); for (;;) { let batch: Record; @@ -1550,7 +1552,7 @@ export async function pollBatchUntilTerminal( batch = await fetchBatch(batchId, apiKey, fetcher); } catch { if (Date.now() >= deadline) return { id: batchId, status: "timeout" }; - await sleep(BROADSIDE_POLL_INTERVAL_MS); + await sleep(intervalMs); continue; } const httpStatus = Number(batch.http_status ?? 200); @@ -1562,10 +1564,43 @@ export async function pollBatchUntilTerminal( opts.onStatus?.(status, counts); if (["completed", "failed", "expired", "cancelled", "auth-failed"].includes(status)) return batch; if (Date.now() >= deadline) return { id: batchId, status: "timeout" }; - await sleep(BROADSIDE_POLL_INTERVAL_MS); + await sleep(intervalMs); } } +/** + * Poll several batch ids in parallel against one shared deadline. Collect + * previously polled one lens at a time, so a slow first lens serialized the + * wall clock for lenses that had already finished server-side (#136). The + * onStatus callback identifies the lens so progress output stays readable + * even while the polls interleave. + */ +export async function pollBatchesConcurrently( + entries: Array<{ lensId: BroadsideLensId; batchId: string }>, + apiKey: string, + opts: { + deadlineMs?: number; + fetcher?: FetchLike; + pollIntervalMs?: number; + onStatus?: (lensId: string, status: string, counts: Record) => void; + } = {}, +): Promise>> { + const results = new Map>(); + const deadlineMs = opts.deadlineMs ?? BROADSIDE_DEFAULT_POLL_BUDGET_MS; + await Promise.all( + entries.map(async ({ lensId, batchId }) => { + const batch = await pollBatchUntilTerminal(batchId, apiKey, { + deadlineMs, + fetcher: opts.fetcher, + pollIntervalMs: opts.pollIntervalMs, + onStatus: (status, counts) => opts.onStatus?.(lensId, status, counts), + }); + results.set(batchId, batch); + }), + ); + return results; +} + // ---------- run orchestration ---------- export async function runBroadsideSubmit( @@ -1923,6 +1958,10 @@ export async function runBroadsideCollect( const allLensResults: StoredLensResult[] = []; + // Terminal entries are settled already; everything else polls in parallel + // against one shared deadline (#136), then results save in lens order so + // output layout stays deterministic. + const inFlight: Array<{ lensId: BroadsideLensId; batchId: string }> = []; for (const lensId of run.lenses) { const entry = run.batches[lensId]; if (!entry || !entry.batchId) { @@ -1935,12 +1974,19 @@ export async function runBroadsideCollect( lensOutcomes[lensId] = { status: entry.status, cost: entry.cost, resultCount: entry.resultCount }; continue; } + inFlight.push({ lensId, batchId: entry.batchId }); + } - const batch = await pollBatchUntilTerminal(entry.batchId, apiKey, { - deadlineMs: Math.max(0, deadline - Date.now()), - onStatus: (status, counts) => opts.onStatus?.(lensId, status, counts), - fetcher: opts.fetcher, - }); + const polled = await pollBatchesConcurrently(inFlight, apiKey, { + deadlineMs: Math.max(0, deadline - Date.now()), + fetcher: opts.fetcher, + onStatus: opts.onStatus, + }); + + for (const { lensId } of inFlight) { + const entry = run.batches[lensId]; + if (!entry) continue; + const batch = polled.get(entry.batchId) ?? { id: entry.batchId, status: "timeout" }; const status = String(batch.status ?? "unknown"); entry.status = status; diff --git a/tests/broadside.test.mjs b/tests/broadside.test.mjs index 62e12a0..347b2dc 100644 --- a/tests/broadside.test.mjs +++ b/tests/broadside.test.mjs @@ -29,6 +29,7 @@ const { loadBroadsideState, listLenses, modelsText, + pollBatchesConcurrently, renderFindingsMarkdown, resolveModelPricing, runBroadsideCollect, @@ -562,6 +563,102 @@ test("modelsText renders pricing, caps, support, and benchmark columns", () => { assert.match(text, /\(default\)/); }); +// ---------- concurrent polling (#136) ---------- + +test("pollBatchesConcurrently polls all batches in parallel against one deadline", async () => { + // Peak-concurrency tracking is deterministic: if polling were sequential, + // the fast batch would hold the loop and peak concurrent GETs would stay + // at 1. Under the fix, the fast batch polls while the slow one is still + // mid-polling. + let inFlightGets = 0; + let peak = 0; + const fetcher = async (url) => { + inFlightGets += 1; + peak = Math.max(peak, inFlightGets); + try { + await new Promise((r) => setTimeout(r, 15)); // overlap window + if (String(url).includes("batch-a")) { + return fakeResponse(200, { id: "batch-a", status: "completed", results: [], usage: { cost: 0.001 } }); + } + return fakeResponse(200, { id: "batch-b", status: "completed", results: [], usage: { cost: 0.002 } }); + } finally { + inFlightGets -= 1; + } + }; + + const results = await pollBatchesConcurrently( + [ + { lensId: "defect", batchId: "batch-a" }, + { lensId: "security", batchId: "batch-b" }, + ], + "sk-fake", + { fetcher, pollIntervalMs: 20, deadlineMs: 5000 }, + ); + assert.equal(results.size, 2); + assert.equal(results.get("batch-a").status, "completed"); + assert.equal(results.get("batch-b").status, "completed"); + assert.ok(peak >= 2, `peak concurrent GETs was ${peak} — polling is sequential`); +}); + +test("collect polls multiple in-flight lenses concurrently", async () => { + const dir = await mkdtemp(join(tmpdir(), "broadside-conc-")); + try { + await writeFile(join(dir, "go.mod"), "module x\n"); + await writeFile(join(dir, "main.go"), "package main\n"); + await mkdir(join(dir, "server")); + await writeFile(join(dir, "server", "routes.go"), "package server\n"); + + let inFlightGets = 0; + let peak = 0; + const fetcher = async (url, init) => { + if (init.method === "POST") { + return fakeResponse(202, { id: "batch-x", status: "validating" }); + } + inFlightGets += 1; + peak = Math.max(peak, inFlightGets); + try { + await new Promise((r) => setTimeout(r, 15)); + const batchId = String(url).split("/").pop(); + return fakeResponse(200, { + id: batchId, + status: "completed", + results: [ + { + custom_id: `${batchId}-1`, + response: { + status_code: 200, + body: { choices: [{ message: { content: JSON.stringify({ module: "x", findings: [], patterns_checked: [], files_scanned: 0 }) } }] }, + }, + error: null, + }, + ], + usage: { cost: 0.001 }, + }); + } finally { + inFlightGets -= 1; + } + }; + + // Two submissions → two lens batches; rewrite state so both are + // in flight under distinct ids, then collect must poll both together. + const result = await runBroadsideSubmit(dir, "sk-fake", { lenses: ["architecture", "security"], fetcher }); + assert.equal(Object.keys(result.batches).length, 2); + const broadsideDir = join(dir, ".codecarto", "broadside"); + const state = await loadBroadsideState(broadsideDir); + state.runs[0].batches.architecture.batchId = "batch-a"; + state.runs[0].batches.architecture.status = "validating"; + state.runs[0].batches.security.batchId = "batch-b"; + state.runs[0].batches.security.status = "validating"; + await saveBroadsideState(broadsideDir, state); + + const collect = await runBroadsideCollect(dir, "sk-fake", { fetcher, includeSynthesis: false, includeTriage: false }); + assert.equal(collect.resultCount, 2); + assert.ok(peak >= 2, `peak concurrent GETs was ${peak} — collect is polling lenses sequentially`); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + // ---------- batch client with a fake fetcher ---------- function fakeResponse(status, body) {