feat(ENG-016): Kappa-style consensus over implementation directions (Phase 1)

.env.example

@@ -32,3 +32,11 @@ GITHUB_DEPLOY_REF=main
 # Persistence backend for agent state (tracker + knowledge base): postgres (default) or github.
 PERSISTENCE_BACKEND=postgres
 GITHUB_DOCS_PREFIX=docs
+
+# Kappa-style consensus (ENG-016 / PRD-001). Off by default; trigger a round per epic via
+# POST /api/epics/:id/consensus. Phase 1 is advisory (records the round + agreement score, no gate).
+CONSENSUS_ENABLED=false
+# CONSENSUS_INPUT_MODE=rank        # "rank" (Borda + Kendall/Krippendorff) or "pick" (Fleiss)
+# CONSENSUS_THRESHOLD=0.6          # agreement gate (Landis–Koch "substantial")
+# CONSENSUS_ADVISORY=true          # false → promote to a real architecture_decision gate (Phase 2)
+# CONSENSUS_TIEBREAKER=human       # "human" (escalate to the gate) or "casting_vote"

apps/server/src/app.ts

@@ -4,8 +4,10 @@ import { type AppEnv, authMiddleware } from "./auth"
 import { persistenceFromEnv } from "./persistence"
 import { artifactPath, SHAPING_STAGES } from "./shaping"
 import {
+  approveArchitecture,
   approveRoadmap,
   approveTicket,
+  startDirectionConsensus,
   startEpicDecomposition,
   startEpicShaping,
   startTicketLifecycle,
@@ -105,6 +107,29 @@ app.get("/api/epics/:id/artifacts", async (c) => {
   return c.json({ artifacts })
 })
 
+// Kappa-style consensus (ENG-016): the senior technical roles independently rate candidate
+// implementation directions; the agreement coefficient gates adopt-vs-tie-break.
+app.post("/api/epics/:id/consensus", async (c) => {
+  const id = c.req.param("id")
+  await startDirectionConsensus(id)
+  return c.json({ epicId: id, consensus: true })
+})
+
+// Architecture-decision tie-break (invariant #4): release the gate so the consensus round resolves.
+app.post("/api/epics/:id/approve-architecture", async (c) => {
+  const id = c.req.param("id")
+  const signaled = await approveArchitecture(id)
+  const by = c.get("actor")
+  await resolveApproval({ kind: "architecture_decision", epicId: id, decidedBy: by })
+  await persistence.audit.append({
+    actor: "human",
+    kind: "approval_decided",
+    ticketId: null,
+    payload: { epicId: id, gate: "architecture", by, signaled },
+  })
+  return c.json({ epicId: id, gate: "architecture", signaled, by })
+})
+
 // Agent-driven decomposition: the Lead Engineer breaks the epic into backlog tickets — but behind a
 // roadmap sign-off gate (the workflow blocks until the human approves the plan, below).
 app.post("/api/epics/:id/decompose", async (c) => {

apps/server/src/consensus.ts

@@ -0,0 +1,30 @@
+import { type ConsensusConfig, DEFAULT_CONSENSUS_CONFIG } from "@eng/core"
+
+/**
+ * Per-unit consensus configuration (ENG-016), resolved from env so the rubric / mode / threshold /
+ * tie-breaker are config, not code (invariant #6). Phase 1 ships **advisory** (record the round + score
+ * but don't block) and behind `CONSENSUS_ENABLED`.
+ */
+export function consensusEnabled(): boolean {
+  return process.env.CONSENSUS_ENABLED === "true"
+}
+
+export function consensusConfigFromEnv(): ConsensusConfig {
+  return {
+    ...DEFAULT_CONSENSUS_CONFIG,
+    inputMode:
+      process.env.CONSENSUS_INPUT_MODE === "pick"
+        ? "pick"
+        : process.env.CONSENSUS_INPUT_MODE === "rank"
+          ? "rank"
+          : DEFAULT_CONSENSUS_CONFIG.inputMode,
+    threshold: process.env.CONSENSUS_THRESHOLD
+      ? Number(process.env.CONSENSUS_THRESHOLD)
+      : DEFAULT_CONSENSUS_CONFIG.threshold,
+    // Advisory by default (Phase 1); set CONSENSUS_ADVISORY=false to promote to a real gate (Phase 2).
+    advisory: process.env.CONSENSUS_ADVISORY
+      ? process.env.CONSENSUS_ADVISORY !== "false"
+      : DEFAULT_CONSENSUS_CONFIG.advisory,
+    tieBreaker: process.env.CONSENSUS_TIEBREAKER === "casting_vote" ? "casting_vote" : "human",
+  }
+}

apps/server/src/temporal/activities.ts

@@ -3,15 +3,23 @@ import {
   draft,
   ensureRepoCloned,
   estimateRunCostCents,
+  proposeDirections,
   proposeFileChanges,
   proposeTickets,
+  rateDirections,
   repoWorkspacePath,
 } from "@eng/agents"
 import {
+  type ConsensusCandidate,
+  type ConsensusConfig,
+  type ConsensusInputMode,
   type DeliveryAdapter,
   type DeployState,
+  decide,
   type PullRequestRef,
+  type Rating,
   ROLES,
+  type RoleId,
   type TicketStatus,
 } from "@eng/core"
 import { createApproval, getBudgetRemaining, reconcileSpend, reserveBudget } from "@eng/db"
@@ -20,6 +28,7 @@ import {
   LocalGitDeliveryAdapter,
   localGitBranchFiles,
 } from "@eng/integrations"
+import { consensusConfigFromEnv } from "../consensus"
 import { persistenceFromEnv } from "../persistence"
 import { artifactPath, SHAPING_STAGES } from "../shaping"
 import { startTicketLifecycle } from "./client"
@@ -661,3 +670,197 @@ export async function recordDeploy(ticketId: string, state: DeployState): Promis
     payload: { state },
   })
 }
+
+// --- Kappa-style consensus (ENG-016 / PRD-001) -------------------------------
+
+/** Candidates + the resolved per-unit config — the workflow needs both to fan out the raters. */
+export interface ConsensusKickoff {
+  candidates: ConsensusCandidate[]
+  config: ConsensusConfig
+}
+
+/**
+ * Generate 2–4 distinct candidate implementation directions (Lead System Design), informed by the
+ * epic's shaping artifacts. Budget-enforced + audited; returns null (no-op) when the runtime is
+ * unavailable or the budget is exhausted, so the lifecycle proceeds normally.
+ */
+export async function generateDirections(epicId: string): Promise<ConsensusKickoff | null> {
+  const config = consensusConfigFromEnv()
+  const role = ROLES.lead_system_design
+  const remaining = (await getBudgetRemaining(role.id)) ?? role.monthlyBudgetCents
+  const held = await reserveBudget(role.id, estimateRunCostCents(remaining))
+  if (held === null) {
+    await persistence.audit.append({
+      actor: "system",
+      kind: "consensus_skipped",
+      ticketId: null,
+      payload: { epicId, stage: "candidates", reason: "budget_exhausted" },
+    })
+    return null
+  }
+  const [epicCtx, artifacts] = await Promise.all([
+    persistence.hierarchy.epicContext(epicId),
+    epicArtifacts(epicId),
+  ])
+  try {
+    const proposed = await proposeDirections({
+      role: role.id,
+      systemPrompt: role.systemPrompt,
+      goalContext: [epicCtx, ...artifacts].join("\n\n"),
+      task: "Enumerate the candidate implementation directions for this epic's core feature.",
+      budgetCentsRemaining: remaining,
+    })
+    await reconcileSpend(role.id, held, proposed.costCents)
+    const candidates: ConsensusCandidate[] = proposed.candidates.map((c, i) => ({
+      id: `c${i + 1}`,
+      title: c.title,
+      summary: c.summary,
+      tradeoffs: c.tradeoffs,
+    }))
+    await persistence.audit.append({
+      actor: role.id,
+      kind: "consensus_candidates",
+      ticketId: null,
+      payload: { epicId, candidates: candidates.length, costCents: proposed.costCents },
+    })
+    return { candidates, config }
+  } catch (err) {
+    await reconcileSpend(role.id, held, 0)
+    await persistence.audit.append({
+      actor: "system",
+      kind: "consensus_skipped",
+      ticketId: null,
+      payload: { epicId, stage: "candidates", error: String(err) },
+    })
+    return null
+  }
+}
+
+/**
+ * One rater scores/ranks the candidates **independently** (no rater sees another's input — that's
+ * what makes the agreement coefficient meaningful). Budget-enforced per role; audited. Returns null
+ * when the runtime is unavailable so `decide` simply has one fewer rating.
+ */
+export async function rateDirection(
+  epicId: string,
+  candidates: ConsensusCandidate[],
+  raterRole: RoleId,
+  inputMode: ConsensusInputMode,
+): Promise<Rating | null> {
+  const role = ROLES[raterRole]
+  const remaining = (await getBudgetRemaining(role.id)) ?? role.monthlyBudgetCents
+  const held = await reserveBudget(role.id, estimateRunCostCents(remaining))
+  if (held === null) {
+    await persistence.audit.append({
+      actor: "system",
+      kind: "consensus_skipped",
+      ticketId: null,
+      payload: { epicId, stage: "rating", raterRole, reason: "budget_exhausted" },
+    })
+    return null
+  }
+  try {
+    const epicCtx = await persistence.hierarchy.epicContext(epicId)
+    const { rating, costCents } = await rateDirections({
+      role: role.id,
+      systemPrompt: role.systemPrompt,
+      goalContext: epicCtx,
+      candidates: candidates.map((c) => ({ id: c.id, title: c.title, summary: c.summary })),
+      criteria: consensusConfigFromEnv().criteria,
+      inputMode,
+      budgetCentsRemaining: remaining,
+    })
+    await reconcileSpend(role.id, held, costCents)
+    await persistence.audit.append({
+      actor: role.id,
+      kind: "consensus_rating",
+      ticketId: null,
+      payload: { epicId, pick: rating.pick, ranking: rating.ranking, costCents },
+    })
+    return rating
+  } catch (err) {
+    await reconcileSpend(role.id, held, 0)
+    await persistence.audit.append({
+      actor: "system",
+      kind: "consensus_skipped",
+      ticketId: null,
+      payload: { epicId, stage: "rating", raterRole, error: String(err) },
+    })
+    return null
+  }
+}
+
+/**
+ * Aggregate the ratings, compute the agreement coefficient, and decide. Persists the outcome to the
+ * audit log and the decision graph (ENG-014), traceable to the originating request. Returns whether
+ * consensus was reached (the workflow branches on it for the tie-breaker).
+ */
+export async function decideConsensus(
+  epicId: string,
+  candidates: ConsensusCandidate[],
+  ratings: (Rating | null)[],
+): Promise<{ consensusReached: boolean; winner: string | null }> {
+  const config = consensusConfigFromEnv()
+  const valid = ratings.filter((r): r is Rating => r !== null)
+  const decision = decide(candidates, valid, config)
+  const winnerCand = candidates.find((c) => c.id === decision.winner) ?? null
+  const totalCost = valid.reduce((s, r) => s + (r.costCents ?? 0), 0)
+
+  const event = await persistence.audit.append({
+    actor: "lead_architect",
+    kind: "consensus_decided",
+    ticketId: null,
+    payload: {
+      epicId,
+      winner: decision.winner,
+      consensusReached: decision.consensusReached,
+      coefficient: decision.coefficient,
+      metric: decision.metric,
+      method: decision.aggregateMethod,
+      threshold: config.threshold,
+      advisory: config.advisory,
+      raters: valid.length,
+    },
+  })
+  await recordStepDecision({
+    epicId,
+    actor: "lead_architect",
+    stage: "architecture",
+    statement: winnerCand
+      ? `Consensus direction: ${winnerCand.title}`
+      : "No clear consensus direction",
+    rationale:
+      `${decision.aggregateMethod} winner from ${valid.length} independent raters; agreement ` +
+      `${decision.metric}=${decision.coefficient.toFixed(2)} vs threshold ${config.threshold} → ` +
+      `${decision.consensusReached ? "consensus reached" : "tie-break required"}`,
+    alternatives: candidates
+      .filter((c) => c.id !== decision.winner)
+      .map((c) => ({ option: c.title, rejectedBecause: "not the consensus winner" })),
+    confidence: Math.max(0, Math.min(1, decision.coefficient)),
+    costCents: totalCost,
+    auditEventId: event.id,
+  })
+  return { consensusReached: decision.consensusReached, winner: decision.winner }
+}
+
+/** Record that an epic's implementation direction awaits the human architecture sign-off (tie-break).
+ *  Creates a first-class pending approval (ENG-006) plus the audit event. */
+export async function requestArchitectureApproval(epicId: string): Promise<void> {
+  await createApproval({ kind: "architecture_decision", epicId, requestedByRole: "lead_architect" })
+  await persistence.audit.append({
+    actor: "lead_architect",
+    kind: "architecture_requested",
+    ticketId: null,
+    payload: { epicId },
+  })
+}
+
+/** Record the human architecture sign-off (the tie-break gate released the consensus). */
+export async function recordArchitectureApproval(epicId: string): Promise<void> {
+  await persistence.audit.append({
+    actor: "human",
+    kind: "architecture_approved",
+    ticketId: null,
+    payload: { epicId },
+  })
+}

apps/server/src/temporal/client.ts

@@ -64,6 +64,34 @@ export async function startEpicDecomposition(epicId: string): Promise<void> {
   }
 }
 
+/** Start Kappa-style direction consensus for an epic (idempotent while one is already running). */
+export async function startDirectionConsensus(epicId: string): Promise<void> {
+  const client = await getTemporalClient()
+  try {
+    await client.workflow.start("directionConsensus", {
+      taskQueue: TASK_QUEUE,
+      workflowId: `epic-consensus-${epicId}`,
+      args: [epicId],
+    })
+  } catch (err) {
+    if ((err as { name?: string })?.name === "WorkflowExecutionAlreadyStartedError") return
+    throw err
+  }
+}
+
+/** Release the architecture-decision tie-break gate for an epic's consensus round (ENG-016). Returns
+ *  false if no consensus round is waiting (advisory, already resolved, or never started). */
+export async function approveArchitecture(epicId: string): Promise<boolean> {
+  const client = await getTemporalClient()
+  const handle = client.workflow.getHandle(`epic-consensus-${epicId}`)
+  try {
+    await handle.signal("architectureApprove")
+    return true
+  } catch {
+    return false
+  }
+}
+
 /** Release the roadmap sign-off gate for an epic. Returns false if no decomposition is waiting
  *  (already decomposed, or never started) — a benign no-op for the UI. */
 export async function approveRoadmap(epicId: string): Promise<boolean> {