feat(oc-plugin): add MACP ACP runtime backend

plugins/macp/src/macp-runtime.ts

@@ -0,0 +1,275 @@
+import path from 'node:path';
+import { access } from 'node:fs/promises';
+
+import type {
+  AcpRuntime,
+  AcpRuntimeCapabilities,
+  AcpRuntimeDoctorReport,
+  AcpRuntimeEnsureInput,
+  AcpRuntimeEvent,
+  AcpRuntimeHandle,
+  AcpRuntimeStatus,
+  AcpRuntimeTurnInput,
+} from '/home/jarvis/.npm-global/lib/node_modules/openclaw/dist/plugin-sdk/acp-runtime.js';
+
+import { formatAssistantEvent, runPiTurn } from './pi-bridge.js';
+
+export interface MacpRuntimeConfig {
+  defaultModel: string;
+  systemPrompt: string;
+  timeoutMs: number;
+  stateDir: string;
+  logger?: {
+    info?: (message: string) => void;
+    warn?: (message: string) => void;
+  };
+}
+
+type HandleState = {
+  name: string;
+  agent: string;
+  cwd: string;
+  model: string;
+  systemPrompt: string;
+  timeoutMs: number;
+};
+
+const MACP_CAPABILITIES: AcpRuntimeCapabilities = {
+  controls: [],
+};
+
+const PI_RUNNER_PATH = '/home/jarvis/src/mosaic-mono-v1/tools/macp/dispatcher/pi_runner.ts';
+
+function encodeHandleState(state: HandleState): string {
+  return JSON.stringify(state);
+}
+
+function decodeHandleState(handle: AcpRuntimeHandle): HandleState {
+  const parsed = JSON.parse(handle.runtimeSessionName) as Partial<HandleState>;
+  if (
+    typeof parsed.name !== 'string' ||
+    typeof parsed.agent !== 'string' ||
+    typeof parsed.cwd !== 'string' ||
+    typeof parsed.model !== 'string' ||
+    typeof parsed.systemPrompt !== 'string' ||
+    typeof parsed.timeoutMs !== 'number'
+  ) {
+    throw new Error('Invalid MACP runtime handle state.');
+  }
+  return parsed as HandleState;
+}
+
+function toSessionName(input: AcpRuntimeEnsureInput): string {
+  return `${input.agent}-${input.sessionKey}`;
+}
+
+export class MacpRuntime implements AcpRuntime {
+  constructor(private readonly config: MacpRuntimeConfig) {}
+
+  async ensureSession(input: AcpRuntimeEnsureInput): Promise<AcpRuntimeHandle> {
+    if (input.agent !== 'pi') {
+      throw new Error(`macp runtime only supports agentId "pi"; received "${input.agent}".`);
+    }
+    if (input.mode !== 'oneshot') {
+      throw new Error(`macp runtime only supports oneshot sessions; received "${input.mode}".`);
+    }
+
+    const cwd = path.resolve(input.cwd ?? process.cwd());
+    const state: HandleState = {
+      name: toSessionName(input),
+      agent: input.agent,
+      cwd,
+      model: this.config.defaultModel,
+      systemPrompt: this.config.systemPrompt,
+      timeoutMs: this.config.timeoutMs,
+    };
+
+    return {
+      sessionKey: input.sessionKey,
+      backend: 'macp',
+      runtimeSessionName: encodeHandleState(state),
+      cwd,
+      backendSessionId: state.name,
+      agentSessionId: state.name,
+    };
+  }
+
+  async *runTurn(input: AcpRuntimeTurnInput): AsyncIterable<AcpRuntimeEvent> {
+    const state = decodeHandleState(input.handle);
+    const logPath = path.join(
+      this.config.stateDir,
+      'macp',
+      `${state.name}-${Date.now()}-${input.requestId}.json`,
+    );
+
+    const streamQueue: AcpRuntimeEvent[] = [];
+    const waiters: Array<() => void> = [];
+    let finished = false;
+
+    const push = (event: AcpRuntimeEvent) => {
+      streamQueue.push(event);
+      const waiter = waiters.shift();
+      waiter?.();
+    };
+
+    const resultPromise = runPiTurn({
+      model: state.model,
+      systemPrompt: state.systemPrompt,
+      prompt: input.text,
+      workDir: state.cwd,
+      timeoutMs: state.timeoutMs,
+      logPath,
+      ...(input.signal ? { signal: input.signal } : {}),
+      onEvent: async (event) => {
+        switch (event.type) {
+          case 'message_update': {
+            const formatted = formatAssistantEvent(event.assistantMessageEvent);
+            if (formatted?.text) {
+              push({
+                type: 'text_delta',
+                text: formatted.text,
+                ...(formatted.stream ? { stream: formatted.stream } : {}),
+                ...(formatted.tag ? { tag: formatted.tag } : {}),
+              });
+            }
+            break;
+          }
+          case 'tool_execution_start':
+            push({
+              type: 'tool_call',
+              text: JSON.stringify(event.args ?? {}),
+              tag: 'tool_call',
+              toolCallId: event.toolCallId,
+              title: event.toolName,
+              status: 'running',
+            });
+            break;
+          case 'tool_execution_update':
+            push({
+              type: 'tool_call',
+              text: JSON.stringify(event.partialResult ?? {}),
+              tag: 'tool_call_update',
+              toolCallId: event.toolCallId,
+              title: event.toolName,
+              status: 'running',
+            });
+            break;
+          case 'tool_execution_end':
+            push({
+              type: 'tool_call',
+              text: JSON.stringify(event.result ?? {}),
+              tag: 'tool_call_update',
+              toolCallId: event.toolCallId,
+              title: event.toolName,
+              status: event.isError ? 'error' : 'completed',
+            });
+            break;
+          case 'turn_end':
+            push({
+              type: 'status',
+              text: 'Pi turn completed.',
+              tag: 'usage_update',
+            });
+            break;
+        }
+      },
+    })
+      .then((result) => {
+        if (result.output) {
+          push({
+            type: 'status',
+            text: result.output,
+            tag: 'session_info_update',
+          });
+        }
+        push({
+          type: 'done',
+          stopReason: result.stopReason,
+        });
+      })
+      .catch((error) => {
+        push({
+          type: 'error',
+          message: error instanceof Error ? error.message : String(error),
+        });
+      })
+      .finally(() => {
+        finished = true;
+        while (waiters.length > 0) {
+          waiters.shift()?.();
+        }
+      });
+
+    try {
+      while (!finished || streamQueue.length > 0) {
+        if (streamQueue.length === 0) {
+          await new Promise<void>((resolve) => waiters.push(resolve));
+          continue;
+        }
+        yield streamQueue.shift() as AcpRuntimeEvent;
+      }
+    } finally {
+      await resultPromise;
+    }
+  }
+
+  getCapabilities(): AcpRuntimeCapabilities {
+    return MACP_CAPABILITIES;
+  }
+
+  async getStatus(input: { handle: AcpRuntimeHandle }): Promise<AcpRuntimeStatus> {
+    const state = decodeHandleState(input.handle);
+    return {
+      summary: 'macp Pi oneshot runtime ready',
+      backendSessionId: state.name,
+      agentSessionId: state.name,
+      details: {
+        mode: 'oneshot',
+        model: state.model,
+        cwd: state.cwd,
+      },
+    };
+  }
+
+  async doctor(): Promise<AcpRuntimeDoctorReport> {
+    try {
+      await access(PI_RUNNER_PATH);
+      const importErrors: string[] = [];
+      await import('@mariozechner/pi-agent-core').catch((error) => {
+        importErrors.push(error instanceof Error ? error.message : String(error));
+      });
+      await import('@mariozechner/pi-ai').catch((error) => {
+        importErrors.push(error instanceof Error ? error.message : String(error));
+      });
+      if (importErrors.length > 0) {
+        return {
+          ok: false,
+          code: 'MACP_DEPS_MISSING',
+          message: 'MACP runtime dependencies are not available.',
+          installCommand: 'pnpm install --frozen-lockfile',
+          details: importErrors,
+        };
+      }
+      return {
+        ok: true,
+        message: 'MACP runtime is ready.',
+        details: [PI_RUNNER_PATH],
+      };
+    } catch (error) {
+      return {
+        ok: false,
+        code: 'MACP_PI_RUNNER_MISSING',
+        message: 'Pi runner was not found.',
+        details: [error instanceof Error ? error.message : String(error)],
+      };
+    }
+  }
+
+  async cancel(_input: { handle: AcpRuntimeHandle; reason?: string }): Promise<void> {
+    this.config.logger?.info?.('macp runtime cancel requested');
+  }
+
+  async close(_input: { handle: AcpRuntimeHandle; reason: string }): Promise<void> {
+    this.config.logger?.info?.('macp runtime close requested');
+  }
+}