diff --git a/.github/workflows/smoke-dispatch-work-coordinator.lock.yml b/.github/workflows/smoke-dispatch-work-coordinator.lock.yml index 822252cf14d..4efbaab4b6c 100644 --- a/.github/workflows/smoke-dispatch-work-coordinator.lock.yml +++ b/.github/workflows/smoke-dispatch-work-coordinator.lock.yml @@ -1188,6 +1188,7 @@ jobs: runs-on: ubuntu-slim permissions: actions: read + contents: read issues: write concurrency: group: "gh-aw-conclusion-smoke-dispatch-work-coordinator" @@ -1257,6 +1258,23 @@ jobs: pattern: safe-outputs-items merge-multiple: true path: /tmp/gh-aw/ + - name: Download activation artifact for work queue summary + if: always() + continue-on-error: true + uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 + with: + name: activation + path: /tmp/gh-aw/ + - name: Summarize work queue activity + if: always() + continue-on-error: true + uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0 + with: + script: | + const path = require('path'); + const actionsDir = path.join(process.env.RUNNER_TEMP, 'gh-aw', 'actions'); + const { main } = require(path.join(actionsDir, 'dispatch_work_coordinator_summary.cjs')); + await main({ core, githubClient: github, context }); - name: Process no-op messages id: noop uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0 diff --git a/actions/setup/js/dispatch_work_coordinator_summary.cjs b/actions/setup/js/dispatch_work_coordinator_summary.cjs new file mode 100644 index 00000000000..7eeda37fb61 --- /dev/null +++ b/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} states + * @param {string} state + */ +function countState(states, state) { + return Object.values(states).filter(value => value === state).length; +} + +/** + * @param {ReturnType} snapshot + * @param {ReturnType} 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.`, + "", + "
", + "Show work queue activity", + "", + "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("", "
", ""); + 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
\nShow work queue activity\n\nActivity is unavailable because the activation snapshot or current queue could not be read.\n\n
\n").write(); + throw new Error("Failed to summarize work queue activity", { cause: error }); + } + await coreApi.summary.addRaw(summary).write(); +} + +module.exports = { main, renderSummary }; diff --git a/actions/setup/js/dispatch_work_coordinator_summary.test.cjs b/actions/setup/js/dispatch_work_coordinator_summary.test.cjs new file mode 100644 index 00000000000..1f6e4975fa9 --- /dev/null +++ b/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("
\nShow work queue activity\n\n"); + expect(summary).not.toContain("
\n$/); + expect(summary.indexOf("| Work state")).toBeGreaterThan(summary.indexOf("
")); + }); + + 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 = "
\n"; + 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("
"); + expect(summary).toContain("Activity is unavailable"); + expect(summary).not.toContain("0 work items"); + expect(summary).not.toContain("sensitive"); + }); +}); diff --git a/docs/adr/64955-git-backed-dispatch-work-coordination.md b/docs/adr/64955-git-backed-dispatch-work-coordination.md index f4cc2d7dadd..599c6bb2161 100644 --- a/docs/adr/64955-git-backed-dispatch-work-coordination.md +++ b/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 diff --git a/pkg/workflow/dispatch_work_coordinator_compilation_integration_test.go b/pkg/workflow/dispatch_work_coordinator_compilation_integration_test.go index 45be64181b2..680baef95cc 100644 --- a/pkg/workflow/dispatch_work_coordinator_compilation_integration_test.go +++ b/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 });") } diff --git a/pkg/workflow/notify_comment.go b/pkg/workflow/notify_comment.go index 5a18dbca098..519c3e41efb 100644 --- a/pkg/workflow/notify_comment.go +++ b/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). diff --git a/pkg/workflow/notify_comment_conclusion_helpers.go b/pkg/workflow/notify_comment_conclusion_helpers.go index 22dea3b102e..52d256a3770 100644 --- a/pkg/workflow/notify_comment_conclusion_helpers.go +++ b/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. diff --git a/pkg/workflow/notify_comment_work_queue_test.go b/pkg/workflow/notify_comment_work_queue_test.go new file mode 100644 index 00000000000..8149e436a65 --- /dev/null +++ b/pkg/workflow/notify_comment_work_queue_test.go @@ -0,0 +1,77 @@ +//go:build !integration + +package workflow + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestConclusionWorkQueueSummary(t *testing.T) { + for _, enabled := range []bool{true, false} { + t.Run(map[bool]string{true: "enabled", false: "disabled"}[enabled], func(t *testing.T) { + data := &WorkflowData{ + Name: "Work queue worker", + Tools: map[string]any{"work-queue": enabled}, + SafeOutputs: &SafeOutputsConfig{}, + } + compiler := NewCompiler() + job, err := compiler.buildConclusionJob(data, "agent", []string{"safe_outputs"}) + require.NoError(t, err) + require.NotNil(t, job) + steps := strings.Join(job.Steps, "") + if !enabled { + require.NotContains(t, steps, "dispatch_work_coordinator_summary.cjs") + require.NotContains(t, steps, "Download activation artifact for work queue summary") + return + } + require.Contains(t, job.Permissions, "contents: read") + require.Contains(t, job.Needs, "safe_outputs") + require.Contains(t, steps, "Download activation artifact for work queue summary\n if: always()") + require.Contains(t, steps, "Summarize work queue activity\n if: always()") + require.Contains(t, steps, "Summarize work queue activity\n if: always()\n continue-on-error: true") + require.Contains(t, steps, "dispatch_work_coordinator_summary.cjs") + require.Contains(t, steps, "await main({ core, githubClient: github, context });") + require.Less(t, strings.Index(steps, "Download activation artifact for work queue summary"), strings.Index(steps, "Summarize work queue activity")) + }) + } +} + +func TestConclusionWorkQueueSummaryPreservesWritePermissions(t *testing.T) { + data := &WorkflowData{ + Tools: map[string]any{"work-queue": true}, + SafeOutputs: &SafeOutputsConfig{ + CreatePullRequests: &CreatePullRequestsConfig{}, + }, + } + level, ok := computeConclusionJobPermissions(data).Get(PermissionContents) + require.True(t, ok) + require.Equal(t, PermissionWrite, level) +} + +func TestConclusionWorkQueueSummaryArtifactPrefix(t *testing.T) { + for _, tc := range []struct { + trigger string + artifactName string + }{ + {trigger: "workflow_dispatch", artifactName: "activation"}, + {trigger: "workflow_call", artifactName: "${{ needs.activation.outputs.artifact_prefix }}activation"}, + } { + t.Run(tc.trigger, func(t *testing.T) { + data := &WorkflowData{ + On: tc.trigger, + Tools: map[string]any{"work-queue": true}, + SafeOutputs: &SafeOutputsConfig{}, + } + job, err := NewCompiler().buildConclusionJob(data, "agent", []string{"safe_outputs"}) + require.NoError(t, err) + require.NotNil(t, job) + require.Contains(t, job.Needs, "activation") + steps := strings.Join(job.Steps, "") + require.Contains(t, steps, " name: "+tc.artifactName+"\n") + require.NotContains(t, steps, "steps.artifact-prefix.outputs.prefix") + }) + } +}