pipeline-operational-loop

Pipeline Operational Loop

Standard template for evaluating any change to the real-time chat pipeline or the report generator pipeline.

Canonical spec: docs/architecture/PIPELINE_OPERATIONAL_STANDARD.md Rule mirror: .claude/rules/pipeline_operational_standard.md

Step 0 — Diagnose first (analyst mindset)

Before writing code, answer:

• Which pipeline layer changed? (args, emit, judge, persistence, UI, dashboard)
• Does it change the contract (EmitProjectionArgs or RunTelemetry)?
• Does it change a judge gate? (if yes → dashboards depend on GATE_ORDER)
• What's the failure mode this change prevents or creates?

Step 1 — Contract

• Update EmitProjectionArgs (server/pipeline/diligenceProjectionWriter.ts) if needed.
• Mirror the shape in convex/schema.ts → diligenceProjections / diligenceRunTelemetry.
• Mirror in Convex args validators (recordTelemetry, recordVerdict).

Step 2 — Instrument

Every emit goes through emitDiligenceProjectionInstrumented(...):

await emitDiligenceProjectionInstrumented(mutationCaller, args, {
  seedTelemetry: { toolCalls, tokensIn, tokensOut, sourceCount },
  onTelemetry: async (t, a) => {
    const { id } = await convex.mutation(api.domains.product.diligenceRunTelemetry.recordTelemetry, { ...mapTelemetry(a, t) });
    const verdict = judgeDiligenceRun({ args: a, telemetry: t });
    await convex.mutation(api.domains.product.diligenceJudge.recordVerdict, {
      telemetryId: id,
      ...verdict,
      gatesJson: JSON.stringify(verdict.gates),
    });
  },
});

Step 3 — Judge

judgeDiligenceRun is a pure deterministic function. Don't call LLMs here. Don't read wall-clock. Don't randomize.

Add a new gate only by:

1. Extending the GateName union in server/pipeline/diligenceJudge.ts.
2. Appending to GATE_ORDER (never reshuffle — dashboards depend on it).
3. Adding the evaluator fn.
4. Adding a row to §3 of PIPELINE_OPERATIONAL_STANDARD.md.
5. Adding scenario tests in diligenceJudge.test.ts covering pass, fail, and skip.