Repository navigation
feat: DAG-based post-processing pipeline - #1147
Cedric Vidal (cedricvidal) wants to merge 27 commits into
Conversation
Refactor post-processing from single-handler to DAG orchestration: Phase 1 - Handler Dispatcher: - Add HandlerDispatcher with DAG topology from services collection - Scheduler REST endpoints: POST /notify/run-terminal, /handler-complete - Workers notify scheduler instead of direct queue enqueue (fallback kept) - Remove triggerReportGeneration (reports now DAG node via pp-report) - DB migrations: services → handler entries, run.handlerStatus field Phase 2 - Taxonomy Handler: - Zod schema for 3-layer taxonomy (scorecard, behavior, actions) - taxonomy-handler worker with Copilot SDK + multi-turn validation loop - Docker-compose service with SCHEDULER_URL wiring DAG structure: pp-atif → pp-taxonomy → pp-report Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Add GET /api/v1/requests/:id/taxonomy endpoint - Bundle taxonomy.json in run download tar.gz archive - Add taxonomyUrl to RunState type and request schema - Update taxonomy worker to persist run.taxonomyUrl - Update OpenAPI snapshot and post-processing docs Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Use multi-stage Dockerfile matching report-generator pattern - Add tsconfig.base.json, proper COPY layers, --prod install in prod stage - Add dev stage for hot-reload with tsx watch - Add taxonomy-handler to docker-compose.dev.yml watch config - Add --profile taxonomy to docker:dev:copilot and docker:dev:all scripts Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Adds a Taxonomy tab to the run detail page that shows: - Scorecard with visual score bars per dimension - Behavior analysis patterns with impact badges - Action items with priority indicators - Collapsible raw JSON view The tab only appears when taxonomy data is available (run.taxonomyUrl set). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…tion jobs Each post-processing handler now owns its own service entry registration via a register-version.ts script (run as K8s Job on deploy): - pp-atif: apps/workers/post-processor/src/register-version.ts - pp-taxonomy: apps/workers/taxonomy/src/register-version.ts - pp-report: apps/workers/report-generator/src/register-version.ts Migration 018 is reduced to only cleaning up the legacy 'post-processor' entry. Each handler defines its own dependencies (dependsOn), queue, version, and selector — decoupling the DAG topology from a single migration. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Replace the three inlined register-version.ts scripts (which wrote handler metadata directly to MongoDB) with a declarative, YAML-driven approach that mirrors the coding-agent registration pattern. - Add per-worker handler.yaml (pp-atif, pp-taxonomy, pp-report) as the single source of truth for handler metadata. - Add generic packages/shared/src/scripts/register-handler.ts that reads HANDLER_YAML_PATH, waits for the scheduler /health, and POSTs to POST /handlers/register with retry. - Add scheduler POST /handlers/register endpoint (notify-routes.ts) + HandlerDispatcher.registerHandler (cycle-checked upsert). Scheduler is now the sole writer of the services collection. - Remove the 3 register-version.ts scripts and the mongodb dep from the taxonomy + report-generator workers. - Wire compose (register-handler-* services) and K8s (scheduler Service, register-handler-post-processor Job, SCHEDULER_URL, /health probes). - Update docs/architecture/post-processing.md. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Reports were modeled as a DAG handler node (pp-report, dependsOn
[pp-atif, pp-taxonomy]), but the report-generator worker is keyed by
reportId and requires a templateId, while the DAG dispatches a generic
{type, requestId, runId} message — so pp-report could never process a
DAG message. Reports are also parameterized, N-per-run, and trigger-gated,
so they don't fit the idempotent per-run DAG-node model.
Take reports off the DAG. The scheduler now detects when a run's handler
DAG has drained (every registered handler terminal: done | failed |
blocked-by-failed-dependency) and calls the existing
POST /api/v1/reports/trigger endpoint, which evaluates each template's
trigger and creates a properly-templated ReportDocument + enqueues
{reportId} per match.
- HandlerDispatcher: isHandlerTerminal/isDagDrained/maybeTriggerReports/
triggerReports + pollTriggerReports safety net. Exactly-once via an
atomic run.reportsTriggeredAt guard, scoped by run._id so a stale notify
or concurrent retry can never trigger reports for — or roll back the
guard of — a different attempt. Fires on run-terminal (zero-handler /
already-drained), both branches of handler-complete (a failure can drain
the DAG), and the 30s poll loop. Rolls back the guard on POST failure.
- Inject API_URL into the dispatcher; add API_URL to scheduler env in
docker-compose.yml and deploy/base/scheduler.yaml.
- Remove pp-report as a DAG node: delete report-generator/handler.yaml,
its Dockerfile COPY lines, the register-handler-report-generator compose
service, the dead idx_handler_dispatch_report index in migration 019, and
stale comments. No migration to delete pp-report is needed — nothing is
deployed, so a fresh environment never registers it.
- Add reportsTriggeredAt to RunState.
- Tests: 13 dispatcher tests (drain detection, exactly-once guard,
attempt-scoped claim/rollback, stale-runId rejection, POST payload).
- Docs: post-processing.md "Report Triggering on DAG Drain" section +
ENV_VARIABLES API_URL.
Verified E2E: run -> pp-atif(done) -> pp-taxonomy(failed) -> DAG drains ->
scheduler POST /reports/trigger -> API matches template -> ReportDocument
(templateId) enqueued -> report-generator -> report done. Confirms reports
fire even when a handler fails.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…-5.4 Extract the Copilot SDK taxonomy-generation loop into a shared, side-effect-free generateTaxonomy() used by both the worker and a new developer CLI (src/cli.ts, `pnpm taxonomy`). The CLI runs generation against any local or integration run, streams session events, and writes the validated taxonomy to a file — making it easy to troubleshoot and A/B models without the full queue/scheduler pipeline. Switch get_atif_trajectory to fetch ATIF through the run-scoped API endpoint instead of reading blob storage directly, so the CLI needs only an API URL + token (no storage connection string). Change the default TAXONOMY_MODEL from gpt-4.1 to gpt-5.4. A/B testing nine models on a multi-iteration run containing a real defect (agent lost context in iteration 2) showed verdicts ranging 72-100: weaker models (incl. gpt-4.1) missed the critical failure entirely, while gpt-5.4 caught it at the lowest token cost of the models tested. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…/pp-dag-orchestration # Conflicts: # ENV_VARIABLES.md # apps/scheduler/src/index.ts # deploy/base/scheduler.yaml # packages/db-migrations/src/check-migrations.test.ts # packages/db-migrations/src/required-migrations.ts
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…/pp-dag-orchestration
…/pp-dag-orchestration # Conflicts: # packages/db-migrations/src/check-migrations.test.ts
…versions The autoBackfill feature re-processes historical runs when a handler version bumps, enabling backfill of new/updated post-processing features across the whole dataset. However, buildDispatchFilter() was excluding 'done' status universally, preventing the version-check branch ($lt version) from matching any historical runs. This change makes status exclusions conditional: - autoBackfill handlers exclude only ['queued', 'processing'], allowing done runs to qualify for version-based re-dispatch - Non-backfill handlers still exclude 'done' (only dispatch on missing handler status) Fixes regression identified in #1147 code review. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
…/pp-dag-orchestration # Conflicts: # packages/db-migrations/src/check-migrations.test.ts
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
loadHandlers() sits on the notify/poll hot path (2-3 reads per notify, one per poll cycle) but the services collection only changes on deploys. Cache the topology in-memory with a 60s TTL (configurable) to eliminate nearly all of those reads — meaningful on CosmosDB where each query costs RUs. registerHandler invalidates the cache after upserting so new topology is picked up immediately; cycle-check validation reads fresh. Addresses jduffney review feedback #4 on #1147. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
…v-configurable The scheduler is a singleton (replicas: 1, Recreate) and the sole writer of the services collection, and registerHandler invalidates the cache on every topology change — so cache coherence rests on invalidation, not expiry. A 60s TTL sat barely above the 30s poll, forcing a needless fresh read on roughly every other poll cycle. Raise the default to 5min (well above the poll interval) and expose SCHEDULER_HANDLER_CACHE_TTL_MS so the TTL is only a coarse backstop for out-of-band writes or a future scale-out beyond one replica. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Make explicit that process-local cache invalidation is only correct while the scheduler runs as a single replica, and document the cross-replica mitigations (Redis pub/sub or change-stream watcher, or a shorter TTL) required before any scale-out. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Guards the singleton scheduler HTTP server against unbounded-memory requests. Rejects oversized payloads with 413 Payload Too Large + Connection: close, using a Content-Length fast-reject plus a streaming byte-count cap for chunked/lying clients. maxBodyBytes is threaded through createHttpServer so tests can use a small deterministic cap. Addresses jduffney review feedback #5 on #1147. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
The enqueue-before-claim refactor made the scheduler write a best-effort handlerStatus.<id>.status="queued" marker in-process immediately after enqueueing. The worker's claim filter excluded "queued", so once that marker won the (near-certain) race against queue delivery, the worker dequeued its own message, matched nothing on the atomic claim, logged "no longer needed, discarding", and deleted the message — stranding the run at "queued" forever (the 30s backfill poll also excludes "queued", so nothing recovered it). This reintroduced the stuck-dispatch failure mode via a new path. Fix: remove "queued" from the worker claim exclusions so queued -> processing is a permitted transition. Idempotency is preserved: still exclude "processing" (another worker in-flight) and, for non-backfill handlers, "done"; autoBackfill "done" stays gated by the version check. Add a regression test asserting a run in the scheduler's "queued" state is claimable and driven to done, update the existing filter-shape assertion, and document that queued -> processing must be permitted. Verified end-to-end against the docker-compose stack: a synthetic status=done run dispatched via /notify/run-terminal now reaches handlerStatus.pp-atif.status=done + reportsTriggeredAt, with duplicate delivery still no-oping and the report trigger firing exactly once. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
…/pp-dag-orchestration
`main` independently claimed 018/019/020 (backfill-criteria-gates, backfill-task-prompt-type, create-codebase-indexes) after this branch had assigned those numbers to the handler-DAG migrations, producing duplicate prefixes once main was merged in. Renumber the three handler migrations to sit after main's highest (024) so prefixes are unique and monotonic: 018-migrate-services-to-handler-dag -> 025-migrate-services-to-handler-dag 019-add-handler-status-index -> 026-add-handler-status-index 020-backfill-handler-status -> 027-backfill-handler-status Class names are unchanged (mongo-migrate-ts tracks applied migrations by className), and these migrations have not been applied to any shared environment (int/prod migrations come from main only), so the renumber is safe. Also updates required-migrations.ts, the internal [NNN] log prefixes, and the db-migrations doc table. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
|
Original author: @github-actions[bot] Test Results (Node.js 22)test: Run #2012
🎉 All tests passed! |
|
Original author: @github-actions[bot] Test Results (Node.js 22)test: Run #2029
🎉 All tests passed! |
|
Original author: @github-actions[bot] Test Results (Node.js 22)test: Run #2030
🎉 All tests passed! |
|
Original author: @github-actions[bot] Test Results (Node.js 22)test: Run #2103
🎉 All tests passed! |
|
Original author: @github-actions[bot] Test Results (Node.js 22)test: Run #2112
🎉 All tests passed! |
|
Original author: @github-actions[bot] Test Results (Node.js 22)test: Run #2192
🎉 All tests passed! |
|
Original author: Wassim Chegham (@manekinekko) Original review: COMMENTED |
|
Original author: Josh Duffney (@duffney) let me review the comment before posting |
|
Original author: Josh Duffney (@duffney) Original review: APPROVED |
|
Original author: Wassim Chegham (@manekinekko) This This is a regression from the legacy Suggest dropping |
Resolve post-processing telemetry and migration sequence conflicts. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: a4e92f08-5d86-4664-b8de-1b99bcccc1cf
Test Results (Node.js 22)test: Run #230
🎉 All tests passed! |
…ricvidal-microsoft/pp-dag-orchestration
Test Results (Node.js 22)test: Run #237
🎉 All tests passed! |
Restore the taxonomy-specific layer after fast-forwarding to PR #1147, reconcile current migrations and OpenAPI artifacts, and adapt the worker to the current Copilot SDK permission contract. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: addea2af-c951-4473-90bc-0a7356c29862
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 247cbe24-e22c-4f4c-b3e4-e71ce03f4669
Test Results (Node.js 22)test: Run #277
🎉 All tests passed! |
Original author: Cedric Vidal (@cedricvidal)
Summary
Introduces a DAG-based post-processing pipeline: post-processors are modelled as nodes in a per-run dependency DAG, the scheduler dispatches each handler when its dependencies are satisfied, tracks per-handler status, and triggers report generation when the DAG drains. This is the foundation that lets the system scale to N independent post-processors per run. (The taxonomy post-processor that consumes this pipeline is split into a separate stacked PR, #1053.)
What's included
DAG-based handler orchestration (
apps/scheduler/,apps/workers/post-processor/,packages/shared/)HandlerDispatcher(apps/scheduler/src/handler-dispatcher.ts) walks a per-run handler DAG, enqueues each handler when its dependencies are satisfied, and tracks per-handler status inrun.handlerStatus[handlerId].apps/scheduler/src/notify-routes.ts):POST /notify/run-terminal,POST /notify/handler-complete, andPOST /handlers/register, plusGET /health.templateId), N-per-run, and trigger-gated, so they don't fit an idempotent per-run DAG node. The scheduler detects when the handler DAG has fully drained (every registered handler reached a terminal state:done/failed/blocked-by-failed-dependency) and callsPOST /api/v1/reports/trigger, which evaluates each template's trigger and enqueues{ reportId }. Exactly-once is guaranteed by an atomic, attempt-scopedrun.reportsTriggeredAtguard with rollback so the poll safety-net can retry.handler.yaml(apps/workers/post-processor/handler.yaml) registered through the scheduler'sPOST /handlers/registerendpoint at worker-registration time via the genericpackages/shared/src/scripts/register-handler.tsscript (with cycle check), replacing migration-baked registration and the old per-workerregister-versionscript.HandlerServiceDocument,HandlerRunStatus,run.handlerStatus, andrun.reportsTriggeredAt; the legacypostProcessorStatus/postProcessorVersionfields are retained (deprecated) for migration compatibility. Queue-processor updated to the handler-DAG status model.Migrations (
packages/db-migrations/)018-migrate-services-to-handler-dag— removes the legacy singlepost-processorservice row (registration is now worker-owned viahandler.yaml).019-add-handler-status-index— adds theidx_handler_dispatch_atifdispatch index for handler-DAG polling.020-backfill-handler-status— backfillsrun.handlerStatus.pp-atiffrom the deprecated flat fields.Deployment (
deploy/,docker-compose*.yml)Servicefor/handlers/register+ notify endpoints; aregister-handler-post-processorJob runs the generic registration script on each deploy. Docker Compose gains scheduler / post-processor /register-handler-post-processorservices wired to the new endpoints.Validation
pnpm buildacross all 27 workspace projects: green.npx vitest run packages/db-migrations apps/scheduler apps/portal packages/shared/src/schemas packages/shared/src/scripts apps/api: 59 files, 886 tests passing (scheduler dispatcher drain/exactly-once/rollback, handler registration + cycle check, queue-processor, db-migrations required-list check, OpenAPI snapshot regenerated taxonomy-free).git diff origin/main...HEADcontains zero taxonomy paths or content.Notes
docs/architecture/post-processing.md,overview.md,db-migrations.md,ENV_VARIABLES.md.