/**
 * Shared hybrid search primitive over the Neo4j knowledge graph.
 *
 * Pre-Task-675 there were two BM25 implementations over the same fulltext
 * index — one in the memory MCP tool, one in the admin Hono route — with
 * divergent filter semantics (soft-delete primitive) and divergent ranking
 * (agent ran hybrid vector+BM25, UI ran BM25-only). collapses both
 * into this lib; memory MCP + admin route now share a single code path so
 * `/graph` UI ranking matches what the agent sees.
 *
 * The fulltext index `knowledge_fulltext` was renamed to `entity_search`
 * and its label union expanded from 3 labels (KnowledgeDocument | Section |
 * Chunk) to the full operator-meaningful label set (~40 labels) on every
 * textual property the schema's writers assign. Task 416 split that single
 * `entity_search` into an admin index plus a public twin; the public twin
 * and its whole read path were retired once public agents became toolless,
 * so the lib now targets the single `entity_search_admin` index via
 * FULLTEXT_INDEX_ADMIN below. The doctrine test at
 * `__tests__/fulltext-coverage.test.ts` parses that declaration and asserts
 * the universal label union.
 *
 *   QUERY -->  EMBED -->  VECTOR SEARCH (per index) --> ┐
 *        │                                               ├--> MERGE --> EXPAND --> RESULTS
 *        └-->  BM25 FULL-TEXT SEARCH -------------------> ┘
 *
 * Hybrid merge: normalise BM25 scores to [0,1] via min-max, then combine
 *   combined = 0.7 * vector_score + 0.3 * normalised_bm25_score
 * Dedup by nodeId, keeping the higher combined score.
 *
 * Trashed nodes (`:Trashed` label + `deletedAt` property) are excluded
 * from every path via `notTrashed()` from `graph-trash`.
 *
 * The lib is STATELESS. Callers pass a `Session` and, for hybrid, an
 * `embed` function. This keeps the lib independent of Ollama config and
 * free of circular imports against each caller's own getSession/embed
 * (memory MCP and the Hono ui build each have their own copies for
 * cross-build-boundary reasons — see neo4j-store.ts:68 comment).
 *
 * Log emission is the caller's responsibility — the lib throws or returns
 * `mode` in the result so the caller can format its own `[graph-search]`
 * line with the prefix and fields it wants.
 */

import { int, type Session } from "neo4j-driver";
import { notTrashed } from "../../graph-trash/dist/index.js";
import { fuseRrf } from "./rrf-fusion.js";
// Task 424 — query-expansion deleted. The calling agent expands its query
// terms in-turn before invoking memory-search; the server no longer makes a
// per-search Haiku call to paraphrase.
import { applyCompiledTruthBoost, applyBacklinkBoost } from "./boosts.js";
import { dedup as dedupHits, DEFAULT_LAYERS, type DedupLayer } from "./dedup.js";
import { pickIndexMix, type RetrievalClass } from "./route.js";

const VECTOR_WEIGHT = 0.7;
const BM25_WEIGHT = 0.3;

/**
 * Per-enhancement feature flags (Task 308). All read from env at the
 * boundary of each `hybrid()` call so flips do not require a process
 * restart. Per-call `HybridParams` fields override the env defaults so
 * tests and A/B harnesses can opt in without touching the environment.
 *
 * Defaults are OFF — unflagged callers see exactly the baseline
 * weighted-sum + nodeId-only-dedup baseline. Flip a flag on at the
 * environment level after the soak window passes.
 */
function envFlag(name: string): boolean {
  const v = process.env[name];
  return v === "1" || v === "true" || v === "yes";
}
function readFlags() {
  return {
    rrf: envFlag("MAXY_GS_RRF"),
    expansion: envFlag("MAXY_GS_EXPANSION"),
    boosts: envFlag("MAXY_GS_BOOSTS"),
    dedup: envFlag("MAXY_GS_DEDUP"),
    route: envFlag("MAXY_GS_ROUTE"),
  };
}

export type { DedupLayer, RetrievalClass };
export { DEFAULT_LAYERS };
/**
 * Name of the fulltext index (Task 416 admin index; the public twin was
 * retired with the public read path). Mirrors the `CREATE FULLTEXT INDEX`
 * declaration in `platform/neo4j/schema.cypher`. The doctrine test asserts
 * the name matches this constant — drift between schema and this constant
 * breaks BM25 silently. `entity_search_admin` carries `n.compiledTruth`.
 */
export const FULLTEXT_INDEX_ADMIN = "entity_search_admin";

/** Detect the public-agent read shape (`allowedScopes` exactly `['public']`).
 *  The BM25 index picker that used this predicate was retired with the public
 *  read path. Its one remaining consumer — the `projectPublicProperties` swap
 *  in the memory MCP's memory-search.ts — is now unreachable: every live read
 *  passes `allowedScopes: undefined`, so this returns false on every call. Kept
 *  as that projection's gate for when a public read surface is restored. */
export function isPublicOnlyScope(
  allowedScopes: readonly string[] | undefined,
): boolean {
  return Array.isArray(allowedScopes) && allowedScopes.length === 1 && allowedScopes[0] === "public";
}

export interface SearchHit {
  nodeId: string;
  labels: string[];
  properties: Record<string, unknown>;
  score: number;
}

export interface SearchResult extends SearchHit {
  /**
   * granular component scores so the /data UI can render
   * `vector N.NN · bm25 N.NN · combined N.NN` per row. `vectorScore` is the
   * raw cosine in [0,1] from the per-label vector index; `bm25Score` is the
   * post-merge normalised value in [0,1] used in the combine (NOT the raw
   * Lucene score, which has no operator-meaningful unit). Either may be 0
   * when only one half of the hybrid hit. In the degraded (bm25-only) path
   * `vectorScore` is always 0 and `bm25Score` is normalised over the
   * local hit set.
   */
  vectorScore: number;
  bm25Score: number;
  related: Array<{
    /**
     * neighbour `elementId`. Required by the /graph canvas to
     * render edges in pipeline-collapse mode (search response IS the canvas
     * data). The MCP memory-search tool ignores it; the field is additive.
     */
    nodeId: string;
    relationship: string;
    direction: string;
    labels: string[];
    properties: Record<string, unknown>;
  }>;
}

export interface ScoredNode {
  nodeId: string;
  labels: string[];
  properties: Record<string, unknown>;
  vectorScore: number;
  bm25Score: number;
  /**
   * true when this node entered the merge map via the BM25 path
   * with a raw Lucene score > 0 (literal-token match against the universal
   * fulltext index), or via the keyword-subscriptions property-lookup path.
   * Set BEFORE bm25 normalisation, so a single-hit set whose normalised
   * value collapses to 0.0 still carries the flag.
   *
   * Read by the vector-threshold filter as the carve-out predicate: rows
   * with weak cosine survive when `bm25Hit` is true (operator's literal
   * tokens override the semantic similarity floor). Without this flag the
   * naive `bm25Score > 0` check would drop every bottom-of-set BM25 row
   * because min-max normalisation pins the lowest score to 0.
   */
  bm25Hit: boolean;
}

/**
 * `"hybrid"` = weighted-sum fusion path (the baseline; the
 * brief calls this `weighted-sum`).
 * `"rrf"` = RRF fusion path (Task 308, flagged via `MAXY_GS_RRF`).
 * `"bm25"` = degraded path after embed failure with `degradeOnEmbedFailure`.
 *
 * UI callers (e.g. `platform/ui/app/graph/page.tsx`) pattern-match on
 * `"bm25"` to render the Ollama-down banner; non-`"bm25"` values collapse
 * to the normal banner, so adding `"rrf"` is wire-compatible.
 */
export type SearchMode = "hybrid" | "rrf" | "bm25";

export interface Bm25OnlyParams {
  query: string;
  accountId?: string;
  limit: number;
  allowedScopes?: string[];
  agentSlug?: string;
  keywords?: string[];
  keywordMatch?: "any" | "all";
  /**
   * gate BM25 hits to nodes carrying at least one of these labels.
   * Mirrors hybrid()'s vector-half label filter so the Ollama-down fallback
   * honours the same gate the operator applied via /graph chips. Empty array
   * is treated as "no gate" (matches hybrid's `labels && labels.length > 0`
   * guard); admin route enforces non-empty `labels` as a precondition.
   */
  labels?: string[];
  /**
   * Task 485 — per-visitor slice ringfence. When set, the scope filter
   * widens from `(scope IS NULL OR scope IN $allowedScopes)` to
   * `(sliceToken = $sliceToken OR scope IS NULL OR scope IN $allowedScopes)`
   * so a gated public-agent read sees its visitor's user-scoped nodes in
   * addition to public-scope rows. Skipped on every admin and open-mode
   * public call; the env var `ACCESS_SLICE_TOKEN` populates it on the
   * memory MCP server side.
   */
  sliceToken?: string;
}

/**
 * Build the scope-and-slice WHERE fragment + params. Centralises the three
 * cypher composition sites (bm25, vector, prop) so the gated/non-gated
 * behaviour does not drift between them. Returns empty strings when no
 * filter applies. Exported so deterministic-read tools outside the ranking
 * stack (e.g. memory-lookup-by-name) reuse the exact same scope predicate
 * rather than re-deriving it — drift here is an account-isolation gap.
 */
export function buildScopeAndSliceClause(
  allowedScopes: readonly string[] | undefined,
  sliceToken: string | undefined,
  alias: string = "node",
): { clause: string; params: Record<string, unknown> } {
  const hasSlice = typeof sliceToken === "string" && sliceToken.length > 0;
  const hasScopes = Array.isArray(allowedScopes) && allowedScopes.length > 0;
  if (!hasSlice && !hasScopes) return { clause: "", params: {} };
  if (hasSlice && hasScopes) {
    return {
      clause: `AND (${alias}.sliceToken = $sliceToken OR ${alias}.scope IS NULL OR ${alias}.scope IN $allowedScopes)`,
      params: { sliceToken, allowedScopes },
    };
  }
  if (hasSlice) {
    return {
      clause: `AND (${alias}.sliceToken = $sliceToken OR ${alias}.scope IS NULL)`,
      params: { sliceToken },
    };
  }
  return {
    clause: `AND (${alias}.scope IS NULL OR ${alias}.scope IN $allowedScopes)`,
    params: { allowedScopes },
  };
}

export interface HybridParams extends Bm25OnlyParams {
  expandHops?: number;
  keywordSubscriptions?: string[];
  /**
   * When true, a failing `embed()` does NOT throw — the lib falls back to
   * `bm25Only()` and returns `mode: "bm25"`. Admin-route callers set this
   * true so Ollama-down degrades to BM25-only; MCP tool callers leave it
   * false so embed failure surfaces loudly.
   */
  degradeOnEmbedFailure?: boolean;
  /**
   * minimum raw vector cosine in [0,1] required for a row to
   * pass the merge step. Rows below the threshold are dropped UNLESS they
   * also entered via the BM25 path (`bm25Hit === true`) — that carve-out
   * preserves literal-token matches whose embedding happened to score
   * mediocre. Counts surface in HybridResponse so callers can render an
   * "N suppressed — show all" affordance.
   *
   * Undefined ⇒ no threshold. The lib itself takes no opinion on the value;
   * every search *surface* opts into the shared `DEFAULT_VECTOR_THRESHOLD`
   * (exported below) as its default, so the same query yields the same node
   * set across chat memory-search, /graph and /data (Task 635). A caller that
   * genuinely wants no floor — the graph-subgraph pivot-narrowing mask — just
   * passes nothing.
   * Zero ⇒ explicit "show all" override (the operator's "show all" click on
   * /data and /graph).
   *
   * Calibration: see `DEFAULT_VECTOR_THRESHOLD` below.
   */
  vectorThreshold?: number;

  // --- Task 308 enhancements (all optional, all default off) ---

  /**
   * Routing hint from the inbound-gateway classifier (Task 304). When
   * the `MAXY_GS_ROUTE` flag is on, the lib picks a label filter and
   * fusion-weight plan per class. Omitted ⇒ `"general"` (unflagged
   * callers see no change).
   */
  retrievalClass?: RetrievalClass;
  /** Per-call override: enable Reciprocal Rank Fusion. */
  enableRrf?: boolean;
  /** Per-call override: enable Haiku multi-query expansion. */
  enableExpansion?: boolean;
  /** Per-call override: enable compiledTruth + backlink boosts. */
  enableBoosts?: boolean;
  /** Per-call override: enable 4-layer dedup (default layers below). */
  enableDedup?: boolean;
  /** Per-call override: enable retrieval-class routing. */
  enableRoute?: boolean;
  /**
   * Layers passed to the 4-layer dedup when `enableDedup` is true.
   * Defaults to `["nodeId","slug","canonicalName","contentHash"]`.
   */
  dedupLayers?: DedupLayer[];
  /** RRF smoothing constant. Default 60 (gbrain default). */
  rrfK?: number;
}

/**
 * Shared default vector-cosine floor applied by every search *surface* — chat
 * memory-search, /graph and /data — so the same query yields the same node set
 * across all of them (Task 635). `hybrid()` keeps the `vectorThreshold ===
 * undefined ⇒ no floor` contract; surfaces opt in by passing this constant as
 * their default. Callers that genuinely want no floor (the graph-subgraph
 * pivot-narrowing mask) pass nothing.
 *
 * Calibration (Real Agent laptop, project_embedding index, account b3b638d9):
 * the `brochure` query splits 13 Project nodes into 6 keepers in
 * [0.8473, 0.8623] and 7 noise rows in [0.7159, 0.8143] — a 0.033-wide gap at
 * ~0.83 with no row inside it. 0.82 sits in that gap: drops all 7 noise rows,
 * keeps all 6 keepers. BM25-positive rows bypass the floor regardless (the
 * `bm25Hit` carve-out above), so proper-noun matches survive a tight floor.
 * Recalibrate by picking a value in the gap between the lowest in-scope keeper
 * and the highest noise row.
 */
export const DEFAULT_VECTOR_THRESHOLD = 0.82;

export interface HybridResponse {
  mode: SearchMode;
  results: SearchResult[];
  /** Populated when degradeOnEmbedFailure fired. Caller logs it. */
  embedError?: string;
  /**
   * milliseconds spent on the batched expand round-trip alone
   * (separate from embed + vector + BM25). The /graph admin route emits
   * this as `expand-ms=N` so a regression on the post-Task-747 batching
   * surfaces in server.log without needing a profiler. Zero when
   * `expandHops === 0` or no merged results.
   */
  expandMs: number;
  /**
   * operator-facing threshold accounting.
   *
   * `rawMerged`     count of nodes in the merge map BEFORE the threshold
   *                 filter ran (vector + BM25 + keyword-subscriptions
   *                 union, deduped by nodeId).
   * `suppressed`    count of nodes the threshold filter dropped.
   *                 Always 0 when `vectorThreshold` is undefined.
   * `bm25Bypass`    count of nodes that would have been dropped by the
   *                 vector-cosine floor BUT survived because `bm25Hit`
   *                 was true. Subset of the rendered set.
   * `threshold`     the actual threshold applied (echoes the param);
   *                 null when no filter ran. Drives the "show all"
   *                 affordance: if null, no banner; if non-null and
   *                 `suppressed > 0`, banner with count.
   */
  rawMerged: number;
  suppressed: number;
  bm25Bypass: number;
  threshold: number | null;
  /**
   * Per-stage counts surfaced for the caller's `[graph-search:hybrid]`
   * log line (Task 308). Always present; values reflect whichever
   * enhancements ran on this call. `expansions` is 1 when expansion
   * was off, `boosted - fused` is 0 when boosts were no-ops, etc.
   */
  stageCounts: {
    expansions: number;
    vector: number;
    bm25: number;
    fused: number;
    boosted: number;
    deduped: number;
    final: number;
  };
}

export type EmbedFn = (text: string) => Promise<number[]>;

/**
 * Lucene-escape special characters per Neo4j's fulltext query grammar.
 * `/[+\-&|!(){}[\]^"~*?:\\/]/g` — matches memory-search.ts:127 and
 * neo4j-store.ts:2521 verbatim; consolidation preserves escape set.
 */
export function escapeLucene(query: string): string {
  return query.replace(/[+\-&|!(){}[\]^"~*?:\\/]/g, "\\$&");
}

/**
 * Normalise BM25 scores to [0, 1] using min-max within the result set.
 * If all scores are equal (or single result), returns 1.0 for all —
 * the range-zero branch prevents divide-by-zero and keeps the combined
 * score meaningful.
 */
export function normaliseBm25Scores(scores: number[]): number[] {
  if (scores.length === 0) return [];
  const min = Math.min(...scores);
  const max = Math.max(...scores);
  const range = max - min;
  if (range === 0) return scores.map(() => 1.0);
  return scores.map((s) => (s - min) / range);
}

// Module-scope vector-index cache — discovered from Neo4j at first query.
// Per-process singleton matches memory-search.ts behaviour.
let indexCache: Map<string, string> | null = null;

export async function discoverIndexes(session: Session): Promise<Map<string, string>> {
  if (indexCache) return indexCache;
  const result = await session.run(
    `SHOW INDEXES YIELD name, labelsOrTypes, type WHERE type = 'VECTOR' RETURN name, labelsOrTypes`
  );
  const fresh = new Map<string, string>();
  for (const record of result.records) {
    const name = record.get("name") as string;
    const labels = record.get("labelsOrTypes") as string[];
    for (const label of labels) fresh.set(label, name);
  }
  indexCache = fresh;
  return fresh;
}

/** Clear the index cache — call after schema changes. */
export function clearIndexCache(): void {
  indexCache = null;
}

interface KeywordFilter {
  clause: string;
  params: Record<string, unknown>;
}

function buildKeywordFilter(
  keywords: string[] | undefined,
  keywordMatch: "any" | "all" = "any",
): KeywordFilter | undefined {
  if (!keywords || keywords.length === 0) return undefined;
  const fn = keywordMatch === "all" ? "ALL" : "ANY";
  return {
    clause: `AND (node.keywords IS NULL OR ${fn}(kw IN $keywords WHERE kw IN node.keywords))`,
    params: { keywords: keywords.map((k) => k.toLowerCase().trim()) },
  };
}

/**
 * BM25 full-text search against the universal `entity_search` index.
 * Returns [] when the index doesn't exist — matches memory-search.ts
 * graceful-fallback semantics so a fresh account or a Pi whose index is
 * in the POPULATING window doesn't 500 the caller.
 */
export async function bm25Only(
  session: Session,
  params: Bm25OnlyParams,
): Promise<SearchHit[]> {
  const { query, accountId, limit, allowedScopes, agentSlug, keywords, keywordMatch, labels, sliceToken } = params;
  const scope = buildScopeAndSliceClause(allowedScopes, sliceToken);
  const scopeClause = scope.clause;
  const agentClause = agentSlug
    ? "AND node.agents IS NOT NULL AND $agentSlug IN node.agents"
    : "";
  const labelClause = labels && labels.length > 0
    ? "AND any(l IN labels(node) WHERE l IN $labels)"
    : "";
  const keywordFilter = buildKeywordFilter(keywords, keywordMatch);
  const kwClause = keywordFilter?.clause ?? "";
  const escaped = escapeLucene(query);

  try {
    const result = await session.run(
      `CALL db.index.fulltext.queryNodes($indexName, $query)
       YIELD node, score
       WHERE node.accountId = $accountId
       ${scopeClause}
       ${agentClause}
       ${labelClause}
       AND ${notTrashed("node")}
       ${kwClause}
       RETURN node, score, labels(node) AS nodeLabels, elementId(node) AS nodeId
       ORDER BY score DESC
       LIMIT $limit`,
      {
        indexName: FULLTEXT_INDEX_ADMIN,
        query: escaped,
        accountId,
        limit: int(limit),
        ...scope.params,
        ...(agentSlug ? { agentSlug } : {}),
        ...(labels && labels.length > 0 ? { labels } : {}),
        ...(keywordFilter?.params ?? {}),
      },
    );
    return result.records.map((r) => {
      const scoreRaw = r.get("score");
      const score = typeof scoreRaw === "number" ? scoreRaw : Number(scoreRaw);
      const node = r.get("node") as { properties: Record<string, unknown> };
      return {
        nodeId: r.get("nodeId") as string,
        labels: r.get("nodeLabels") as string[],
        properties: plainProperties(node.properties),
        score,
      };
    });
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    if (msg.includes("index") || msg.includes("fulltext") || msg.includes("not found")) {
      return [];
    }
    throw err;
  }
}

/**
 * Hybrid vector + BM25 search with graph expansion.
 *
 * Sequence:
 *  1. Resolve flags + route plan. Route may short-circuit (`retrievalClass=none`).
 *  2. Multi-query expansion (Task 308 flag): Haiku produces 3-5 paraphrases.
 *     Each paraphrase is embedded and run through vector+BM25 in parallel.
 *  3. embed(query) — may throw if Ollama is unreachable. When
 *     `degradeOnEmbedFailure=true`, caller receives bm25Only() result
 *     with `mode: "bm25"`; otherwise the throw propagates.
 *  4. Vector search per label (one query per vector index discovered at
 *     boot). Nodes-by-label filter short-circuits when the requested
 *     labels have no index.
 *  5. BM25 search on the universal `entity_search` index —
 *     same filter semantics as vector half (scope/agent/keyword/trashed).
 *  6. Keyword subscriptions (when set): BM25 per keyword + property
 *     lookup against `node.keywords` array. Both bypass the agentSlug
 *     filter — subscriptions are scope-inclusive by design (matches
 *     memory-search.ts comment L319). Runs once, not per expansion.
 *  7. Fusion: weighted-sum (`vw * vector + bw * bm25_norm`) by default,
 *     RRF when `MAXY_GS_RRF` is on. Route plan supplies fusion weights.
 *  8. Boosts (Task 308 flag): compiledTruth bump + backlink log-bump.
 *  9. Dedup (Task 308 flag): 4-layer dedup over the union, default layers
 *     `[nodeId, slug, canonicalName, contentHash]`.
 * 10. Threshold + sort + slice. Counts (rawMerged / suppressed / bm25Bypass)
 *     reflect the pre-threshold and post-threshold populations.
 * 11. Graph expand (1 hop by default, 0 to skip) — notTrashed filter +
 *     scope + agent clauses mirrored.
 */
export async function hybrid(
  session: Session,
  embed: EmbedFn,
  params: HybridParams,
): Promise<HybridResponse> {
  const {
    query,
    labels,
    accountId,
    limit,
    allowedScopes,
    sliceToken,
    keywords,
    keywordMatch = "any",
    agentSlug,
    keywordSubscriptions,
    expandHops = 1,
    degradeOnEmbedFailure = false,
    retrievalClass,
    dedupLayers = DEFAULT_LAYERS,
    rrfK = 60,
  } = params;

  const envFlags = readFlags();
  const useRoute = params.enableRoute ?? envFlags.route;
  const useRrf = params.enableRrf ?? envFlags.rrf;
  const useExpansion = params.enableExpansion ?? envFlags.expansion;
  const useBoosts = params.enableBoosts ?? envFlags.boosts;
  const useDedup = params.enableDedup ?? envFlags.dedup;

  // --- Route plan (Task 308): pick label filter + fusion weights from class ---
  const plan = useRoute
    ? pickIndexMix(retrievalClass)
    : { skip: false, vectorWeight: VECTOR_WEIGHT, bm25Weight: BM25_WEIGHT, labelFilter: undefined as string[] | undefined };

  const emptyStageCounts = {
    expansions: 1,
    vector: 0,
    bm25: 0,
    fused: 0,
    boosted: 0,
    deduped: 0,
    final: 0,
  };

  if (plan.skip) {
    return {
      mode: useRrf ? "rrf" : "hybrid",
      results: [],
      expandMs: 0,
      rawMerged: 0,
      suppressed: 0,
      bm25Bypass: 0,
      threshold: params.vectorThreshold ?? null,
      stageCounts: emptyStageCounts,
    };
  }

  // Per-call labels win over route-supplied label filter — operator-set
  // filters are intentional; the route plan is a soft hint.
  const effectiveLabels = (labels && labels.length > 0)
    ? labels
    : plan.labelFilter;

  // --- Degraded path: embed fails, fall back to BM25-only ---
  // Run a single embed up front so degradation is detected before any
  // expansion / multi-query work spins up.
  let queryEmbedding: number[];
  try {
    queryEmbedding = await embed(query);
  } catch (err) {
    if (!degradeOnEmbedFailure) throw err;
    const msg = err instanceof Error ? err.message : String(err);
    const bm25Hits = await bm25Only(session, { ...params, labels: effectiveLabels });
    const normalised = normaliseBm25Scores(bm25Hits.map((h) => h.score));
    const results: SearchResult[] = bm25Hits.map((h, i) => ({
      ...h,
      vectorScore: 0,
      bm25Score: normalised[i] ?? 0,
      related: [],
    }));
    return {
      mode: "bm25",
      results,
      embedError: msg,
      expandMs: 0,
      rawMerged: results.length,
      suppressed: 0,
      bm25Bypass: 0,
      threshold: null,
      stageCounts: {
        expansions: 1,
        vector: 0,
        bm25: results.length,
        fused: results.length,
        boosted: results.length,
        deduped: results.length,
        final: results.length,
      },
    };
  }

  const labelToIndex = await discoverIndexes(session);

  // --- Resolve vector indexes from the effective label set ---
  //
  // Task 478: when `effectiveLabels` is set but none resolve to a vector index
  // (e.g. `labels:["Organization"]` — universal BM25 covers it, no vector
  // index declared), proceed with `indexesToQuery=[]`. The vector loop below
  // runs zero iterations and the BM25 half still runs with the label filter,
  // so search returns rows whose existence the rest of the platform asserts.
  // Pre-fix this branch returned an empty result, hiding ingest-created
  // Organization rows that BM25 would have found — the deadlock recorded in
  // .tasks/archive/478-*.md. The labels-without-vector list is emitted as an
  // observability line so a future analogous gap stays visible.
  let indexesToQuery: string[];
  if (effectiveLabels && effectiveLabels.length > 0) {
    indexesToQuery = effectiveLabels
      .map((l) => labelToIndex.get(l))
      .filter((idx): idx is string => idx !== undefined);
    if (indexesToQuery.length === 0) {
      const missing = effectiveLabels.filter((l) => !labelToIndex.has(l));
      process.stderr.write(
        `[graph-search] labels-without-vector-index=${missing.join(",")} ` +
          `accountId=${(accountId ?? "").slice(0, 8)} — falling through to BM25-only for these labels\n`,
      );
    }
  } else {
    indexesToQuery = [...new Set(labelToIndex.values())];
  }

  const hybridScope = buildScopeAndSliceClause(allowedScopes, sliceToken);
  const scopeClause = hybridScope.clause;
  const scopeParams = hybridScope.params;
  const agentClause = agentSlug
    ? "AND node.agents IS NOT NULL AND $agentSlug IN node.agents"
    : "";
  const agentParams = agentSlug ? { agentSlug } : {};

  const keywordFilter = buildKeywordFilter(keywords, keywordMatch);
  const keywordClause = keywordFilter?.clause ?? "";
  const keywordParams = keywordFilter?.params ?? {};

  // --- Query expansion (Task 308 / removed in Task 424) ---
  // Server-side Haiku query paraphrasing was removed in Task 424. Callers
  // (the calling agent) construct expanded query terms themselves before
  // invoking memory-search. The `useExpansion` flag is now a no-op; the
  // single original query is the only pass.
  void useExpansion;
  const expansions = [query];

  type Pass = {
    queryText: string;
    embedding: number[] | null;
    vectorRanked: ScoredNode[];
    bm25Ranked: ScoredNode[];
  };

  const passes: Pass[] = await Promise.all(
    expansions.map(async (qText, idx) => {
      let pEmbed: number[] | null;
      if (idx === 0) {
        pEmbed = queryEmbedding;
      } else {
        try {
          pEmbed = await embed(qText);
        } catch {
          pEmbed = null;
        }
      }

      const vectorRanked: ScoredNode[] = [];
      const vectorSeen = new Map<string, number>();
      if (pEmbed) {
        for (const indexName of indexesToQuery) {
          const vectorResult = await session.run(
            `CALL db.index.vector.queryNodes($indexName, $limit, $embedding)
             YIELD node, score
             WHERE node.accountId = $accountId
             ${scopeClause}
             ${agentClause}
             AND ${notTrashed("node")}
             ${keywordClause}
             RETURN node, score, labels(node) AS nodeLabels, elementId(node) AS nodeId
             ORDER BY score DESC
             LIMIT $limit`,
            {
              indexName,
              embedding: pEmbed,
              limit: int(limit),
              accountId,
              ...scopeParams,
              ...agentParams,
              ...keywordParams,
            },
          );
          for (const record of vectorResult.records) {
            const nodeId = record.get("nodeId") as string;
            const scoreRaw = record.get("score");
            const score = typeof scoreRaw === "number" ? scoreRaw : Number(scoreRaw);
            const existingIdx = vectorSeen.get(nodeId);
            if (existingIdx !== undefined) {
              const existing = vectorRanked[existingIdx];
              if (score > existing.vectorScore) existing.vectorScore = score;
              continue;
            }
            const node = record.get("node") as { properties: Record<string, unknown> };
            vectorSeen.set(nodeId, vectorRanked.length);
            vectorRanked.push({
              nodeId,
              labels: record.get("nodeLabels") as string[],
              properties: plainProperties(node.properties),
              vectorScore: score,
              bm25Score: 0,
              bm25Hit: false,
            });
          }
        }
        vectorRanked.sort((a, b) => b.vectorScore - a.vectorScore);
      }

      const bm25Hits = await bm25Only(session, {
        ...params,
        query: qText,
        labels: effectiveLabels,
      });
      const bm25Norm = normaliseBm25Scores(bm25Hits.map((h) => h.score));
      const bm25Ranked: ScoredNode[] = bm25Hits.map((h, i) => ({
        nodeId: h.nodeId,
        labels: h.labels,
        properties: h.properties,
        vectorScore: 0,
        bm25Score: bm25Norm[i] ?? 0,
        bm25Hit: true,
      }));

      return { queryText: qText, embedding: pEmbed, vectorRanked, bm25Ranked };
    }),
  );

  // --- Fuse passes into the score map ---
  // Weighted-sum: union per nodeId, taking the max vector score and max
  // bm25-normalised score across passes, then combine with the route's
  // weights. This preserves the baseline single-query behaviour exactly
  // when expansions.length === 1.
  // RRF: each pass contributes one vector-ranked list and one bm25-ranked
  // list. Fusion is computed across the full union of ranked lists.
  const scoreMap = new Map<string, ScoredNode>();
  let totalVector = 0;
  let totalBm25 = 0;

  for (const pass of passes) {
    totalVector += pass.vectorRanked.length;
    totalBm25 += pass.bm25Ranked.length;
    for (const v of pass.vectorRanked) {
      const existing = scoreMap.get(v.nodeId);
      if (existing) {
        existing.vectorScore = Math.max(existing.vectorScore, v.vectorScore);
      } else {
        scoreMap.set(v.nodeId, { ...v });
      }
    }
    for (const b of pass.bm25Ranked) {
      mergeBm25Hit(scoreMap, {
        nodeId: b.nodeId,
        labels: b.labels,
        properties: b.properties,
        score: b.bm25Score,
      }, b.bm25Score);
    }
  }

  // --- Keyword subscriptions (scope-inclusive, agentSlug-bypassing) ---
  // Runs once regardless of expansion count — subscriptions are keyed by
  // operator config, not by the inbound query.
  if (keywordSubscriptions && keywordSubscriptions.length > 0) {
    for (const kw of keywordSubscriptions) {
      const kwHits = await bm25Only(session, {
        query: kw,
        accountId,
        limit,
        allowedScopes,
        sliceToken,
      });
      if (kwHits.length === 0) continue;
      const rawScores = kwHits.map((h) => h.score);
      const normalised = normaliseBm25Scores(rawScores);
      for (let i = 0; i < kwHits.length; i++) {
        mergeBm25Hit(scoreMap, kwHits[i], normalised[i]);
      }
    }

    const propScope = buildScopeAndSliceClause(allowedScopes, sliceToken);
    const propResult = await session.run(
      `MATCH (node)
       WHERE node.accountId = $accountId
       AND ${notTrashed("node")}
       AND node.keywords IS NOT NULL
       AND ANY(kw IN $kwSubs WHERE ANY(nk IN node.keywords WHERE toLower(nk) = kw))
       ${propScope.clause}
       RETURN node, labels(node) AS nodeLabels, elementId(node) AS nodeId
       LIMIT $limit`,
      {
        accountId,
        kwSubs: keywordSubscriptions,
        limit: int(limit),
        ...propScope.params,
      },
    );
    for (const record of propResult.records) {
      const nodeId = record.get("nodeId") as string;
      const existing = scoreMap.get(nodeId);
      if (existing) {
        existing.bm25Score = Math.max(existing.bm25Score, 1.0);
        existing.bm25Hit = true;
      } else {
        const node = record.get("node") as { properties: Record<string, unknown> };
        scoreMap.set(nodeId, {
          nodeId,
          labels: record.get("nodeLabels") as string[],
          properties: plainProperties(node.properties),
          vectorScore: 0,
          bm25Score: 1.0,
          bm25Hit: true,
        });
      }
    }
  }

  // --- Keyword subscriptions (scope-inclusive, agentSlug-bypassing) ---
  if (keywordSubscriptions && keywordSubscriptions.length > 0) {
    for (const kw of keywordSubscriptions) {
      const kwHits = await bm25Only(session, {
        query: kw,
        accountId,
        limit,
        allowedScopes,
        sliceToken,
      });
      if (kwHits.length === 0) continue;
      const rawScores = kwHits.map((h) => h.score);
      const normalised = normaliseBm25Scores(rawScores);
      for (let i = 0; i < kwHits.length; i++) {
        mergeBm25Hit(scoreMap, kwHits[i], normalised[i]);
      }
    }

    const propScope = buildScopeAndSliceClause(allowedScopes, sliceToken);
    const propResult = await session.run(
      `MATCH (node)
       WHERE node.accountId = $accountId
       AND ${notTrashed("node")}
       AND node.keywords IS NOT NULL
       AND ANY(kw IN $kwSubs WHERE ANY(nk IN node.keywords WHERE toLower(nk) = kw))
       ${propScope.clause}
       RETURN node, labels(node) AS nodeLabels, elementId(node) AS nodeId
       LIMIT $limit`,
      {
        accountId,
        kwSubs: keywordSubscriptions,
        limit: int(limit),
        ...propScope.params,
      },
    );
    for (const record of propResult.records) {
      const nodeId = record.get("nodeId") as string;
      const existing = scoreMap.get(nodeId);
      if (existing) {
        existing.bm25Score = Math.max(existing.bm25Score, 1.0);
        existing.bm25Hit = true;
      } else {
        const node = record.get("node") as { properties: Record<string, unknown> };
        scoreMap.set(nodeId, {
          nodeId,
          labels: record.get("nodeLabels") as string[],
          properties: plainProperties(node.properties),
          vectorScore: 0,
          bm25Score: 1.0,
          bm25Hit: true,
        });
      }
    }
  }

  // --- Fusion ---
  //
  // Weighted-sum (default): combined = vw * vector + bw * bm25_norm, with
  // weights from the route plan (or the 0.7/0.3 lib defaults when routing
  // is off). This preserves the baseline single-query single-fusion path
  // when no flags are flipped on.
  //
  // RRF (flagged): contributions from each pass's vector and BM25 ranked
  // lists are summed across all passes; combinedScore becomes the RRF
  // score. Per-half components on `ScoredNode` (vectorScore / bm25Score)
  // are retained so the /data UI's score-line renders consistently.
  let allMerged: Array<ScoredNode & { combinedScore: number }>;
  if (useRrf) {
    const rankedLists: ScoredNode[][] = [];
    for (const pass of passes) {
      if (pass.vectorRanked.length > 0) rankedLists.push(pass.vectorRanked);
      if (pass.bm25Ranked.length > 0) rankedLists.push(pass.bm25Ranked);
    }
    const fused = fuseRrf(rankedLists, rrfK);
    allMerged = fused.map((f) => {
      // Use the score-map representative so vectorScore / bm25Score reflect
      // the cross-pass max, not just whichever list ranked first. The
      // representative is guaranteed to exist (scoreMap aggregated the
      // union); fall back to the fusion entry if not (defensive).
      const rep = scoreMap.get(f.nodeId) ?? f;
      return { ...rep, combinedScore: f.rrfScore };
    });
  } else {
    allMerged = [...scoreMap.values()].map((node) => ({
      ...node,
      combinedScore: plan.vectorWeight * node.vectorScore + plan.bm25Weight * node.bm25Score,
    }));
  }
  const fusedCount = allMerged.length;

  // --- Boosts (Task 308) ---
  // Both are no-ops on the current corpus until Task 305 (typed-edge
  // backlinks) and Task 306 (compiledTruth property) land. Reading a
  // missing property as zero is the correct behaviour — the boost is
  // additive over hits that have the field.
  if (useBoosts) {
    allMerged = applyCompiledTruthBoost(
      applyBacklinkBoost(allMerged.map((m) => ({
        ...m,
        score: m.combinedScore,
      }))),
    ).map((m) => ({ ...m, combinedScore: m.score }));
  }
  const boostedCount = allMerged.length;

  // --- Dedup (Task 308) ---
  // 4-layer dedup is a strict superset of the baseline nodeId-only dedup.
  // The post-fusion scoreMap is already nodeId-deduped, so when no flag
  // is set this step is a no-op.
  if (useDedup) {
    allMerged = dedupHits(
      allMerged.map((m) => ({ ...m, score: m.combinedScore })),
      dedupLayers,
    ).map((m) => ({ ...m, combinedScore: m.score }));
  }
  const dedupedCount = allMerged.length;

  // --- Threshold + sort + slice ---
  //
  // vectorThreshold filters BEFORE sort+slice. Order matters:
  // slicing first would consume the slot budget with rows the threshold
  // would have suppressed, under-filling the rendered set. Sorting before
  // filtering would waste cycles ordering rows we are about to drop. So:
  // filter -> sort -> slice. Counts (rawMerged / suppressed / bm25Bypass)
  // are computed against the pre-filter and post-filter populations so the
  // route can render "N suppressed — show all" without a second query.
  //
  // Carve-out: a row with vector cosine below the threshold survives iff
  // bm25Hit is true. The flag (not bm25Score) is the gate — min-max
  // normalisation pins the lowest BM25 row's score to 0.0, so checking
  // bm25Score > 0 would silently drop a literal-token match that just
  // happens to be the worst-scored BM25 hit in the set.
  const rawMerged = allMerged.length;
  const threshold = params.vectorThreshold;
  let kept = allMerged;
  let bm25Bypass = 0;
  if (threshold !== undefined) {
    kept = [];
    for (const node of allMerged) {
      if (node.vectorScore >= threshold) {
        kept.push(node);
      } else if (node.bm25Hit) {
        kept.push(node);
        bm25Bypass++;
      }
    }
  }
  const suppressed = rawMerged - kept.length;
  const merged = kept
    .sort((a, b) => b.combinedScore - a.combinedScore)
    .slice(0, limit);

  // --- Graph expand (single batched round-trip) ---
  //
  // Pre-Task-747: one Cypher per merged node, in a JS for-loop. At the admin
  // route's slider=2000 this produced 2000 driver round-trips per search.
  //
  // Post-Task-747: one UNWIND-driven query for all merged nodeIds. The
  // `WITH nid, collect({...})[0..20]` clause preserves the per-result cap of
  // 20 neighbours that the per-node query enforced via `LIMIT 20`. Slice
  // notation is order-preserving over the rows it consumes (no upstream
  // ORDER BY changes that), so canvas-edge density per hit is unchanged.
  //
  // Each `related` entry now carries `relatedNodeId` (the neighbour's
  // elementId) so the /graph canvas can render edges in pipeline-collapse
  // mode (search response IS the canvas data when `q` is set).
  const results: SearchResult[] = merged.map((node) => ({
    nodeId: node.nodeId,
    labels: node.labels,
    properties: node.properties,
    score: node.combinedScore,
    // carry the per-half components so the /data UI can render
    // a per-row breakdown. These are already computed on `ScoredNode` from
    // the vector half (raw cosine) and the bm25 half (post-merge normalised
    // value, NOT raw Lucene). Memory MCP ignores both; admin route forwards
    // them through SearchResult to the wire.
    vectorScore: node.vectorScore,
    bm25Score: node.bm25Score,
    related: [],
  }));

  let expandMs = 0;
  if (expandHops > 0 && results.length > 0) {
    const expandScopeClause = allowedScopes
      ? "AND (related.scope IS NULL OR related.scope IN $allowedScopes)"
      : "";
    const expandAgentClause = agentSlug
      ? "AND (related.agents IS NULL OR $agentSlug IN related.agents)"
      : "";
    const expandStart = Date.now();
    const expandResult = await session.run(
      `UNWIND $nodeIds AS nid
       MATCH (n)-[r]-(related)
       WHERE elementId(n) = nid
       AND ${notTrashed("related")}
       ${expandScopeClause}
       ${expandAgentClause}
       WITH nid, n, r, related
       WITH nid, collect({
         relType: type(r),
         direction: CASE WHEN startNode(r) = n THEN 'outgoing' ELSE 'incoming' END,
         relatedNodeId: elementId(related),
         relatedLabels: labels(related),
         related: related
       })[0..20] AS items
       RETURN nid, items`,
      { nodeIds: results.map((r) => r.nodeId), ...scopeParams, ...agentParams },
    );
    expandMs = Date.now() - expandStart;
    const byNodeId = new Map<string, SearchResult>(results.map((r) => [r.nodeId, r]));
    for (const rec of expandResult.records) {
      const nid = rec.get("nid") as string;
      const target = byNodeId.get(nid);
      if (!target) continue;
      const items = rec.get("items") as Array<{
        relType: string;
        direction: string;
        relatedNodeId: string;
        relatedLabels: string[];
        related: { properties: Record<string, unknown> };
      }>;
      for (const item of items) {
        target.related.push({
          nodeId: item.relatedNodeId,
          relationship: item.relType,
          direction: item.direction,
          labels: item.relatedLabels,
          properties: plainProperties(item.related.properties),
        });
      }
    }
  }

  return {
    mode: useRrf ? "rrf" : "hybrid",
    results,
    expandMs,
    rawMerged,
    suppressed,
    bm25Bypass,
    threshold: threshold ?? null,
    stageCounts: {
      expansions: expansions.length,
      vector: totalVector,
      bm25: totalBm25,
      fused: fusedCount,
      boosted: boostedCount,
      deduped: dedupedCount,
      final: results.length,
    },
  };
}

function mergeBm25Hit(
  map: Map<string, ScoredNode>,
  hit: SearchHit,
  normalisedScore: number,
): void {
  const existing = map.get(hit.nodeId);
  if (existing) {
    existing.bm25Score = Math.max(existing.bm25Score, normalisedScore);
    // set bm25Hit irrespective of normalised value. The hit
    // came from db.index.fulltext.queryNodes (raw Lucene score > 0 by
    // construction); min-max normalisation pinning the lowest row to
    // 0.0 is a presentation artefact, not a "no match" signal.
    existing.bm25Hit = true;
  } else {
    map.set(hit.nodeId, {
      nodeId: hit.nodeId,
      labels: hit.labels,
      properties: hit.properties,
      vectorScore: 0,
      bm25Score: normalisedScore,
      bm25Hit: true,
    });
  }
}

/** Strip `embedding` and unwrap Neo4j Integers into JS numbers. */
function plainProperties(properties: Record<string, unknown>): Record<string, unknown> {
  const plain: Record<string, unknown> = {};
  for (const [key, value] of Object.entries(properties)) {
    if (key === "embedding") continue;
    if (value && typeof value === "object" && "toNumber" in value) {
      plain[key] = (value as { toNumber(): number }).toNumber();
    } else {
      plain[key] = value;
    }
  }
  return plain;
}