fix(pi): end the run on a skill's [ABORT] marker

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();

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;

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<typeof bindPiCancellation> | undefined;
     let timedOut = false;
     let timeoutAbort: Promise<void> | 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',

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<AgentResult> {
   let mcpCleanup: (() => void) | undefined;
   let aioFailed = true;
   let cancellation: ReturnType<typeof bindPiCancellation> | 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<AgentResult> {
           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<AgentResult> {
         !security.state.criticalViolation &&
         !inputs.signal?.aborted &&
         !terminal &&
-        !abortReason &&
+        !abort &&
         !setupSignalFailure(signals) &&
         !isSettled(orchestrator)
       ) {
@@ -547,7 +553,7 @@ export async function runPiTask(inputs: TaskRunInputs): Promise<AgentResult> {
         requestRemark &&
         !security.state.criticalViolation &&
         !terminal &&
-        !abortReason &&
+        !abort &&
         !setupSignalFailure(signals) &&
         !inputs.signal?.aborted
       ) {
@@ -574,15 +580,15 @@ export async function runPiTask(inputs: TaskRunInputs): Promise<AgentResult> {
       };
     }
 
-    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',

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 —