feat(needle): comprehensive integration of local pre-query planner bridge, cross-platform python runners, robust provider fallbacks, and REPL/agent routing

docs/advanced-setup.md

@@ -938,3 +938,29 @@ project's `disabledMcpServers` key to restore inheritance. Builtin servers that
 are disabled by default keep their separate project-level `enabledMcpServers`
 allowlist. Existing running sessions are not forcibly disconnected by a toggle
 in another project.
+
+## Needle local planner
+
+Needle is an optional local pre-query planner and is disabled by default. Install Python 3 and Git first, then install the upstream package from source.
+
+Linux and macOS:
+
+```bash
+git clone https://github.com/cactus-compute/needle.git
+cd needle
+python3 -m pip install --user -e '.[jax]'
+```
+
+Windows PowerShell:
+
+```powershell
+git clone https://github.com/cactus-compute/needle.git
+Set-Location needle
+py -3 -m pip install --user -e ".[jax]"
+```
+
+OpenClaude uses `python3` by default on Linux/macOS and `py -3` on Windows. Set `OPENCLAUDE_NEEDLE_PYTHON` to an interpreter path when your setup differs.
+
+Enable it in an OpenClaude session with `/needle enable`. OpenClaude verifies that the selected interpreter can import `needle`; if not, enablement is refused with an installation message. Use `/needle mode worker` for AgentTool workers or `/needle mode main` for the main conversation; `/needle status` reports the current settings, and `/needle disable` turns it off. In either mode, Needle proposes at most one tool call. OpenClaude executes it using normal permissions and tool handling, then continues the same conversation with the configured provider. The provider remains responsible for the final response, summaries, interpreting WebFetch results, redirects, and any subsequent tool calls. Needle does not summarize the conversation or execute tools itself.
+
+The `needle-worker` model route selects Needle for AgentTool workers when enabled; main-conversation use is controlled by the `main` mode. The TypeScript bridge exchanges JSONL requests with Python. If Needle fails after enablement, OpenClaude skips the local prelude and continues with the configured provider. Set `OPENCLAUDE_NEEDLE_TIMEOUT_MS` to change the bounded planner timeout.

src/commands.ts

@@ -26,6 +26,7 @@ import ctx_viz from './commands/ctx_viz/index.js'
 import doctor from './commands/doctor/index.js'
 import onboardGithub from './commands/onboard-github/index.js'
 import knowledge from './commands/knowledge/index.js'
+import needle from './commands/needle/index.js'
 import memory from './commands/memory/index.js'
 import repomap from './commands/repomap/index.js'
 import help from './commands/help/index.js'
@@ -326,6 +327,7 @@ const COMMANDS = memoize((): Command[] => [
   init,
   keybindings,
   knowledge,
+  needle,
   lsp,
   installGitHubApp,
   installSlackApp,

src/commands/needle/index.ts

@@ -0,0 +1,12 @@
+import type { Command } from '../../commands.js'
+
+const needle: Command = {
+  type: 'local',
+  name: 'needle',
+  description: 'Enable or disable the local Needle tool-calling planner',
+  supportsNonInteractive: true,
+  argumentHint: 'enable | disable | status | mode <worker|main>',
+  load: () => import('./needle.js'),
+}
+
+export default needle

src/commands/needle/needle.test.ts

@@ -0,0 +1,64 @@
+import { describe, expect, it, beforeEach, afterEach, spyOn } from 'bun:test'
+import { mkdtempSync, rmSync } from 'fs'
+import { tmpdir } from 'os'
+import { join } from 'path'
+import { call as needleCall } from './needle.js'
+import * as needleClient from '../../services/needle/needleClient.js'
+import { getGlobalConfig, saveGlobalConfig } from '../../utils/config.js'
+import { setClaudeConfigHomeDirForTesting } from '../../utils/envUtils.js'
+import type { LocalJSXCommandContext } from '../../types/command.js'
+import {
+  acquireSharedMutationLock,
+  releaseSharedMutationLock,
+} from '../../test/sharedMutationLock.js'
+
+describe('needle command', () => {
+  let configDir: string
+  beforeEach(async () => {
+    await acquireSharedMutationLock('commands/needle.test.ts')
+    configDir = mkdtempSync(join(tmpdir(), 'openclaude-needle-command-'))
+    process.env.CLAUDE_CONFIG_DIR = configDir
+    setClaudeConfigHomeDirForTesting(configDir)
+  })
+  afterEach(() => {
+    saveGlobalConfig(current => ({ ...current, needleEnabled: false, needleMode: 'worker' }))
+    setClaudeConfigHomeDirForTesting(undefined)
+    delete process.env.CLAUDE_CONFIG_DIR
+    rmSync(configDir, { recursive: true, force: true })
+    releaseSharedMutationLock()
+  })
+
+  it('refuses enablement when Needle is unavailable', async () => {
+    const previous = process.env.OPENCLAUDE_NEEDLE_PYTHON
+    process.env.OPENCLAUDE_NEEDLE_PYTHON = process.platform === 'win32' ? 'missing-needle-python.exe' : '/bin/false'
+    const context = {} as unknown as LocalJSXCommandContext
+    try {
+      const result = await needleCall('enable', context)
+      expect(getGlobalConfig().needleEnabled).toBe(false)
+      expect(result.type).toBe('text')
+      if (result.type === 'text') expect(result.value).toContain('was not enabled')
+    } finally {
+      if (previous === undefined) delete process.env.OPENCLAUDE_NEEDLE_PYTHON
+      else process.env.OPENCLAUDE_NEEDLE_PYTHON = previous
+    }
+  })
+
+  it('toggles the worker and persists its mode', async () => {
+    const availability = spyOn(needleClient, 'checkNeedleAvailability').mockResolvedValue({ available: true })
+    const context = {} as unknown as LocalJSXCommandContext
+    try {
+      await needleCall('enable', context)
+      expect(getGlobalConfig().needleEnabled).toBe(true)
+      expect(getGlobalConfig().needleMode).toBe('worker')
+
+      await needleCall('mode main', context)
+      expect(getGlobalConfig().needleMode).toBe('main')
+
+      const status = await needleCall('status', context)
+      expect(status.type).toBe('text')
+      if (status.type === 'text') expect(status.value).toContain('main')
+    } finally {
+      availability.mockRestore()
+    }
+  })
+})

src/commands/needle/needle.ts

@@ -0,0 +1,62 @@
+import type { LocalCommandCall } from '../../types/command.js'
+import { getGlobalConfig, saveGlobalConfig } from '../../utils/config.js'
+import chalk from 'chalk'
+import { checkNeedleAvailability } from '../../services/needle/needleClient.js'
+
+export const call: LocalCommandCall = async args => {
+  const parts = String(args ?? '').trim().toLowerCase().split(/\s+/).filter(Boolean)
+  const command = parts[0] ?? 'status'
+  const config = getGlobalConfig()
+  const enabled = config.needleEnabled === true
+  const mode = config.needleMode ?? 'worker'
+
+  if (command === 'status') {
+    return {
+      type: 'text',
+      value: `Needle: ${enabled ? chalk.green('enabled') : chalk.red('disabled')} (${mode} mode)`,
+    }
+  }
+
+  if (command === 'enable' || command === 'disable') {
+    const nextEnabled = command === 'enable'
+    if (nextEnabled) {
+      const availability = await checkNeedleAvailability()
+      if (!availability.available) {
+        return {
+          type: 'text',
+          value: `Needle was not enabled because its Python package is unavailable. Install it from source, then retry /needle enable. (${availability.error ?? 'import failed'})`,
+        }
+      }
+    }
+    saveGlobalConfig(current => ({
+      ...current,
+      needleEnabled: nextEnabled,
+      needleMode: current.needleMode ?? 'worker',
+    }))
+    return {
+      type: 'text',
+      value: `Needle ${nextEnabled ? chalk.green('enabled') : chalk.red('disabled')} (${mode} mode).`,
+    }
+  }
+
+  if (command === 'mode') {
+    const nextMode = parts[1]
+    if (nextMode !== 'worker' && nextMode !== 'main') {
+      return { type: 'text', value: 'Usage: /needle mode <worker|main>' }
+    }
+    saveGlobalConfig(current => ({ ...current, needleMode: nextMode }))
+    return {
+      type: 'text',
+      value: `Needle mode set to ${chalk.cyan(nextMode)}. ${
+        nextMode === 'main'
+          ? 'The main conversation will use Needle when enabled.'
+          : 'AgentTool workers will use Needle when enabled.'
+      }`,
+    }
+  }
+
+  return {
+    type: 'text',
+    value: 'Usage: /needle <enable|disable|status|mode worker|main>',
+  }
+}

src/screens/REPL.tsx

@@ -148,6 +148,7 @@ import { useMailboxBridge } from '../hooks/useMailboxBridge.js';
 import { queryCheckpoint, logQueryProfileReport, clearQueryProfile } from '../utils/queryProfiler.js';
 import type { Message as MessageType, UserMessage, ProgressMessage, HookResultMessage, PartialCompactDirection } from '../types/message.js';
 import { query, type QueryTurnBudget } from '../query.js';
+import { closeNeedleClient, runNeedleWorkerTurn } from '../services/needle/runNeedleWorkerTurn.js';
 import type { Terminal as QueryTerminal } from '../query/transitions.js';
 import type { AutoCompactTrackingState } from '../services/compact/autoCompact.js';
 import { mergeClients, useMergedClients } from '../hooks/useMergedClients.js';
@@ -681,6 +682,7 @@ export function REPL({
   // without requiring a REPL remount. CLI prop still wins over env/config.
   const isRemoteSession = !!remoteSessionConfig;
   const foregroundTurnBudgetRef = useRef<ForegroundTurnBudgetHandoff | null>(null);
+  const needleClientOwnerRef = useRef({});
   const backgroundHandoffStartedRef = useRef(false);
   const [backgroundHandoffPreparing, setBackgroundHandoffPreparing] = useState(false);
 
@@ -1017,6 +1019,10 @@ export function REPL({
   // REPL bridge to abort the active query when a remote interrupt arrives.
   const abortControllerRef = useRef<AbortController | null>(null);
   abortControllerRef.current = abortController;
+  useEffect(() => () => {
+    abortControllerRef.current?.abort();
+    closeNeedleClient(needleClientOwnerRef.current);
+  }, []);
 
   // Ref for the bridge result callback — set after useReplBridge initializes,
   // read in the onQuery finally block to notify mobile clients that a turn ended.
@@ -3292,7 +3298,8 @@ export function REPL({
     resetTurnToolDuration();
     resetTurnClassifierDuration();
     let expectedAutoCompactTracking = queryAutoCompactTracking;
-    const queryGenerator = query({
+    const needleConfig = getGlobalConfig();
+    const queryParams = {
       messages: messagesIncludingNewMessages,
       requestOnlyMessages,
       onModelRequestStart: interruptionCorrectionQueryId
@@ -3301,8 +3308,8 @@ export function REPL({
               shouldQuery,
               isInterruptionCorrectionEligible: true,
               queryId: interruptionCorrectionQueryId,
-            })
-            onModelRequestStart?.()
+            });
+            onModelRequestStart?.();
           }
         : undefined,
       systemPrompt,
@@ -3318,13 +3325,34 @@ export function REPL({
         if (setAutoCompactTrackingForSessionIfUnchanged(querySessionId, expectedAutoCompactTracking, tracking)) {
           expectedAutoCompactTracking = tracking;
         }
-      }
-    });
+      },
+    };
+    const currentUserMessage = newMessages.findLast(selectableUserMessagesFilter);
+    const currentUserContent = currentUserMessage?.message.content;
+    const needlePrompt = currentUserContent ? getContentText(currentUserContent) ?? '' : '';
+    const hasImageContent =
+      Array.isArray(currentUserContent) && currentUserContent.some(block => block.type === 'image');
+    const queryGenerator =
+      needleConfig.needleEnabled === true &&
+      needleConfig.needleMode === 'main' &&
+      needlePrompt &&
+      !hasImageContent
+        ? runNeedleWorkerTurn({
+            prompt: needlePrompt,
+            queryParams,
+            tools: freshTools,
+            toolUseContext,
+            clientOwner: needleClientOwnerRef.current,
+            canUseTool,
+            signal: abortController.signal,
+          })
+        : query(queryParams);
     const queryTerminal = await driveQueryEvents(
       queryGenerator,
       reason => queryGuard.registerActivity(reason, queryGeneration),
       onQueryEvent,
     );
+
     if (isBuddyEnabled()) {
       void fireCompanionObserver(messagesRef.current, reaction => setAppState(prev => prev.companionReaction === reaction ? prev : {
         ...prev,

src/services/needle/needleClient.test.ts

@@ -0,0 +1,108 @@
+import { describe, expect, test } from 'bun:test'
+import { chmodSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import {
+  buildNeedleToolPayload,
+  DEFAULT_NEEDLE_TIMEOUT_MS,
+  getNeedlePythonCommand,
+  getNeedleTimeoutMs,
+  NEEDLE_WORKER_MODEL,
+} from './needleClient.js'
+
+describe('Needle client configuration', () => {
+  test('uses the dedicated worker model identifier', () => {
+    expect(NEEDLE_WORKER_MODEL).toBe('needle-worker')
+  })
+
+  test('uses platform-appropriate Python launchers', () => {
+    const previous = process.env.OPENCLAUDE_NEEDLE_PYTHON
+    delete process.env.OPENCLAUDE_NEEDLE_PYTHON
+    try {
+      expect(getNeedlePythonCommand('linux')).toEqual({ command: 'python3', args: [] })
+      expect(getNeedlePythonCommand('darwin')).toEqual({ command: 'python3', args: [] })
+      expect(getNeedlePythonCommand('win32')).toEqual({ command: 'py', args: ['-3'] })
+    } finally {
+      if (previous === undefined) delete process.env.OPENCLAUDE_NEEDLE_PYTHON
+      else process.env.OPENCLAUDE_NEEDLE_PYTHON = previous
+    }
+  })
+
+  test('converts Anthropic-style tool schemas to Needle payloads', () => {
+    expect(buildNeedleToolPayload([
+      { name: 'read', description: 'Read a file', input_schema: { type: 'object', properties: {} } },
+    ])).toEqual([
+      { name: 'read', description: 'Read a file', parameters: { type: 'object', properties: {} } },
+    ])
+  })
+
+  test.skipIf(process.platform === 'win32')('restarts the worker after a child exit before serving later requests', async () => {
+    const previous = process.env.OPENCLAUDE_NEEDLE_PYTHON
+    process.env.OPENCLAUDE_NEEDLE_PYTHON = '/bin/false'
+    const client = new (await import('./needleClient.js')).NeedleClient(100)
+    try {
+      await expect(client.complete({ prompt: 'first', tools: [] })).rejects.toThrow('exited with code 1')
+      await expect(client.complete({ prompt: 'second', tools: [] })).rejects.toThrow('exited with code 1')
+    } finally {
+      client.close()
+      if (previous === undefined) delete process.env.OPENCLAUDE_NEEDLE_PYTHON
+      else process.env.OPENCLAUDE_NEEDLE_PYTHON = previous
+    }
+  })
+
+  test.skipIf(process.platform === 'win32')('restarts the worker after a zero child exit before serving later requests', async () => {
+    const previous = process.env.OPENCLAUDE_NEEDLE_PYTHON
+    process.env.OPENCLAUDE_NEEDLE_PYTHON = '/bin/true'
+    const client = new (await import('./needleClient.js')).NeedleClient(100)
+    try {
+      await expect(client.complete({ prompt: 'first', tools: [] })).rejects.toThrow('exited with code 0')
+      await expect(client.complete({ prompt: 'second', tools: [] })).rejects.toThrow('exited with code 0')
+    } finally {
+      client.close()
+      if (previous === undefined) delete process.env.OPENCLAUDE_NEEDLE_PYTHON
+      else process.env.OPENCLAUDE_NEEDLE_PYTHON = previous
+    }
+  })
+
+  test.skipIf(process.platform === 'win32')('rejects malformed matching responses instead of leaving requests pending', async () => {
+    const directory = mkdtempSync(join(tmpdir(), 'needle-runtime-'))
+    const runtime = join(directory, 'runtime.py')
+    const command = join(directory, 'python')
+    writeFileSync(runtime, `import json, sys\nfor line in sys.stdin:\n    request = json.loads(line)\n    print(json.dumps({"id": request["id"]}), flush=True)\n`)
+    writeFileSync(command, `#!/bin/sh\nexec python3 /${runtime.replace(/^\//, '')}\n`)
+    chmodSync(command, 0o755)
+    const previous = process.env.OPENCLAUDE_NEEDLE_PYTHON
+    process.env.OPENCLAUDE_NEEDLE_PYTHON = command
+    const client = new (await import('./needleClient.js')).NeedleClient(1_000)
+    try {
+      await expect(client.complete({ prompt: 'malformed', tools: [] })).rejects.toThrow('Needle worker returned no result')
+    } finally {
+      client.close()
+      if (previous === undefined) delete process.env.OPENCLAUDE_NEEDLE_PYTHON
+      else process.env.OPENCLAUDE_NEEDLE_PYTHON = previous
+      rmSync(directory, { recursive: true, force: true })
+    }
+  })
+
+  test.skipIf(process.platform === 'win32')('restarts the worker after a timeout before serving later requests', async () => {
+    const directory = mkdtempSync(join(tmpdir(), 'needle-runtime-'))
+    const runtime = join(directory, 'runtime.py')
+    const command = join(directory, 'python')
+    writeFileSync(runtime, `import json, sys, time\nfor line in sys.stdin:\n    request = json.loads(line)\n    if request["prompt"] == "delay": time.sleep(0.2)\n    print(json.dumps({"id": request["id"], "result": {"function_calls": [{"name": "read", "arguments": {"prompt": request["prompt"]}}]}}), flush=True)\n`)
+    writeFileSync(command, `#!/bin/sh\nexec python3 /${runtime.replace(/^\//, '')}\n`)
+    chmodSync(command, 0o755)
+    const previous = process.env.OPENCLAUDE_NEEDLE_PYTHON
+    process.env.OPENCLAUDE_NEEDLE_PYTHON = command
+    const client = new (await import('./needleClient.js')).NeedleClient(30)
+    try {
+      await expect(client.complete({ prompt: 'delay', tools: [] })).rejects.toThrow('timed out')
+      const response = await client.complete({ prompt: 'fresh', tools: [] })
+      expect(response.functionCalls[0]?.arguments.prompt).toBe('fresh')
+    } finally {
+      client.close()
+      if (previous === undefined) delete process.env.OPENCLAUDE_NEEDLE_PYTHON
+      else process.env.OPENCLAUDE_NEEDLE_PYTHON = previous
+      rmSync(directory, { recursive: true, force: true })
+    }
+  })
+})