Skip to content

feat: DAG-based post-processing pipeline - #1147

Draft
Cedric Vidal (cedricvidal) wants to merge 36 commits into
feat/portal-copilot-aifrom
cedricvidal-microsoft/pp-dag-orchestration
Draft

Cedric Vidal (cedricvidal) wants to merge 36 commits into
feat/portal-copilot-aifrom
cedricvidal-microsoft/pp-dag-orchestration

Conversation

@cedricvidal

Copy link
Copy Markdown
Collaborator

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/)

  • New 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 in run.handlerStatus[handlerId].
  • New scheduler notify/registration HTTP server (apps/scheduler/src/notify-routes.ts): POST /notify/run-terminal, POST /notify/handler-complete, and POST /handlers/register, plus GET /health.
  • Report generation is taken off the DAG. Reports are parameterized (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 calls POST /api/v1/reports/trigger, which evaluates each template's trigger and enqueues { reportId }. Exactly-once is guaranteed by an atomic, attempt-scoped run.reportsTriggeredAt guard with rollback so the poll safety-net can retry.
  • YAML-driven handler registration: handlers declare themselves via a handler.yaml (apps/workers/post-processor/handler.yaml) registered through the scheduler's POST /handlers/register endpoint at worker-registration time via the generic packages/shared/src/scripts/register-handler.ts script (with cycle check), replacing migration-baked registration and the old per-worker register-version script.
  • Shared types gain HandlerServiceDocument, HandlerRunStatus, run.handlerStatus, and run.reportsTriggeredAt; the legacy postProcessorStatus/postProcessorVersion fields 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 single post-processor service row (registration is now worker-owned via handler.yaml).
  • 019-add-handler-status-index — adds the idx_handler_dispatch_atif dispatch index for handler-DAG polling.
  • 020-backfill-handler-status — backfills run.handlerStatus.pp-atif from the deprecated flat fields.

Deployment (deploy/, docker-compose*.yml)

  • Scheduler manifest exposes a Service for /handlers/register + notify endpoints; a register-handler-post-processor Job runs the generic registration script on each deploy. Docker Compose gains scheduler / post-processor / register-handler-post-processor services wired to the new endpoints.

Validation

  • pnpm build across 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...HEAD contains zero taxonomy paths or content.

Notes

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

# 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>
`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>
@cedricvidal

Copy link
Copy Markdown
Collaborator Author

Original author: @github-actions[bot]

Test Results (Node.js 22)

test: Run #2012

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
2213 2213 0 0 0 0 0 2m28s

🎉 All tests passed!

Github Test Reporter

@cedricvidal

Copy link
Copy Markdown
Collaborator Author

Original author: @github-actions[bot]

Test Results (Node.js 22)

test: Run #2029

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
2271 2271 0 0 0 0 0 2m25s

🎉 All tests passed!

Github Test Reporter

@cedricvidal

Copy link
Copy Markdown
Collaborator Author

Original author: @github-actions[bot]

Test Results (Node.js 22)

test: Run #2030

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
2271 2271 0 0 0 0 0 2m20s

🎉 All tests passed!

Github Test Reporter

@cedricvidal

Copy link
Copy Markdown
Collaborator Author

Original author: @github-actions[bot]

Test Results (Node.js 22)

test: Run #2103

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
2436 2436 0 0 0 0 0 2m16s

🎉 All tests passed!

Github Test Reporter

@github-actions

Copy link
Copy Markdown

Test Results (Node.js 22)

test: Run #320

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
3289 3289 0 0 0 0 0 1m22s

🎉 All tests passed!

Github Test Reporter

Carry the common readability guidance and Claude loader forward without changing DAG pipeline code.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: a4e92f08-5d86-4664-b8de-1b99bcccc1cf
@github-actions

Copy link
Copy Markdown

Test Results (Node.js 22)

test: Run #334

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
3289 3289 0 0 0 0 0 1m1s

🎉 All tests passed!

Github Test Reporter

@github-actions

Copy link
Copy Markdown

Test Results (Node.js 22)

test: Run #347

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
3289 3289 0 0 0 0 0 1m4s

🎉 All tests passed!

Github Test Reporter

@github-actions

Copy link
Copy Markdown

Test Results (Node.js 22)

test: Run #428

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
3289 3289 0 0 0 0 0 59.1s

🎉 All tests passed!

Github Test Reporter

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>
Resolve the native-stack add/add conflicts by preserving the canonical generated license headers.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
@github-actions github-actions Bot added area: cicd Build, test, release, and deployment pipelines. type: dependencies Dependency additions, removals, and version updates. topic: security Security protections, vulnerabilities, and risk reduction. topic: ui Portal, visual presentation, layout, and interface components. labels Nov 4, 2026
@github-actions

github-actions Bot commented Nov 4, 2026

Copy link
Copy Markdown

Test Results (Node.js 22)

test: Run #464

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
3289 3289 0 0 0 0 0 1m13s

🎉 All tests passed!

Github Test Reporter

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
@github-actions github-actions Bot removed area: cicd Build, test, release, and deployment pipelines. type: dependencies Dependency additions, removals, and version updates. topic: security Security protections, vulnerabilities, and risk reduction. topic: ui Portal, visual presentation, layout, and interface components. labels Nov 5, 2026
@github-actions

github-actions Bot commented Nov 5, 2026

Copy link
Copy Markdown

Test Results (Node.js 22)

test: Run #506

Tests 📝 Passed ✅ Failed ❌ Skipped ⏭️ Pending ⏳ Other ❓ Flaky 🍂 Duration ⏱️
3316 3316 0 0 0 0 0 1m14s

🎉 All tests passed!

Github Test Reporter

Cedric Vidal (cedricvidal) added a commit that referenced this pull request Nov 5, 2026
Refresh stacked checks against parent #1147 after its latest-main CI cycle. Preserve the committed standalone foundation and parent changes without including unpublished local documentation.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area: infrastructure Cloud resources, hosting, networking, and shared infrastructure. area: post-processing ATIF generation and the run-output post-processing pipeline. area: scheduler area: shared Shared package: types, DB models, and queue/blob/redis clients used by all apps. area: worker Coding agent worker runtime, task execution, and lifecycle. language: javascript Work involving JavaScript code, tooling, or dependencies. topic: testing Test coverage, test infrastructure, and validation quality. type: documentation Documentation additions, corrections, and improvements. type: enhancement New functionality or improvements to existing behavior.

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants