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..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,6 +545,57 @@ describe.each(['linear', 'task'] as const)('%s setup signals', (mode) => { }); }); +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 = 'Installed the SDK'; + 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(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('Installed the SDK'); + 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(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(run(abortingRun())).resolves.toEqual({ kind: 'success' }); + }); +}); + describe('a task turn after complete_task', () => { const REFUSAL = '[posthog:abort] Request is out of scope'; const runOrchestrated = (orchestrator: { @@ -559,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 77d688045..5dd7b6dc4 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'; @@ -165,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 @@ -314,14 +320,17 @@ 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'); // 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. @@ -639,6 +648,7 @@ export const piBackend: AgentHarness = { const blocks = textBlocks(event.message); const assistant = blocks.join('').trim(); policyAbort(assistant); + 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) { @@ -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/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', 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 —