diff --git a/.github/workflows/warden.yml b/.github/workflows/warden.yml index c8156fed..76b83b65 100644 --- a/.github/workflows/warden.yml +++ b/.github/workflows/warden.yml @@ -11,6 +11,10 @@ on: pull_request: types: [opened, synchronize, reopened] +concurrency: + group: warden-${{ github.workflow }}-${{ github.event.pull_request.number }} + cancel-in-progress: true + jobs: review: runs-on: ubuntu-latest @@ -39,15 +43,50 @@ jobs: with: mode: analyze + - name: Verify cancelled findings + if: ${{ cancelled() }} + env: + FINDINGS_FILE: ${{ steps.warden-analyze.outputs.findings-file }} + run: | + if [ -z "$FINDINGS_FILE" ]; then + echo "Analyze was cancelled without producing a findings file" >&2 + exit 1 + fi + + FINDINGS_FILE="$FINDINGS_FILE" node --input-type=module -e ' + const fs = await import("node:fs"); + const path = process.env.FINDINGS_FILE; + const findings = JSON.parse(fs.readFileSync(path, "utf8")); + if (findings.outcome !== "cancelled") { + throw new Error(`Expected cancelled outcome, received ${findings.outcome}`); + } + if (!fs.existsSync(`${path}.done`)) { + throw new Error(`Missing finalization marker: ${path}.done`); + } + console.log(`Preserved ${findings.summary.totalFindings} finding(s)`); + ' + + - name: Upload cancelled findings + if: ${{ cancelled() && steps.warden-analyze.outputs.findings-file != '' }} + uses: actions/upload-artifact@v4 + with: + name: cancelled-findings-${{ github.run_id }}-${{ github.run_attempt }} + path: | + ${{ steps.warden-analyze.outputs.findings-file }} + ${{ steps.warden-analyze.outputs.findings-file }}.done + if-no-files-found: error + - uses: actions/create-github-app-token@v1 id: app-token + if: ${{ always() && steps.warden-analyze.outputs.findings-file != '' }} with: app-id: ${{ secrets.WARDEN_APP_ID }} private-key: ${{ secrets.WARDEN_PRIVATE_KEY }} - name: Report + if: ${{ always() && steps.warden-analyze.outputs.findings-file != '' && steps.app-token.outcome == 'success' }} uses: ./ with: mode: report findings-file: ${{ steps.warden-analyze.outputs.findings-file }} - github-token: ${{ steps.app-token.outputs.token }} + github-token: ${{ steps.app-token.outputs.token }} \ No newline at end of file diff --git a/action.yml b/action.yml index 333e51a3..c7eb7b5e 100644 --- a/action.yml +++ b/action.yml @@ -120,4 +120,4 @@ runs: INPUT_SERVICE_DATA: ${{ inputs.service-data }} INPUT_SERVICE_MEMORY: ${{ inputs.service-memory }} INPUT_SERVICE_TIMEOUT_MS: ${{ inputs.service-timeout-ms }} - run: node ${{ github.action_path }}/dist/action/index.js + run: exec node "${{ github.action_path }}/dist/action/index.js" diff --git a/packages/warden/src/action/cancellation.test.ts b/packages/warden/src/action/cancellation.test.ts new file mode 100644 index 00000000..c54a1b2e --- /dev/null +++ b/packages/warden/src/action/cancellation.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it, vi } from 'vitest'; +import { ActionCancellation, createActionSignalHandler } from './cancellation.js'; + +describe('Action cancellation signals', () => { + it('aborts gracefully on the first cancellation signal', () => { + const cancellation = new ActionCancellation(); + const exit = vi.fn(); + const handler = createActionSignalHandler({ cancellation, exit }); + + handler('SIGTERM'); + + expect(cancellation.requested).toBe(true); + expect(cancellation.signalName).toBe('SIGTERM'); + expect(cancellation.abortController.signal.aborted).toBe(true); + expect(cancellation.exitCode).toBe(143); + expect(exit).not.toHaveBeenCalled(); + }); + + it('ignores duplicate delivery before forcing a later exit', () => { + const cancellation = new ActionCancellation(); + const exit = vi.fn(); + let now = 1_000; + const handler = createActionSignalHandler({ cancellation, exit, now: () => now }); + + handler('SIGINT'); + now += 100; + handler('SIGTERM'); + expect(exit).not.toHaveBeenCalled(); + + now += 1_000; + handler('SIGTERM'); + expect(exit).toHaveBeenCalledWith(143); + }); +}); diff --git a/packages/warden/src/action/cancellation.ts b/packages/warden/src/action/cancellation.ts new file mode 100644 index 00000000..4c4ce8e7 --- /dev/null +++ b/packages/warden/src/action/cancellation.ts @@ -0,0 +1,57 @@ +export type ActionCancelSignal = 'SIGINT' | 'SIGTERM'; + +const DEFAULT_DUPLICATE_SIGNAL_WINDOW_MS = 750; + +/** Run-scoped cancellation state shared by the Action entrypoint and workflows. */ +export class ActionCancellation { + readonly abortController = new AbortController(); + signalName: ActionCancelSignal | undefined; + + get requested(): boolean { + return this.signalName !== undefined; + } + + request(signalName: ActionCancelSignal): boolean { + if (this.requested) return false; + + this.signalName = signalName; + this.abortController.abort(new Error(`Action cancelled by ${signalName}`)); + return true; + } + + get exitCode(): number { + return this.signalName === 'SIGTERM' ? 143 : 130; + } +} + +interface ActionSignalHandlerOptions { + cancellation: ActionCancellation; + now?: () => number; + exit?: (code: number) => void; + duplicateWindowMs?: number; +} + +/** Create a shared SIGINT/SIGTERM handler with graceful-first, force-second behavior. */ +export function createActionSignalHandler( + options: ActionSignalHandlerOptions +): (signalName: ActionCancelSignal) => void { + const duplicateWindowMs = options.duplicateWindowMs ?? DEFAULT_DUPLICATE_SIGNAL_WINDOW_MS; + const now = options.now ?? (() => Date.now()); + const exit = options.exit ?? ((code) => process.exit(code)); + let lastSignalAt = 0; + + return (signalName) => { + const receivedAt = now(); + if (options.cancellation.requested && receivedAt - lastSignalAt < duplicateWindowMs) { + return; + } + + lastSignalAt = receivedAt; + if (!options.cancellation.request(signalName)) { + exit(signalName === 'SIGTERM' ? 143 : 130); + return; + } + + console.warn(`Cancellation requested by ${signalName}; finalizing partial results`); + }; +} diff --git a/packages/warden/src/action/run.ts b/packages/warden/src/action/run.ts index c3c42cfe..36d29a26 100644 --- a/packages/warden/src/action/run.ts +++ b/packages/warden/src/action/run.ts @@ -8,22 +8,47 @@ import { initSentry, flushSentry } from '../sentry.js'; import { ActionFailedError } from './workflow/base.js'; import { runAction } from './runner.js'; +import { ActionCancellation, createActionSignalHandler } from './cancellation.js'; -async function flushActionTelemetry(): Promise { - if (!(await flushSentry())) { +const CANCELLATION_TELEMETRY_FLUSH_TIMEOUT_MS = 3_000; + +async function flushActionTelemetry(timeoutMs?: number): Promise { + if (!(await flushSentry(timeoutMs))) { console.warn('::warning::Timed out while flushing Sentry telemetry'); } } -initSentry('action'); -runAction() - .then(() => flushActionTelemetry()) - .catch(async (error) => { +async function main(): Promise { + const cancellation = new ActionCancellation(); + const handleSignal = createActionSignalHandler({ cancellation }); + const onSigint = () => handleSignal('SIGINT'); + const onSigterm = () => handleSignal('SIGTERM'); + process.on('SIGINT', onSigint); + process.on('SIGTERM', onSigterm); + + try { + await runAction(cancellation); + await flushActionTelemetry( + cancellation.requested ? CANCELLATION_TELEMETRY_FLUSH_TIMEOUT_MS : undefined, + ); + if (cancellation.requested) { + process.exitCode = cancellation.exitCode; + } + } catch (error) { if (error instanceof ActionFailedError) { console.error(`::error::${error.message}`); } else { console.error(`::error::Unexpected error: ${error}`); } - await flushActionTelemetry(); - process.exit(1); - }); + await flushActionTelemetry( + cancellation.requested ? CANCELLATION_TELEMETRY_FLUSH_TIMEOUT_MS : undefined, + ); + process.exitCode = cancellation.requested ? cancellation.exitCode : 1; + } finally { + process.off('SIGINT', onSigint); + process.off('SIGTERM', onSigterm); + } +} + +initSentry('action'); +void main(); diff --git a/packages/warden/src/action/runner.test.ts b/packages/warden/src/action/runner.test.ts index bedb9e37..db18261d 100644 --- a/packages/warden/src/action/runner.test.ts +++ b/packages/warden/src/action/runner.test.ts @@ -40,6 +40,7 @@ vi.mock('./workflow/schedule.js', () => ({ })); import { runAction } from './runner.js'; +import { ActionCancellation } from './cancellation.js'; const baseInputs: ActionInputs = { anthropicApiKey: 'test-api-key', @@ -79,7 +80,8 @@ describe('runAction without telemetry', () => { expect(mocks.runScheduleWorkflow).toHaveBeenCalledWith( mocks.octokit, baseInputs, - '/tmp/workspace' + '/tmp/workspace', + expect.any(Object), ); }); }); @@ -173,7 +175,8 @@ describe('runAction', () => { baseInputs, 'push', '/tmp/event.json', - '/tmp/workspace' + '/tmp/workspace', + expect.any(Object), ); expect(emit).toHaveBeenCalledWith( 'processMetric', @@ -200,6 +203,47 @@ describe('runAction', () => { ); }); + it('records a requested cancellation as the Action outcome', async () => { + const cancellation = new ActionCancellation(); + cancellation.request('SIGTERM'); + const emit = spyOnClientEmit(); + + await runAction(cancellation); + await Sentry.flush(1000); + + expect(emit).toHaveBeenCalledWith( + 'processMetric', + expect.objectContaining({ + name: 'warden.action.runs', + attributes: expect.objectContaining({ + 'warden.action.outcome': 'cancelled', + }), + }), + ); + }); + + it('keeps the cancelled outcome when cleanup throws', async () => { + const cancellation = new ActionCancellation(); + cancellation.request('SIGTERM'); + const error = new Error('cleanup failed'); + mocks.runScheduleWorkflow.mockRejectedValueOnce(error); + const emit = spyOnClientEmit(); + + await expect(runAction(cancellation)).rejects.toBe(error); + await Sentry.flush(1000); + + expect(emit).toHaveBeenCalledWith( + 'processMetric', + expect.objectContaining({ + name: 'warden.action.runs', + attributes: expect.objectContaining({ + 'warden.action.outcome': 'cancelled', + }), + }), + ); + expect(capturedEvents).toHaveLength(0); + }); + it('attributes input parsing failures before capturing them', async () => { const error = new Error('Invalid mode "later"'); const setTag = vi.spyOn(Sentry.getIsolationScope(), 'setTag'); diff --git a/packages/warden/src/action/runner.ts b/packages/warden/src/action/runner.ts index 789103f3..8fbcdad6 100644 --- a/packages/warden/src/action/runner.ts +++ b/packages/warden/src/action/runner.ts @@ -13,13 +13,14 @@ import { parseActionInputs, setupAuthEnv, validateInputs } from './inputs.js'; import { ActionFailedError, setFailed } from './workflow/base.js'; import { runPRWorkflow } from './workflow/pr-workflow.js'; import { runScheduleWorkflow } from './workflow/schedule.js'; +import { ActionCancellation } from './cancellation.js'; function isPullRequestEvent(eventName: string): boolean { return eventName === 'pull_request'; } /** Run the GitHub Action dispatcher once. */ -export async function runAction(): Promise { +export async function runAction(cancellation = new ActionCancellation()): Promise { const eventName = process.env['GITHUB_EVENT_NAME']; const actionAttributes = setGitHubActionScope(eventName); @@ -48,18 +49,27 @@ export async function runAction(): Promise { if (inputs.mode !== 'run') { setFailed(`${inputs.mode} mode is only supported for pull request workflows`); } - await runScheduleWorkflow(octokit, inputs, repoPath); + await runScheduleWorkflow(octokit, inputs, repoPath, cancellation); } else { if (inputs.mode !== 'run' && !isPullRequestEvent(eventName)) { setFailed(`${inputs.mode} mode is only supported for pull request workflows`); } - await runPRWorkflow(octokit, inputs, eventName, eventPath, repoPath); + await runPRWorkflow(octokit, inputs, eventName, eventPath, repoPath, cancellation); } - span.setAttribute('warden.action.outcome', 'success'); + const outcome = cancellation.requested ? 'cancelled' : 'success'; + span.setAttribute('warden.action.outcome', outcome); span.setStatus({ code: SPAN_STATUS_OK }); - emitActionRunMetric('success', stage); + emitActionRunMetric(outcome, stage); } catch (error) { + if (cancellation.requested) { + span.setAttribute('warden.action.outcome', 'cancelled'); + span.setAttribute('warden.action.stage', stage); + span.setStatus({ code: SPAN_STATUS_OK }); + emitActionRunMetric('cancelled', stage); + throw error; + } + const { code } = classifyError(error); span.setAttribute('warden.action.outcome', 'failure'); span.setAttribute('warden.action.stage', stage); diff --git a/packages/warden/src/action/service.test.ts b/packages/warden/src/action/service.test.ts index 304a3302..5a419046 100644 --- a/packages/warden/src/action/service.test.ts +++ b/packages/warden/src/action/service.test.ts @@ -74,6 +74,24 @@ afterEach(() => { }); describe('Action service integration', () => { + it('preserves a cancelled findings outcome in the service envelope', () => { + const output = buildFindingsOutput([report], context, [], { + runId: 'cancelled-action-run', + timestamp: '2026-08-12T12:00:01.000Z', + outcome: 'cancelled', + }); + + const envelope = buildFindingsServiceRunEnvelope(output, { + url: 'https://warden.example.com', + token: 'service-token', + data: 'findings', + memory: false, + timeoutMs: 2_000, + }, 'action'); + + expect(envelope.outcome).toBe('cancelled'); + }); + it('defaults a URL-and-token-only Action setup to findings and memory', () => { expect(resolveActionServiceOptions(inputs({ serviceUrl: 'https://warden.example.com', diff --git a/packages/warden/src/action/triggers/executor.test.ts b/packages/warden/src/action/triggers/executor.test.ts index d0e64cae..3bc3d392 100644 --- a/packages/warden/src/action/triggers/executor.test.ts +++ b/packages/warden/src/action/triggers/executor.test.ts @@ -96,6 +96,12 @@ describe('executeTrigger', () => { ...checkOptions, ...options, }), + cancel: (report: SkillReport) => + updateSkillCheck(mockOctokit, check.checkRunId, report, { + ...checkOptions, + conclusion: 'cancelled', + title: 'Analysis cancelled', + }), fail: (error: unknown) => failSkillCheck(mockOctokit, check.checkRunId, error, checkOptions), }; @@ -320,6 +326,26 @@ describe('executeTrigger', () => { expect(result.error).toBeUndefined(); }); + it('cancels the skill check when Action cancellation was requested', async () => { + const mockReport = createReport(); + const cancellation = new AbortController(); + cancellation.abort(); + vi.mocked(runSkillTask).mockResolvedValue({ name: 'test-trigger', report: mockReport }); + vi.mocked(createSkillCheck).mockResolvedValue({ checkRunId: 123, url: 'https://github.com/check/123' }); + vi.mocked(updateSkillCheck).mockResolvedValue(undefined); + + await executeTrigger(mockTrigger, { + ...mockDeps, + cancellationSignal: cancellation.signal, + }); + + expect(updateSkillCheck).toHaveBeenCalledWith(mockOctokit, 123, mockReport, { + ...checkOptions, + conclusion: 'cancelled', + title: 'Analysis cancelled', + }); + }); + it('handles skill resolution failure', async () => { vi.mocked(runSkillTask).mockResolvedValue({ name: 'test-trigger', error: new Error('Skill not found') }); vi.mocked(createSkillCheck).mockResolvedValue({ checkRunId: 123, url: 'https://github.com/check/123' }); diff --git a/packages/warden/src/action/triggers/executor.ts b/packages/warden/src/action/triggers/executor.ts index 20ae18e2..45c61d2f 100644 --- a/packages/warden/src/action/triggers/executor.ts +++ b/packages/warden/src/action/triggers/executor.ts @@ -62,6 +62,7 @@ export interface TriggerCheckRun { url?: string; checkRunId?: number; complete(report: SkillReport, options: TriggerCheckCompleteOptions): Promise; + cancel(report: SkillReport): Promise; fail(error: unknown): Promise; } @@ -104,6 +105,8 @@ export interface TriggerExecutorDeps { analysisQueue?: AsyncWorkQueue; /** Shared controller for stopping the whole action run */ abortController?: AbortController; + /** User-requested Action cancellation, distinct from provider circuit-breaker aborts. */ + cancellationSignal?: AbortSignal; /** Shared circuit breaker for auth/provider failures */ circuitBreaker?: ProviderFailureCircuitBreaker; /** Optional context-bound check writer. Omit for analyze mode. */ @@ -134,6 +137,8 @@ export interface TriggerResult { auxiliaryModel?: string; synthesisModel?: string; error?: unknown; + /** The trigger matched but cancellation stopped it before dispatch. */ + pending?: boolean; /** Verification/merge events captured during post-processing, for provenance export. */ findingProcessingEvents?: FindingProcessingEvent[]; /** @@ -263,12 +268,16 @@ export async function executeTrigger( // Update skill check with results if (skillCheck && context.pullRequest) { try { - await skillCheck.complete(report, { - failOn, - reportOn, - minConfidence, - failCheck, - }); + if (deps.cancellationSignal?.aborted) { + await skillCheck.cancel(report); + } else { + await skillCheck.complete(report, { + failOn, + reportOn, + minConfidence, + failCheck, + }); + } } catch (error) { console.error(`::warning::Failed to update skill check for ${trigger.skill}: ${error}`); } diff --git a/packages/warden/src/action/workflow/base.test.ts b/packages/warden/src/action/workflow/base.test.ts index 02be566e..80747544 100644 --- a/packages/warden/src/action/workflow/base.test.ts +++ b/packages/warden/src/action/workflow/base.test.ts @@ -103,6 +103,22 @@ describe('findings output', () => { expect(payload.findingObservations).toHaveLength(1); }); + it('writes a completed cancelled findings artifact', () => { + process.env['GITHUB_WORKSPACE'] = tempDir; + + const filePath = writeFindingsOutput( + [createReport()], + createContext(tempDir), + [], + { outcome: 'cancelled' }, + ); + + const payload = FindingsOutputSchema.parse(JSON.parse(readFileSync(filePath, 'utf-8'))); + expect(payload.outcome).toBe('cancelled'); + expect(payload.summary.totalFindings).toBe(1); + expect(existsSync(`${filePath}.done`)).toBe(true); + }); + it('falls back to RUNNER_TEMP when no repo path is provided', () => { const runnerTemp = join(tempDir, 'runner-temp'); mkdirSync(runnerTemp); diff --git a/packages/warden/src/action/workflow/pr-workflow.test.ts b/packages/warden/src/action/workflow/pr-workflow.test.ts index 95a59741..b7e88826 100644 --- a/packages/warden/src/action/workflow/pr-workflow.test.ts +++ b/packages/warden/src/action/workflow/pr-workflow.test.ts @@ -78,6 +78,22 @@ vi.mock('../fix-evaluation/index.js', () => ({ postThreadReply: vi.fn(() => Promise.resolve()), })); +vi.mock('../service.js', async () => { + const actual: Record = await vi.importActual('../service.js'); + return { + ...actual, + recallActionMemoryFailOpen: vi.fn(), + }; +}); + +vi.mock('@sentry/node', async () => { + const actual: Record = await vi.importActual('@sentry/node'); + return { + ...actual, + captureException: vi.fn(), + }; +}); + // Mock base utilities that call process.exit or need system access vi.mock('./base.js', async () => { const actual = await vi.importActual('./base.js'); @@ -132,6 +148,9 @@ import { runPRWorkflow } from './pr-workflow.js'; import { clearSkillsCache } from '../../skills/loader.js'; import { AsyncWorkQueue } from '../../utils/index.js'; import { buildFindingsOutput } from '../../reporting/output.js'; +import { ActionCancellation } from '../cancellation.js'; +import { recallActionMemoryFailOpen } from '../service.js'; +import { Sentry } from '../../sentry.js'; // Type the mocks const mockRunSkillTask = vi.mocked(runSkillTask); @@ -143,6 +162,7 @@ const mockSetFailed = vi.mocked(setFailed); const mockWriteFindingsOutput = vi.mocked(writeFindingsOutput); const mockWriteFindingsOutputLive = vi.mocked(writeFindingsOutputLive); const mockClearStaleFindingsOutput = vi.mocked(clearStaleFindingsOutput); +const mockRecallActionMemoryFailOpen = vi.mocked(recallActionMemoryFailOpen); // Type helper for mocking Octokit responses type GetPullResponse = Awaited>; @@ -362,6 +382,58 @@ describe('runPRWorkflow', () => { }); describe('split action modes', () => { + it('does not publish when analyze mode is cancelled before execution', async () => { + const cancellation = new ActionCancellation(); + cancellation.request('SIGINT'); + const fetchSpy = vi.spyOn(globalThis, 'fetch'); + + await runPRWorkflow( + mockOctokit, + createDefaultInputs({ + mode: 'analyze', + serviceUrl: 'https://warden.example.com', + serviceToken: 'service-token', + serviceData: 'metrics', + serviceMemory: false, + }), + 'pull_request', + EVENT_PAYLOAD_PATH, + FIXTURES_DIR, + cancellation, + ); + + expect(fetchSpy).not.toHaveBeenCalled(); + expect(mockWriteFindingsOutput.mock.calls.at(-1)?.[3]?.outcome).toBe('cancelled'); + }); + + it('preserves recalled-memory linkage when analyze mode is cancelled after initialization', async () => { + const cancellation = new ActionCancellation(); + mockRecallActionMemoryFailOpen.mockImplementationOnce(async () => { + cancellation.request('SIGINT'); + return { + clientRecallId: 'recall-1', + memories: [{ id: 'memory-1', version: 2, kind: 'convention', content: 'Use stable identifiers.' }], + }; + }); + + await runPRWorkflow( + mockOctokit, + createDefaultInputs({ mode: 'analyze' }), + 'pull_request', + EVENT_PAYLOAD_PATH, + FIXTURES_DIR, + cancellation, + ); + + expect(mockWriteFindingsOutput.mock.calls.at(-1)?.[3]).toEqual( + expect.objectContaining({ + outcome: 'cancelled', + recalledMemories: [{ id: 'memory-1', version: 2 }], + memoryRecallId: 'recall-1', + }), + ); + }); + it('analyze mode writes findings without creating GitHub checks or reviews', async () => { const finding = createFinding(); const report = createSkillReport({ findings: [finding] }); @@ -494,6 +566,33 @@ describe('runPRWorkflow', () => { expect(finalOptions?.resolvedDefaults).toBeDefined(); }); + it('does not post checks or reviews from a cancelled analyze artifact', async () => { + const report = createSkillReport({ findings: [createFinding()] }); + const findingsFile = writeFindingsArtifact( + [report], + [{ triggerName: 'test-skill', skillName: 'test-skill', report }], + (output) => { + output.outcome = 'cancelled'; + }, + ); + + try { + await runPRWorkflow( + mockOctokit, + createDefaultInputs({ mode: 'report', findingsFile }), + 'pull_request', + EVENT_PAYLOAD_PATH, + FIXTURES_DIR, + ); + } finally { + rmSync(dirname(findingsFile), { recursive: true, force: true }); + } + + expect(mockWriteFindingsOutput.mock.calls.at(-1)?.[3]?.outcome).toBe('cancelled'); + expect(mockOctokit.checks.create).not.toHaveBeenCalled(); + expect(mockOctokit.pulls.createReview).not.toHaveBeenCalled(); + }); + it('preserves the canonical analyze artifact when report mode starts', async () => { const report = createSkillReport({ findings: [createFinding()] }); const sourceFindingsFile = writeFindingsArtifact([report], [ @@ -1692,6 +1791,119 @@ describe('runPRWorkflow', () => { ]); }); + it('flushes completed findings and finalizes as cancelled before posting reviews', async () => { + const cancellation = new ActionCancellation(); + const finding = createFinding(); + let notifyStarted!: () => void; + const started = new Promise((resolve) => { + notifyStarted = resolve; + }); + mockRunSkillTask.mockImplementation(async (taskOptions) => { + notifyStarted(); + await new Promise((resolve) => { + taskOptions.runnerOptions?.abortController?.signal.addEventListener('abort', () => resolve(), { + once: true, + }); + }); + return { + name: taskOptions.name, + report: createSkillReport({ skill: 'test-skill', findings: [finding] }), + }; + }); + + const workflow = runPRWorkflow( + mockOctokit, + createDefaultInputs({ parallel: 1 }), + 'pull_request', + EVENT_PAYLOAD_PATH, + DUPLICATE_TRIGGER_FIXTURES_DIR, + cancellation, + ); + await started; + cancellation.request('SIGTERM'); + await workflow; + + const finalCall = mockWriteFindingsOutput.mock.calls.at(-1); + expect(finalCall?.[0]).toEqual([ + expect.objectContaining({ findings: [finding] }), + ]); + expect(finalCall?.[3]?.outcome).toBe('cancelled'); + expect(finalCall?.[3]?.triggerResults).toEqual( + expect.arrayContaining([expect.objectContaining({ pending: true })]), + ); + expect(finalCall?.[3]?.skippedTriggers).toEqual( + expect.arrayContaining([expect.objectContaining({ reason: 'pending' })]), + ); + expect(mockFetchExistingComments).not.toHaveBeenCalled(); + expect(vi.mocked(mockOctokit.checks.update).mock.calls).toEqual( + expect.arrayContaining([ + [expect.objectContaining({ conclusion: 'cancelled' })], + [expect.objectContaining({ conclusion: 'cancelled' })], + ]), + ); + }); + + it('preserves memory linkage and captures a core-check failure when cancelled after setup', async () => { + const cancellation = new ActionCancellation(); + const checkError = new Error('check update failed'); + mockRecallActionMemoryFailOpen.mockResolvedValueOnce({ + clientRecallId: 'recall-1', + memories: [{ id: 'memory-1', version: 2, kind: 'convention', content: 'Use stable identifiers.' }], + }); + vi.mocked(mockOctokit.checks.create).mockImplementationOnce(async () => { + cancellation.request('SIGTERM'); + return { + data: { id: 1, html_url: 'https://example.com/check/1' }, + } as Awaited>; + }); + vi.mocked(mockOctokit.checks.update).mockRejectedValue(checkError); + + await runPRWorkflow( + mockOctokit, + createDefaultInputs(), + 'pull_request', + EVENT_PAYLOAD_PATH, + FIXTURES_DIR, + cancellation, + ); + expect(mockWriteFindingsOutput.mock.calls.at(-1)?.[3]).toEqual( + expect.objectContaining({ + outcome: 'cancelled', + recalledMemories: [{ id: 'memory-1', version: 2 }], + memoryRecallId: 'recall-1', + }), + ); + expect(Sentry.captureException).toHaveBeenCalledWith(checkError, { + tags: { operation: 'cancel_core_check' }, + }); + }); + + it('finalizes as cancelled when cancellation arrives while skipped checks are created', async () => { + const cancellation = new ActionCancellation(); + let nextCheckRunId = 1; + vi.mocked(mockOctokit.checks.create).mockImplementation(async () => { + const checkRunId = nextCheckRunId++; + if (checkRunId === 2) { + cancellation.request('SIGINT'); + } + return { + data: { id: checkRunId, html_url: `https://example.com/check/${checkRunId}` }, + } as Awaited>; + }); + + await runPRWorkflow( + mockOctokit, + createDefaultInputs(), + 'pull_request', + EVENT_PAYLOAD_PATH, + NO_MATCH_FIXTURES_DIR, + cancellation, + ); + + expect(mockWriteFindingsOutput.mock.calls.at(-1)?.[3]?.outcome).toBe('cancelled'); + expect(mockFetchExistingComments).not.toHaveBeenCalled(); + }); + it('clears stale findings before the first trigger settles, not lazily on the first live write', async () => { mockRunSkillTask.mockResolvedValue({ name: 'test-trigger', report: createSkillReport({ skill: 'test-skill' }) }); diff --git a/packages/warden/src/action/workflow/pr-workflow.ts b/packages/warden/src/action/workflow/pr-workflow.ts index 290bc27c..151be2d5 100644 --- a/packages/warden/src/action/workflow/pr-workflow.ts +++ b/packages/warden/src/action/workflow/pr-workflow.ts @@ -36,7 +36,7 @@ import { evaluateFixAttempts, postThreadReply } from '../fix-evaluation/index.js import type { EvaluateFixAttemptsResult, FixEvaluation } from '../fix-evaluation/index.js'; import { aggregateUsage } from '../../sdk/usage.js'; import { logAction, warnAction } from '../../cli/output/tty.js'; -import { formatCost, formatTokens, formatDuration } from '../../cli/output/formatters.js'; +import { formatCost, formatTokens, formatDuration, totalUsageStats } from '../../cli/output/formatters.js'; import { findBotReviewState } from '../review-state.js'; import type { BotReviewInfo } from '../review-state.js'; import type { ActionInputs } from '../inputs.js'; @@ -105,6 +105,7 @@ import { type SkillExecutionMeta, type BuildFindingsOutputOptions, } from '../../reporting/output.js'; +import { ActionCancellation } from '../cancellation.js'; // ----------------------------------------------------------------------------- // Phase Result Types @@ -238,23 +239,19 @@ function toSkippedTriggers( } /** - * A trigger that threw before producing a report has no `report`, so - * `toSkillExecutions`'s filter (which requires one) can never include it — - * without this, an errored trigger vanishes from the export entirely aside - * from a console warning and (in analyze/report mode) a `triggerResults` - * row. Surfacing it here instead keeps it visible in the same place a - * schedule-mode trigger error is now surfaced. + * Reportless results cannot appear in `skillExecutions`, so preserve them in + * `skippedTriggers`: failures as errors and cancellation-before-dispatch as pending. */ -function toErroredSkippedTriggers( +function toUnfinishedSkippedTriggers( results: TriggerResult[] ): NonNullable { return results - .filter((r) => r.error && !r.report) + .filter((r) => (r.pending || r.error) && !r.report) .map((r) => ({ skillName: r.skillName, triggerId: r.triggerId, triggerName: r.triggerName, - reason: 'error' as const, + reason: r.pending ? 'pending' as const : 'error' as const, })); } @@ -613,6 +610,12 @@ function createTriggerCheckReporter( ...checkOptions, ...options, }), + cancel: (report) => + updateSkillCheck(octokit, check.checkRunId, report, { + ...checkOptions, + conclusion: 'cancelled', + title: 'Analysis cancelled', + }), fail: (error) => failSkillCheck(octokit, check.checkRunId, error, checkOptions), }; }, @@ -627,6 +630,7 @@ async function executeAllTriggers( options: { checks?: TriggerCheckReporter; memoryRecall?: ActionMemoryRecall; + cancellation?: ActionCancellation; /** Fired after each trigger settles, with every result settled so far (completion order, not input order). */ onTriggerComplete?: (completedSoFar: TriggerResult[]) => void; } = {} @@ -639,32 +643,47 @@ async function executeAllTriggers( const circuitBreaker = new ProviderFailureCircuitBreaker({ abortController }); const completedSoFar: TriggerResult[] = []; - // Limit trigger dispatch too; the analysis queue only gates work after a trigger starts. - const results = await runPool( - matchedTriggers, - concurrency, - async (trigger) => { - const result = await executeTrigger(trigger, { - context, - anthropicApiKey: inputs.anthropicApiKey, - claudePath: runtimeEnv.pathToClaudeCodeExecutable, - globalFailOn: inputs.failOn, - globalReportOn: inputs.reportOn, - globalMaxFindings: inputs.maxFindings, - globalRequestChanges: inputs.requestChanges, - globalFailCheck: inputs.failCheck, - analysisQueue, - abortController, - circuitBreaker, - checks: options.checks, - historicalEvidence: options.memoryRecall?.historicalEvidence, - }); - completedSoFar.push(result); - options.onTriggerComplete?.([...completedSoFar]); - return result; - }, - { shouldAbort: () => abortController.signal.aborted }, + const cancelAnalysis = (): void => abortController.abort( + options.cancellation?.abortController.signal.reason, ); + if (options.cancellation?.requested) { + cancelAnalysis(); + } else { + options.cancellation?.abortController.signal.addEventListener('abort', cancelAnalysis, { once: true }); + } + + // Limit trigger dispatch too; the analysis queue only gates work after a trigger starts. + let results: TriggerResult[]; + try { + results = await runPool( + matchedTriggers, + concurrency, + async (trigger) => { + const result = await executeTrigger(trigger, { + context, + anthropicApiKey: inputs.anthropicApiKey, + claudePath: runtimeEnv.pathToClaudeCodeExecutable, + globalFailOn: inputs.failOn, + globalReportOn: inputs.reportOn, + globalMaxFindings: inputs.maxFindings, + globalRequestChanges: inputs.requestChanges, + globalFailCheck: inputs.failCheck, + analysisQueue, + abortController, + cancellationSignal: options.cancellation?.abortController.signal, + circuitBreaker, + checks: options.checks, + historicalEvidence: options.memoryRecall?.historicalEvidence, + }); + completedSoFar.push(result); + options.onTriggerComplete?.([...completedSoFar]); + return result; + }, + { shouldAbort: () => abortController.signal.aborted }, + ); + } finally { + options.cancellation?.abortController.signal.removeEventListener('abort', cancelAnalysis); + } // `runPool` never dispatches work items past an abort, so a matched trigger // the circuit breaker aborted before it started doesn't appear in `results` @@ -682,7 +701,9 @@ async function executeAllTriggers( skillExecutionId: trigger.skillExecutionId, triggerName: trigger.name, skillName: trigger.skill, - error: new Error('Trigger execution aborted before dispatch (circuit breaker tripped)'), + ...(options.cancellation?.requested + ? { pending: true } + : { error: new Error('Trigger execution aborted before dispatch (circuit breaker tripped)') }), })); completedSoFar.push(...abortedResults); options.onTriggerComplete?.([...completedSoFar]); @@ -1203,7 +1224,7 @@ async function finalizeWorkflow( triggerResults: toReplayTriggerResults(results), ...buildBaseOutputOptions(inputs, [ ...toSkippedTriggers(skippedTriggers, context), - ...toErroredSkippedTriggers(results), + ...toUnfinishedSkippedTriggers(results), ]), skillExecutions: toSkillExecutions(results), recalledMemories: memoryRecall?.memories.map(({ id, version }) => ({ id, version })), @@ -1394,6 +1415,38 @@ async function failCoreCheck( } } +/** Mark an in-progress core check as cancelled while preserving partial results. */ +async function cancelCoreCheck( + octokit: Octokit, + context: EventContext, + coreCheckId: number | undefined, + results: TriggerResult[], + postChecks: boolean, +): Promise { + const options = checkOptionsForPullRequest(context, postChecks); + if (!coreCheckId || !options) { + return; + } + + const reports = results.flatMap((result) => (result.report ? [result.report] : [])); + try { + await updateCoreCheck( + octokit, + coreCheckId, + { + ...buildCoreSummaryData(results, reports), + title: 'Warden cancelled', + message: 'Analysis was cancelled. Partial results are shown below.', + }, + 'cancelled', + options, + ); + } catch (error) { + Sentry.captureException(error, { tags: { operation: 'cancel_core_check' } }); + warnAction(`Failed to mark core check as cancelled: ${error}`); + } +} + async function runOrFailCore( octokit: Octokit, context: EventContext, @@ -1499,6 +1552,7 @@ function toReplayTriggerResults(results: TriggerResult[]): ReplayTriggerResult[] skillName: result.skillName, report: result.report, error: result.error, + pending: result.pending, findingProcessingEvents: result.findingProcessingEvents, auxiliaryModel: result.auxiliaryModel, synthesisModel: result.synthesisModel, @@ -1613,6 +1667,13 @@ function buildReportModeResults( }; } + if (outputResult.status === 'pending') { + return { + ...baseResult, + pending: true, + }; + } + if (outputResult.status === 'error' || !outputResult.report) { return { ...baseResult, @@ -1852,7 +1913,7 @@ async function finalizeReportWorkflow( triggerResults: toReplayTriggerResults(results), ...buildBaseOutputOptions(options.inputs, [ ...toSkippedTriggers(options.skippedTriggers ?? [], context), - ...toErroredSkippedTriggers(results), + ...toUnfinishedSkippedTriggers(results), ]), skillExecutions: toSkillExecutions(results), recalledMemories: options.recalledMemories, @@ -1979,6 +2040,92 @@ async function cleanupOrphanedComments( return findingObservations; } +interface CancelledPRFinalizationOptions { + inputs: ActionInputs; + context: EventContext; + results: TriggerResult[]; + skippedTriggers: ResolvedTrigger[]; + resolvedTriggers: ResolvedTrigger[]; + matchedTriggers: ResolvedTrigger[]; + findingObservations?: FindingObservation[]; + recalledMemories?: readonly { id: string; version: number }[]; + memoryRecallId?: string; + service?: ResolvedServiceOptions; + publish: boolean; + failOnWriteError?: boolean; +} + +function buildCancelledPRFinalizationBase(inputs: ActionInputs, initResult: InitResult) { + return { + inputs, + context: initResult.context, + skippedTriggers: initResult.skippedTriggers, + resolvedTriggers: initResult.resolvedTriggers, + matchedTriggers: initResult.matchedTriggers, + recalledMemories: initResult.memoryRecall?.memories.map(({ id, version }) => ({ id, version })), + memoryRecallId: initResult.memoryRecall?.clientRecallId, + service: initResult.service, + }; +} + +/** Finalize a cancelled PR run without performing any GitHub reporting writes. */ +async function finalizeCancelledPRRun( + options: CancelledPRFinalizationOptions, +): Promise<{ findingsCount: number; highCount: number; summary: string }> { + const reports = options.results.flatMap((result) => (result.report ? [result.report] : [])); + const outputs = computeWorkflowOutputs(reports); + setWorkflowOutputs(outputs); + const findingsOptions: BuildFindingsOutputOptions = { + outcome: 'cancelled', + triggerResults: toReplayTriggerResults(options.results), + ...buildBaseOutputOptions(options.inputs, [ + ...toSkippedTriggers(options.skippedTriggers, options.context), + ...toUnfinishedSkippedTriggers(options.results), + ]), + skillExecutions: toSkillExecutions(options.results), + recalledMemories: options.recalledMemories, + memoryRecallId: options.memoryRecallId, + configuredSkills: buildConfiguredSkillsList({ + allTriggers: options.resolvedTriggers, + matchedTriggers: options.matchedTriggers, + }), + }; + const findingObservations = options.findingObservations ?? []; + try { + const findingsPath = writeFindingsOutput( + reports, + options.context, + findingObservations, + findingsOptions, + ); + logAction(`Findings written to ${findingsPath}`); + } catch (error) { + const message = `Failed to write cancelled findings output: ${error}`; + if (options.failOnWriteError) { + setFailed(message); + } + warnAction(message); + } + + if (options.publish) { + await publishActionRunFailOpen( + options.service, + () => buildFindingsOutput(reports, options.context, findingObservations, findingsOptions), + ); + } + return outputs; +} + +function formatCancelledPreservation(findingsCount: number, results: TriggerResult[]): string { + const usages = results.flatMap((result) => { + const usage = totalUsageStats(result.report?.usage, result.report?.auxiliaryUsage); + return usage ? [usage] : []; + }); + const cost = aggregateUsage(usages).costUSD; + const executionLabel = results.length === 1 ? 'execution' : 'executions'; + return `${findingsCount} findings and ${formatCost(cost)} of usage from ${results.length} skill ${executionLabel}`; +} + /** * Run the analysis phase without GitHub reporting writes. * It executes matched triggers and writes the replay artifact for report mode. @@ -1986,7 +2133,8 @@ async function cleanupOrphanedComments( async function runAnalyzeMode( inputs: ActionInputs, initResult: InitResult, - span: { setAttribute: (name: string, value: number) => void } + span: { setAttribute: (name: string, value: number) => void }, + cancellation: ActionCancellation, ): Promise { const { context, @@ -2024,12 +2172,13 @@ async function runAnalyzeMode( }, () => executeAllTriggers(matchedTriggers, context, runnerConcurrency, inputs, { memoryRecall, + cancellation, onTriggerComplete: (completedSoFar) => { const reportsSoFar = completedSoFar.flatMap((r) => (r.report ? [r.report] : [])); writeFindingsOutputLive(reportsSoFar, context, [], { ...buildBaseOutputOptions(inputs, [ ...toSkippedTriggers(skippedTriggers, context), - ...toErroredSkippedTriggers(completedSoFar), + ...toUnfinishedSkippedTriggers(completedSoFar), ]), skillExecutions: toSkillExecutions(completedSoFar), configuredSkills: buildConfiguredSkillsList({ allTriggers: resolvedTriggers, matchedTriggers }), @@ -2040,15 +2189,26 @@ async function runAnalyzeMode( const reports = results.flatMap((result) => (result.report ? [result.report] : [])); const outputs = computeWorkflowOutputs(reports); - setWorkflowOutputs(outputs); span.setAttribute('warden.finding.count', reports.flatMap((r) => r.findings).length); + if (cancellation.requested) { + await finalizeCancelledPRRun({ + ...buildCancelledPRFinalizationBase(inputs, initResult), + results, + publish: false, + failOnWriteError: true, + }); + logAction(`Analysis cancelled: preserved ${formatCancelledPreservation(outputs.findingsCount, results)}`); + return; + } + + setWorkflowOutputs(outputs); try { const findingsPath = writeFindingsOutput(reports, context, [], { triggerResults: toReplayTriggerResults(results), ...buildBaseOutputOptions(inputs, [ ...toSkippedTriggers(skippedTriggers, context), - ...toErroredSkippedTriggers(results), + ...toUnfinishedSkippedTriggers(results), ]), skillExecutions: toSkillExecutions(results), recalledMemories: memoryRecall?.memories.map(({ id, version }) => ({ id, version })), @@ -2073,7 +2233,8 @@ async function runReportMode( inputs: ActionInputs, initResult: InitResult, repoPath: string, - span: { setAttribute: (name: string, value: number) => void } + span: { setAttribute: (name: string, value: number) => void }, + cancellation: ActionCancellation, ): Promise { const { context, @@ -2092,6 +2253,26 @@ async function runReportMode( memoryRecallId: findingsOutput.memoryRecallId, }; + const finalizeCancelledReport = async ( + cancelledResults: TriggerResult[], + findingObservations: FindingObservation[] = findingsOutput.findingObservations, + ): Promise => { + const outputs = await finalizeCancelledPRRun({ + inputs, + context, + results: cancelledResults, + skippedTriggers, + resolvedTriggers, + matchedTriggers, + findingObservations, + ...replayMemoryOptions, + service, + publish: true, + }); + span.setAttribute('warden.finding.count', outputs.findingsCount); + logAction(`Reporting cancelled: preserved ${formatCancelledPreservation(outputs.findingsCount, cancelledResults)}`); + }; + let results: TriggerResult[] = []; let previousReviewInfo: BotReviewInfo | null = null; let reviewPhase!: ReviewPhaseResult; @@ -2099,8 +2280,23 @@ async function runReportMode( let canResolveStale!: boolean; try { + if (findingsOutput.outcome === 'cancelled') { + const cancelledResults = findingsOutput.triggerResults?.length + ? buildReportModeResults(findingsOutput, matchedTriggers, inputs) + : []; + await finalizeCancelledReport(cancelledResults); + return; + } results = buildReportModeResults(findingsOutput, matchedTriggers, inputs); + if (cancellation.requested) { + await finalizeCancelledReport(results); + return; + } await createCompletedSkippedSkillChecks(octokit, context, skippedTriggers, postChecks); + if (cancellation.requested) { + await finalizeCancelledReport(results); + return; + } if (skipCoreCheck) { const outputs = { findingsCount: 0, highCount: 0, summary: skipCoreCheck.title }; @@ -2131,6 +2327,10 @@ async function runReportMode( }, 'neutral' ); + if (cancellation.requested) { + await finalizeCancelledReport(results); + return; + } await publishActionRunFailOpen(service, () => buildFindingsOutput([], context, [], findingsOptions)); logAction('Analysis complete: 0 total findings'); return; @@ -2144,6 +2344,10 @@ async function runReportMode( auxiliaryOptions, { failOnWriteError: true } ); + if (cancellation.requested) { + await finalizeCancelledReport(results, cleanupFindingObservations); + return; + } const outputs = { findingsCount: 0, highCount: 0, summary: 'No triggers matched' }; setWorkflowOutputs(outputs); const findingsOptions = { @@ -2172,6 +2376,10 @@ async function runReportMode( }, 'neutral' ); + if (cancellation.requested) { + await finalizeCancelledReport(results, cleanupFindingObservations); + return; + } await publishActionRunFailOpen( service, () => buildFindingsOutput([], context, cleanupFindingObservations, findingsOptions), @@ -2181,8 +2389,16 @@ async function runReportMode( } results = await createCompletedSkillChecksForReport(octokit, context, results, postChecks); + if (cancellation.requested) { + await finalizeCancelledReport(results); + return; + } previousReviewInfo = await fetchPreviousReviewInfo(octokit, context); + if (cancellation.requested) { + await finalizeCancelledReport(results); + return; + } if (previousReviewInfo) { logAction(`Previous Warden review state: ${previousReviewInfo.state}`); } @@ -2194,6 +2410,10 @@ async function runReportMode( failOnPostError: true, }), ); + if (cancellation.requested) { + await finalizeCancelledReport(results, reviewPhase.findingObservations); + return; + } triggerErrors = collectTriggerErrors(results); canResolveStale = shouldResolveStaleComments(results); @@ -2221,6 +2441,10 @@ async function runReportMode( reviewPhase.findingObservations.push(...resolutionResult.findingObservations); }, ); + if (cancellation.requested) { + await finalizeCancelledReport(results, reviewPhase.findingObservations); + return; + } await finalizeReportWorkflow( octokit, context, previousReviewInfo, @@ -2264,7 +2488,8 @@ export async function runPRWorkflow( inputs: ActionInputs, eventName: string, eventPath: string, - repoPath: string + repoPath: string, + cancellation = new ActionCancellation(), ): Promise { const reportInputPath = inputs.mode === 'report' && inputs.findingsFile ? resolveFindingsFilePath(inputs.findingsFile, repoPath) @@ -2319,12 +2544,23 @@ export async function runPRWorkflow( 'trace.id': traceId, }); + if (cancellation.requested && inputs.mode !== 'report') { + await finalizeCancelledPRRun({ + ...buildCancelledPRFinalizationBase(inputs, initResult), + results: [], + publish: inputs.mode === 'run', + failOnWriteError: inputs.mode === 'analyze', + }); + span.setAttribute('warden.finding.count', 0); + return; + } + if (inputs.mode === 'analyze') { - return runAnalyzeMode(inputs, initResult, span); + return runAnalyzeMode(inputs, initResult, span, cancellation); } if (inputs.mode === 'report') { - return runReportMode(octokit, inputs, initResult, repoPath, span); + return runReportMode(octokit, inputs, initResult, repoPath, span, cancellation); } const { coreCheckId, previousReviewInfo } = await Sentry.startSpan( @@ -2332,7 +2568,33 @@ export async function runPRWorkflow( () => setupGitHubState(octokit, context, postChecks), ); + const finalizeCancelledRun = async ( + results: TriggerResult[] = [], + findingObservations: FindingObservation[] = [], + ): Promise => { + if (!cancellation.requested) { + return false; + } + await cancelCoreCheck(octokit, context, coreCheckId, results, postChecks); + const outputs = await finalizeCancelledPRRun({ + ...buildCancelledPRFinalizationBase(inputs, initResult), + results, + findingObservations, + publish: true, + }); + span.setAttribute('warden.finding.count', outputs.findingsCount); + logAction(`Analysis cancelled: preserved ${formatCancelledPreservation(outputs.findingsCount, results)}`); + return true; + }; + + if (await finalizeCancelledRun()) { + return; + } + await completeSkippedSkillChecks(octokit, context, skippedTriggers, postChecks); + if (await finalizeCancelledRun()) { + return; + } if (skipCoreCheck) { setOutput('findings-count', 0); @@ -2348,6 +2610,9 @@ export async function runPRWorkflow( warnAction(`Failed to write findings output: ${error}`); } await completeSkippedCoreCheck(octokit, context, coreCheckId, skipCoreCheck, postChecks); + if (await finalizeCancelledRun()) { + return; + } await publishActionRunFailOpen(service, () => buildFindingsOutput([], context, [], findingsOptions)); return; } @@ -2360,6 +2625,9 @@ export async function runPRWorkflow( inputs, auxiliaryOptions ); + if (await finalizeCancelledRun([], cleanupFindingObservations)) { + return; + } setOutput('findings-count', 0); setOutput('high-count', 0); setOutput('summary', 'No triggers matched'); @@ -2376,6 +2644,9 @@ export async function runPRWorkflow( title: 'No triggers matched', message: 'No triggers matched for this event.', }, postChecks); + if (await finalizeCancelledRun([], cleanupFindingObservations)) { + return; + } await publishActionRunFailOpen( service, () => buildFindingsOutput([], context, cleanupFindingObservations, findingsOptions), @@ -2395,12 +2666,13 @@ export async function runPRWorkflow( () => executeAllTriggers(matchedTriggers, context, runnerConcurrency, inputs, { checks: createTriggerCheckReporter(octokit, context, postChecks), memoryRecall, + cancellation, onTriggerComplete: (completedSoFar) => { const reportsSoFar = completedSoFar.flatMap((r) => (r.report ? [r.report] : [])); writeFindingsOutputLive(reportsSoFar, context, [], { ...buildBaseOutputOptions(inputs, [ ...toSkippedTriggers(skippedTriggers, context), - ...toErroredSkippedTriggers(completedSoFar), + ...toUnfinishedSkippedTriggers(completedSoFar), ]), skillExecutions: toSkillExecutions(completedSoFar), configuredSkills: buildConfiguredSkillsList({ allTriggers: resolvedTriggers, matchedTriggers }), @@ -2442,6 +2714,10 @@ export async function runPRWorkflow( throw error; } + if (await finalizeCancelledRun(results)) { + return; + } + const gate = new ReviewFeedbackGate(octokit, context); const reviewPhase = await runOrFailCore( octokit, @@ -2453,6 +2729,9 @@ export async function runPRWorkflow( () => postReviewsAndTrackFailures(octokit, context, results, inputs, auxiliaryOptions, gate), ), ); + if (await finalizeCancelledRun(results, reviewPhase.findingObservations)) { + return; + } const triggerErrors = collectTriggerErrors(results); const canResolveStale = shouldResolveStaleComments(results); @@ -2485,6 +2764,9 @@ export async function runPRWorkflow( }, ), ); + if (await finalizeCancelledRun(results, reviewPhase.findingObservations)) { + return; + } await finalizeWorkflow( octokit, context, previousReviewInfo, coreCheckId, diff --git a/packages/warden/src/action/workflow/schedule.test.ts b/packages/warden/src/action/workflow/schedule.test.ts index 4d8b721f..b69eecc3 100644 --- a/packages/warden/src/action/workflow/schedule.test.ts +++ b/packages/warden/src/action/workflow/schedule.test.ts @@ -113,6 +113,7 @@ import { } from './base.js'; import { runScheduleWorkflow } from './schedule.js'; import { clearSkillsCache } from '../../skills/loader.js'; +import { ActionCancellation } from '../cancellation.js'; // Type the mocks const mockRunSkill = vi.mocked(runSkill); @@ -815,6 +816,39 @@ describe('runScheduleWorkflow', () => { expect(finalOptions?.skippedTriggers?.some((t) => t.reason === 'pending')).toBe(false); }); + it('flushes an in-progress report and finalizes as cancelled before creating an issue', async () => { + const cancellation = new ActionCancellation(); + const finding = createFinding(); + let notifyStarted!: () => void; + const started = new Promise((resolve) => { + notifyStarted = resolve; + }); + mockRunSkill.mockImplementation(async (_skill, _context, options) => { + notifyStarted(); + await new Promise((resolve) => { + options?.abortController?.signal.addEventListener('abort', () => resolve(), { once: true }); + }); + return createSkillReport({ findings: [finding] }); + }); + + const workflow = runScheduleWorkflow( + mockOctokit, + createDefaultInputs(), + SCHEDULE_FIXTURES, + cancellation, + ); + await started; + cancellation.request('SIGTERM'); + await workflow; + + const finalCall = mockWriteFindingsOutput.mock.calls.at(-1); + expect(finalCall?.[0]).toEqual([ + expect.objectContaining({ findings: [finding] }), + ]); + expect(finalCall?.[3]?.outcome).toBe('cancelled'); + expect(mockCreateOrUpdateIssue).not.toHaveBeenCalled(); + }); + it('marks a trigger with no matching files as skipped for no_changes', async () => { mockResolveSkillAsync .mockResolvedValueOnce({ name: 'test-skill-a', description: 'Test skill A', prompt: 'Review code' }) diff --git a/packages/warden/src/action/workflow/schedule.ts b/packages/warden/src/action/workflow/schedule.ts index 1076e28c..0692d6f3 100644 --- a/packages/warden/src/action/workflow/schedule.ts +++ b/packages/warden/src/action/workflow/schedule.ts @@ -46,6 +46,7 @@ import { writeFindingsOutputLive, } from './base.js'; import { captureActionTriggerError } from '../error-reporting.js'; +import { ActionCancellation } from '../cancellation.js'; interface SkippedScheduleTrigger { skillName: string; @@ -67,6 +68,7 @@ async function emitEmptyScheduleRun( inputs: ActionInputs, repoPath: string, service: ResolvedServiceOptions | undefined, + cancellation?: ActionCancellation, ): Promise { const fullName = process.env['GITHUB_REPOSITORY'] ?? ''; const [owner = '', name = ''] = fullName.split('/'); @@ -76,7 +78,10 @@ async function emitEmptyScheduleRun( repository: { owner, name, fullName, defaultBranch: '' }, repoPath, }; - const findingsOptions = buildBaseOutputOptions(inputs, []); + const findingsOptions = { + ...(cancellation?.requested ? { outcome: 'cancelled' as const } : {}), + ...buildBaseOutputOptions(inputs, []), + }; try { writeFindingsOutput([], context, [], findingsOptions); } catch (error) { @@ -91,11 +96,12 @@ async function emitEmptyScheduleRun( export async function runScheduleWorkflow( octokit: Octokit, inputs: ActionInputs, - repoPath: string + repoPath: string, + cancellation = new ActionCancellation(), ): Promise { return Sentry.startSpan( { op: 'workflow.run', name: 'review schedule' }, - (span) => runScheduleWorkflowInner(octokit, inputs, repoPath, span), + (span) => runScheduleWorkflowInner(octokit, inputs, repoPath, span, cancellation), ); } @@ -103,7 +109,8 @@ async function runScheduleWorkflowInner( octokit: Octokit, inputs: ActionInputs, repoPath: string, - workflowSpan: WorkflowSpan + workflowSpan: WorkflowSpan, + cancellation: ActionCancellation, ): Promise { const githubRepository = process.env['GITHUB_REPOSITORY']; setRepositoryScope(githubRepository); @@ -154,7 +161,7 @@ async function runScheduleWorkflowInner( setOutput('summary', 'No warden.toml found'); workflowSpan.setAttribute('warden.trigger.count', 0); workflowSpan.setAttribute('warden.finding.count', 0); - await emitEmptyScheduleRun(inputs, repoPath, service); + await emitEmptyScheduleRun(inputs, repoPath, service, cancellation); return; } throw error; @@ -174,7 +181,7 @@ async function runScheduleWorkflowInner( setOutput('high-count', 0); setOutput('summary', 'No schedule triggers configured'); workflowSpan.setAttribute('warden.finding.count', 0); - await emitEmptyScheduleRun(inputs, repoPath, service); + await emitEmptyScheduleRun(inputs, repoPath, service, cancellation); return; } @@ -208,7 +215,7 @@ async function runScheduleWorkflowInner( }; let memoryRecall: ActionMemoryRecall | undefined; - if (service?.memory) { + if (service?.memory && !cancellation.requested) { try { const recallContext = await buildScheduleEventContext({ patterns: [...new Set(scheduleTriggers.flatMap((trigger) => @@ -253,6 +260,16 @@ async function runScheduleWorkflowInner( // Process each schedule trigger for (const [triggerIndex, resolved] of scheduleTriggers.entries()) { + if (cancellation.requested) { + skippedTriggers.push(...scheduleTriggers.slice(triggerIndex).map((trigger) => ({ + skillName: trigger.skill, + triggerId: trigger.id, + triggerName: trigger.name, + reason: 'pending' as const, + }))); + break; + } + logGroup(`Running trigger: ${resolved.name} (skill: ${resolved.skill})`); const findingProcessingEvents: FindingProcessingEvent[] = []; let executionRecorded = false; @@ -317,6 +334,7 @@ async function runScheduleWorkflowInner( triggerName: resolved.name, historicalEvidence: memoryRecall?.historicalEvidence, pathToClaudeCodeExecutable: runtimeEnv.pathToClaudeCodeExecutable, + abortController: cancellation.abortController, callbacks: { onFindingProcessing: (event) => findingProcessingEvents.push(event), }, @@ -342,6 +360,18 @@ async function runScheduleWorkflowInner( skillExecutions.push(executionMeta); executionRecorded = true; + if (cancellation.requested) { + skippedTriggers.push(...scheduleTriggers.slice(triggerIndex + 1).map((trigger) => ({ + skillName: trigger.skill, + triggerId: trigger.id, + triggerName: trigger.name, + reason: 'pending' as const, + }))); + logGroupEnd(); + writeLiveSnapshot(triggerIndex); + break; + } + // Create/update issue with findings const scheduleConfig: Partial = resolved.schedule ?? {}; const issueTitle = scheduleConfig.issueTitle ?? `Warden: ${resolved.name}`; @@ -373,6 +403,19 @@ async function runScheduleWorkflowInner( writeLiveSnapshot(triggerIndex); } catch (error) { if (error instanceof ActionFailedError) throw error; + if (cancellation.requested) { + skippedTriggers.push(...scheduleTriggers.slice( + executionRecorded ? triggerIndex + 1 : triggerIndex, + ).map((trigger) => ({ + skillName: trigger.skill, + triggerId: trigger.id, + triggerName: trigger.name, + reason: 'pending' as const, + }))); + logGroupEnd(); + writeLiveSnapshot(triggerIndex); + break; + } captureActionTriggerError(error, { triggerName: resolved.name, skillName: resolved.skill, @@ -402,6 +445,7 @@ async function runScheduleWorkflowInner( // must land even when every trigger failed, or a terminated run is left // looking permanently in-progress to a follower of the live snapshots. const findingsOptions = { + ...(cancellation.requested ? { outcome: 'cancelled' as const } : {}), ...buildBaseOutputOptions(inputs, skippedTriggers), skillExecutions, recalledMemories: memoryRecall?.memories.map(({ id, version }) => ({ id, version })), @@ -419,6 +463,11 @@ async function runScheduleWorkflowInner( () => buildFindingsOutput(allReports, scheduleContext, [], findingsOptions), ); + if (cancellation.requested) { + console.log(`\nScheduled analysis cancelled: preserved ${totalFindings} findings`); + return; + } + handleTriggerErrors(triggerErrors, scheduleTriggers.length); if (shouldFailAction) { diff --git a/packages/warden/src/reporting/output.test.ts b/packages/warden/src/reporting/output.test.ts index 779de9bd..d316ede3 100644 --- a/packages/warden/src/reporting/output.test.ts +++ b/packages/warden/src/reporting/output.test.ts @@ -156,6 +156,11 @@ describe('findings output schema', () => { skillName: 'failed-skill', error: new Error('Token expired'), }, + { + triggerName: 'pending-trigger', + skillName: 'pending-skill', + pending: true, + }, ], }); @@ -176,6 +181,11 @@ describe('findings output schema', () => { message: 'Token expired', }, }, + { + triggerName: 'pending-trigger', + skillName: 'pending-skill', + status: 'pending', + }, ]); }); diff --git a/packages/warden/src/reporting/output.ts b/packages/warden/src/reporting/output.ts index f0928c2e..c2133ece 100644 --- a/packages/warden/src/reporting/output.ts +++ b/packages/warden/src/reporting/output.ts @@ -142,11 +142,18 @@ export const TriggerRunResultSchema = z.discriminatedUnion('status', [ report: z.never().optional(), error: TriggerErrorSchema, }), + TriggerRunResultBaseSchema.extend({ + status: z.literal('pending'), + report: z.never().optional(), + error: z.never().optional(), + }), ]); export const FindingsOutputSchema = z.object({ version: z.literal('1'), timestamp: z.string().datetime(), + /** Final run disposition. Optional while older analyze artifacts remain replayable. */ + outcome: z.enum(['success', 'failure', 'cancelled', 'skipped']).optional(), runAttempt: z.string().optional(), /** Which build of Warden produced this run. */ harness: HarnessSchema.optional(), @@ -243,6 +250,7 @@ export interface ReplayTriggerResult { skillName: string; report?: SkillReport; error?: unknown; + pending?: boolean; findingProcessingEvents?: FindingProcessingEvent[]; auxiliaryModel?: string; synthesisModel?: string; @@ -269,6 +277,7 @@ export interface BuildFindingsOutputOptions { timestamp?: string; runId?: string; runAttempt?: string; + outcome?: NonNullable; actionRef?: string; triggerResults?: ReplayTriggerResult[]; resolvedDefaults?: z.infer; @@ -374,6 +383,15 @@ function serializeTriggerResult(result: ReplayTriggerResult): z.infer { expect(result.mergedCount).toBe(0); }); + it('does not start consolidation after cancellation', async () => { + const findings = [ + makeFinding({ id: 'f1', location: { path: 'src/a.ts', startLine: 1 } }), + makeFinding({ id: 'f2', location: { path: 'src/b.ts', startLine: 1 } }), + ]; + const abortController = new AbortController(); + abortController.abort(); + + const result = await mergeCrossLocationFindings(findings, { + apiKey: 'test-key', + abortController, + }); + + expect(result.findings).toEqual(findings); + expect(mockCallHaiku).not.toHaveBeenCalled(); + }); + it('returns unchanged when LLM says no groups', async () => { const findings = [ makeFinding({ id: 'f1', title: 'Issue A', location: { path: 'src/a.ts', startLine: 1 } }), @@ -188,6 +205,7 @@ describe('mergeCrossLocationFindings', () => { runSynthesis, } as unknown as ReturnType); + const abortController = new AbortController(); try { const result = await mergeCrossLocationFindings(findings, { apiKey: 'test-key', @@ -195,11 +213,13 @@ describe('mergeCrossLocationFindings', () => { runtime: 'pi', model: 'openrouter/openai/gpt-5.6-luna', effort: 'high', + abortController, }); expect(result.mergedCount).toBe(0); expect(runSynthesis).toHaveBeenCalledWith(expect.objectContaining({ model: 'openrouter/openai/gpt-5.6-luna', effort: 'high', + abortController, })); } finally { getRuntimeSpy.mockRestore(); diff --git a/packages/warden/src/sdk/extract.ts b/packages/warden/src/sdk/extract.ts index 7ed9605d..e5c03b95 100644 --- a/packages/warden/src/sdk/extract.ts +++ b/packages/warden/src/sdk/extract.ts @@ -34,6 +34,7 @@ export interface AuxiliaryCallOptions { effort?: Effort; maxRetries?: number; agentName?: string; + abortController?: AbortController; } /** Return true when the selected runtime can authenticate outside a legacy Anthropic API key. */ @@ -540,6 +541,10 @@ export async function mergeCrossLocationFindings( const apiKey = options?.apiKey; const repoPath = options?.repoPath ?? '.'; + if (options?.abortController?.signal.aborted) { + return { findings, mergedCount: 0 }; + } + // Early exit: need at least 2 located findings to merge const withLocations = findings.filter((f) => f.location); if (withLocations.length < 2 || !canUseRuntimeAuth(options)) { @@ -575,6 +580,7 @@ Singletons should not appear. Return [] if no findings describe the same issue.` effort: options?.effort, maxTokens: 512, maxRetries: options?.maxRetries, + abortController: options?.abortController, }); if (!result.success) { diff --git a/packages/warden/src/sdk/haiku.ts b/packages/warden/src/sdk/haiku.ts index 5217b088..26c949ec 100644 --- a/packages/warden/src/sdk/haiku.ts +++ b/packages/warden/src/sdk/haiku.ts @@ -170,6 +170,7 @@ export interface CallHaikuOptions { maxTokens?: number; timeout?: number; maxRetries?: number; + abortController?: AbortController; } /** @@ -188,7 +189,7 @@ function inferPrefill(schema: z.ZodType): string | undefined { * Auto-prefills based on Zod schema type, extracts JSON, validates with Zod. */ export async function callHaiku(options: CallHaikuOptions): Promise> { - const { apiKey, prompt, schema, agentName, task, model = HAIKU_MODEL, maxTokens = DEFAULT_MAX_TOKENS, timeout = DEFAULT_TIMEOUT_MS, maxRetries = DEFAULT_AUXILIARY_MAX_RETRIES } = options; + const { apiKey, prompt, schema, agentName, task, model = HAIKU_MODEL, maxTokens = DEFAULT_MAX_TOKENS, timeout = DEFAULT_TIMEOUT_MS, maxRetries = DEFAULT_AUXILIARY_MAX_RETRIES, abortController } = options; return startTracedSpan( { @@ -218,11 +219,14 @@ export async function callHaiku(options: CallHaikuOptions): Promise( tools?: AuxiliaryTool[]; executeTool?: (name: string, input: Record) => Promise; maxIterations?: number; + abortController?: AbortController; } ): Promise> { if (!request.apiKey) { @@ -181,6 +182,7 @@ async function runStructured( maxTokens: request.maxTokens, timeout: request.timeout, maxRetries: request.maxRetries, + abortController: request.abortController, }); } diff --git a/packages/warden/src/sdk/runtimes/pi.ts b/packages/warden/src/sdk/runtimes/pi.ts index dcb3dfa5..f05a7c35 100644 --- a/packages/warden/src/sdk/runtimes/pi.ts +++ b/packages/warden/src/sdk/runtimes/pi.ts @@ -889,6 +889,7 @@ async function runStructured( tools?: AuxiliaryTool[]; executeTool?: (name: string, input: Record) => Promise; maxIterations?: number; + abortController?: AbortController; } ): Promise> { const customTools = toPiCustomTools(request.tools, request.executeTool); @@ -928,6 +929,7 @@ async function runStructured( maxTurns: request.maxIterations, maxRetries: request.maxRetries, timeout: request.timeout, + abortController: request.abortController, parentSpan: span, }); const result = normalizePiResult(run); diff --git a/packages/warden/src/sdk/runtimes/types.ts b/packages/warden/src/sdk/runtimes/types.ts index 83ff0157..4c474cb1 100644 --- a/packages/warden/src/sdk/runtimes/types.ts +++ b/packages/warden/src/sdk/runtimes/types.ts @@ -141,6 +141,7 @@ export interface SynthesisRunRequest { maxTokens?: number; timeout?: number; maxRetries?: number; + abortController?: AbortController; } export interface Runtime { diff --git a/packages/warden/src/sentry.ts b/packages/warden/src/sentry.ts index 4db3937d..4b77e725 100644 --- a/packages/warden/src/sentry.ts +++ b/packages/warden/src/sentry.ts @@ -291,7 +291,7 @@ export function emitRunMetric(): void { /** Emit the final outcome of a GitHub Action invocation, including startup failures. */ export function emitActionRunMetric( - outcome: 'success' | 'failure', + outcome: 'success' | 'failure' | 'cancelled', stage: 'input' | 'environment' | 'dispatch', errorCode?: ErrorCode ): void { diff --git a/packages/warden/src/service/findings.ts b/packages/warden/src/service/findings.ts index c4ee2465..ab691164 100644 --- a/packages/warden/src/service/findings.ts +++ b/packages/warden/src/service/findings.ts @@ -162,7 +162,7 @@ export function buildFindingsServiceRunEnvelope( wardenVersion: output.harness?.version ?? getVersion(), startedAt: new Date(completedAt.getTime() - durationMs), completedAt, - outcome: reports.some((item) => item.report.error) ? 'failure' : 'success', + outcome: output.outcome ?? (reports.some((item) => item.report.error) ? 'failure' : 'success'), repository: { provider: 'github', owner: output.repository.owner,