fusion/plugins/fusion-plugin-acp-runtime/src/provider.ts

// ACP connection layer: spawn → ClientSideConnection → initialize handshake.
//
// U2 establishes the transport and completes the `initialize` handshake with
// integer protocol-version negotiation (KTD2) and a readiness timeout. Session
// driving (`session/new`, `session/prompt`, cancel, load) is U3 — this unit only
// exposes the live `conn` on the returned handle so later units can drive it.
//
// Security posture (KTD6): filesystem client capabilities are advertised ONLY
// when the caller's `advertiseFs` toggle is true — never hardcoded. Teardown is
// registry-SIGKILL-authoritative (KTD4a): `dispose()` force-kills the child via
// the process registry; that kill is the no-orphan guarantee, not a graceful
// round-trip.

import { Readable, Writable } from "node:stream";
import type { ChildProcess } from "node:child_process";
import {
  ClientSideConnection,
  ndJsonStream,
  PROTOCOL_VERSION,
  type Agent,
  type AgentCapabilities,
  type Client,
  type ContentBlock,
  type RequestPermissionResponse,
  type StopReason,
} from "@agentclientprotocol/sdk";
import { spawnAgent, captureStderr, forceKill, unregisterProcess } from "./process-manager.js";
import { createEventBridge } from "./event-bridge.js";
import { resolvePermission, type ResolvePermissionOptions } from "./control-handler.js";
import { createFsHandlers } from "./fs-capabilities.js";
import { boundIdentifier } from "./sanitize.js";
import type { AcpCallbacks, AcpMcpServer, PermissionGate } from "./types.js";

/** Options enabling the U7 fs client capabilities on the bridging handler. */
export interface FsHandlerBuildOptions {
  /** Confinement root — the session cwd / task worktree. */
  cwd: string;
  /** Register `readTextFile` (advertised iff true). */
  allowRead: boolean;
  /** Register `writeTextFile` (default OFF — KTD6; advertised iff true). */
  allowWrite: boolean;
}

/** Default bound for the `initialize` handshake. */
export const DEFAULT_INITIALIZE_TIMEOUT_MS = 30_000;

/** Thrown when the agent negotiates an integer protocol version we don't support. */
export class IncompatibleProtocolError extends Error {
  readonly code = "incompatible_protocol" as const;
  constructor(
    readonly agentProtocolVersion: number,
    readonly expected: number = PROTOCOL_VERSION,
  ) {
    super(
      `ACP agent negotiated incompatible protocol version ${agentProtocolVersion} (client supports ${expected})`,
    );
    this.name = "IncompatibleProtocolError";
  }
}

/** Thrown when the `initialize` handshake does not complete within the bound. */
export class HandshakeTimeoutError extends Error {
  readonly code = "handshake_timeout" as const;
  constructor(readonly timeoutMs: number) {
    super(`ACP initialize handshake timed out after ${timeoutMs}ms`);
    this.name = "HandshakeTimeoutError";
  }
}

/*
FNXC:GrokAcp 2026-07-12-07:00:
Grok ACP emits vendor extension notifications such as `_x.ai/session_notification`
and `_x.ai/session/update` for hook_execution status (post_tool_use, etc.).
@agentclientprotocol/sdk routes non-spec methods to Client.extNotification /
extMethod; when those are absent the SDK logs Method not found (-32601) for
every hook even though hooks themselves succeeded. Accept extensions as no-ops
so Fusion clients stay forward-compatible without claiming Grok-only features.
Do not forward into sessionUpdate: hook_execution is not a standard
sessionUpdate tag and would fail zSessionNotification validation.
*/

/** No-op ACP extension handlers so vendor notifications do not Method-not-found. */
function acceptAcpExtensions(): Pick<Client, "extMethod" | "extNotification"> {
  return {
    async extMethod(_method: string, _params: Record<string, unknown>) {
      return {};
    },
    async extNotification(_method: string, _params: Record<string, unknown>) {
      // Intentionally empty — informational Grok hooks / x.ai session extensions.
    },
  };
}

/**
 * Minimal default client handler. Later units (U3/U4/U5/U7) supply the real one
 * that bridges `session/update` into Fusion callbacks and routes permission
 * requests through the action gate. The default cancels every permission request
 * (never auto-allows an untrusted agent) and ignores updates.
 */
export function createDefaultClientHandler(): Client {
  return {
    async sessionUpdate() {
      // no-op until the U4 event bridge is wired
    },
    async requestPermission() {
      return { outcome: { outcome: "cancelled" } };
    },
    ...acceptAcpExtensions(),
  };
}

/** A bridging client handler plus a drain control for its in-flight permissions. */
export interface BridgingClientHandler {
  /** The ACP `Client` impl handed to `ClientSideConnection`. */
  handler: Client;
  /**
   * Resolve every in-flight `requestPermission` with `{ cancelled }` and mark the
   * handler cancelled so any request arriving afterward is answered cancelled
   * immediately (U5 cancel-drain — KTD4a). Idempotent.
   */
  cancelPending(): void;
  /**
   * Reset the event bridge's PER-TURN state (tool correlation, delta
   * accumulators, cumulative-output counter, output-cap latch). MUST be called
   * at the start of each prompt turn so a turn that trips the per-turn output cap
   * does not silently suppress every subsequent turn (FIX 1).
   */
  resetTurn(): void;
}

/**
 * The real client handler (U4 + U5): bridges every `session/update` notification
 * into the engine callbacks, AND answers `session/request_permission` through the
 * per-category action gate (U5 — the SECURITY FLOOR).
 *
 * Permission requests are routed to `resolvePermission`, which classifies each
 * call per-category against the live `gate` and selects `allow_once` only (never
 * `*_always`). When no `gate` is supplied the resolver default-denies.
 *
 * Cancel-drain (KTD4a / Risk: in-flight permission deadlock): every pending
 * `requestPermission` promise is tracked; `cancelPending()` resolves them all
 * with `{ cancelled }`. A request that arrives AFTER cancel is answered
 * `{ cancelled }` immediately so the agent never blocks on teardown.
 */
export function createBridgingClientHandler(
  callbacks: AcpCallbacks,
  gate?: PermissionGate,
  fsOpts?: FsHandlerBuildOptions,
  permissionOpts?: ResolvePermissionOptions,
): BridgingClientHandler {
  const bridge = createEventBridge(callbacks);

  // U7: build the fs handlers, returning only the enabled ones. They are added
  // to the handler below ONLY when present, keeping the advertised-capability /
  // registered-handler invariant consistent (KTD6).
  const fsHandlers = fsOpts
    ? createFsHandlers({
        cwd: fsOpts.cwd,
        gate,
        allowRead: fsOpts.allowRead,
        allowWrite: fsOpts.allowWrite,
        allowUnrestricted: permissionOpts?.allowUnrestricted,
      })
    : {};

  const cancelledResponse: RequestPermissionResponse = {
    outcome: { outcome: "cancelled" },
  };

  let cancelled = false;
  // Each entry resolves its pending requestPermission with a cancelled outcome.
  const pending = new Set<(response: RequestPermissionResponse) => void>();

  function cancelPending(): void {
    cancelled = true;
    for (const resolveCancelled of [...pending]) {
      resolveCancelled(cancelledResponse);
    }
    pending.clear();
  }

  const handler: Client = {
    async sessionUpdate(params) {
      bridge.handleSessionUpdate(params.update);
    },
    async requestPermission(params): Promise<RequestPermissionResponse> {
      // A request arriving after cancel is answered cancelled immediately.
      if (cancelled) return cancelledResponse;

      // Race the real gate resolution against a cancel-drain so an in-flight
      // request is answered the moment teardown drains it (never deadlocks).
      return await new Promise<RequestPermissionResponse>((resolve) => {
        let settled = false;
        const finish = (response: RequestPermissionResponse) => {
          if (settled) return;
          settled = true;
          pending.delete(drain);
          resolve(response);
        };
        const drain = (response: RequestPermissionResponse) => finish(response);
        pending.add(drain);

        resolvePermission(params.toolCall, params.options, gate, permissionOpts).then(
          (response) => finish(response),
          // resolvePermission never rejects, but stay safe: deny-by-cancel.
          () => finish(cancelledResponse),
        );
      });
    },
    // FNXC:GrokAcp 2026-07-12-07:00: swallow `_x.ai/*` extension notifications
    // (hook_execution, session admin, …) so ACP SDK does not log -32601.
    ...acceptAcpExtensions(),
  };

  // Register fs handlers ONLY when enabled, so the advertised capability and the
  // present handler stay consistent (KTD6). If a capability is disabled the
  // method is absent → an agent calling it gets a JSON-RPC method-not-found
  // error (never a silent success).
  if (fsHandlers.readTextFile) handler.readTextFile = fsHandlers.readTextFile;
  if (fsHandlers.writeTextFile) handler.writeTextFile = fsHandlers.writeTextFile;

  return { handler, cancelPending, resetTurn: () => bridge.reset() };
}

export interface AcpConnection {
  /** Live ACP connection — later units drive session/new, prompt, cancel, load. */
  conn: ClientSideConnection;
  child: ChildProcess;
  agentCapabilities?: AgentCapabilities;
  /** Auth methods the agent advertised; non-empty means auth is required. */
  authMethods: Array<{ id: string }>;
  /** Current redacted stderr buffer. */
  stderr(): string;
  /** Force-kill the agent via the registry (KTD4a — SIGKILL is authoritative). */
  dispose(): void;
}

export interface ConnectOptions {
  binaryPath: string;
  args: string[];
  cwd: string;
  env: NodeJS.ProcessEnv;
  clientHandler?: Client;
  /** Advertise fs capabilities ONLY where the toggle is true (KTD6). */
  advertiseFs: { read: boolean; write: boolean };
  initializeTimeoutMs?: number;
  /**
   * FNXC:GrokAcp 2026-07-11-15:00:
   * Optional post-initialize authenticate (xAI Grok docs: initialize → authenticate
   * → session/new). Prefer methods listed in preferMethods that the agent
   * advertised; when require is true, missing auth fails closed.
   * See https://docs.x.ai/build/cli/headless-scripting#acp
   */
  authenticate?: {
    preferMethods?: string[];
    methodId?: string;
    meta?: Record<string, unknown>;
    require?: boolean;
  };
}

function withTimeout<T>(promise: Promise<T>, ms: number, onTimeout: () => Error): Promise<T> {
  return new Promise<T>((resolve, reject) => {
    const timer = setTimeout(() => reject(onTimeout()), ms);
    timer.unref?.();
    promise.then(
      (value) => {
        clearTimeout(timer);
        resolve(value);
      },
      (err) => {
        clearTimeout(timer);
        reject(err);
      },
    );
  });
}

/**
 * Spawn the agent, establish a `ClientSideConnection` over its stdio, and
 * complete the `initialize` handshake under a timeout.
 *
 * Throws `HandshakeTimeoutError` on timeout, `IncompatibleProtocolError` when
 * the negotiated integer protocol version mismatches — in both cases the
 * subprocess is force-killed before throwing (no orphans, KTD4a). On `initialize`
 * the fs capability flags are gated by `advertiseFs` and never hardcoded (KTD6).
 */
export async function connect(opts: ConnectOptions): Promise<AcpConnection> {
  const timeoutMs = opts.initializeTimeoutMs ?? DEFAULT_INITIALIZE_TIMEOUT_MS;
  const child = spawnAgent({
    binaryPath: opts.binaryPath,
    args: opts.args,
    cwd: opts.cwd,
    env: opts.env,
  });
  const stderr = captureStderr(child);

  let disposed = false;
  const dispose = () => {
    if (disposed) return;
    disposed = true;
    forceKill(child);
    unregisterProcess(child);
  };

  // If the binary is missing, spawn emits "error" asynchronously. Surface that
  // as a rejection of the handshake rather than an unhandled event-loop error.
  let spawnError: Error | undefined;
  const spawnErrored = new Promise<never>((_resolve, reject) => {
    child.once("error", (err: Error) => {
      spawnError = err;
      reject(err);
    });
  });
  // Avoid an unhandled rejection if the handshake resolves/throws first.
  spawnErrored.catch(() => undefined);

  // output = the agent's stdin; input = the agent's stdout.
  const stream = ndJsonStream(
    Writable.toWeb(child.stdin!) as unknown as WritableStream<Uint8Array>,
    Readable.toWeb(child.stdout!) as unknown as ReadableStream<Uint8Array>,
  );

  const handler = opts.clientHandler ?? createDefaultClientHandler();
  const conn = new ClientSideConnection((_agent: Agent) => handler, stream);

  let initResult: Awaited<ReturnType<ClientSideConnection["initialize"]>>;
  try {
    initResult = await Promise.race([
      withTimeout(
        conn.initialize({
          protocolVersion: PROTOCOL_VERSION,
          clientCapabilities: {
            fs: {
              readTextFile: opts.advertiseFs.read === true,
              writeTextFile: opts.advertiseFs.write === true,
            },
          },
        }),
        timeoutMs,
        () => new HandshakeTimeoutError(timeoutMs),
      ),
      spawnErrored,
    ]);
  } catch (err) {
    dispose();
    if (spawnError && err === spawnError) throw spawnError;
    throw err;
  }

  // Compare the negotiated integer protocol version; do NOT assume the agent
  // errors first (KTD2).
  if (initResult.protocolVersion !== PROTOCOL_VERSION) {
    dispose();
    throw new IncompatibleProtocolError(initResult.protocolVersion);
  }

  const authMethods = Array.isArray(initResult.authMethods)
    ? initResult.authMethods.map((m) => ({ id: m.id }))
    : [];

  /*
  FNXC:GrokAcp 2026-07-11-15:00:
  Official Grok ACP scripting requires authenticate after initialize (method
  xai.api_key when XAI_API_KEY is set, else cached_token) with
  `_meta: { headless: true }` before session/new. Generic ACP agents that
  advertise no preferred methods skip this step.
  */
  if (opts.authenticate) {
    try {
      await authenticateAcpConnection(
        { conn, authMethods },
        opts.authenticate,
      );
    } catch (err) {
      dispose();
      throw err;
    }
  }

  return {
    conn,
    child,
    agentCapabilities: initResult.agentCapabilities,
    authMethods,
    stderr,
    dispose,
  };
}

export class AcpAuthRequiredError extends Error {
  readonly code = "acp_auth_required" as const;
  constructor(readonly availableMethodIds: string[]) {
    super(
      availableMethodIds.length > 0
        ? `ACP agent requires authentication but no preferred method matched (available: ${availableMethodIds.join(", ")})`
        : "ACP agent requires authentication but advertised no auth methods",
    );
    this.name = "AcpAuthRequiredError";
  }
}

/**
 * Call ACP `authenticate` with the first preferred method the agent advertised.
 * No-ops when neither methodId nor a preferred method is available and require
 * is false.
 */
export async function authenticateAcpConnection(
  connection: Pick<AcpConnection, "conn" | "authMethods">,
  opts: {
    preferMethods?: string[];
    methodId?: string;
    meta?: Record<string, unknown>;
    require?: boolean;
  },
): Promise<{ methodId: string } | undefined> {
  const available = connection.authMethods.map((m) => m.id);
  const availableSet = new Set(available);
  let methodId = opts.methodId?.trim();
  if (methodId && !availableSet.has(methodId)) {
    methodId = undefined;
  }
  if (!methodId) {
    for (const candidate of opts.preferMethods ?? []) {
      if (availableSet.has(candidate)) {
        methodId = candidate;
        break;
      }
    }
  }
  if (!methodId) {
    if (opts.require) {
      throw new AcpAuthRequiredError(available);
    }
    return undefined;
  }
  await connection.conn.authenticate({
    methodId,
    _meta: opts.meta ?? { headless: true },
  });
  return { methodId };
}

// --- U3: session driving on top of connect() -------------------------------
//
// These helpers wrap the `ClientSideConnection` session methods so the runtime
// adapter drives one shape (open → prompt → cancel/resume) without touching SDK
// types directly. v1 always sends an empty `mcpServers` (KTD5).

function readsLoadSession(connection: AcpConnection): boolean {
  // `agentCapabilities` is already typed as `AgentCapabilities | undefined`.
  return connection.agentCapabilities?.loadSession === true;
}

export interface NewAcpSessionResult {
  sessionId: string;
  /** Initial session mode state, when the agent reports one. */
  modes?: unknown;
}

/**
 * Open a fresh ACP session via `session/new`. Forwards `opts.mcpServers` (U10 —
 * Route A): when present and non-empty, the agent can call those Fusion tools and
 * each call still routes through the U5 permission floor. Defaults to `[]` so
 * Route B read-only ask turns keep their no-tools posture.
 */
export async function newAcpSession(
  connection: AcpConnection,
  opts: {
    cwd: string;
    mcpServers?: AcpMcpServer[];
    /**
     * FNXC:GrokAcp 2026-07-11-14:00:
     * Optional ACP `_meta` bag for agent-specific session setup (Grok uses
     * `pluginDirs`, `rules`, `systemPromptOverride`). Opaque to the generic
     * ACP client — agents interpret their own keys.
     */
    meta?: Record<string, unknown>;
  },
): Promise<NewAcpSessionResult> {
  const res = await connection.conn.newSession({
    cwd: opts.cwd,
    mcpServers: (opts.mcpServers ?? []) as never,
    ...(opts.meta && Object.keys(opts.meta).length > 0 ? { _meta: opts.meta } : {}),
  });
  // `sessionId` is agent-supplied/untrusted (U6/Risk S7): bound its length and
  // strip path separators / NUL bytes before it is stored on the session or
  // could ever touch a resume-file path.
  return { sessionId: boundIdentifier(res.sessionId), modes: res.modes ?? undefined };
}

/*
FNXC:AcpRuntime 2026-07-15-18:20:
An ACP turn that fails server-side arrives as a JSON-RPC error object whose `message` is the
bare protocol-standard text — `-32603` renders as literally "Internal error". Rethrowing the
SDK error as-is discards `code`/`data`, the only fields identifying the fault as provider-side
and retryable. Downstream transient classification then sees an unclassifiable string and parks
the task permanently (FN-8004: a 20s Grok blip terminally failed an auto-merge).

`describeAcpTurnError` re-shapes the error into a stable, greppable diagnostic carrying the
numeric code, so classifiers anchor on an ACP-specific signature instead of pattern-matching the
dangerously generic phrase "Internal error" (which could appear in unrelated application output).
Format is load-bearing — `ACP_TRANSIENT_ERROR_PATTERNS` in the engine's transient-error-detector
matches it. Keep the two in sync.
*/

/** JSON-RPC error codes that indicate a provider-side, retryable fault rather than a bad request. */
const RETRYABLE_JSONRPC_CODES = new Set([
  -32603, // Internal error — the agent blew up server-side.
  -32000, // Server error (generic, reserved implementation-defined range).
  -32001,
  -32002,
  -32003,
]);

/** Structured view of a thrown ACP/JSON-RPC error; all fields best-effort. */
export interface AcpTurnErrorDetail {
  message: string;
  code?: number;
  data?: unknown;
  retryable: boolean;
}

/**
 * Extract JSON-RPC `code`/`data` from a thrown ACP error.
 *
 * The SDK surfaces request errors in more than one shape depending on build: a flat
 * `{ code, message, data }` and a nested `{ error: { code, message, data } }`. Read both.
 */
export function inspectAcpTurnError(error: unknown): AcpTurnErrorDetail {
  type RpcShape = { code?: unknown; data?: unknown; message?: unknown };
  const raw = error as (RpcShape & { error?: RpcShape }) | null;
  const nested = raw && typeof raw === "object" ? raw.error : undefined;

  /*
  Read message/code/data from ONE source. Reading `code` from the nested payload while taking
  `message` from the outer Error yields "request failed (acp rpc code -32603)" — the outer
  wrapper text, not the provider's actual reason. The flat shape wins when it carries a numeric
  code; otherwise the nested envelope is authoritative.
  */
  const source: RpcShape | undefined =
    typeof raw?.code === "number" ? raw : typeof nested?.code === "number" ? nested : undefined;

  const code = typeof source?.code === "number" ? source.code : undefined;
  const data = source?.data;
  const message =
    (typeof source?.message === "string" && source.message)
    || (typeof raw?.message === "string" && raw.message)
    || (error instanceof Error ? error.message : String(error ?? "unknown error"));

  return { message, code, data, retryable: code !== undefined && RETRYABLE_JSONRPC_CODES.has(code) };
}

/**
 * Render a thrown ACP turn error as a single-line diagnostic that preserves the JSON-RPC code
 * and any `data` payload, so transient classification has something to anchor on.
 *
 * Shape: `Internal error (acp rpc code -32603, retryable) [data: {...}]`
 */
export function describeAcpTurnError(error: unknown): string {
  const { message, code, data, retryable } = inspectAcpTurnError(error);
  if (code === undefined) return message;
  const suffix = retryable ? ", retryable" : "";
  let out = `${message} (acp rpc code ${code}${suffix})`;
  if (data !== undefined) {
    let rendered: string;
    try {
      rendered = typeof data === "string" ? data : JSON.stringify(data);
    } catch {
      rendered = String(data);
    }
    if (rendered && rendered !== "{}") out += ` [data: ${rendered.slice(0, 500)}]`;
  }
  return out;
}

/**
 * Send a prompt turn via `session/prompt` and return the terminal `stopReason`.
 *
 * The SDK prompt promise resolves only AFTER every `session/update` for the turn
 * has been delivered to the client handler — so resolving here is the correct
 * "turn complete" signal (no extra draining required).
 *
 * FNXC:AcpRuntime 2026-07-15-18:20: rethrows with `describeAcpTurnError` so the JSON-RPC code
 * survives to the engine's transient classifier (FN-8004). `cause` retains the original error.
 */
export async function promptAcpSession(
  connection: AcpConnection,
  sessionId: string,
  blocks: ContentBlock[],
): Promise<StopReason> {
  try {
    const res = await connection.conn.prompt({ sessionId, prompt: blocks });
    return res.stopReason;
  } catch (error) {
    throw new Error(describeAcpTurnError(error), { cause: error });
  }
}

/**
 * Best-effort cancel of the active turn via the `session/cancel` notification.
 *
 * This is fire-and-forget (no ack in the protocol). Errors are swallowed — it
 * runs during teardown where the registry SIGKILL is the authoritative guarantee
 * (KTD4a).
 */
/** Upper bound on how long `cancelAcpSession` waits on the cancel write (FIX 7). */
const CANCEL_TIMEOUT_MS = 2_000;

export async function cancelAcpSession(
  connection: AcpConnection,
  sessionId: string,
): Promise<void> {
  // `conn.cancel` writes to the agent's stdin pipe; a dead or full pipe can
  // back-pressure and stall teardown (the adapter awaits this BEFORE the
  // authoritative registry SIGKILL). Bound it so the kill still runs promptly
  // (FIX 7). Errors are swallowed — this is already best-effort.
  try {
    await Promise.race([
      connection.conn.cancel({ sessionId }),
      new Promise<void>((resolve) => {
        const timer = setTimeout(resolve, CANCEL_TIMEOUT_MS);
        timer.unref?.();
      }),
    ]);
  } catch {
    // fire-and-forget; teardown's SIGKILL is authoritative
  }
}

/**
 * Resume a session. Prefers `session/load` (history replay) when the agent
 * advertised the `loadSession` capability; otherwise falls back to opening a
 * fresh `session/new`. There is no separate `resume` method in this SDK build —
 * `loadSession` IS the resume path.
 *
 * NOTE (v1): engine-driven resume wiring is intentionally deferred — the
 * runtime adapter always opens a fresh session via `newAcpSession`. This helper
 * exists (and is unit-tested for the id-sanitization invariant) so resume can be
 * wired in by passing a `sessionId` through `AgentRuntimeOptions` later without
 * building new resume machinery.
 */
export async function loadAcpSession(
  connection: AcpConnection,
  opts: { sessionId: string; cwd: string },
): Promise<NewAcpSessionResult> {
  if (readsLoadSession(connection)) {
    // Bound the (agent-originated) resume id before it is used as a protocol /
    // potential path component (U6/Risk S7).
    const safeId = boundIdentifier(opts.sessionId);
    const res = await connection.conn.loadSession({
      sessionId: safeId,
      cwd: opts.cwd,
      mcpServers: [],
    });
    return { sessionId: safeId, modes: res.modes ?? undefined };
  }
  return newAcpSession(connection, { cwd: opts.cwd });
}