feat: add interrupt agent (PLA-7)

- Add WebSocket action interrupt_agent to stop current agent run
- Strategies: optional interruptAgent() implemented for opencode, claude-code,
  gemini, openai-codex (kill process), mock (clear timeout)
- Orchestrator: handle interrupt, finish stream with accumulated text and
  stream_end; extract finishStream() to avoid duplication
- Chat: Stop button (muted style) when streaming, interruptAgent in hook
- Docs: document interrupt_agent in API.md
- Tests: orchestrator (interrupt when idle/processing), MockStrategy
  (interruptAgent rejects with INTERRUPTED), useChatWebSocket (interruptAgent)
- Fix: opencode ToolEvent kind tool_call, remove unused _hasSession
- Bump version to 1.6.8

Author: vyakymenko

apps/api/src/app/orchestrator/orchestrator.service.test.ts

@@ -166,4 +166,25 @@ describe('OrchestratorService', () => {
     expect(thinkingStep?.data.title).toBe('Generating response');
     expect(thinkingStep?.data.status).toBe('processing');
   });
+
+  test('handleClientMessage interrupt_agent when not processing does nothing', async () => {
+    const orch = await createOrchestrator();
+    const events: Array<{ type: string }> = [];
+    orch.outbound.subscribe((ev) => events.push(ev));
+    orch.handleClientMessage({ action: WS_ACTION.INTERRUPT_AGENT });
+    expect(events.length).toBe(0);
+  });
+
+  test('handleClientMessage interrupt_agent when processing sends stream_end with accumulated', async () => {
+    const orch = await createOrchestrator();
+    orch.isAuthenticated = true;
+    const events: Array<{ type: string }> = [];
+    orch.outbound.subscribe((ev) => events.push(ev));
+    const promise = orch.handleClientMessage({ action: WS_ACTION.SEND_CHAT_MESSAGE, text: 'hi' });
+    orch.handleClientMessage({ action: WS_ACTION.INTERRUPT_AGENT });
+    await promise;
+    expect(events.some((e) => e.type === WS_EVENT.STREAM_START)).toBe(true);
+    expect(events.some((e) => e.type === WS_EVENT.STREAM_END)).toBe(true);
+    expect(orch.isProcessing).toBe(false);
+  });
 });

apps/api/src/app/orchestrator/orchestrator.service.ts

@@ -20,6 +20,7 @@ import type {
   ThinkingStep,
   ToolEvent,
 } from '../strategies/strategy.types';
+import { INTERRUPTED_MESSAGE } from '../strategies/strategy.types';
 import { StrategyRegistryService } from '../strategies/strategy-registry.service';
 import {
   AUTH_STATUS as AUTH_STATUS_VAL,
@@ -83,6 +84,24 @@ export class OrchestratorService implements OnModuleInit {
     this.outbound$.next({ type, data });
   }
 
+  private finishStream(
+    accumulated: string,
+    stepId: string,
+    step: ThinkingStep
+  ): void {
+    const finalText = accumulated || 'The agent produced no visible output.';
+    this.messageStore.add('assistant', finalText);
+    void this.phoenixSync.syncMessages(JSON.stringify(this.messageStore.all()));
+    this._send(WS_EVENT.THINKING_STEP, {
+      id: stepId,
+      title: step.title,
+      status: 'complete',
+      details: step.details,
+      timestamp: new Date().toISOString(),
+    });
+    this._send(WS_EVENT.STREAM_END, {});
+  }
+
   ensureStrategySettings(): void {
     this.strategy.ensureSettings?.();
   }
@@ -138,6 +157,11 @@ export class OrchestratorService implements OnModuleInit {
       case WS_ACTION.SET_MODEL:
         this.handleSetModel(msg.model ?? '');
         break;
+      case WS_ACTION.INTERRUPT_AGENT:
+        if (this.isProcessing) {
+          this.strategy.interruptAgent?.();
+        }
+        break;
       default:
         this.logger.warn(`Unknown action: ${action}`);
     }
@@ -374,24 +398,14 @@ export class OrchestratorService implements OnModuleInit {
         accumulated += chunk;
         this._send(WS_EVENT.STREAM_CHUNK, { text: chunk });
       }, callbacks, systemPrompt || undefined);
-
-      const finalText =
-        accumulated || 'The agent produced no visible output.';
-      this.messageStore.add('assistant', finalText);
-      void this.phoenixSync.syncMessages(
-        JSON.stringify(this.messageStore.all())
-      );
-      this._send(WS_EVENT.THINKING_STEP, {
-        id: syntheticStepId,
-        title: syntheticStep.title,
-        status: 'complete',
-        details: syntheticStep.details,
-        timestamp: new Date().toISOString(),
-      });
-      this._send(WS_EVENT.STREAM_END, {});
+      this.finishStream(accumulated, syntheticStepId, syntheticStep);
     } catch (err) {
       const message = err instanceof Error ? err.message : String(err);
-      this._send(WS_EVENT.ERROR, { message });
+      if (message === INTERRUPTED_MESSAGE) {
+        this.finishStream(accumulated, syntheticStepId, syntheticStep);
+      } else {
+        this._send(WS_EVENT.ERROR, { message });
+      }
     } finally {
       this.isProcessing = false;
     }

apps/api/src/app/strategies/claude-code.strategy.ts

@@ -1,8 +1,8 @@
-import { spawn } from 'node:child_process';
+import { spawn, type ChildProcess } from 'node:child_process';
 import { existsSync, mkdirSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
 import { join } from 'node:path';
 import type { AuthConnection, LogoutConnection, ToolEvent } from './strategy.types';
-import type { AgentStrategy } from './strategy.types';
+import { INTERRUPTED_MESSAGE, type AgentStrategy } from './strategy.types';
 
 const CLAUDE_CONFIG_DIR = join(process.env.HOME ?? '/home/node', '.claude');
 const TOKEN_FILE_PATH = join(CLAUDE_CONFIG_DIR, 'agent_token.txt');
@@ -76,6 +76,8 @@ export function toolUseToEvent(
 export class ClaudeCodeStrategy implements AgentStrategy {
   private currentConnection: AuthConnection | null = null;
   private _hasSession = false;
+  private currentStreamProcess: ChildProcess | null = null;
+  private streamInterrupted = false;
 
   private getToken(): string | null {
     if (existsSync(TOKEN_FILE_PATH)) {
@@ -195,6 +197,11 @@ export class ClaudeCodeStrategy implements AgentStrategy {
     });
   }
 
+  interruptAgent(): void {
+    this.streamInterrupted = true;
+    this.currentStreamProcess?.kill();
+  }
+
   executePromptStreaming(
     prompt: string,
     _model: string,
@@ -203,6 +210,7 @@ export class ClaudeCodeStrategy implements AgentStrategy {
     systemPrompt?: string
   ): Promise<void> {
     return new Promise((resolve, reject) => {
+      this.streamInterrupted = false;
       if (!existsSync(PLAYGROUND_DIR)) {
         mkdirSync(PLAYGROUND_DIR, { recursive: true });
       }
@@ -233,6 +241,7 @@ export class ClaudeCodeStrategy implements AgentStrategy {
         cwd: PLAYGROUND_DIR,
         shell: false,
       });
+      this.currentStreamProcess = claudeProcess;
       claudeProcess.stdin?.end();
 
       let errorResult = '';
@@ -304,7 +313,12 @@ export class ClaudeCodeStrategy implements AgentStrategy {
       });
 
       claudeProcess.on('close', (code) => {
+        this.currentStreamProcess = null;
         if (useStreamJson && stdoutBuffer.trim()) handleStreamJsonLine(stdoutBuffer);
+        if (this.streamInterrupted) {
+          reject(new Error(INTERRUPTED_MESSAGE));
+          return;
+        }
         if (code !== 0 && errorResult.trim()) {
           reject(new Error(errorResult || `Process exited with code ${code}`));
         } else {
@@ -313,7 +327,10 @@ export class ClaudeCodeStrategy implements AgentStrategy {
         }
       });
 
-      claudeProcess.on('error', reject);
+      claudeProcess.on('error', (err) => {
+        this.currentStreamProcess = null;
+        reject(err);
+      });
     });
   }
 

apps/api/src/app/strategies/gemini.strategy.ts

@@ -1,9 +1,9 @@
 import { Logger } from '@nestjs/common';
-import { spawn } from 'node:child_process';
+import { spawn, type ChildProcess } from 'node:child_process';
 import { existsSync, mkdirSync, readFileSync, rmSync, unlinkSync, writeFileSync } from 'node:fs';
 import { join } from 'node:path';
 import type { AuthConnection, LogoutConnection } from './strategy.types';
-import type { AgentStrategy } from './strategy.types';
+import { INTERRUPTED_MESSAGE, type AgentStrategy } from './strategy.types';
 import { runAuthProcess } from './auth-process-helper';
 
 const GEMINI_CONFIG_DIR = join(process.env.HOME ?? '/home/node', '.gemini');
@@ -14,6 +14,8 @@ export class GeminiStrategy implements AgentStrategy {
   private currentConnection: AuthConnection | null = null;
   private authCancel: (() => void) | null = null;
   private _hasSession = false;
+  private currentStreamProcess: ChildProcess | null = null;
+  private streamInterrupted = false;
 
   ensureSettings(): void {
     if (!existsSync(GEMINI_CONFIG_DIR)) {
@@ -193,6 +195,11 @@ export class GeminiStrategy implements AgentStrategy {
     return ['-m', model];
   }
 
+  interruptAgent(): void {
+    this.streamInterrupted = true;
+    this.currentStreamProcess?.kill();
+  }
+
   executePromptStreaming(
     prompt: string,
     model: string,
@@ -201,6 +208,7 @@ export class GeminiStrategy implements AgentStrategy {
     systemPrompt?: string
   ): Promise<void> {
     return new Promise((resolve, reject) => {
+      this.streamInterrupted = false;
       this.ensureSettings();
       const playgroundDir = join(process.cwd(), 'playground');
       if (!existsSync(playgroundDir)) {
@@ -221,6 +229,7 @@ export class GeminiStrategy implements AgentStrategy {
         cwd: playgroundDir,
         shell: false,
       });
+      this.currentStreamProcess = geminiProcess;
 
       let errorResult = '';
 
@@ -234,6 +243,11 @@ export class GeminiStrategy implements AgentStrategy {
       });
 
       geminiProcess.on('close', (code) => {
+        this.currentStreamProcess = null;
+        if (this.streamInterrupted) {
+          reject(new Error(INTERRUPTED_MESSAGE));
+          return;
+        }
         const modelNotFound =
           errorResult.includes('ModelNotFoundError') ||
           errorResult.includes('Requested entity was not found');
@@ -267,7 +281,10 @@ export class GeminiStrategy implements AgentStrategy {
         }
       });
 
-      geminiProcess.on('error', reject);
+      geminiProcess.on('error', (err) => {
+        this.currentStreamProcess = null;
+        reject(err);
+      });
     });
   }
 }

apps/api/src/app/strategies/mock.strategy.test.ts

@@ -0,0 +1,28 @@
+import { describe, test, expect, beforeEach } from 'bun:test';
+import { MockStrategy } from './mock.strategy';
+import { INTERRUPTED_MESSAGE } from './strategy.types';
+
+describe('MockStrategy', () => {
+  let strategy: MockStrategy;
+
+  beforeEach(() => {
+    strategy = new MockStrategy();
+  });
+
+  test('interruptAgent rejects executePromptStreaming promise with INTERRUPTED', async () => {
+    const chunks: string[] = [];
+    const promise = strategy.executePromptStreaming(
+      'prompt',
+      'model',
+      (chunk) => chunks.push(chunk),
+      undefined,
+      undefined
+    );
+    strategy.interruptAgent();
+    await expect(promise).rejects.toThrow(INTERRUPTED_MESSAGE);
+  });
+
+  test('interruptAgent when no stream is running does not throw', () => {
+    expect(() => strategy.interruptAgent()).not.toThrow();
+  });
+});

apps/api/src/app/strategies/mock.strategy.ts

@@ -1,12 +1,13 @@
 import { Logger } from '@nestjs/common';
 import type { AuthConnection, LogoutConnection } from './strategy.types';
-import type { AgentStrategy } from './strategy.types';
+import { INTERRUPTED_MESSAGE, type AgentStrategy } from './strategy.types';
 
 const MOCK_AUTH_DELAY_MS = 1000;
 const MOCK_LOGOUT_DELAY_MS = 500;
 
 export class MockStrategy implements AgentStrategy {
   private readonly logger = new Logger(MockStrategy.name);
+  private streamCancel: (() => void) | null = null;
 
   executeAuth(connection: AuthConnection): void {
     this.logger.log('executeAuth: Mocking auth success in 1s');
@@ -40,14 +41,24 @@ export class MockStrategy implements AgentStrategy {
     return Promise.resolve(true);
   }
 
+  interruptAgent(): void {
+    this.streamCancel?.();
+  }
+
   executePromptStreaming(
     _prompt: string,
     _model: string,
     onChunk: (chunk: string) => void,
     callbacks?: import('./strategy.types').StreamingCallbacks,
     _systemPrompt?: string
   ): Promise<void> {
-    return new Promise((resolve) => {
+    return new Promise((resolve, reject) => {
+      const timeoutRef: { id: ReturnType<typeof setTimeout> | null } = { id: null };
+      this.streamCancel = () => {
+        this.streamCancel = null;
+        if (timeoutRef.id !== null) clearTimeout(timeoutRef.id);
+        reject(new Error(INTERRUPTED_MESSAGE));
+      };
       callbacks?.onReasoningChunk?.('Considering the request...\n');
       callbacks?.onReasoningChunk?.('Preparing a helpful response.\n');
       callbacks?.onReasoningEnd?.();
@@ -57,7 +68,9 @@ export class MockStrategy implements AgentStrategy {
         path: 'example-output.ts',
         summary: 'Sample file created by mock agent',
       });
-      setTimeout(() => {
+      timeoutRef.id = setTimeout(() => {
+        this.streamCancel = null;
+        timeoutRef.id = null;
         const timestamp = new Date().toISOString();
         onChunk('[MOCKED RESPONSE] Hello! ');
         onChunk(`The current timestamp is ${timestamp}`);