fix(aggregator): orchestration provenance contract — accept the publisher's object form (#1110, #1140)

pm-stack/aggregator/scripts/collect-fleet-feed.ts

@@ -43,7 +43,7 @@ import { homedir } from "node:os";
 import { dirname, join } from "node:path";
 import { fileURLToPath } from "node:url";
 import { dnsSectionFromRegister } from "../src/dns.ts";
-import { isOrchestrationSection, type FleetFeed } from "../src/feed.ts";
+import { normalizeOrchestrationSection, orchestrationSectionProblem, type FleetFeed } from "../src/feed.ts";
 import type {
   DnsSection,
   JulesSummary,
@@ -83,8 +83,7 @@ const dnsRegisterPath =
   join(HOME, "workspace/goals/stand-up-lab-cluster-talm-talos-cozystack-postgres-rustfs/files/dns-domain-register.md");
 const launchRecordsPath = process.env.CLAUDE_LAUNCH_RECORDS_PATH ?? join(HIDDEN, "claude-launch-records.jsonl");
 const quotaRegistryPath = process.env.QUOTA_REGISTRY_PATH ?? join(HIDDEN, "provider-quota-registry.md");
-const snapshotOutboxPath =
-  process.env.USAGE_SNAPSHOT_OUTBOX_PATH ?? join(HIDDEN, "usage-snapshot-outbox.jsonl");
+const snapshotOutboxPath = process.env.USAGE_SNAPSHOT_OUTBOX_PATH ?? join(HIDDEN, "usage-snapshot-outbox.jsonl");
 const serverUrl = process.env.SANDBOX_OPENCODE_URL ?? "http://127.0.0.1:4096";
 const passwordFile =
   process.env.OPENCODE_SERVER_PASSWORD_FILE ?? join(HOME, "tools/secrets.d/opencode-server-password");
@@ -187,8 +186,7 @@ async function resolveProjectToken(): Promise<string | null> {
   const fromEnv = process.env.GITHUB_PROJECT_TOKEN ?? process.env.GH_TOKEN;
   if (fromEnv && fromEnv.trim()) return fromEnv.trim();
   const bwsBin = process.env.BWS_BIN ?? join(HOME, "tools/bin/bws");
-  const accessTokenFile =
-    process.env.BWS_ACCESS_TOKEN_FILE ?? join(HOME, ".config/bitwarden/access-token");
+  const accessTokenFile = process.env.BWS_ACCESS_TOKEN_FILE ?? join(HOME, ".config/bitwarden/access-token");
   try {
     await access(bwsBin);
     const accessToken = (await readFile(accessTokenFile, "utf8")).trim();
@@ -325,10 +323,16 @@ async function main(): Promise<void> {
   let orchestration: OrchestrationSection | undefined;
   try {
     const parsed: unknown = JSON.parse(await readFile(orchestrationPath, "utf8"));
-    if (isOrchestrationSection(parsed)) {
-      orchestration = parsed;
+    // Embed the normalized form (#1110): the feed carries the section
+    // in the one canonical shape the aggregator serves, whatever
+    // provenance form the publisher emitted.
+    const normalized = normalizeOrchestrationSection(parsed);
+    if (normalized) {
+      orchestration = normalized;
     } else {
-      console.error("collector: orchestration.json failed its structural check; section omitted");
+      console.error(
+        `collector: orchestration.json failed its structural check (${orchestrationSectionProblem(parsed)}); section omitted`,
+      );
     }
   } catch {
     // No orchestration file yet (the job has not run) is the normal
@@ -352,7 +356,9 @@ async function main(): Promise<void> {
       }
     }
   } catch (err) {
-    console.error(`collector: GitHub Project read failed (${err instanceof Error ? err.message : "error"}); projects section omitted`);
+    console.error(
+      `collector: GitHub Project read failed (${err instanceof Error ? err.message : "error"}); projects section omitted`,
+    );
     projects = undefined;
   }
 

pm-stack/aggregator/src/feed.ts

@@ -9,6 +9,7 @@ import type {
   DnsSection,
   GoalMeasurement,
   JulesSummary,
+  OrchestrationProvenance,
   OrchestrationSection,
   PaygLine,
   ProjectPoolItem,
@@ -82,25 +83,68 @@ function isGoalMeasurement(v: unknown): boolean {
   );
 }
 
-export function isOrchestrationSection(v: unknown): v is OrchestrationSection {
+/**
+ * The section's provenance (#1110). The contract is the publisher's
+ * object form — `{ sources: string[], runId?: string }` — which every
+ * orchestration run actually emits. The legacy plain-string form is
+ * also accepted here and normalized at the boundary (see
+ * normalizeOrchestrationSection), the same string-or-richer tolerance
+ * derive.ts gives `progressedSinceLastRun`. Anything else — a number,
+ * an object without a sources array, non-string sources — fails.
+ */
+function isProvenance(v: unknown): boolean {
+  if (typeof v === "string") return v.length > 0;
   if (typeof v !== "object" || v === null) return false;
-  const s = v as Partial<OrchestrationSection>;
+  const p = v as { sources?: unknown; runId?: unknown };
   return (
-    s.version === 1 &&
-    typeof s.generatedAt === "string" &&
-    typeof s.runId === "string" &&
-    typeof s.provenance === "string" &&
-    typeof s.tiers === "object" &&
-    s.tiers !== null &&
-    Array.isArray(s.tiers.needsDrew) &&
-    Array.isArray(s.tiers.needsOrchestrator) &&
-    Array.isArray(s.dependencies) &&
-    Array.isArray(s.goals) &&
-    s.goals.every(isGoalMeasurement) &&
-    Array.isArray(s.actions)
+    Array.isArray(p.sources) &&
+    p.sources.every((x) => typeof x === "string") &&
+    (p.runId === undefined || typeof p.runId === "string")
   );
 }
 
+/**
+ * The first structural check an orchestration section fails, as a short
+ * field name — or null when the section is valid (#1140: a present-but-
+ * invalid section must be diagnosable, not silently "absent"). This is
+ * the single source of truth behind isOrchestrationSection, so the
+ * diagnosis can never drift from the gate itself.
+ */
+export function orchestrationSectionProblem(v: unknown): string | null {
+  if (typeof v !== "object" || v === null) return "not an object";
+  const s = v as Partial<OrchestrationSection>;
+  if (s.version !== 1) return "version";
+  if (typeof s.generatedAt !== "string") return "generatedAt";
+  if (typeof s.runId !== "string") return "runId";
+  if (!isProvenance(s.provenance)) return "provenance";
+  if (typeof s.tiers !== "object" || s.tiers === null) return "tiers";
+  if (!Array.isArray(s.tiers.needsDrew)) return "tiers.needsDrew";
+  if (!Array.isArray(s.tiers.needsOrchestrator)) return "tiers.needsOrchestrator";
+  if (!Array.isArray(s.dependencies)) return "dependencies";
+  if (!Array.isArray(s.goals)) return "goals";
+  if (!s.goals.every(isGoalMeasurement)) return "goals (per-goal shape)";
+  if (!Array.isArray(s.actions)) return "actions";
+  return null;
+}
+
+export function isOrchestrationSection(v: unknown): v is OrchestrationSection {
+  return orchestrationSectionProblem(v) === null;
+}
+
+/**
+ * A valid section with its provenance in the canonical object form:
+ * a legacy plain-string provenance becomes `{ sources: [it] }`.
+ * Returns null for anything isOrchestrationSection rejects. The
+ * collector embeds this normalized form in the feed and the live
+ * source serves it, so downstream consumers see exactly one shape.
+ */
+export function normalizeOrchestrationSection(v: unknown): OrchestrationSection | null {
+  if (!isOrchestrationSection(v)) return null;
+  const s = v as OrchestrationSection & { provenance: OrchestrationProvenance | string };
+  if (typeof s.provenance !== "string") return s;
+  return { ...s, provenance: { sources: [s.provenance] } };
+}
+
 export function isFleetFeed(v: unknown): v is FleetFeed {
   if (typeof v !== "object" || v === null) return false;
   const f = v as Partial<FleetFeed>;

pm-stack/aggregator/src/model.ts

@@ -446,14 +446,30 @@ export interface GoalMeasurement {
 }
 
 /** The whole orchestration section, as embedded in the fleet feed. */
+/**
+ * The orchestration section's own provenance (#1110): the sources the
+ * producing run read, as the orchestration job publishes them. The
+ * object form is the contract — the publisher has always emitted it,
+ * and it is strictly richer than a joined string. The aggregator's
+ * feed boundary also accepts the legacy plain-string form older runs
+ * published and normalizes it to this shape (see feed.ts), so every
+ * consumer sees exactly this type.
+ */
+export interface OrchestrationProvenance {
+  /** One human-readable note per source the run read. */
+  sources: string[];
+  /** The producing run, when the publisher records it here too. */
+  runId?: string;
+}
+
 export interface OrchestrationSection {
   version: 1;
   /** ISO timestamp of the orchestrator run that produced this section. */
   generatedAt: string;
   /** "orch-<ISO>" — the producing run. */
   runId: string;
-  /** Sources + run, human-readable (the section's own provenance). */
-  provenance: string;
+  /** What the producing run read (the section's own provenance). */
+  provenance: OrchestrationProvenance;
   tiers: { needsDrew: TierItem[]; needsOrchestrator: TierItem[] };
   /** Flat cross-item dependency view for one-place rendering. */
   dependencies: (DependencyRef & { taskRef?: RefKey })[];

pm-stack/aggregator/src/sources/live.ts

@@ -7,7 +7,13 @@ import { mkdir, readFile, rename, writeFile } from "node:fs/promises";
 import { dirname } from "node:path";
 import type { Config } from "../config.ts";
 import { isDnsSection } from "../dns.ts";
-import { feedAgeText, isOrchestrationSection, loadFeed, type FleetFeed } from "../feed.ts";
+import {
+  feedAgeText,
+  normalizeOrchestrationSection,
+  orchestrationSectionProblem,
+  loadFeed,
+  type FleetFeed,
+} from "../feed.ts";
 import { isSessionRecord, mergeRecords, recordFromSessionEntry } from "../registry.ts";
 import { buildAttentionFlags } from "../derive.ts";
 import { summarizeDns } from "../dns.ts";
@@ -397,7 +403,11 @@ const slugify = (s: string): string =>
     .replace(/[^a-z0-9]+/g, "-")
     .replace(/^-+|-+$/g, "")
     .slice(0, 60) || "task";
-const normTitle = (s: string): string => s.toLowerCase().replace(/[^a-z0-9]+/g, " ").trim();
+const normTitle = (s: string): string =>
+  s
+    .toLowerCase()
+    .replace(/[^a-z0-9]+/g, " ")
+    .trim();
 
 /**
  * The synthetic in-flight entries for one orchestration section,
@@ -584,10 +594,11 @@ export function liveSource(cfg: Config): Source {
           // fails soft: no feed / no section = the Kaneo board alone.
           try {
             const feed = await getFeed();
-            if (feed?.orchestration && isOrchestrationSection(feed.orchestration)) {
-              const inFlight = inFlightBoardItems(feed.orchestration, merged);
+            const orchSection = feed?.orchestration ? normalizeOrchestrationSection(feed.orchestration) : null;
+            if (orchSection) {
+              const inFlight = inFlightBoardItems(orchSection, merged);
               if (inFlight.length) {
-                provenance.board += ` + orchestrator in-flight (${inFlight.length} lined-up, section ${feed.orchestration.generatedAt})`;
+                provenance.board += ` + orchestrator in-flight (${inFlight.length} lined-up, section ${orchSection.generatedAt})`;
                 merged = [...inFlight, ...merged];
               }
             }
@@ -648,7 +659,9 @@ export function liveSource(cfg: Config): Source {
       if (poolServed > 0 && feed) {
         provenance.items =
           `live (project pool · fleet feed · ${feedAgeText(feed.generatedAt)})` +
-          (remaining.length ? ` + ${projectorServed ? "projector" : "fixtures"} for ${remaining.length} ref(s) outside the pool` : "");
+          (remaining.length
+            ? ` + ${projectorServed ? "projector" : "fixtures"} for ${remaining.length} ref(s) outside the pool`
+            : "");
       } else if (projectorServed) {
         provenance.items = "live (projector)";
       }
@@ -787,10 +800,21 @@ export function liveSource(cfg: Config): Source {
       // #899: the orchestration section rides the fleet feed, embedded
       // by the collector from the orchestrator's own output file. The
       // section carries its own provenance; absent = not measured yet.
+      // #1140: three states, told apart — live (accepted, served in
+      // the normalized contract shape, #1110), invalid (a section is
+      // present but failed validation: name the failed check, never
+      // claim the job "has not published"), absent (no section key).
       const feed = await getFeed();
-      if (feed?.orchestration && isOrchestrationSection(feed.orchestration)) {
-        provenance.orchestration = `live (orchestration section ${feed.orchestration.runId} · ${feedAgeText(feed.orchestration.generatedAt)})`;
-        return feed.orchestration;
+      if (feed?.orchestration) {
+        const section = normalizeOrchestrationSection(feed.orchestration);
+        if (section) {
+          provenance.orchestration = `live (orchestration section ${section.runId} · ${feedAgeText(section.generatedAt)})`;
+          return section;
+        }
+        const problem = orchestrationSectionProblem(feed.orchestration);
+        console.error(`aggregator: orchestration section present but invalid (${problem}); the today view derives`);
+        provenance.orchestration = `invalid — orchestration section present but failed validation (${problem}); the today view derives`;
+        return null;
       }
       provenance.orchestration = feed
         ? "none yet — the orchestration job (#899) has not published a section; the today view derives"

pm-stack/aggregator/test/board-in-flight.test.ts

@@ -10,16 +10,28 @@ import { inFlightBoardItems, titleRefs } from "../src/sources/live.ts";
 
 const GENERATED = "2026-10-26T16:15:23Z";
 
-function section(linedUp: OrchestrationSection["goals"][number]["linedUp"], goal = "Keep the four repos healthy"): OrchestrationSection {
+function section(
+  linedUp: OrchestrationSection["goals"][number]["linedUp"],
+  goal = "Keep the four repos healthy",
+): OrchestrationSection {
   return {
     version: 1,
     generatedAt: GENERATED,
     runId: `orch-${GENERATED}`,
-    provenance: "test section",
+    provenance: { sources: ["test section"] },
     tiers: { needsDrew: [], needsOrchestrator: [] },
     dependencies: [],
     goals: [
-      { goalId: null, goal, momentum: "moving", summary: "", linedUp, progressedSinceLastRun: [], stalled: [], gaps: [] },
+      {
+        goalId: null,
+        goal,
+        momentum: "moving",
+        summary: "",
+        linedUp,
+        progressedSinceLastRun: [],
+        stalled: [],
+        gaps: [],
+      },
     ],
     actions: [],
   };
@@ -46,7 +58,10 @@ describe("titleRefs", () => {
       "developing-today/code#944",
       "developing-today/code#985",
     ]);
-    expect(titleRefs("nix #889 queued; nix #848 bump task 1127 re-queued")).toEqual(["dezren39/nix#889", "dezren39/nix#848"]);
+    expect(titleRefs("nix #889 queued; nix #848 bump task 1127 re-queued")).toEqual([
+      "dezren39/nix#889",
+      "dezren39/nix#848",
+    ]);
     expect(titleRefs("Jules PR #927 for code #811 - still OPEN draft")).toEqual([
       "developing-today/code#927",
       "developing-today/code#811",
@@ -69,7 +84,9 @@ describe("inFlightBoardItems", () => {
       section([
         task("code #944 fix - PR #985 opened", "in-progress"),
         task("code #968 - queue task 1116 revived", "queued"),
-        task("Plaid/Citi sign-in", "blocked", { blockedBy: [{ task: "Drew's Plaid/Citi sign-in", status: "stalled" }] }),
+        task("Plaid/Citi sign-in", "blocked", {
+          blockedBy: [{ task: "Drew's Plaid/Citi sign-in", status: "stalled" }],
+        }),
         task("code #983 fix - done via PR #986 merged", "done-recently"),
       ]),
       [],
@@ -85,13 +102,19 @@ describe("inFlightBoardItems", () => {
   });
 
   test("an explicit linedUp ref is linked even when the title carries none", () => {
-    const out = inFlightBoardItems(section([task("mcpx slice work", "in-progress", { ref: "developing-today/code#958" })]), []);
+    const out = inFlightBoardItems(
+      section([task("mcpx slice work", "in-progress", { ref: "developing-today/code#958" })]),
+      [],
+    );
     expect(out).toHaveLength(1);
     expect(out[0].links).toEqual([{ ref: "developing-today/code#958", relation: "tracks" }]);
   });
 
   test("dedupes against Kaneo cards by ref and by title", () => {
-    const existing = [kaneoCard("Some finished card", ["developing-today/code#944"]), kaneoCard("Daily morning readout", [])];
+    const existing = [
+      kaneoCard("Some finished card", ["developing-today/code#944"]),
+      kaneoCard("Daily morning readout", []),
+    ];
     const out = inFlightBoardItems(
       section([
         task("code #944 fix - PR #985 opened", "in-progress"), // ref already on a Kaneo card
@@ -104,8 +127,15 @@ describe("inFlightBoardItems", () => {
   });
 
   test("dedupes repeats inside the section", () => {
-    const s = section([task("code #968 - queue task 1116 revived", "queued"), task("code #968 - queue task 1116 revived", "queued")]);
-    s.goals.push({ ...s.goals[0], goal: "Another goal", linedUp: [task("code #968 - queue task 1116 revived", "in-progress")] });
+    const s = section([
+      task("code #968 - queue task 1116 revived", "queued"),
+      task("code #968 - queue task 1116 revived", "queued"),
+    ]);
+    s.goals.push({
+      ...s.goals[0],
+      goal: "Another goal",
+      linedUp: [task("code #968 - queue task 1116 revived", "in-progress")],
+    });
     expect(inFlightBoardItems(s, [])).toHaveLength(1);
   });
 
@@ -123,15 +153,22 @@ describe("inFlightBoardItems", () => {
           "nix #889 queued (park condition cleared - PR #887 merged); nix #848 bump task 1127 re-queued after lease expiry (labels truthful, no repair needed)",
           "queued",
         ),
-        task("openchamber #492 - task 1124 claim-evaporation repaired by monitor, labels reset to queued with handoff note", "queued"),
+        task(
+          "openchamber #492 - task 1124 claim-evaporation repaired by monitor, labels reset to queued with handoff note",
+          "queued",
+        ),
       ]),
       [],
     );
-    expect(out.map((i) => i.title)).not.toContain("code #983 R2-secret-in-argv fix - done via PR #986 merged, issue CLOSED");
+    expect(out.map((i) => i.title)).not.toContain(
+      "code #983 R2-secret-in-argv fix - done via PR #986 merged, issue CLOSED",
+    );
     const by944 = out.find((i) => i.title.includes("#944"));
     expect(by944?.links.map((l) => l.ref)).toContain("developing-today/code#985");
     expect(out.find((i) => i.title.includes("#889"))?.repo).toBe("dezren39/nix");
-    expect(out.find((i) => i.title.includes("#492"))?.links.map((l) => l.ref)).toEqual(["developing-today/openchamber#492"]);
+    expect(out.find((i) => i.title.includes("#492"))?.links.map((l) => l.ref)).toEqual([
+      "developing-today/openchamber#492",
+    ]);
   });
 });