Add work queue activity summary to the conclusion job

actions/setup/js/dispatch_work_coordinator_summary.cjs

@@ -0,0 +1,91 @@
+// @ts-check
+"use strict";
+
+const { loadDispatchCoordinatorSnapshot } = require("./dispatch_work_coordinator_mcp_server.cjs");
+const { replayTransactions } = require("./dispatch_work_coordinator_replay.cjs");
+const { readCoordinatorLog } = require("./dispatch_work_coordinator_store.cjs");
+
+/**
+ * @param {Record<string, string>} states
+ * @param {string} state
+ */
+function countState(states, state) {
+  return Object.values(states).filter(value => value === state).length;
+}
+
+/**
+ * @param {ReturnType<typeof loadDispatchCoordinatorSnapshot>} snapshot
+ * @param {ReturnType<typeof replayTransactions>} current
+ */
+function renderSummary(snapshot, current) {
+  const before = snapshot.projection;
+  const baseline = new Set(before.transactions.map(transaction => JSON.stringify(transaction)));
+  const added = current.transactions.filter(transaction => !baseline.has(JSON.stringify(transaction)));
+  const workCount = Object.keys(current.work).length;
+  const lines = [
+    "### Work queue activity",
+    "",
+    `${workCount} work item${workCount === 1 ? "" : "s"} in the queue; ${added.length} new transaction${added.length === 1 ? "" : "s"} observed since activation.`,
+    "",
+    "<details>",
+    "<summary>Show work queue activity</summary>",
+    "",
+    "This compares the activation snapshot with the queue read during conclusion. The queue is shared: changes may include activity from other workflow runs.",
+    "",
+    "| Work state | At activation | At conclusion |",
+    "| --- | ---: | ---: |",
+  ];
+  for (const state of ["available", "claimed", "completed", "cancelled"]) {
+    lines.push(`| ${state} | ${countState(before.work, state)} | ${countState(current.work, state)} |`);
+  }
+  lines.push("", "| Claim state | At activation | At conclusion |", "| --- | ---: | ---: |");
+  for (const state of ["effective", "superseded", "cancelled"]) {
+    lines.push(`| ${state} | ${countState(before.claim, state)} | ${countState(current.claim, state)} |`);
+  }
+  lines.push("", "| Activity since activation | Transactions |", "| --- | ---: |");
+  for (const [kind, label] of [
+    ["Work", "Work added"],
+    ["Claim", "Claims created"],
+    ["ClaimCancellation", "Claims cancelled"],
+    ["Completion", "Work completed"],
+    ["WorkCancellation", "Work cancelled"],
+  ]) {
+    lines.push(`| ${label} | ${added.filter(transaction => transaction.kind === kind).length} |`);
+  }
+  lines.push("", `Unique transactions: ${before.transactions.length} at activation; ${current.transactions.length} at conclusion.`);
+  if (snapshot.worker) {
+    const work = Object.hasOwn(current.work, snapshot.worker.work_id) ? current.work[snapshot.worker.work_id] : "absent";
+    const claim = Object.hasOwn(current.claim, snapshot.worker.claim_id) ? current.claim[snapshot.worker.claim_id] : "absent";
+    lines.push("", `Assigned worker: work **${work}**; claim **${claim}**.`);
+  } else {
+    lines.push("", "No worker claim was assigned to this run.");
+  }
+  lines.push("", "</details>", "");
+  return lines.join("\n");
+}
+
+async function main(options = {}) {
+  const coreApi = options.core || core;
+  let summary;
+  try {
+    const snapshot = loadDispatchCoordinatorSnapshot(options.snapshotPath);
+    const githubClient = options.githubClient || github;
+    const repositoryContext = options.context || context;
+    const readLog = options.readCoordinatorLog || readCoordinatorLog;
+    const latest = await readLog({
+      githubClient,
+      owner: repositoryContext.repo.owner,
+      repo: repositoryContext.repo.repo,
+      publishUpgrades: false,
+      core: coreApi,
+    });
+    summary = renderSummary(snapshot, replayTransactions(latest.transactions));
+  } catch (error) {
+    coreApi.warning("Work queue activity summary is unavailable; the activation snapshot or current queue could not be read.");
+    await coreApi.summary.addRaw("### Work queue activity\n\n<details>\n<summary>Show work queue activity</summary>\n\nActivity is unavailable because the activation snapshot or current queue could not be read.\n\n</details>\n").write();
+    throw new Error("Failed to summarize work queue activity", { cause: error });
+  }
+  await coreApi.summary.addRaw(summary).write();
+}
+
+module.exports = { main, renderSummary };

actions/setup/js/dispatch_work_coordinator_summary.test.cjs

@@ -0,0 +1,184 @@
+// @ts-check
+import { afterEach, describe, expect, it, vi } from "vitest";
+import fs from "fs";
+import os from "os";
+import path from "path";
+import { main, renderSummary } from "./dispatch_work_coordinator_summary.cjs";
+import { replayTransactions, serializeTransactionLog } from "./dispatch_work_coordinator_replay.cjs";
+
+const work = id => ({ version: 1, kind: "Work", work: id, claim: null, attempt: null });
+const claim = (id, claimId) => ({ version: 1, kind: "Claim", work: id, claim: claimId, attempt: null });
+const initial = [work("done"), claim("done", "claim-a"), work("retry"), claim("retry", "claim-b"), work("cancel"), work("waiting")];
+const latest = [
+  ...initial,
+  { version: 1, kind: "Completion", work: "done", claim: "claim-a", attempt: "run-1" },
+  { version: 1, kind: "ClaimCancellation", work: "retry", claim: "claim-b", attempt: null },
+  { version: 1, kind: "WorkCancellation", work: "cancel", claim: null, attempt: null },
+  work("new"),
+  claim("new", "claim-c"),
+  claim("new", "claim-d"),
+];
+let tempDirectory;
+
+function snapshot(transactions = initial, worker = null) {
+  return { sha: "head", worker, projection: replayTransactions(transactions) };
+}
+
+function setup() {
+  tempDirectory = fs.mkdtempSync(path.join(os.tmpdir(), "dispatch-coordinator-summary-"));
+  const snapshotPath = path.join(tempDirectory, "snapshot.json");
+  fs.writeFileSync(snapshotPath, JSON.stringify({ version: 2, sha: "head", worker: null, transactionLog: serializeTransactionLog(initial) }));
+  const summary = { addRaw: vi.fn(), write: vi.fn().mockResolvedValue(undefined) };
+  summary.addRaw.mockReturnValue(summary);
+  const core = { summary, info: vi.fn(), warning: vi.fn() };
+  return { snapshotPath, core, githubClient: {}, context: { repo: { owner: "owner", repo: "repo" } } };
+}
+
+afterEach(() => {
+  if (tempDirectory) {
+    fs.rmSync(tempDirectory, { recursive: true, force: true });
+    tempDirectory = undefined;
+  }
+});
+
+describe("dispatch coordinator conclusion summary", () => {
+  it("shows queue state changes and every transaction kind in collapsed details", () => {
+    const summary = renderSummary(snapshot(), replayTransactions(latest));
+    expect(summary).toContain("### Work queue activity\n\n5 work items in the queue; 6 new transactions observed since activation.");
+    expect(summary).toContain("<details>\n<summary>Show work queue activity</summary>\n\n");
+    expect(summary).not.toContain("<details open");
+    expect(summary).toContain("changes may include activity from other workflow runs");
+    expect(summary).toContain("| available | 2 | 2 |");
+    expect(summary).toContain("| claimed | 2 | 1 |");
+    expect(summary).toContain("| completed | 0 | 1 |");
+    expect(summary).toContain("| cancelled | 0 | 1 |");
+    expect(summary).toContain("| effective | 2 | 2 |");
+    expect(summary).toContain("| superseded | 0 | 1 |");
+    expect(summary).toContain("| Claims created | 2 |");
+    for (const label of ["Work added", "Claims cancelled", "Work completed", "Work cancelled"]) {
+      expect(summary).toContain(`| ${label} | 1 |`);
+    }
+    expect(summary).toContain("Unique transactions: 6 at activation; 12 at conclusion.");
+    expect(summary).toContain("No worker claim was assigned to this run.");
+    expect(summary).toMatch(/<\/details>\n$/);
+    expect(summary.indexOf("| Work state")).toBeGreaterThan(summary.indexOf("<details>"));
+  });
+
+  it("deduplicates facts and does not count reordering as activity", () => {
+    const summary = renderSummary(snapshot(), replayTransactions([...initial].reverse().concat(initial)));
+    expect(summary).toContain("0 new transactions observed");
+    expect(summary).toContain("Unique transactions: 6 at activation; 6 at conclusion.");
+    expect(summary).toContain("| Claims created | 0 |");
+  });
+
+  it("reports an empty queue explicitly", () => {
+    const summary = renderSummary(snapshot([]), replayTransactions([]));
+    expect(summary).toContain("0 work items in the queue; 0 new transactions");
+    expect(summary).toContain("| available | 0 | 0 |");
+  });
+
+  it.each([
+    [[], [work("one")], "1 work item in the queue; 1 new transaction observed since activation."],
+    [[work("one")], [work("one")], "1 work item in the queue; 0 new transactions observed since activation."],
+    [[work("one")], [work("one"), work("two")], "2 work items in the queue; 1 new transaction observed since activation."],
+  ])("pluralizes work items and new transactions independently", (before, after, expected) => {
+    expect(renderSummary(snapshot(before), replayTransactions(after))).toContain(expected);
+  });
+
+  it.each([
+    ["done", "claim-a", "completed", "effective"],
+    ["retry", "claim-b", "available", "cancelled"],
+    ["new", "claim-d", "claimed", "superseded"],
+    ["missing", "__proto__", "absent", "absent"],
+  ])("reports worker state without disclosing identifiers for %s", (workId, claimId, workState, claimState) => {
+    const summary = renderSummary(snapshot(initial, { work_id: workId, claim_id: claimId }), replayTransactions(latest));
+    expect(summary).toContain(`Assigned worker: work **${workState}**; claim **${claimState}**.`);
+    expect(summary).not.toContain(claimId);
+  });
+
+  it("never renders untrusted queue identifiers or attempt data", () => {
+    const id = "</details>\n<script>secret</script>";
+    const summary = renderSummary(snapshot([]), replayTransactions([work(id), claim(id, "`unsafe|claim`"), { version: 1, kind: "Completion", work: id, claim: "`unsafe|claim`", attempt: "sensitive-attempt" }]));
+    expect(summary).not.toContain(id);
+    expect(summary).not.toContain("unsafe");
+    expect(summary).not.toContain("sensitive-attempt");
+    expect(summary.match(/<\/details>/g)).toHaveLength(1);
+  });
+
+  it("writes the summary with a read-only refresh of the durable queue", async () => {
+    const options = setup();
+    const readCoordinatorLog = vi.fn().mockResolvedValue({ sha: "latest", transactions: latest });
+    await main({ ...options, readCoordinatorLog });
+    expect(readCoordinatorLog).toHaveBeenCalledWith({
+      githubClient: options.githubClient,
+      owner: "owner",
+      repo: "repo",
+      publishUpgrades: false,
+      core: options.core,
+    });
+    expect(options.core.summary.addRaw).toHaveBeenCalledWith(renderSummary(snapshot(), replayTransactions(latest)));
+    expect(options.core.summary.write).toHaveBeenCalledOnce();
+    expect(options.core.warning).not.toHaveBeenCalled();
+  });
+
+  it("reads legacy queue data without publishing upgrades", async () => {
+    const options = setup();
+    const legacyWork = { kind: "Work", work: "legacy", claim: null, attempt: null };
+    const createBlob = vi.fn();
+    const githubClient = {
+      rest: {
+        git: {
+          getRef: async () => ({ data: { object: { sha: "legacy-head" } } }),
+          getCommit: async () => ({ data: { tree: { sha: "tree" } } }),
+          getTree: async () => ({ data: { tree: [{ path: "dispatch-work-coordinator.jsonl", type: "blob", sha: "blob" }] } }),
+          getBlob: async () => ({ data: { encoding: "base64", content: Buffer.from(`${JSON.stringify(legacyWork)}\n`).toString("base64") } }),
+          createBlob,
+        },
+      },
+    };
+    await main({ ...options, githubClient });
+    expect(createBlob).not.toHaveBeenCalled();
+    expect(options.core.summary.addRaw).toHaveBeenCalledWith(expect.stringContaining("1 work item in the queue; 1 new transaction"));
+  });
+
+  it("treats a missing coordinator branch as an empty queue", async () => {
+    const options = setup();
+    const githubClient = {
+      rest: {
+        git: {
+          getRef: async () => {
+            throw Object.assign(new Error("Not Found"), { status: 404 });
+          },
+        },
+      },
+    };
+    await main({ ...options, githubClient });
+    expect(options.core.summary.addRaw).toHaveBeenCalledWith(expect.stringContaining("0 work items in the queue; 0 new transactions"));
+    expect(options.core.warning).not.toHaveBeenCalled();
+  });
+
+  it("does not misreport a summary write failure as unavailable queue data", async () => {
+    const options = setup();
+    const readCoordinatorLog = vi.fn().mockResolvedValue({ transactions: initial });
+    options.core.summary.write.mockRejectedValue(new Error("summary write failed"));
+    await expect(main({ ...options, readCoordinatorLog })).rejects.toThrow("summary write failed");
+    expect(options.core.warning).not.toHaveBeenCalled();
+    expect(options.core.summary.addRaw).toHaveBeenCalledOnce();
+  });
+
+  it.each(["missing snapshot", "malformed snapshot", "failed refresh", "invalid queue"])("reports %s as unavailable rather than an empty queue", async failure => {
+    const options = setup();
+    const readCoordinatorLog = vi.fn().mockResolvedValue({ sha: null, transactions: [] });
+    if (failure === "missing snapshot") fs.unlinkSync(options.snapshotPath);
+    if (failure === "malformed snapshot") fs.writeFileSync(options.snapshotPath, '{"version":0}');
+    if (failure === "failed refresh") readCoordinatorLog.mockRejectedValue(new Error("sensitive API error"));
+    if (failure === "invalid queue") readCoordinatorLog.mockResolvedValue({ transactions: [{ kind: "unknown", work: "sensitive" }] });
+    await expect(main({ ...options, readCoordinatorLog })).rejects.toThrow("Failed to summarize work queue activity");
+    expect(options.core.warning).toHaveBeenCalledOnce();
+    const summary = options.core.summary.addRaw.mock.calls[0][0];
+    expect(summary).toContain("<details>");
+    expect(summary).toContain("Activity is unavailable");
+    expect(summary).not.toContain("0 work items");
+    expect(summary).not.toContain("sensitive");
+  });
+});

docs/adr/64955-git-backed-dispatch-work-coordination.md

@@ -28,6 +28,8 @@ Implement `tools.work-queue` as a first-class, compiler-aware tool backed by one
 
 The [formal verification notes and reproducible checker](../../specs/dispatch-work-coordinator/README.md) document the invariants and assumptions. The bounded model checks support a parameterized inductive proof argument, not a mechanically checked unbounded proof or proof that a future runtime implementation refines the model.
 
+The conclusion job writes a work queue activity step summary for workflows using `tools.work-queue`. It compares the activation snapshot with a read-only refresh of the durable queue, showing work and claim state counts, new transaction counts by kind, and the assigned worker's current state in a collapsed details section. These are shared-queue observations since activation, not activity attributed exclusively to the current run. Work, claim, and attempt identifiers are omitted. An unreadable snapshot or queue is reported as unavailable, not as an empty queue.
+
 For review, the [trace walkthrough](../../specs/dispatch-work-coordinator/README.md#inspect-execution-traces) generates bounded textual TLC executions and counterexamples to deliberately false reachability witnesses. These expose competing Claims, orphan recovery, and the finalization-to-effect sequence under the guarded protocol; they are evidence of modeled possibilities, not runtime conformance or liveness guarantees.
 
 ### Alternatives Considered

pkg/workflow/dispatch_work_coordinator_compilation_integration_test.go

@@ -85,4 +85,11 @@ Compile each dispatch-coordinator workflow phase.
 		handlerEnd += len("      - name:")
 	}
 	require.Contains(t, safeOutputs[handlerStart:handlerStart+handlerEnd], gate)
+
+	conclusion := extractJobSection(compiled, "conclusion")
+	require.Contains(t, conclusion, "contents: read")
+	require.Contains(t, conclusion, "Download activation artifact for work queue summary")
+	require.Contains(t, conclusion, "Summarize work queue activity\n        if: always()")
+	require.Contains(t, conclusion, "dispatch_work_coordinator_summary.cjs")
+	require.Contains(t, conclusion, "await main({ core, githubClient: github, context });")
 }

pkg/workflow/notify_comment.go

@@ -66,6 +66,7 @@ func (c *Compiler) buildConclusionJob(data *WorkflowData, mainJobName string, sa
 // failed-jobs reporting, the optional status-comment update, and the steering-issue step.
 func (c *Compiler) buildConclusionJobSteps(data *WorkflowData, mainJobName string, safeOutputJobNames []string) ([]string, error) {
 	steps := c.buildConclusionSetupSteps(data)
+	steps = append(steps, c.buildConclusionWorkQueueSummaryStep(data)...)
 	steps = append(steps, c.buildConclusionNoOpStep(data, mainJobName)...)
 	steps = append(steps, c.buildConclusionDetectionRunsStep(data, mainJobName)...)
 	steps = append(steps, c.buildConclusionMissingToolStep(data, mainJobName)...)
@@ -109,6 +110,11 @@ func (c *Compiler) buildConclusionJobSteps(data *WorkflowData, mainJobName strin
 // conclusion mechanism.
 func computeConclusionJobPermissions(data *WorkflowData) *Permissions {
 	conclusionPerms := ComputePermissionsForSafeOutputs(data.SafeOutputs)
+	if isDispatchWorkCoordinatorEnabled(data) {
+		if level, ok := conclusionPerms.Get(PermissionContents); !ok || level == PermissionNone {
+			conclusionPerms.Set(PermissionContents, PermissionRead)
+		}
+	}
 	// When observability.otlp.github-app is configured without app-id/private-key
 	// credentials, id-token: write is needed so the conclusion job can mint the OTLP
 	// OIDC token via core.getIDToken(audience) (mirrors threat_detection_job.go).

pkg/workflow/notify_comment_conclusion_helpers.go

@@ -58,6 +58,28 @@ func (c *Compiler) buildConclusionSetupSteps(data *WorkflowData) []string {
 	return steps
 }
 
+func (c *Compiler) buildConclusionWorkQueueSummaryStep(data *WorkflowData) []string {
+	if !isDispatchWorkCoordinatorEnabled(data) {
+		return nil
+	}
+	steps := buildArtifactDownloadSteps(ArtifactDownloadConfig{
+		ArtifactName: artifactPrefixExprForDownstreamJob(data) + constants.ActivationArtifactName.String(),
+		DownloadPath: constants.TmpGhAwDirSlash,
+		StepName:     "Download activation artifact for work queue summary",
+		IfCondition:  "always()",
+	}, c.getActionPin)
+	return append(steps,
+		"      - name: Summarize work queue activity\n",
+		"        if: always()\n",
+		"        continue-on-error: true\n",
+		fmt.Sprintf("        uses: %s\n", c.getActionPin("actions/github-script")),
+		"        with:\n",
+		"          script: |\n",
+		"            const { main } = require('${{ runner.temp }}/gh-aw/actions/dispatch_work_coordinator_summary.cjs');\n",
+		"            await main({ core, githubClient: github, context });\n",
+	)
+}
+
 // buildConclusionNoOpStep builds the merged no-op handler step.
 func (c *Compiler) buildConclusionNoOpStep(data *WorkflowData, mainJobName string) []string {
 	// Add noop processing step if noop is configured.