Skip to content

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

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

Cedric Vidal (cedricvidal) wants to merge 27 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

@cedricvidal

Copy link
Copy Markdown
Collaborator Author

Original author: @github-actions[bot]

Test Results (Node.js 22)

test: Run #2112

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

🎉 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 #2192

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

🎉 All tests passed!

Github Test Reporter

@cedricvidal

Copy link
Copy Markdown
Collaborator Author

Original author: Wassim Chegham (@manekinekko)

Original review: COMMENTED

@cedricvidal

Copy link
Copy Markdown
Collaborator Author

Original author: Josh Duffney (@duffney)

let me review the comment before posting

@cedricvidal

Copy link
Copy Markdown
Collaborator Author

Original author: Josh Duffney (@duffney)

Original review: APPROVED

@cedricvidal

Copy link
Copy Markdown
Collaborator Author

Original author: Wassim Chegham (@manekinekko)

This $nin excludes "done", which breaks autoBackfill version bumps. Any run that has a version is already done, so the { version: { $lt: handler.version } } branch below can never match — and since the poll path is the only backfill mechanism, previously-completed runs are never re-processed when a handler's version is bumped.

This is a regression from the legacy PostProcessorDispatcher, whose filter excluded only ["queued", "processing"] (not done), so version bumps did backfill historical runs. It also contradicts the documented behavior in post-processing.md ("Requests with missing or outdated handler state are re-dispatched ... automatically applied to historical data") and the migration 020 backfill rationale.

Suggest dropping "done" from the $nin when autoBackfill is true (mirroring the legacy filter) so the $lt version branch becomes reachable. Worth adding a test for the done-at-old-version case, which is currently uncovered.

@manekinekko Wassim Chegham (manekinekko) added area: infrastructure Cloud resources, hosting, networking, and shared infrastructure. area: post-processing ATIF generation and the run-output post-processing pipeline. area: scheduler 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. area: shared Shared package: types, DB models, and queue/blob/redis clients used by all apps. and removed author: cedricvidal labels Sep 25, 2026
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
@github-actions

Copy link
Copy Markdown

Test Results (Node.js 22)

test: Run #230

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

🎉 All tests passed!

Github Test Reporter

@cedricvidal
Cedric Vidal (cedricvidal) changed the base branch from main to feat/portal-copilot-ai October 12, 2026 15:48
@cedricvidal
Cedric Vidal (cedricvidal) added this pull request to stack #1485 October 12, 2026 15:49
@github-actions

Copy link
Copy Markdown

Test Results (Node.js 22)

test: Run #237

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

🎉 All tests passed!

Github Test Reporter

Cedric Vidal (cedricvidal) added a commit that referenced this pull request Oct 12, 2026
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
Cedric Vidal (cedricvidal) added a commit that referenced this pull request Oct 12, 2026
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 247cbe24-e22c-4f4c-b3e4-e71ce03f4669
@cedricvidal
Cedric Vidal (cedricvidal) marked this pull request as draft October 14, 2026 16:00
@cedricvidal
Cedric Vidal (cedricvidal) removed this pull request from stack #1485 October 17, 2026 00:57
@cedricvidal
Cedric Vidal (cedricvidal) added this pull request to stack #1518 October 17, 2026 00:58
@github-actions

Copy link
Copy Markdown

Test Results (Node.js 22)

test: Run #277

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

🎉 All tests passed!

Github Test Reporter

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