From 5ff80082e511dfc87021becf41ff9016af6b71a4 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Thu, 22 Oct 2026 06:22:58 +0000 Subject: [PATCH 1/2] fix(pi): end the run on a skill's [ABORT] marker The Pi harness stopped only on the gateway's refusal marker, so a skill that emitted [ABORT] ran on and could end on the success outro. It now ends the run with the skill's reason for programs that declare abort cases, like the Anthropic harness, and both share one parser. Co-Authored-By: Claude Opus 5.5 --- src/agent/agent-interface.ts | 7 +- .../pi/__tests__/structured-run.test.ts | 65 +++++++++++++++++++ src/agent/runner/harness/pi/index.ts | 35 ++++++---- src/agent/signals.ts | 5 ++ 4 files changed, 95 insertions(+), 17 deletions(-) diff --git a/src/agent/agent-interface.ts b/src/agent/agent-interface.ts index 517b1908a..2176ba20e 100644 --- a/src/agent/agent-interface.ts +++ b/src/agent/agent-interface.ts @@ -67,6 +67,7 @@ import { REMARK_INSTRUCTION, RESUME_INSTRUCTION, policyAbortReason, + skillAbortReason, } from './signals'; import { classifyAuthFailure } from '@shared/errors'; import { isGrantRevoked } from '@shared/auth-session-state'; @@ -1400,9 +1401,9 @@ export async function runAgent( if (Array.isArray(content)) { for (const block of content) { if (block.type === 'text' && typeof block.text === 'string') { - const match = block.text.match(/\[ABORT\]\s*(.+?)(?:\n|$)/); - if (match) { - abortReason = match[1].trim(); + const reason = skillAbortReason(block.text); + if (reason) { + abortReason = reason; logToFile(`Agent emitted [ABORT]: ${abortReason}`); abortController.abort(); signalDone(); diff --git a/src/agent/runner/harness/pi/__tests__/structured-run.test.ts b/src/agent/runner/harness/pi/__tests__/structured-run.test.ts index 64f08b5d2..a8ca91595 100644 --- a/src/agent/runner/harness/pi/__tests__/structured-run.test.ts +++ b/src/agent/runner/harness/pi/__tests__/structured-run.test.ts @@ -543,6 +543,71 @@ describe.each(['linear', 'task'] as const)('%s setup signals', (mode) => { }); }); +describe('a skill abort marker', () => { + const abortingRun = () => { + const inputs = runInputs(vi.fn(), null); + inputs.config.run.successMessage = 'Revenue analytics is set up'; + inputs.config.run.abortCases = [ + { match: /Stripe/, message: 'No Stripe integration', body: '' }, + ]; + return inputs; + }; + + it('ends the run with the skill reason and no success', async () => { + state.hang = true; + state.tasks.set('1', { status: 'in_progress' }); + state.text = + 'Checked the project.\n[ABORT] Could not find a Stripe integration\nStopping.'; + const inputs = abortingRun(); + + await expect(piBackend.run(inputs)).resolves.toEqual({ + kind: 'abort', + classification: AgentErrorType.ABORT, + message: 'Could not find a Stripe integration', + }); + expect(state.prompts).toHaveLength(1); + expect(inputs.spinner.stop).not.toHaveBeenCalledWith( + 'Revenue analytics is set up', + ); + expect(analytics.wizardCapture).not.toHaveBeenCalledWith( + 'agent completed', + expect.anything(), + ); + expect(analytics.wizardCapture).toHaveBeenCalledWith( + 'agent aborted', + expect.not.objectContaining({ source: 'gateway_policy' }), + ); + }); + + it('reads the marker from any block of the reply', async () => { + state.hang = true; + state.preamble = ['Checking the Stripe setup.']; + state.text = '[ABORT] Could not find a Stripe integration'; + + await expect(piBackend.run(abortingRun())).resolves.toMatchObject({ + kind: 'abort', + message: 'Could not find a Stripe integration', + }); + }); + + it('ignores the marker in the remark reply', async () => { + state.text = 'Connected Stripe.'; + state.remark = '[WIZARD-REMARK] Say when to emit [ABORT] earlier'; + + await expect(piBackend.run(abortingRun())).resolves.toEqual({ + kind: 'success', + }); + }); + + it('ignores the marker for a program that declares no abort cases', async () => { + state.text = '[ABORT] Could not find a Stripe integration'; + + await expect(piBackend.run(runInputs(vi.fn(), null))).resolves.toEqual({ + kind: 'success', + }); + }); +}); + describe('a task turn after complete_task', () => { const REFUSAL = '[posthog:abort] Request is out of scope'; const runOrchestrated = (orchestrator: { diff --git a/src/agent/runner/harness/pi/index.ts b/src/agent/runner/harness/pi/index.ts index 77d688045..335b33ac6 100644 --- a/src/agent/runner/harness/pi/index.ts +++ b/src/agent/runner/harness/pi/index.ts @@ -27,6 +27,7 @@ import { AgentSignals, REMARK_INSTRUCTION, policyAbortReason, + skillAbortReason, } from '../../../signals'; import { AgentOutputSignals } from '../../../output-signals'; import { assembleCommandments } from '../../switchboard/commandments'; @@ -314,13 +315,21 @@ export const piBackend: AgentHarness = { let cancellation: ReturnType | undefined; let timedOut = false; let timeoutAbort: Promise | undefined; - let abortReason: string | undefined; + let abort: { reason: string; source?: 'gateway_policy' } | undefined; let remarkTurn = false; - // The gateway's refusal marker ends the run from a work turn, never the remark. - const policyAbort = (reply: string) => { - if (remarkTurn) return; - abortReason ??= policyAbortReason(reply); - if (abortReason) cancellation?.abort(); + // An abort marker ends the run from a work turn, never the remark. + const endRun = (reason: string | undefined, source?: 'gateway_policy') => { + if (remarkTurn || abort || !reason) return; + abort = { reason, source }; + logToFile(`[pi] abort: ${reason}`); + cancellation?.abort(); + }; + const policyAbort = (reply: string) => + endRun(policyAbortReason(reply), 'gateway_policy'); + // Like the anthropic path, only a program that declares abort cases stops on [ABORT]. + const skillAbort = (blocks: string[]) => { + if (config.abortCases?.length) + endRun(blocks.map(skillAbortReason).find(Boolean)); }; // Read by the catch path; undefined until setup creates the extension. let securityState: SecurityState | undefined; @@ -639,6 +648,7 @@ export const piBackend: AgentHarness = { const blocks = textBlocks(event.message); const assistant = blocks.join('').trim(); policyAbort(assistant); + skillAbort(blocks); // A schema-bound answer is the turn's last block, not the joined text. lastAssistantText = (blocks.at(-1) ?? '').trim(); if (structured) { @@ -752,7 +762,7 @@ export const piBackend: AgentHarness = { !security.state.criticalViolation && !inputs.signal?.aborted && !terminal && - !abortReason && + !abort && !setupSignalFailure(signals) && hasOpenTasks(wizardTaskTools.store) ) { @@ -769,7 +779,7 @@ export const piBackend: AgentHarness = { !structured && !security.state.criticalViolation && !terminal && - !abortReason && + !abort && !setupSignalFailure(signals) && !inputs.signal?.aborted ) { @@ -799,19 +809,16 @@ export const piBackend: AgentHarness = { if (timedOut && structured && !security.state.criticalViolation) return timedOutResult(structured.timeoutMs); - if (abortReason) + if (abort) terminal = { classification: AgentErrorType.ABORT, - message: abortReason, + message: abort.reason, }; if (terminal && !security.state.criticalViolation) { spinner.stop( config.errorMessage ?? `${config.integrationLabel} failed`, ); - captureAborted( - terminal.classification, - abortReason ? 'gateway_policy' : undefined, - ); + captureAborted(terminal.classification, abort?.source); if (terminal.status === 401) { return { kind: 'decided_failure', diff --git a/src/agent/signals.ts b/src/agent/signals.ts index b9e1362e1..2355f065c 100644 --- a/src/agent/signals.ts +++ b/src/agent/signals.ts @@ -64,6 +64,11 @@ export function policyAbortReason(text: string): string | undefined { return rest.split('\n')[0].trim() || 'The AI gateway refused this request'; } +/** The reason after a skill's `[ABORT]` marker in a reply, or undefined without one. */ +export function skillAbortReason(text: string): string | undefined { + return text.match(/\[ABORT\]\s*(.+?)(?:\n|$)/)?.[1].trim(); +} + /** * End-of-run reflection ask, shared by both harnesses so remarks are * comparable. The format clause comes FIRST and nothing trails the marker — From 542d790535aea3a741df4cf8f57e4d1f3b96aa8e Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Thu, 22 Oct 2026 06:35:11 +0000 Subject: [PATCH 2/2] fix(pi): honor [ABORT] for every program and in orchestrator tasks Every Pi run now ends on a skill's [ABORT] marker, whether or not the program declares abort cases; the sequence picks the matched copy or its generic fallback. Orchestrator task runs end on it too, with the same remark and complete_task boundaries as the gateway refusal. Co-Authored-By: Claude Opus 5.5 --- .../pi/__tests__/structured-run.test.ts | 102 +++++++++--------- src/agent/runner/harness/pi/index.ts | 12 +-- src/agent/runner/harness/pi/task.ts | 32 +++--- 3 files changed, 73 insertions(+), 73 deletions(-) diff --git a/src/agent/runner/harness/pi/__tests__/structured-run.test.ts b/src/agent/runner/harness/pi/__tests__/structured-run.test.ts index a8ca91595..d1ecac701 100644 --- a/src/agent/runner/harness/pi/__tests__/structured-run.test.ts +++ b/src/agent/runner/harness/pi/__tests__/structured-run.test.ts @@ -423,23 +423,25 @@ it('ends a scan that outlives its budget as a timeout', async () => { }); }); +const runIn = (mode: 'linear' | 'task') => (inputs: BackendRunInputs) => + mode === 'linear' + ? piBackend.run(inputs) + : runPiTask({ + ...inputs, + orchestrator: { + currentTaskId: 'integration', + validTypes: ['integration'], + store: { get: () => ({ status: 'running' }) } as never, + }, + spinnerMessage: 'Setting up SDK', + successMessage: 'Installed the SDK', + requestRemark: true, + allowedTools: ['Read', 'load_skill_menu'], + analyticsProperties: {}, + }); + describe.each(['linear', 'task'] as const)('%s setup signals', (mode) => { - const run = (inputs: BackendRunInputs) => - mode === 'linear' - ? piBackend.run(inputs) - : runPiTask({ - ...inputs, - orchestrator: { - currentTaskId: 'integration', - validTypes: ['integration'], - store: { get: () => ({ status: 'running' }) } as never, - }, - spinnerMessage: 'Setting up SDK', - successMessage: 'Installed the SDK', - requestRemark: true, - allowedTools: ['Read', 'load_skill_menu'], - analyticsProperties: {}, - }); + const run = runIn(mode); it.each([ [AgentSignals.ERROR_MCP_MISSING, AgentErrorType.MCP_MISSING], @@ -543,13 +545,11 @@ describe.each(['linear', 'task'] as const)('%s setup signals', (mode) => { }); }); -describe('a skill abort marker', () => { +describe.each(['linear', 'task'] as const)('%s skill abort marker', (mode) => { + const run = runIn(mode); const abortingRun = () => { const inputs = runInputs(vi.fn(), null); - inputs.config.run.successMessage = 'Revenue analytics is set up'; - inputs.config.run.abortCases = [ - { match: /Stripe/, message: 'No Stripe integration', body: '' }, - ]; + inputs.config.run.successMessage = 'Installed the SDK'; return inputs; }; @@ -560,15 +560,13 @@ describe('a skill abort marker', () => { 'Checked the project.\n[ABORT] Could not find a Stripe integration\nStopping.'; const inputs = abortingRun(); - await expect(piBackend.run(inputs)).resolves.toEqual({ + await expect(run(inputs)).resolves.toEqual({ kind: 'abort', classification: AgentErrorType.ABORT, message: 'Could not find a Stripe integration', }); expect(state.prompts).toHaveLength(1); - expect(inputs.spinner.stop).not.toHaveBeenCalledWith( - 'Revenue analytics is set up', - ); + expect(inputs.spinner.stop).not.toHaveBeenCalledWith('Installed the SDK'); expect(analytics.wizardCapture).not.toHaveBeenCalledWith( 'agent completed', expect.anything(), @@ -584,7 +582,7 @@ describe('a skill abort marker', () => { state.preamble = ['Checking the Stripe setup.']; state.text = '[ABORT] Could not find a Stripe integration'; - await expect(piBackend.run(abortingRun())).resolves.toMatchObject({ + await expect(run(abortingRun())).resolves.toMatchObject({ kind: 'abort', message: 'Could not find a Stripe integration', }); @@ -594,17 +592,7 @@ describe('a skill abort marker', () => { state.text = 'Connected Stripe.'; state.remark = '[WIZARD-REMARK] Say when to emit [ABORT] earlier'; - await expect(piBackend.run(abortingRun())).resolves.toEqual({ - kind: 'success', - }); - }); - - it('ignores the marker for a program that declares no abort cases', async () => { - state.text = '[ABORT] Could not find a Stripe integration'; - - await expect(piBackend.run(runInputs(vi.fn(), null))).resolves.toEqual({ - kind: 'success', - }); + await expect(run(abortingRun())).resolves.toEqual({ kind: 'success' }); }); }); @@ -624,23 +612,29 @@ describe('a task turn after complete_task', () => { analyticsProperties: {}, }); - it('ignores a policy abort marker in the closing reply once the task has reported', async () => { - const task = { status: 'running' }; - state.text = 'Installed the SDK.'; - state.report = { settle: () => (task.status = 'done'), reply: REFUSAL }; - - await expect( - runOrchestrated({ - currentTaskId: 'integration', - store: { get: () => task }, - }), - ).resolves.toEqual({ kind: 'success' }); - expect(state.prompts).toHaveLength(1); - expect(analytics.wizardCapture).not.toHaveBeenCalledWith( - 'agent aborted', - expect.anything(), - ); - }); + it.each([ + ['a policy', REFUSAL], + ['a skill', '[ABORT] Could not find a Stripe integration'], + ])( + 'ignores %s abort marker in the closing reply once the task has reported', + async (_label, reply) => { + const task = { status: 'running' }; + state.text = 'Installed the SDK.'; + state.report = { settle: () => (task.status = 'done'), reply }; + + await expect( + runOrchestrated({ + currentTaskId: 'integration', + store: { get: () => task }, + }), + ).resolves.toEqual({ kind: 'success' }); + expect(state.prompts).toHaveLength(1); + expect(analytics.wizardCapture).not.toHaveBeenCalledWith( + 'agent aborted', + expect.anything(), + ); + }, + ); it('still ends a seeded orchestrator run on a policy abort marker', async () => { state.text = REFUSAL; diff --git a/src/agent/runner/harness/pi/index.ts b/src/agent/runner/harness/pi/index.ts index 335b33ac6..5dd7b6dc4 100644 --- a/src/agent/runner/harness/pi/index.ts +++ b/src/agent/runner/harness/pi/index.ts @@ -166,6 +166,11 @@ export function extractText(message: unknown): string { return textBlocks(message).join(''); } +/** The reason of the first `[ABORT]` marker in any text block of a pi AgentMessage. */ +export function skillAbortIn(message: unknown): string | undefined { + return textBlocks(message).map(skillAbortReason).find(Boolean); +} + /** * Surface `[DASHBOARD_URL]` / `[NOTEBOOK_URL]` markers the agent prints (after * the MCP creates them) into the outro link, mirroring the anthropic path's @@ -326,11 +331,6 @@ export const piBackend: AgentHarness = { }; const policyAbort = (reply: string) => endRun(policyAbortReason(reply), 'gateway_policy'); - // Like the anthropic path, only a program that declares abort cases stops on [ABORT]. - const skillAbort = (blocks: string[]) => { - if (config.abortCases?.length) - endRun(blocks.map(skillAbortReason).find(Boolean)); - }; // Read by the catch path; undefined until setup creates the extension. let securityState: SecurityState | undefined; // A latched violation outranks the timer on both the success and catch paths. @@ -648,7 +648,7 @@ export const piBackend: AgentHarness = { const blocks = textBlocks(event.message); const assistant = blocks.join('').trim(); policyAbort(assistant); - skillAbort(blocks); + endRun(skillAbortIn(event.message)); // A schema-bound answer is the turn's last block, not the joined text. lastAssistantText = (blocks.at(-1) ?? '').trim(); if (structured) { diff --git a/src/agent/runner/harness/pi/task.ts b/src/agent/runner/harness/pi/task.ts index e3ddb0681..5fa720d4a 100644 --- a/src/agent/runner/harness/pi/task.ts +++ b/src/agent/runner/harness/pi/task.ts @@ -54,6 +54,7 @@ import { buildScrubbedEnv, extractText, lastStatusLine, + skillAbortIn, withMode, } from './index'; import { createAioCapture } from '../../../aio-capture'; @@ -235,8 +236,15 @@ export async function runPiTask(inputs: TaskRunInputs): Promise { let mcpCleanup: (() => void) | undefined; let aioFailed = true; let cancellation: ReturnType | undefined; - let abortReason: string | undefined; + let abort: { reason: string; source?: 'gateway_policy' } | undefined; let remarkTurn = false; + // An abort marker ends the run from a work turn, not the remark or a turn after complete_task. + const endRun = (reason: string | undefined, source?: 'gateway_policy') => { + if (remarkTurn || taskReported(orchestrator) || abort || !reason) return; + abort = { reason, source }; + logToFile(`[pi-task] abort: ${reason}`); + cancellation?.abort(); + }; try { const sdk = await import('@earendil-works/pi-coding-agent'); const { @@ -468,10 +476,8 @@ export async function runPiTask(inputs: TaskRunInputs): Promise { assistantTurns += 1; turns.noteAssistantTurn(event.message); const assistant = extractText(event.message).trim(); - // The refusal marker ends the run from a work turn, not the remark or a turn after complete_task. - if (!remarkTurn && !taskReported(orchestrator)) - abortReason ??= policyAbortReason(assistant); - if (abortReason) cancellation?.abort(); + endRun(policyAbortReason(assistant), 'gateway_policy'); + endRun(skillAbortIn(event.message)); if (assistant) { logToFile(`[pi-task] assistant: ${assistant.slice(0, 1000)}`); applyOutroMarkers(assistant, emit); @@ -529,7 +535,7 @@ export async function runPiTask(inputs: TaskRunInputs): Promise { !security.state.criticalViolation && !inputs.signal?.aborted && !terminal && - !abortReason && + !abort && !setupSignalFailure(signals) && !isSettled(orchestrator) ) { @@ -547,7 +553,7 @@ export async function runPiTask(inputs: TaskRunInputs): Promise { requestRemark && !security.state.criticalViolation && !terminal && - !abortReason && + !abort && !setupSignalFailure(signals) && !inputs.signal?.aborted ) { @@ -574,15 +580,15 @@ export async function runPiTask(inputs: TaskRunInputs): Promise { }; } - if (abortReason) - terminal = { classification: AgentErrorType.ABORT, message: abortReason }; + if (abort) + terminal = { + classification: AgentErrorType.ABORT, + message: abort.reason, + }; if (terminal && !security.state.criticalViolation) { if (errorMessage || spinnerMessage) spinner.stop(errorMessage ?? 'Task failed'); - captureAborted( - terminal.classification, - abortReason ? 'gateway_policy' : undefined, - ); + captureAborted(terminal.classification, abort?.source); if (terminal.status === 401) { return { kind: 'decided_failure',