#!/usr/bin/env node
/**
 * Graph MCP shim — adopts upstream `mcp-neo4j-cypher` per brand.
 *
 * Claude Code spawns this file as the `graph` MCP server; it
 * starts the upstream Python server under `uvx` and proxies the JSON-RPC
 * stdio stream byte-for-byte. While proxying, it:
 *   1. Resolves `NEO4J_URI/USERNAME/PASSWORD` from the brand's own env
 *      (falls back to `${PLATFORM_ROOT}/config/.neo4j-password`, matching
 *      plugins/memory/mcp/src/lib/neo4j.ts:10-21).
 *   2. Installs the Maxy stderr tee so upstream stderr lands in the
 *      per-conversation stream log alongside every other plugin. The raw
 *      per-server sink is the mcp-spawn-tee shim's, not this module's.
 *   3. Locks the upstream server into read-only mode (`NEO4J_READ_ONLY=true`)
 *      and a stable Maxy namespace (`NEO4J_NAMESPACE=maxy-graph`).
 *   4. Parses JSON-RPC lines on the request path to record the cypher and
 *      start time per `id`, and on the response path to emit one
 *      `[graph-query] op=... brand=... port=... cypher="..." rows=... ms=...`
 *      line per tool call. Forwarding is byte-accurate — the parser
 *      observes, never mutates.
 *
 * Isolation model per MAXY-PRD.md:627 and create-maxy src/index.ts:504-824:
 * each brand already owns its own Neo4j instance on its own port with its
 * own password. This shim trusts that — no query-layer tenant filtering.
 */

import { execFileSync, spawn } from "node:child_process";
import { accessSync, constants, readFileSync, statSync } from "node:fs";
import { resolve } from "node:path";
import { StringDecoder } from "node:string_decoder";
import {
  validate as validateCypher,
  type ForbiddenPattern,
  type UnknownToken,
} from "./cypher-validate.js";
import {
  auditCypherWrite,
  formatAuditLine,
  type AuditWarning,
} from "../../graph-write/dist/audit.js";
import { SchemaCache, getSharedDriver, neo4jSchemaFetcher } from "./schema-cache.js";
import { rewriteWithProvenanceStamps } from "./cypher-rewrite-stamp.js";
import {
  OrphanRollbackError,
  runWriteTxBody,
  synthesiseOrphanRollback,
  synthesiseShimError,
  synthesiseWriteResponse,
  type GraphDriver,
} from "./cypher-shim-write.js";
import {
  filterEnvelopeNotifications,
  runReadProbe,
  stitchWarningsIntoResponse,
  type ProbeDriver,
} from "./cypher-shim-read.js";
import { createHash } from "node:crypto";

const SERVER_NAME = "graph";
const UPSTREAM_PACKAGE = "mcp-neo4j-cypher@0.6.0";


// Emit a failure line on exit paths (missing password, uvx unresolvable,
// upstream spawn/exit error). The mcp-spawn-tee shim spawns this entry (see
// PLUGIN.md) and installs an appendFileSync-backed stderr writer, so this
// process.stderr.write reaches the raw sink synchronously and survives an
// immediate process.exit() — an async buffer would not flush in time.
function syncEmit(line: string): void {
  const msg = `[graph-mcp] ${line}`;
  try {
    process.stderr.write(`${msg}\n`);
  } catch {
    /* stderr closed — nothing else to do */
  }
}

function resolvePassword(): string {
  if (process.env.NEO4J_PASSWORD) return process.env.NEO4J_PASSWORD;
  // __dirname points at `lib/graph-mcp/dist` after compilation; the platform
  // root is three levels up (lib/graph-mcp/dist -> lib/graph-mcp -> lib -> platform).
  const root = process.env.PLATFORM_ROOT ?? resolve(__dirname, "../../..");
  const file = resolve(root, "config/.neo4j-password");
  try {
    return readFileSync(file, "utf-8").trim();
  } catch {
    syncEmit(`password unavailable — file=${file} env=NEO4J_PASSWORD=unset`);
    process.exit(1);
  }
}

// Resolve `uvx` to an absolute path at boot so the systemd-user default PATH
// (which excludes ~/.local/bin where the installer's installUv drops uvx)
// does not ENOENT the spawn. Precedence: UVX_PATH env override > ~/.local/bin/uvx
// (direct filesystem check, skips PATH lookup) > `which uvx` (catches custom
// installs). Each source is recorded in the diagnostic log line so a later
// "uvx unresolvable" emission names every checked location, not just a bare
// "not found".
interface UvxResolution {
  path: string | null;
  source: "env" | "home-local" | "which" | null;
  envChecked: string;
  homeLocalChecked: string;
  whichChecked: string;
}

function resolveUvxPath(): UvxResolution {
  const envPath = process.env.UVX_PATH ?? "";
  if (envPath) {
    try {
      accessSync(envPath, constants.X_OK);
      return { path: envPath, source: "env", envChecked: envPath, homeLocalChecked: "skipped", whichChecked: "skipped" };
    } catch {
      /* fall through */
    }
  }
  const home = process.env.HOME ?? "";
  const homeLocalPath = home ? resolve(home, ".local/bin/uvx") : "";
  if (homeLocalPath) {
    try {
      const stats = statSync(homeLocalPath);
      if (stats.isFile()) {
        accessSync(homeLocalPath, constants.X_OK);
        return { path: homeLocalPath, source: "home-local", envChecked: envPath || "unset", homeLocalChecked: homeLocalPath, whichChecked: "skipped" };
      }
    } catch {
      /* fall through */
    }
  }
  let whichPath = "";
  try {
    whichPath = execFileSync("which", ["uvx"], { encoding: "utf-8", stdio: ["ignore", "pipe", "ignore"] }).trim();
  } catch {
    whichPath = "";
  }
  if (whichPath) {
    // Verify the result is still executable. `which` prints the first PATH
    // hit by name; if the target was removed or the symlink dangles, the
    // spawn would ENOENT and reintroduce the silent-fail class
    // closes.
    try {
      accessSync(whichPath, constants.X_OK);
      return { path: whichPath, source: "which", envChecked: envPath || "unset", homeLocalChecked: homeLocalPath || "unset", whichChecked: whichPath };
    } catch {
      return { path: null, source: null, envChecked: envPath || "unset", homeLocalChecked: homeLocalPath || "unset", whichChecked: `${whichPath} (not-executable)` };
    }
  }
  return { path: null, source: null, envChecked: envPath || "unset", homeLocalChecked: homeLocalPath || "unset", whichChecked: "not-found" };
}

const brand = process.env.BRAND ?? "maxy";
// NEO4J_URI must be explicit. graph-mcp is spawned from
// claude-agent.ts:getMcpServers which now propagates NEO4J_URI via baseEnv.
const neo4jUri = process.env.NEO4J_URI;
if (!neo4jUri) {
  throw new Error(
    "[graph-mcp] NEO4J_URI unset — refusing to default to bolt://localhost:7687"
  );
}
console.error(`[graph-mcp] resolved neo4j_uri=${neo4jUri}`);
const neo4jUser = process.env.NEO4J_USERNAME ?? process.env.NEO4J_USER ?? "neo4j";
const neo4jPassword = resolvePassword();
const portMatch = /:(\d+)$/.exec(neo4jUri);
const neo4jPort = portMatch ? portMatch[1] : "?";
const namespace = process.env.NEO4J_NAMESPACE ?? "maxy-graph";
// default flipped from "true" to "false" so the upstream
// `mcp-neo4j-cypher` registers `write_neo4j_cypher` alongside the read tool.
// Per-agent gating via the `tools:` frontmatter list confines the surface:
// this server is packaged as the `graph` plugin (Task 502), so its tools
// surface under the long prefix `mcp__plugin_graph_graph__maxy-graph-*`. The
// write tool is not admin-allowlisted in `plugins/graph/PLUGIN.md`; only the
// `database-operator` / `librarian` specialists list it. Operators who set
// NEO4J_READ_ONLY=true explicitly (e.g. dev sandboxes) still get the
// upstream's read-only mode.
const readOnly = process.env.NEO4J_READ_ONLY ?? "false";
const responseTokenLimit = process.env.NEO4J_RESPONSE_TOKEN_LIMIT ?? "20000";

const childEnv: NodeJS.ProcessEnv = {
  ...process.env,
  NEO4J_URI: neo4jUri,
  NEO4J_URL: neo4jUri,
  NEO4J_USERNAME: neo4jUser,
  NEO4J_PASSWORD: neo4jPassword,
  NEO4J_READ_ONLY: readOnly,
  NEO4J_NAMESPACE: namespace,
  NEO4J_RESPONSE_TOKEN_LIMIT: responseTokenLimit,
};

console.error(
  `[graph-mcp] boot brand=${brand} uri=${neo4jUri} user=${neo4jUser} ` +
    `namespace=${namespace} readOnly=${readOnly} tokenLimit=${responseTokenLimit}`,
);

// async schema cache. The validator fails OPEN until the first
// refresh resolves; all cypher calls during that window are forwarded
// unvalidated and the existing [graph-query] line gains `validated=false`
// so operators see the bypass. One Neo4j driver is shared across refreshes;
// loaded lazily so test harnesses can exercise SchemaCache without pulling
// neo4j-driver into the module graph.
// neo4jUri is narrowed to `string` by the guard throw above, but the narrow
// doesn't flow into the closure below. Pin it to a local const whose type
// is `string` by construction.
const resolvedNeo4jUri: string = neo4jUri;
let schemaFetcherReady: Promise<Awaited<ReturnType<typeof neo4jSchemaFetcher>>> | null = null;
function getSchemaFetcher(): Promise<Awaited<ReturnType<typeof neo4jSchemaFetcher>>> {
  if (!schemaFetcherReady) {
    schemaFetcherReady = neo4jSchemaFetcher(resolvedNeo4jUri, neo4jUser, neo4jPassword);
  }
  return schemaFetcherReady;
}
const schemaCache = new SchemaCache({
  async labels() {
    const f = await getSchemaFetcher();
    return f.labels();
  },
  async relationshipTypes() {
    const f = await getSchemaFetcher();
    return f.relationshipTypes();
  },
});
void schemaCache.start().catch((err) => {
  const msg = err instanceof Error ? err.message : String(err);
  console.error(`[schema-cache] start failed error="${msg.replace(/"/g, "'")}"`);
});

const uvx = resolveUvxPath();
if (!uvx.path) {
  syncEmit(
    `uvx unresolvable — checked env=${uvx.envChecked} home-local=${uvx.homeLocalChecked} which=${uvx.whichChecked}`,
  );
  syncEmit(
    "install uv (if missing): curl -LsSf https://astral.sh/uv/install.sh | sh",
  );
  process.exit(127);
}
console.error(`[graph-mcp] uvx resolved path=${uvx.path} source=${uvx.source}`);

const child = spawn(uvx.path, [UPSTREAM_PACKAGE, "--transport", "stdio"], {
  stdio: ["pipe", "pipe", "pipe"],
  env: childEnv,
});

child.on("spawn", () => {
  console.error(`[graph-mcp] spawned ${UPSTREAM_PACKAGE} pid=${child.pid}`);
});

child.on("error", (err) => {
  const msg = err instanceof Error ? err.message : String(err);
  syncEmit(`spawn error: ${msg} uvx=${uvx.path ?? "unresolved"}`);
  process.exit(127);
});

child.on("exit", (code, signal) => {
  syncEmit(`uvx exited code=${code ?? "null"} signal=${signal ?? "null"}`);
  process.exit(code ?? (signal ? 1 : 0));
});

// Forward child stderr to our own stderr so the stderr tee picks it up.
child.stderr.on("data", (chunk: Buffer) => {
  process.stderr.write(chunk);
});

// --- JSON-RPC call correlation + validation ---
// tools/call is the only method we time or validate. For read/write cypher
// calls, the line is validated against the schema cache before forwarding.
// Write-path rejection: synthesised MCP tool-error response on stdout, NOT
// forwarded — fires only on forbidden DDL/admin patterns
// unknown labels/types in writes are warnings, not rejections, since
// operators legitimately introduce new labels via REMOVE/SET). Read-path
// rejection: forwarded, with warnings appendix prepended to
// response.content[0].text.
interface PendingCall {
  method: string;
  cypherPrefix: string | null;
  /** Full cypher body: needed for the post-write audit's static parse. */
  cypherFull: string | null;
  isWrite: boolean;
  /** Caller-supplied sessionId param, used by the post-write audit emission. */
  sessionIdParam: string | null;
  startMs: number;
  validated: boolean;
  readWarnings: UnknownToken[];
  /** unknown relationship tokens carried from request to response so
   *  the audit emits one `unknown-type-warning` line per unknown after the
   *  upstream commits (the validator only detected them; emission waits for
   *  the post-write line family). */
  writeUnknownTokens: UnknownToken[];
  /** full operator-supplied params (e.g. for $-bound names), kept so
   *  the post-response envelope-warning probe runs the same cypher with the
   *  same params as the upstream call. Reads only. */
  cypherParams: Record<string, unknown> | null;
}
const pending = new Map<string | number, PendingCall>();

function truncate(s: string, n: number): string {
  return s.length <= n ? s : `${s.slice(0, n)}...`;
}

function stripNamespace(toolName: string | undefined): string {
  if (!toolName) return "(unknown)";
  const prefix = `${namespace}_`;
  return toolName.startsWith(prefix) ? toolName.slice(prefix.length) : toolName;
}

const READ_CYPHER_TOOL = "read_neo4j_cypher";
const WRITE_CYPHER_TOOL = "write_neo4j_cypher";

type JsonRpcMessage = {
  jsonrpc?: string;
  id?: string | number;
  method?: string;
  params?: { name?: string; arguments?: Record<string, unknown> };
  result?: {
    content?: Array<{ text?: string; type?: string }>;
    isError?: boolean;
    tools?: Array<Record<string, unknown>>;
  };
  error?: { message?: string; code?: number };
};

// track outstanding `tools/list` request ids so the response-path
// interceptor can rewrite the matching response. The upstream Python
// `mcp-neo4j-cypher` server does not expose tool-registration `_meta`, and we
// cannot patch the upstream from TypeScript — so the shim mints the
// `anthropic/alwaysLoad: true` flag on every namespaced tool here, on the
// wire, between upstream and Claude Code. Without this, the graph tools
// (read_neo4j_cypher / write_neo4j_cypher / get_neo4j_schema) stay deferred
// even when every in-process plugin has migrated to `eagerTool`, and the
// database-operator pays one ToolSearch hit per session for its three core
// tools.
const pendingListRequests = new Set<string | number>();

function extractCypherFull(args: Record<string, unknown> | undefined): string | null {
  if (!args) return null;
  const q = args["query"] ?? args["cypher"];
  return typeof q === "string" ? q : null;
}

// the operator's Graph Stewardship Doctrine (Rule 3) requires
// `r.createdBySession = $sessionId` on every write. The post-write audit
// emits the `[graph-cypher-write] accepted` line with this sessionId so
// later forensic queries can join the audit log to the persisted node set
// via `WHERE n.createdBySession = $sessionId`. Returns null if the operator
// did not pass a sessionId param — the audit then emits sessionId=unknown
// and the missing-provenance warning fires from the static parse.
function extractSessionIdParam(args: Record<string, unknown> | undefined): string | null {
  if (!args) return null;
  const params = args["params"];
  if (!params || typeof params !== "object") return null;
  const sid = (params as Record<string, unknown>)["sessionId"];
  return typeof sid === "string" ? sid : null;
}

// parse the upstream `mcp-neo4j-cypher` write response counters.
// The upstream emits human-readable summary phrases ("3 nodes created", "4
// relationships created") in the result.content[0].text. Defensive regex —
// missing phrases coerce to 0 so the accepted line still emits with
// observable nodesCreated/relsCreated fields.
function parseWriteCounters(
  result: JsonRpcMessage["result"],
): { nodes: number; rels: number } {
  if (!result?.content || !Array.isArray(result.content)) return { nodes: 0, rels: 0 };
  const text = result.content[0]?.text ?? "";
  const nodesMatch = text.match(/(\d+)\s*nodes?\s*created/i);
  const relsMatch = text.match(/(\d+)\s*relationships?\s*created/i);
  return {
    nodes: nodesMatch ? parseInt(nodesMatch[1], 10) : 0,
    rels: relsMatch ? parseInt(relsMatch[1], 10) : 0,
  };
}

function truncateForLog(cypher: string): string {
  return truncate(cypher.replace(/\s+/g, " ").trim(), 80);
}

function countRows(result: JsonRpcMessage["result"]): string {
  if (!result?.content || !Array.isArray(result.content)) return "?";
  const text = result.content[0]?.text ?? "";
  const rowsMatch = /(\d+)\s+(?:rows?|records?)/i.exec(text);
  if (rowsMatch) return rowsMatch[1];
  return String(result.content.length);
}

/**
 * Render the admin-facing rejection or warnings text. Plain prose so the
 * agent's Tool Failure Discipline reads it the same way it reads any tool
 * error — no special-case structured JSON parser required.
 */
function renderUnknownTokens(
  unknown: UnknownToken[],
  mode: "rejected" | "warning",
): string {
  const heading = mode === "rejected"
    ? "schema-validation rejected — cypher NOT executed"
    : "schema-validation warning — cypher executed but referenced unknown tokens";
  const lines = unknown.map((u) => {
    const nearest = u.nearest.length > 0 ? ` nearest=[${u.nearest.join(", ")}]` : "";
    return `  - ${u.kind} '${u.token}':${nearest} — ${u.hint}`;
  });
  return `${heading}\n${lines.join("\n")}`;
}

function synthesiseRejection(id: string | number, unknown: UnknownToken[]): string {
  const text = `${renderUnknownTokens(unknown, "rejected")}\n\nFix the token names, or if the schema has just changed, invoke ${namespace}_get_neo4j_schema to refresh your view.`;
  const envelope = {
    jsonrpc: "2.0",
    id,
    result: {
      content: [{ type: "text", text }],
      isError: true,
    },
  };
  return JSON.stringify(envelope);
}

// synthesised rejection for forbidden DDL/admin patterns. Fired
// from the write-mode validator when the operator (or a regression in admin
// dispatch) tries `DROP DATABASE`, `CREATE INDEX/CONSTRAINT`, `CALL dbms.*`,
// or `CALL db.create*`. The text names every matched pattern so the operator
// reads exactly which clause tripped the gate.
function synthesiseForbiddenRejection(
  id: string | number,
  forbidden: ForbiddenPattern[],
): string {
  const lines = forbidden.map(
    (f) => `  - ${f.kind}: matched "${f.match.replace(/"/g, "'")}" — ${f.hint}`,
  );
  const text =
    `forbidden cypher pattern — write NOT executed\n${lines.join("\n")}\n\n` +
    `Index/constraint DDL is admin-owned (seed-neo4j.sh). Security CALL surface ` +
    `(dbms.*, db.create*) is not exposed to the operator role.`;
  const envelope = {
    jsonrpc: "2.0",
    id,
    result: {
      content: [{ type: "text", text }],
      isError: true,
    },
  };
  return JSON.stringify(envelope);
}

function wrapReadWarnings(msg: JsonRpcMessage, warnings: UnknownToken[]): string {
  const warningText = `${renderUnknownTokens(warnings, "warning")}\n\n--- results below (executed despite unknown tokens) ---\n`;
  const original = msg.result?.content ?? [];
  const wrapped: JsonRpcMessage = {
    ...msg,
    result: {
      ...(msg.result ?? {}),
      content: [{ type: "text", text: warningText }, ...original],
    },
  };
  return JSON.stringify(wrapped);
}

// --- shim-owned write path ----------------------------------------------
// write_neo4j_cypher is intercepted (not forwarded to upstream). The shim
// runs the operator's cypher inside its own driver session under
// `executeWrite`, so provenance auto-stamping and orphan-rollback are
// transactionally atomic. Reads still pass through to upstream unchanged.
// Pure helpers (runWriteTxBody, synthesise*, OrphanRollbackError) live in
// cypher-shim-write.ts so they can be unit-tested without booting this
// module's stdin/stdout pipe and uvx spawn.

async function runShimWriteCypher(
  id: string | number,
  cypherFull: string,
  cypherPrefix: string,
  sessionIdParam: string | null,
  operatorArgs: Record<string, unknown> | undefined,
  startMs: number,
  writeUnknownTokens: UnknownToken[],
  validated: boolean,
): Promise<void> {
  const agentName = process.env.AGENT_SLUG ?? "unknown";
  const sessionIdField =
    sessionIdParam ?? process.env.SESSION_ID ?? "unknown";
  const operatorParams =
    (operatorArgs?.["params"] as Record<string, unknown> | undefined) ?? {};
  const safePrefix = cypherPrefix.replace(/"/g, "'");

  let rewrite: ReturnType<typeof rewriteWithProvenanceStamps>;
  try {
    rewrite = rewriteWithProvenanceStamps(cypherFull);
  } catch (err) {
    // Defensive — the rewriter is regex-only with no catastrophic-backtrack
    // patterns, but a thrown error before driver acquisition would otherwise
    // emit no [graph-query] line at all (review F3).
    const errMsg = err instanceof Error ? err.message : String(err);
    console.error(
      `[graph-query] op=${WRITE_CYPHER_TOOL} brand=${brand} port=${neo4jPort} cypher="${safePrefix}" error="rewrite-failed: ${errMsg.replace(/"/g, "'")}" validated=${validated} ms=${Date.now() - startMs}`,
    );
    process.stdout.write(
      `${synthesiseShimError(id, "cypher rewrite failed — write NOT executed", errMsg)}\n`,
    );
    return;
  }
  if (rewrite.stampsAppended > 0) {
    console.error(
      `[graph-cypher-write] auto-stamp applied query="${safePrefix}" creates=${rewrite.stampsAppended}`,
    );
  }

  let driver: GraphDriver;
  try {
    driver = (await getSharedDriver(resolvedNeo4jUri, neo4jUser, neo4jPassword)) as GraphDriver;
  } catch (err) {
    const errMsg = err instanceof Error ? err.message : String(err);
    console.error(
      `[graph-query] op=${WRITE_CYPHER_TOOL} brand=${brand} port=${neo4jPort} cypher="${safePrefix}" error="driver-unavailable: ${errMsg.replace(/"/g, "'")}" validated=${validated} ms=${Date.now() - startMs}`,
    );
    process.stdout.write(
      `${synthesiseShimError(id, "graph driver unavailable — write NOT executed", errMsg)}\n`,
    );
    return;
  }

  const session = driver.session();
  let outcome: { nodesCreated: number; relsCreated: number; propertiesSet: number; serializedRecords: Array<Record<string, unknown>> } | null = null;

  try {
    outcome = await session.executeWrite((tx) =>
      runWriteTxBody(tx, rewrite.cypher, operatorParams, {
        agent: agentName,
        session: sessionIdField,
      }),
    );
  } catch (err) {
    if (err instanceof OrphanRollbackError) {
      console.error(
        `[graph-cypher-write] orphan-rollback query="${safePrefix}" orphanLabels=${err.sampleLabels.join(",")}`,
      );
      console.error(
        formatAuditLine({
          kind: "orphan-warning",
          cypherPrefix,
          orphanIds: err.orphanIds,
        }),
      );
      console.error(
        `[graph-query] op=${WRITE_CYPHER_TOOL} brand=${brand} port=${neo4jPort} cypher="${safePrefix}" rolledBack=true orphans=${err.orphanIds.length} validated=${validated} ms=${Date.now() - startMs}`,
      );
      process.stdout.write(
        `${synthesiseOrphanRollback(id, err.orphanIds.length, err.sampleLabels)}\n`,
      );
    } else {
      const errMsg = err instanceof Error ? err.message : String(err);
      console.error(
        `[graph-query] op=${WRITE_CYPHER_TOOL} brand=${brand} port=${neo4jPort} cypher="${safePrefix}" error="${errMsg.replace(/"/g, "'")}" validated=${validated} ms=${Date.now() - startMs}`,
      );
      process.stdout.write(
        `${synthesiseShimError(id, "Neo4j error", errMsg)}\n`,
      );
    }
    return;
  } finally {
    await session.close().catch(() => {
      /* session already closed by the driver on commit/rollback */
    });
  }

  // outcome is non-null on the success path (the catch above returns early
  // on rollback / shim error). Defensive narrow.
  if (!outcome) return;

  console.error(
    `[graph-query] op=${WRITE_CYPHER_TOOL} brand=${brand} port=${neo4jPort} cypher="${safePrefix}" rows=${outcome.serializedRecords.length} validated=${validated} ms=${Date.now() - startMs}`,
  );
  console.error(
    formatAuditLine({
      kind: "accepted",
      cypherPrefix,
      nodesCreated: outcome.nodesCreated,
      relsCreated: outcome.relsCreated,
      agentName,
      sessionId: sessionIdField,
    }),
  );

  for (const t of writeUnknownTokens.filter((u) => u.kind === "relationship")) {
    console.error(
      formatAuditLine({
        kind: "unknown-type-warning",
        cypherPrefix,
        type: t.token,
      }),
    );
  }

  const auditWarnings: AuditWarning[] = auditCypherWrite({
    cypher: rewrite.cypher,
    schema: schemaCache.snapshot(),
    agentName,
    sessionId: sessionIdField,
    nodesCreated: outcome.nodesCreated,
    relsCreated: outcome.relsCreated,
    orphanIds: [],
  });
  for (const w of auditWarnings) {
    if (w.kind === "missing-provenance-warning") {
      console.error(
        formatAuditLine({
          kind: "missing-provenance-warning",
          cypherPrefix,
          created: w.created,
          stamped: w.stamped,
        }),
      );
    }
    // unknown-type already emitted from the validator's writeUnknownTokens above;
    // orphan-warning fires only on the rollback path.
  }

  process.stdout.write(
    `${synthesiseWriteResponse(id, {
      nodesCreated: outcome.nodesCreated,
      relsCreated: outcome.relsCreated,
      propertiesSet: outcome.propertiesSet,
      records: outcome.serializedRecords,
    })}\n`,
  );
}

type RequestDecision = "forward" | "intercepted";

function handleRequestLine(line: string): RequestDecision {
  let msg: JsonRpcMessage;
  try {
    msg = JSON.parse(line) as JsonRpcMessage;
  } catch {
    return "forward";
  }
  // register `tools/list` ids for the response interceptor.
  // Upstream still answers; we mutate the answer in flight.
  if (msg.method === "tools/list" && msg.id !== undefined) {
    pendingListRequests.add(msg.id);
    return "forward";
  }
  if (msg.method !== "tools/call" || msg.id === undefined) return "forward";

  const methodName = stripNamespace(msg.params?.name);
  const cypherFull = extractCypherFull(msg.params?.arguments);
  const cypherPrefix = cypherFull ? truncateForLog(cypherFull) : null;
  const isCypherCall =
    methodName === READ_CYPHER_TOOL || methodName === WRITE_CYPHER_TOOL;
  const isWriteCall = methodName === WRITE_CYPHER_TOOL;
  const sessionIdParam = isWriteCall
    ? extractSessionIdParam(msg.params?.arguments)
    : null;

  const operatorArgs = msg.params?.arguments ?? {};
  const paramsArg = operatorArgs["params"];
  const cypherParams: Record<string, unknown> | null =
    paramsArg && typeof paramsArg === "object" && !Array.isArray(paramsArg)
      ? (paramsArg as Record<string, unknown>)
      : null;

  const entry: PendingCall = {
    method: methodName,
    cypherPrefix,
    cypherFull,
    isWrite: isWriteCall,
    sessionIdParam,
    startMs: Date.now(),
    validated: false,
    readWarnings: [],
    writeUnknownTokens: [],
    cypherParams: isWriteCall ? null : cypherParams,
  };

  if (!isCypherCall || !cypherFull) {
    pending.set(msg.id, entry);
    return "forward";
  }

  // writes are intercepted into the shim-owned driver path.
  // Helper so each branch below routes consistently.
  const interceptWrite = (): RequestDecision => {
    void runShimWriteCypher(
      msg.id as string | number,
      cypherFull,
      cypherPrefix ?? "",
      sessionIdParam,
      msg.params?.arguments,
      entry.startMs,
      entry.writeUnknownTokens,
      entry.validated,
    ).catch((err) => {
      // Defensive catch — runShimWriteCypher already handles its own errors,
      // but never let an unexpected throw leak past this point.
      const errMsg = err instanceof Error ? err.message : String(err);
      console.error(
        `[graph-mcp] runShimWriteCypher unhandled error: ${errMsg.replace(/"/g, "'")}`,
      );
      try {
        process.stdout.write(
          `${synthesiseShimError(msg.id as string | number, "shim-internal error", errMsg)}\n`,
        );
      } catch {
        /* stdout closed */
      }
    });
    return "intercepted";
  };

  const snapshot = schemaCache.snapshot();
  const cacheReady = schemaCache.ready();

  if (!cacheReady) {
    console.error(
      `[cypher-validate] tool=${isWriteCall ? "write" : "read"} outcome=skipped reason=cache-not-ready cypher="${(cypherPrefix ?? "").replace(/"/g, "'")}"`,
    );
    if (isWriteCall) {
      // still rewrite + run inside shim tx even when validator
      // is cold. Auto-stamp + orphan rollback don't depend on schema.
      return interceptWrite();
    }
    pending.set(msg.id, entry);
    return "forward";
  }

  const result = validateCypher(cypherFull, snapshot, {
    mode: isWriteCall ? "write" : "read",
  });
  entry.validated = true;

  // forbidden DDL/admin → hard reject in write mode (only path
  // that produces forbidden entries; read mode never populates them).
  if (isWriteCall && result.forbidden.length > 0) {
    const forbiddenSummary = result.forbidden.map((f) => f.kind).join(",");
    console.error(
      `[cypher-validate] tool=write outcome=rejected forbidden=${forbiddenSummary} cypher="${(cypherPrefix ?? "").replace(/"/g, "'")}"`,
    );
    const response = synthesiseForbiddenRejection(msg.id, result.forbidden);
    process.stdout.write(`${response}\n`);
    console.error(
      `[graph-query] op=${methodName} brand=${brand} port=${neo4jPort} cypher="${(cypherPrefix ?? "").replace(/"/g, "'")}" rejected=true forbidden=${forbiddenSummary} validated=true ms=${Date.now() - entry.startMs}`,
    );
    return "intercepted";
  }

  if (result.ok) {
    console.error(
      `[cypher-validate] tool=${isWriteCall ? "write" : "read"} outcome=accepted labels=${result.labelTokens.length} relationships=${result.edgeTokens.length}`,
    );
    if (isWriteCall) return interceptWrite();
    pending.set(msg.id, entry);
    return "forward";
  }

  void schemaCache.maybeRebuildOnStaleMiss(result.unknown);
  const tokenSummary = result.unknown
    .map((u) => `${u.kind}:${u.token}`)
    .join(",");

  if (isWriteCall) {
    // write mode treats unknown tokens as soft warnings — operators
    // legitimately introduce new labels (REMOVE n:Old SET n:New) and edges
    // pending an ontology update. The post-write audit emits one
    // unknown-type-warning per unknown so operators see the gap; the cypher
    // still commits via the shim-owned write path.
    entry.writeUnknownTokens = result.unknown;
    console.error(
      `[cypher-validate] tool=write outcome=warned unknown=${tokenSummary} cypher="${(cypherPrefix ?? "").replace(/"/g, "'")}"`,
    );
    return interceptWrite();
  }

  entry.readWarnings = result.unknown;
  console.error(
    `[cypher-validate] tool=read outcome=warned unknown=${tokenSummary} cypher="${(cypherPrefix ?? "").replace(/"/g, "'")}"`,
  );
  pending.set(msg.id, entry);
  return "forward";
}

async function handleResponseLine(line: string): Promise<string | null> {
  let msg: JsonRpcMessage;
  try {
    msg = JSON.parse(line) as JsonRpcMessage;
  } catch {
    return null;
  }
  // `tools/list` response interceptor. Inject
  // `_meta["anthropic/alwaysLoad"]: true` on every upstream tool so the
  // Claude Code SDK's `isDeferredTool` short-circuit fires and the model
  // can call the graph tools without paying a ToolSearch round-trip.
  // Failure here forwards the original response unchanged: a bug in the
  // injection path must not break Cypher availability.
  if (msg.id !== undefined && pendingListRequests.has(msg.id)) {
    pendingListRequests.delete(msg.id);
    const tools = msg.result?.tools;
    if (!Array.isArray(tools) || tools.length === 0) return null;
    let mutated = 0;
    for (const tool of tools) {
      const existingMeta = (tool["_meta"] && typeof tool["_meta"] === "object")
        ? (tool["_meta"] as Record<string, unknown>)
        : {};
      tool["_meta"] = { ...existingMeta, "anthropic/alwaysLoad": true };
      mutated += 1;
    }
    console.error(`[graph-mcp] tools/list eager-flagged count=${mutated}`);
    try {
      return JSON.stringify(msg);
    } catch (err) {
      console.error(
        `[graph-mcp] tools/list rewrite failed error="${(err instanceof Error ? err.message : String(err)).replace(/"/g, "'")}" — forwarding upstream response unchanged`,
      );
      return null;
    }
  }
  if (msg.id === undefined || !pending.has(msg.id)) return null;
  const p = pending.get(msg.id)!;
  pending.delete(msg.id);
  const elapsed = Date.now() - p.startMs;
  const cypherField = p.cypherPrefix ? `cypher="${p.cypherPrefix.replace(/"/g, "'")}"` : "";
  const validatedField = `validated=${p.validated}`;
  if (msg.error) {
    const errText = (msg.error.message ?? JSON.stringify(msg.error)).replace(/"/g, "'");
    console.error(
      `[graph-query] op=${p.method} brand=${brand} port=${neo4jPort} ${cypherField} error="${errText}" ${validatedField} ms=${elapsed}`,
    );
    return null;
  }
  const rows = countRows(msg.result);
  const warnedField = p.readWarnings.length > 0 ? ` warned=${p.readWarnings.length}` : "";
  console.error(
    `[graph-query] op=${p.method} brand=${brand} port=${neo4jPort} ${cypherField} rows=${rows} ${validatedField}${warnedField} ms=${elapsed}`,
  );

  // post-write audit. The shim emits the `[graph-cypher-write]`
  // family alongside the existing `[graph-query]` line so operators have a
  // dedicated stream to grep without read-side noise. Static-only audit:
  // unknown-type and missing-provenance warnings come from regex over the
  // cypher body. Dynamic orphan detection (querying Neo4j for nodes
  // matching createdBySession=$id AND NOT (n)--()) is deferred to
  // — the audit module's `orphanIds` input stays empty here, the
  // `[graph-cypher-write] orphan-warning` line never fires from production
  // until that wires up.
  if (p.isWrite && p.cypherFull && msg.result && !msg.result.isError) {
    const counters = parseWriteCounters(msg.result);
    const cypherPrefix = p.cypherPrefix ?? "";
    const agentName = process.env.AGENT_SLUG ?? "unknown";
    const sessionIdField = p.sessionIdParam ?? "unknown";
    const snapshot = schemaCache.snapshot();
    console.error(
      formatAuditLine({
        kind: "accepted",
        cypherPrefix,
        nodesCreated: counters.nodes,
        relsCreated: counters.rels,
        agentName,
        sessionId: sessionIdField,
      }),
    );
    const auditWarnings: AuditWarning[] = auditCypherWrite({
      cypher: p.cypherFull,
      schema: snapshot,
      agentName,
      sessionId: sessionIdField,
      nodesCreated: counters.nodes,
      relsCreated: counters.rels,
      orphanIds: [],
    });
    for (const w of auditWarnings) {
      switch (w.kind) {
        case "unknown-type-warning":
          console.error(
            formatAuditLine({
              kind: "unknown-type-warning",
              cypherPrefix,
              type: w.type,
            }),
          );
          break;
        case "missing-provenance-warning":
          console.error(
            formatAuditLine({
              kind: "missing-provenance-warning",
              cypherPrefix,
              created: w.created,
              stamped: w.stamped,
            }),
          );
          break;
        case "orphan-warning":
          console.error(
            formatAuditLine({
              kind: "orphan-warning",
              cypherPrefix,
              orphanIds: w.orphanIds,
            }),
          );
          break;
      }
    }
  }

  // envelope-warning probe. Reads only, after upstream succeeds.
  // Runs the same cypher through the shim's own driver to harvest
  // `summary.notifications` matching ^0[12]N5\d$ (property/label-missing
  // codes the upstream Python server drops before serialising). Best-effort:
  // any probe failure leaves the upstream response untouched.
  let envelopeStitched: string | null = null;
  if (
    !p.isWrite &&
    p.cypherFull &&
    msg.result &&
    !msg.result.isError &&
    p.method === READ_CYPHER_TOOL
  ) {
    try {
      const driver = (await getSharedDriver(
        resolvedNeo4jUri,
        neo4jUser,
        neo4jPassword,
      )) as ProbeDriver;
      const notifications = await runReadProbe(
        driver,
        p.cypherFull,
        p.cypherParams ?? {},
      );
      const envelopeWarnings = filterEnvelopeNotifications(notifications);
      if (envelopeWarnings.length > 0) {
        const queryHash = createHash("sha1")
          .update(p.cypherFull)
          .digest("hex")
          .slice(0, 12);
        for (const w of envelopeWarnings) {
          console.error(
            `[mcp:graph] envelope-warning gql_status=${w.gql_status} query_hash=${queryHash} description="${w.description.replace(/"/g, "'")}"`,
          );
        }
        envelopeStitched = stitchWarningsIntoResponse(msg, envelopeWarnings);
      }
    } catch (err) {
      const errMsg = err instanceof Error ? err.message : String(err);
      console.error(
        `[mcp:graph] probe-error op=${p.method} error="${errMsg.replace(/"/g, "'")}" — forwarding response without envelope stitch`,
      );
    }
  }

  // Compose with the existing cypher-validate warning prefix. If
  // both are present, the envelope warnings appear first (outermost prepend),
  // then the validation warnings, then the upstream's rendered rows.
  if (p.readWarnings.length > 0) {
    try {
      const validationWrapped = wrapReadWarnings(msg, p.readWarnings);
      if (envelopeStitched) {
        // Re-stitch envelope warnings ONTO the validation-wrapped message.
        const parsed = JSON.parse(envelopeStitched) as JsonRpcMessage;
        const valParsed = JSON.parse(validationWrapped) as JsonRpcMessage;
        const merged: JsonRpcMessage = {
          ...valParsed,
          result: {
            ...(valParsed.result ?? {}),
            content: [
              ...(parsed.result?.content ?? []).slice(0, 1),
              ...(valParsed.result?.content ?? []),
            ],
          },
        };
        return JSON.stringify(merged);
      }
      return validationWrapped;
    } catch (err) {
      const errMsg = err instanceof Error ? err.message : String(err);
      console.error(
        `[cypher-validate] warning-wrap failed op=${p.method} error="${errMsg.replace(/"/g, "'")}" — forwarding response unwrapped`,
      );
      return envelopeStitched;
    }
  }
  return envelopeStitched;
}

/**
 * Per-stream buffering splitter. Yields complete lines (without trailing \n)
 * as they accumulate; leaves any partial tail in the buffer until more bytes
 * arrive. The caller decides per-line whether to forward or intercept.
 */
function makeLineBuffer(): { push: (chunk: Buffer) => string[] } {
  const decoder = new StringDecoder("utf8");
  let buf = "";
  return {
    push(chunk: Buffer): string[] {
      buf += decoder.write(chunk);
      const out: string[] = [];
      let idx: number;
      while ((idx = buf.indexOf("\n")) !== -1) {
        out.push(buf.slice(0, idx));
        buf = buf.slice(idx + 1);
      }
      return out;
    },
  };
}

const requestBuffer = makeLineBuffer();
process.stdin.on("data", (chunk: Buffer) => {
  for (const line of requestBuffer.push(chunk)) {
    if (line.length === 0) {
      child.stdin.write("\n");
      continue;
    }
    const decision = handleRequestLine(line);
    if (decision === "forward") {
      child.stdin.write(`${line}\n`);
    }
    // "intercepted" — synthesised response already written to stdout.
  }
});
process.stdin.on("end", () => {
  child.stdin.end();
});

const responseBuffer = makeLineBuffer();
// handleResponseLine is now async (envelope-warning probe). The
// per-line work is serialised through a write-chain so multi-line chunks
// preserve their original byte order on the way to stdout. The chain only
// adds latency when the probe actually runs (reads with a result); other
// lines resolve synchronously and append immediately.
let responseWriteChain: Promise<void> = Promise.resolve();
child.stdout.on("data", (chunk: Buffer) => {
  for (const line of responseBuffer.push(chunk)) {
    responseWriteChain = responseWriteChain.then(async () => {
      if (line.length === 0) {
        process.stdout.write("\n");
        return;
      }
      const rewritten = await handleResponseLine(line);
      process.stdout.write(`${rewritten ?? line}\n`);
    });
  }
});
child.stdout.on("end", () => {
  // handleResponseLine is async; if a probe is still in flight when
  // upstream closes its stdout, ending process.stdout synchronously would drop
  // the final response (Node's Writable silently discards writes after end()).
  // Await the write-chain so every queued line reaches stdout first.
  responseWriteChain.then(() => process.stdout.end());
});

for (const sig of ["SIGTERM", "SIGINT", "SIGHUP"] as const) {
  process.on(sig, () => {
    console.error(`[graph-mcp] ${sig} received — forwarding to child`);
    child.kill(sig);
  });
}