diff --git a/.github/workflows/claude.yaml b/.github/workflows/claude.yaml
new file mode 100644
index 00000000..732de78c
--- /dev/null
+++ b/.github/workflows/claude.yaml
@@ -0,0 +1,68 @@
+name: Claude Code
+
+on:
+  issue_comment:
+    types: [created]
+  pull_request_review_comment:
+    types: [created]
+  issues:
+    types: [opened]
+  pull_request_review:
+    types: [submitted]
+  pull_request_target:
+    types: [opened, synchronize]
+
+jobs:
+  claude:
+    # This simplified condition is more robust and correctly checks permissions.
+    if: >
+      (contains(github.event.comment.body, '@claude') ||
+      contains(github.event.review.body, '@claude') ||
+      contains(github.event.issue.body, '@claude') ||
+      contains(github.event.pull_request.body, '@claude')) &&
+      (github.event.sender.type == 'User' && (
+        github.event.comment.author_association == 'OWNER' ||
+        github.event.comment.author_association == 'MEMBER' ||
+        github.event.comment.author_association == 'COLLABORATOR'
+      ))
+    runs-on: ubuntu-latest
+    permissions:
+      # CRITICAL: Write permissions are required for the action to push branches and update issues/PRs.
+      contents: write
+      pull-requests: write
+      issues: write
+      id-token: write # Required for OIDC token exchange
+      actions: read  # Required for Claude to read CI results on PRs
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          # This correctly checks out the PR's head commit for pull_request_target events.
+          ref: ${{ github.event.pull_request.head.sha }}
+
+      - name: Create Claude settings file
+        run: |
+          mkdir -p /home/runner/.claude
+          cat > /home/runner/.claude/settings.json << 'EOF'
+          {
+            "env": {
+              "ANTHROPIC_BASE_URL": "https://api.z.ai/api/anthropic",
+              "ANTHROPIC_AUTH_TOKEN": "${{ secrets.CUSTOM_ENDPOINT_API_KEY }}"
+            }
+          }
+          EOF
+
+      - name: Run Claude Code
+        id: claude
+        uses: anthropics/claude-code-action@v1
+        with:
+          # Still need this to satisfy the action's validation
+          anthropic_api_key: ${{ secrets.CUSTOM_ENDPOINT_API_KEY }}
+          
+          # Use the same variable names as your local setup
+          settings: '{"env": {"ANTHROPIC_BASE_URL": "https://api.z.ai/api/anthropic", "ANTHROPIC_AUTH_TOKEN": "${{ secrets.CUSTOM_ENDPOINT_API_KEY }}"}}'
+          
+          track_progress: true
+          claude_args: |
+            --allowedTools "Bash,Edit,Read,Write,Glob,Grep"
diff --git a/ctx-mcp-bridge/src/mcpServer.js b/ctx-mcp-bridge/src/mcpServer.js
index 53cb05b7..7180afff 100644
--- a/ctx-mcp-bridge/src/mcpServer.js
+++ b/ctx-mcp-bridge/src/mcpServer.js
@@ -8,7 +8,13 @@ import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"
 import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
 import { Client } from "@modelcontextprotocol/sdk/client/index.js";
 import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
-import { CallToolRequestSchema, ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js";
+import {
+  CallToolRequestSchema,
+  ListToolsRequestSchema,
+  ListResourcesRequestSchema,
+  ListResourceTemplatesRequestSchema,
+  ReadResourceRequestSchema,
+} from "@modelcontextprotocol/sdk/types.js";
 import { loadAnyAuthEntry, loadAuthEntry, readConfig, saveAuthEntry } from "./authConfig.js";
 import { maybeRemapToolArgs, maybeRemapToolResult } from "./resultPathMapping.js";
 import * as oauthHandler from "./oauthHandler.js";
@@ -58,14 +64,46 @@ function dedupeTools(tools) {
   return out;
 }
 
+function dedupeResources(resources) {
+  const seen = new Set();
+  const out = [];
+  for (const resource of resources) {
+    const uri = resource && typeof resource.uri === "string" ? resource.uri : "";
+    if (!uri || seen.has(uri)) {
+      continue;
+    }
+    seen.add(uri);
+    out.push(resource);
+  }
+  return out;
+}
+
+function dedupeResourceTemplates(templates) {
+  const seen = new Set();
+  const out = [];
+  for (const template of templates) {
+    const uri =
+      template && typeof template.uriTemplate === "string"
+        ? template.uriTemplate
+        : "";
+    if (!uri || seen.has(uri)) {
+      continue;
+    }
+    seen.add(uri);
+    out.push(template);
+  }
+  return out;
+}
+
 async function listMemoryTools(client) {
   if (!client) {
     return [];
   }
   try {
+    const timeoutMs = getBridgeListTimeoutMs();
     const remote = await withTimeout(
       client.listTools(),
-      5000,
+      timeoutMs,
       "memory tools/list",
     );
     return Array.isArray(remote?.tools) ? remote.tools.slice() : [];
@@ -75,6 +113,42 @@ async function listMemoryTools(client) {
   }
 }
 
+async function listResourcesSafe(client, label) {
+  if (!client) {
+    return [];
+  }
+  try {
+    const timeoutMs = getBridgeListTimeoutMs();
+    const remote = await withTimeout(
+      client.listResources(),
+      timeoutMs,
+      `${label} resources/list`,
+    );
+    return Array.isArray(remote?.resources) ? remote.resources.slice() : [];
+  } catch (err) {
+    debugLog(`[ctxce] Error calling ${label} resources/list: ` + String(err));
+    return [];
+  }
+}
+
+async function listResourceTemplatesSafe(client, label) {
+  if (!client) {
+    return [];
+  }
+  try {
+    const timeoutMs = getBridgeListTimeoutMs();
+    const remote = await withTimeout(
+      client.listResourceTemplates(),
+      timeoutMs,
+      `${label} resources/templates/list`,
+    );
+    return Array.isArray(remote?.resourceTemplates) ? remote.resourceTemplates.slice() : [];
+  } catch (err) {
+    debugLog(`[ctxce] Error calling ${label} resources/templates/list: ` + String(err));
+    return [];
+  }
+}
+
 function withTimeout(promise, ms, label) {
   return new Promise((resolve, reject) => {
     let settled = false;
@@ -125,6 +199,25 @@ function getBridgeToolTimeoutMs() {
   }
 }
 
+function getBridgeListTimeoutMs() {
+  try {
+    // Keep list operations on a separate budget from tools/call.
+    // Some streamable-http clients (including Codex) probe tools/resources early,
+    // and a short timeout here can make the bridge appear unavailable.
+    const raw = process.env.CTXCE_LIST_TIMEOUT_MSEC;
+    if (!raw) {
+      return 60000;
+    }
+    const parsed = Number.parseInt(String(raw), 10);
+    if (!Number.isFinite(parsed) || parsed <= 0) {
+      return 60000;
+    }
+    return parsed;
+  } catch {
+    return 60000;
+  }
+}
+
 function selectClientForTool(name, indexerClient, memoryClient) {
   if (!name) {
     return indexerClient;
@@ -651,6 +744,7 @@ async function createBridgeServer(options) {
     {
       capabilities: {
         tools: {},
+        resources: {},
       },
     },
   );
@@ -664,9 +758,10 @@ async function createBridgeServer(options) {
       if (!indexerClient) {
         throw new Error("Indexer MCP client not initialized");
       }
+      const timeoutMs = getBridgeListTimeoutMs();
       remote = await withTimeout(
         indexerClient.listTools(),
-        10000,
+        timeoutMs,
         "indexer tools/list",
       );
     } catch (err) {
@@ -693,6 +788,57 @@ async function createBridgeServer(options) {
     return { tools };
   });
 
+  server.setRequestHandler(ListResourcesRequestSchema, async () => {
+    // Proxy resource discovery/read-through so clients that use MCP resources
+    // (not only tools) can access upstream indexer/memory resources directly.
+    await initializeRemoteClients(false);
+    const indexerResources = await listResourcesSafe(indexerClient, "indexer");
+    const memoryResources = await listResourcesSafe(memoryClient, "memory");
+    const resources = dedupeResources([...indexerResources, ...memoryResources]);
+    debugLog(`[ctxce] resources/list: returning ${resources.length} resources`);
+    return { resources };
+  });
+
+  server.setRequestHandler(ListResourceTemplatesRequestSchema, async () => {
+    await initializeRemoteClients(false);
+    const indexerTemplates = await listResourceTemplatesSafe(indexerClient, "indexer");
+    const memoryTemplates = await listResourceTemplatesSafe(memoryClient, "memory");
+    const resourceTemplates = dedupeResourceTemplates([...indexerTemplates, ...memoryTemplates]);
+    debugLog(`[ctxce] resources/templates/list: returning ${resourceTemplates.length} templates`);
+    return { resourceTemplates };
+  });
+
+  server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
+    await initializeRemoteClients(false);
+    const params = request.params || {};
+    const timeoutMs = getBridgeToolTimeoutMs();
+    const uri =
+      params && typeof params.uri === "string" ? params.uri : "<missing-uri>";
+    debugLog(`[ctxce] resources/read: ${uri}`);
+
+    const tryRead = async (client, label) => {
+      if (!client) {
+        return null;
+      }
+      try {
+        return await client.readResource(params, { timeout: timeoutMs });
+      } catch (err) {
+        debugLog(`[ctxce] resources/read failed on ${label}: ` + String(err));
+        return null;
+      }
+    };
+
+    const indexerResult = await tryRead(indexerClient, "indexer");
+    if (indexerResult) {
+      return indexerResult;
+    }
+    const memoryResult = await tryRead(memoryClient, "memory");
+    if (memoryResult) {
+      return memoryResult;
+    }
+    throw new Error(`Resource ${uri} not available on any configured MCP server`);
+  });
+
   // tools/call → proxied to indexer or memory server
   server.setRequestHandler(CallToolRequestSchema, async (request) => {
     const params = request.params || {};
@@ -843,6 +989,13 @@ export async function runHttpMcpServer(options) {
     typeof options.port === "number"
       ? options.port
       : Number.parseInt(process.env.CTXCE_HTTP_PORT || "30810", 10) || 30810;
+  // TODO(auth): replace this boolean toggle with explicit auth modes (none|required).
+  // In required mode, enforce Bearer auth on /mcp with consistent 401 challenges and
+  // only advertise OAuth metadata/endpoints when authentication is mandatory.
+  // In local/dev mode, leaving OAuth discovery off avoids clients entering an
+  // unnecessary OAuth path for otherwise unauthenticated bridge usage.
+  const oauthEnabled = String(process.env.CTXCE_ENABLE_OAUTH || "").trim().toLowerCase();
+  const oauthEndpointsEnabled = oauthEnabled === "1" || oauthEnabled === "true" || oauthEnabled === "yes";
 
   const transport = new StreamableHTTPServerTransport({
     sessionIdGenerator: undefined,
@@ -865,34 +1018,36 @@ export async function runHttpMcpServer(options) {
       // OAuth 2.0 Endpoints (RFC9728 Protected Resource Metadata + RFC7591)
       // ================================================================
 
-      // OAuth metadata endpoint (RFC9728)
-      if (parsedUrl.pathname === "/.well-known/oauth-authorization-server") {
-        oauthHandler.handleOAuthMetadata(req, res, issuerUrl);
-        return;
-      }
+      if (oauthEndpointsEnabled) {
+        // OAuth metadata endpoint (RFC9728)
+        if (parsedUrl.pathname === "/.well-known/oauth-authorization-server") {
+          oauthHandler.handleOAuthMetadata(req, res, issuerUrl);
+          return;
+        }
 
-      // OAuth Dynamic Client Registration endpoint (RFC7591)
-      if (parsedUrl.pathname === "/oauth/register" && req.method === "POST") {
-        oauthHandler.handleOAuthRegister(req, res);
-        return;
-      }
+        // OAuth Dynamic Client Registration endpoint (RFC7591)
+        if (parsedUrl.pathname === "/oauth/register" && req.method === "POST") {
+          oauthHandler.handleOAuthRegister(req, res);
+          return;
+        }
 
-      // OAuth authorize endpoint
-      if (parsedUrl.pathname === "/oauth/authorize") {
-        oauthHandler.handleOAuthAuthorize(req, res, parsedUrl.searchParams);
-        return;
-      }
+        // OAuth authorize endpoint
+        if (parsedUrl.pathname === "/oauth/authorize") {
+          oauthHandler.handleOAuthAuthorize(req, res, parsedUrl.searchParams);
+          return;
+        }
 
-      // Store session endpoint (helper for login page)
-      if (parsedUrl.pathname === "/oauth/store-session" && req.method === "POST") {
-        oauthHandler.handleOAuthStoreSession(req, res);
-        return;
-      }
+        // Store session endpoint (helper for login page)
+        if (parsedUrl.pathname === "/oauth/store-session" && req.method === "POST") {
+          oauthHandler.handleOAuthStoreSession(req, res);
+          return;
+        }
 
-      // OAuth token endpoint
-      if (parsedUrl.pathname === "/oauth/token" && req.method === "POST") {
-        oauthHandler.handleOAuthToken(req, res);
-        return;
+        // OAuth token endpoint
+        if (parsedUrl.pathname === "/oauth/token" && req.method === "POST") {
+          oauthHandler.handleOAuthToken(req, res);
+          return;
+        }
       }
 
       // ================================================================
@@ -1058,4 +1213,3 @@ function detectRepoName(workspace, config) {
   const leaf = workspace ? path.basename(workspace) : "";
   return leaf && SLUGGED_REPO_RE.test(leaf) ? leaf : null;
 }
-
diff --git a/scripts/collection_admin.py b/scripts/collection_admin.py
index d970e941..149dadf0 100644
--- a/scripts/collection_admin.py
+++ b/scripts/collection_admin.py
@@ -1,12 +1,15 @@
+import logging
 import os
 import json
 import re
 import shutil
 import time
-from pathlib import Path
 from datetime import datetime
+from pathlib import Path
 from typing import Any, Dict, Optional, List
 
+logger = logging.getLogger(__name__)
+
 from scripts.auth_backend import mark_collection_deleted
 
 try:
@@ -193,6 +196,7 @@ def delete_collection_everywhere(
     out: Dict[str, Any] = {
         "collection": name,
         "qdrant_deleted": False,
+        "qdrant_graph_deleted": False,
         "registry_marked_deleted": False,
         "deleted_state_files": 0,
         "deleted_managed_workspaces": 0,
@@ -209,6 +213,14 @@ def delete_collection_everywhere(
                     out["qdrant_deleted"] = True
                 except Exception:
                     out["qdrant_deleted"] = False
+                # Best-effort: also delete companion graph edges collection when present.
+                # This branch stores file-level edges in `<collection>_graph`.
+                if not name.endswith("_graph"):
+                    try:
+                        cli.delete_collection(collection_name=f"{name}_graph")
+                        out["qdrant_graph_deleted"] = True
+                    except Exception:
+                        out["qdrant_graph_deleted"] = False
     except Exception:
         out["qdrant_deleted"] = False
 
@@ -359,8 +371,10 @@ def _manual_copy_points() -> None:
                 vectors_config = None
                 sparse_vectors_config = None
 
+            # Support vector-less collections (e.g. payload-only graph edge collections).
             if vectors_config is None:
-                raise RuntimeError(f"Cannot determine vectors config for source collection {src}")
+                vectors_config = {}
+            vectorless = isinstance(vectors_config, dict) and not vectors_config
 
             try:
                 cli.create_collection(
@@ -401,7 +415,7 @@ def _manual_copy_points() -> None:
                         limit=batch_limit,
                         offset=offset,
                         with_payload=True,
-                        with_vectors=True,
+                        with_vectors=(not vectorless),
                     )
                 except Exception as exc:
                     raise RuntimeError(f"Failed to scroll points from {src}: {exc}") from exc
@@ -414,7 +428,9 @@ def _manual_copy_points() -> None:
                         point_id = getattr(record, "id", None)
                         payload = getattr(record, "payload", None)
                         vector = None
-                        if hasattr(record, "vector") and getattr(record, "vector") is not None:
+                        if vectorless:
+                            vector = {}
+                        elif hasattr(record, "vector") and getattr(record, "vector") is not None:
                             vector = getattr(record, "vector")
                         elif hasattr(record, "vectors") and getattr(record, "vectors") is not None:
                             vector = getattr(record, "vectors")
@@ -477,4 +493,23 @@ def _count_points(name: str) -> Optional[int]:
         # The manual path guarantees the destination gets the exact same points/payloads/vectors.
         _manual_copy_points()
 
+    # Best-effort: copy the companion graph collection when copying a base collection.
+    # Graph edges are derived data and can be rebuilt, but copying avoids a cold-start window
+    # during staging cutovers where the clone has no graph.
+    if not src.endswith("_graph") and not dest.endswith("_graph"):
+        try:
+            copy_collection_qdrant(
+                source=f"{src}_graph",
+                target=f"{dest}_graph",
+                qdrant_url=base_url,
+                overwrite=overwrite,
+            )
+        except Exception as exc:
+            logger.debug(
+                "Best-effort graph collection copy %s_graph -> %s_graph failed: %s",
+                src,
+                dest,
+                exc,
+            )
+
     return dest
diff --git a/scripts/indexing_admin.py b/scripts/indexing_admin.py
index f3bb69d8..4ab8932d 100644
--- a/scripts/indexing_admin.py
+++ b/scripts/indexing_admin.py
@@ -927,6 +927,17 @@ def delete_collection_qdrant(*, qdrant_url: str, api_key: Optional[str], collect
         return
     try:
         cli.delete_collection(collection_name=name)
+        # Best-effort: also delete companion graph edges collection when present.
+        if not name.endswith("_graph"):
+            try:
+                cli.delete_collection(collection_name=f"{name}_graph")
+            except Exception as exc:
+                try:
+                    print(
+                        f"[indexing_admin] best-effort graph collection delete failed for {name}_graph: {exc}"
+                    )
+                except Exception:
+                    pass
     except Exception:
         pass
     finally:
@@ -951,6 +962,17 @@ def recreate_collection_qdrant(*, qdrant_url: str, api_key: Optional[str], colle
             cli.delete_collection(collection_name=name)
         except Exception as delete_error:
             raise RuntimeError(f"Failed to delete existing collection '{name}' in Qdrant: {delete_error}") from delete_error
+        # Best-effort: also delete companion graph edges collection when present.
+        if not name.endswith("_graph"):
+            try:
+                cli.delete_collection(collection_name=f"{name}_graph")
+            except Exception as exc:
+                try:
+                    print(
+                        f"[indexing_admin] best-effort graph collection delete failed for {name}_graph: {exc}"
+                    )
+                except Exception:
+                    pass
     finally:
         try:
             cli.close()
@@ -984,12 +1006,9 @@ def spawn_ingest_code(
                 env.pop(k, None)
             else:
                 env[str(k)] = str(v)
-        # When we provide env overrides for a run (e.g. staging rebuild), we also want to
-        # force ingest_code to honor the explicit COLLECTION_NAME instead of routing based
-        # on per-repo state/serving_collection in multi-repo mode.
-        # CTXCE_FORCE_COLLECTION_NAME is only used for these subprocess runs; normal watcher
-        # and indexer flows do not set it.
-        env["CTXCE_FORCE_COLLECTION_NAME"] = "1"  # Force ingest_code to use COLLECTION_NAME for staging/pending env overrides
+    # For admin-triggered subprocess runs (recreate/reindex/staging), force ingest_code to
+    # honor explicit COLLECTION_NAME and avoid multi-repo enumeration.
+    env["CTXCE_FORCE_COLLECTION_NAME"] = "1"
     env["COLLECTION_NAME"] = collection
     env["WATCH_ROOT"] = work_dir
     env["WORKSPACE_PATH"] = work_dir
diff --git a/scripts/ingest/cli.py b/scripts/ingest/cli.py
index 8561a9c2..676ee77a 100644
--- a/scripts/ingest/cli.py
+++ b/scripts/ingest/cli.py
@@ -15,6 +15,8 @@
     is_multi_repo_mode,
     get_collection_name,
 )
+from scripts import workspace_state as _ws
+from scripts.collection_health import clear_indexing_caches as _clear_indexing_caches_impl
 from scripts.ingest.pipeline import index_repo
 from scripts.ingest.pseudo import generate_pseudo_tags
 
@@ -40,6 +42,11 @@ def parse_args():
         action="store_true",
         help="Do not skip files whose content hash matches existing index",
     )
+    parser.add_argument(
+        "--clear-indexing-caches",
+        action="store_true",
+        help="Clear local indexing caches (file hash/symbol caches) before indexing",
+    )
     parser.add_argument(
         "--schema-mode",
         type=str,
@@ -186,13 +193,25 @@ def main():
         )
         return
 
+    def _clear_indexing_caches(workspace_root: Path, repo_name: str | None) -> None:
+        try:
+            _clear_indexing_caches_impl(str(workspace_root), repo_name=repo_name)
+        except Exception:
+            pass
+
     qdrant_url = os.environ.get("QDRANT_URL", "http://localhost:6333")
     api_key = os.environ.get("QDRANT_API_KEY")
     collection = os.environ.get("COLLECTION_NAME") or os.environ.get("DEFAULT_COLLECTION") or "codebase"
     model_name = os.environ.get("EMBEDDING_MODEL", "BAAI/bge-base-en-v1.5")
 
     # Resolve collection name based on multi-repo mode
-    multi_repo = bool(is_multi_repo_mode and is_multi_repo_mode())
+    force_collection = (os.environ.get("CTXCE_FORCE_COLLECTION_NAME") or "").strip().lower() in {
+        "1",
+        "true",
+        "yes",
+        "on",
+    }
+    multi_repo = bool(is_multi_repo_mode and is_multi_repo_mode()) and not force_collection
     if multi_repo:
         print("[multi_repo] Multi-repo mode enabled - will create separate collections per repository")
 
@@ -231,6 +250,9 @@ def main():
             if not repo_collection:
                 repo_collection = "codebase"
 
+            if args.clear_indexing_caches:
+                _clear_indexing_caches(root_path, repo_name)
+
             index_repo(
                 repo_root,
                 qdrant_url,
@@ -249,7 +271,7 @@ def main():
             try:
                 resolved = get_collection_name(str(Path(args.root).resolve()))
                 placeholders = {"", "default-collection", "my-collection", "codebase"}
-                if resolved and collection in placeholders:
+                if resolved and collection in placeholders and not force_collection:
                     collection = resolved
             except Exception:
                 pass
@@ -260,6 +282,9 @@ def main():
     flag = (os.environ.get("PSEUDO_DEFER_TO_WORKER") or "").strip().lower()
     pseudo_mode = "off" if flag in {"1", "true", "yes", "on"} else "full"
 
+    if args.clear_indexing_caches:
+        _clear_indexing_caches(Path(args.root).resolve(), None)
+
     index_repo(
         Path(args.root).resolve(),
         qdrant_url,
diff --git a/scripts/ingest/graph_edges.py b/scripts/ingest/graph_edges.py
new file mode 100644
index 00000000..e3b5cb66
--- /dev/null
+++ b/scripts/ingest/graph_edges.py
@@ -0,0 +1,397 @@
+#!/usr/bin/env python3
+"""
+ingest/graph_edges.py - Materialized graph edges in Qdrant.
+
+This is a small, MIT-safe reimplementation of the "graph edges collection" idea:
+- Maintain a dedicated Qdrant collection named `<base_collection>_graph`
+- Store payload-only edge docs for fast lookups:
+  - callers/importers queries become simple keyword filters on an indexed payload field
+
+Design goals for this branch:
+- Keep this as an *accelerator* (symbol_graph still works without it)
+- Avoid Neo4j/PageRank/GraphRAG complexity
+- Avoid CLI flags; watcher can backfill opportunistically
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import os
+from typing import Any, Dict, Iterable, List, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+GRAPH_COLLECTION_SUFFIX = "_graph"
+
+EDGE_TYPE_CALLS = "calls"
+EDGE_TYPE_IMPORTS = "imports"
+
+GRAPH_INDEX_FIELDS: Tuple[str, ...] = (
+    "caller_path",
+    "callee_symbol",
+    "edge_type",
+    "repo",
+)
+
+_ENSURED_GRAPH_COLLECTIONS: set[str] = set()
+_GRAPH_VECTOR_MODE: dict[str, str] = {}
+_MISSING_GRAPH_COLLECTIONS: set[str] = set()
+_BACKFILL_OFFSETS: dict[tuple[str, Optional[str]], Any] = {}
+
+_EDGE_VECTOR_NAME = "_edge"
+_EDGE_VECTOR_VALUE = [0.0]
+
+
+def _normalize_path(path: str) -> str:
+    if not path:
+        return ""
+    try:
+        normalized = os.path.normpath(str(path))
+    except Exception:
+        normalized = str(path)
+    return normalized.replace("\\", "/")
+
+
+def get_graph_collection_name(base_collection: str) -> str:
+    return f"{base_collection}{GRAPH_COLLECTION_SUFFIX}"
+
+
+def _edge_vector_for_upsert(graph_collection: str) -> dict:
+    mode = _GRAPH_VECTOR_MODE.get(graph_collection)
+    if mode == "named":
+        return {_EDGE_VECTOR_NAME: _EDGE_VECTOR_VALUE}
+    return {}
+
+
+def ensure_graph_collection(client: Any, base_collection: str) -> Optional[str]:
+    """Ensure `<base_collection>_graph` exists and has payload indexes."""
+    from qdrant_client import models as qmodels
+    from qdrant_client.http.exceptions import UnexpectedResponse
+
+    if not base_collection:
+        return None
+    graph_coll = get_graph_collection_name(base_collection)
+    if graph_coll in _ENSURED_GRAPH_COLLECTIONS:
+        return graph_coll
+
+    def _detect_vector_mode(info: Any) -> str:
+        try:
+            vectors = getattr(
+                getattr(getattr(info, "config", None), "params", None), "vectors", None
+            )
+            if isinstance(vectors, dict):
+                return "none" if not vectors else "named"
+            return "none" if vectors is None else "named"
+        except Exception:
+            return "named"
+
+    try:
+        info = client.get_collection(graph_coll)
+        _GRAPH_VECTOR_MODE[graph_coll] = _detect_vector_mode(info)
+        _ENSURED_GRAPH_COLLECTIONS.add(graph_coll)
+        _MISSING_GRAPH_COLLECTIONS.discard(graph_coll)
+        return graph_coll
+    except UnexpectedResponse as e:
+        # Only a 404 means "missing"; any other HTTP failure should be visible.
+        if getattr(e, "status_code", None) != 404:
+            logger.exception(
+                "Failed to get graph collection %s (status=%s): %s",
+                graph_coll,
+                getattr(e, "status_code", None),
+                e,
+            )
+            return None
+    except Exception as e:
+        logger.exception("Failed to get graph collection %s: %s", graph_coll, e)
+        return None
+
+    try:
+        # Prefer vector-less collection when supported by server/client.
+        try:
+            client.create_collection(
+                collection_name=graph_coll,
+                vectors_config={},
+            )
+            _GRAPH_VECTOR_MODE[graph_coll] = "none"
+        except Exception as vec_exc:
+            logger.debug(
+                "Vector-less creation failed for %s, trying named vector: %s",
+                graph_coll,
+                vec_exc,
+            )
+            client.create_collection(
+                collection_name=graph_coll,
+                vectors_config={
+                    _EDGE_VECTOR_NAME: qmodels.VectorParams(
+                        size=1, distance=qmodels.Distance.COSINE
+                    )
+                },
+            )
+            _GRAPH_VECTOR_MODE[graph_coll] = "named"
+
+        # Create payload indexes (best-effort).
+        for field in GRAPH_INDEX_FIELDS:
+            try:
+                client.create_payload_index(
+                    collection_name=graph_coll,
+                    field_name=field,
+                    field_schema=qmodels.PayloadSchemaType.KEYWORD,
+                )
+            except Exception:
+                pass
+
+        _ENSURED_GRAPH_COLLECTIONS.add(graph_coll)
+        _MISSING_GRAPH_COLLECTIONS.discard(graph_coll)
+        return graph_coll
+    except Exception as e:
+        logger.debug("Failed to ensure graph collection %s: %s", graph_coll, e)
+        return None
+
+
+def _edge_id(edge_type: str, repo: str, caller_path: str, callee_symbol: str) -> str:
+    key = f"{edge_type}\x00{repo}\x00{caller_path}\x00{callee_symbol}"
+    return hashlib.sha256(key.encode("utf-8", errors="ignore")).hexdigest()[:32]
+
+
+def _iter_edges(
+    *,
+    caller_path: str,
+    repo: str,
+    calls: Iterable[str] = (),
+    imports: Iterable[str] = (),
+) -> List[Dict[str, Any]]:
+    norm_path = _normalize_path(caller_path)
+    repo_s = (repo or "").strip() or "default"
+
+    edges: List[Dict[str, Any]] = []
+    for sym in calls or []:
+        s = str(sym).strip()
+        if not s:
+            continue
+        edges.append(
+            {
+                "id": _edge_id(EDGE_TYPE_CALLS, repo_s, norm_path, s),
+                "payload": {
+                    "caller_path": norm_path,
+                    "callee_symbol": s,
+                    "edge_type": EDGE_TYPE_CALLS,
+                    "repo": repo_s,
+                },
+            }
+        )
+    for sym in imports or []:
+        s = str(sym).strip()
+        if not s:
+            continue
+        edges.append(
+            {
+                "id": _edge_id(EDGE_TYPE_IMPORTS, repo_s, norm_path, s),
+                "payload": {
+                    "caller_path": norm_path,
+                    "callee_symbol": s,
+                    "edge_type": EDGE_TYPE_IMPORTS,
+                    "repo": repo_s,
+                },
+            }
+        )
+    return edges
+
+
+def upsert_file_edges(
+    client: Any,
+    base_collection: str,
+    *,
+    caller_path: str,
+    repo: str | None,
+    calls: List[str] | None = None,
+    imports: List[str] | None = None,
+) -> int:
+    graph_coll = ensure_graph_collection(client, base_collection)
+    if not graph_coll:
+        return 0
+    edges = _iter_edges(
+        caller_path=caller_path,
+        repo=repo or "default",
+        calls=calls or [],
+        imports=imports or [],
+    )
+    if not edges:
+        return 0
+
+    from qdrant_client import models as qmodels
+
+    points = [
+        qmodels.PointStruct(
+            id=e["id"],
+            vector=_edge_vector_for_upsert(graph_coll),
+            payload=e["payload"],
+        )
+        for e in edges
+    ]
+    try:
+        client.upsert(collection_name=graph_coll, points=points, wait=True)
+        return len(points)
+    except Exception as e:
+        logger.debug("Graph edge upsert failed for %s: %s", caller_path, e)
+        return 0
+
+
+def delete_edges_by_path(
+    client: Any,
+    base_collection: str,
+    *,
+    caller_path: str,
+    repo: str | None = None,
+) -> int:
+    graph_coll = get_graph_collection_name(base_collection)
+    if graph_coll in _MISSING_GRAPH_COLLECTIONS:
+        return 0
+
+    from qdrant_client import models as qmodels
+
+    norm_path = _normalize_path(caller_path)
+    must: list[Any] = [
+        qmodels.FieldCondition(
+            key="caller_path", match=qmodels.MatchValue(value=norm_path)
+        )
+    ]
+    if repo:
+        r = str(repo).strip()
+        if r and r != "*":
+            must.append(
+                qmodels.FieldCondition(key="repo", match=qmodels.MatchValue(value=r))
+            )
+
+    try:
+        client.delete(
+            collection_name=graph_coll,
+            points_selector=qmodels.FilterSelector(filter=qmodels.Filter(must=must)),
+        )
+        return 1
+    except Exception as e:
+        err = str(e).lower()
+        if "404" in err or "doesn't exist" in err or "not found" in err:
+            _MISSING_GRAPH_COLLECTIONS.add(graph_coll)
+        return 0
+
+
+def graph_edges_backfill_tick(
+    client: Any,
+    base_collection: str,
+    *,
+    repo_name: str | None = None,
+    max_files: int = 128,
+) -> int:
+    """Best-effort incremental backfill from `<base_collection>` into `<base_collection>_graph`.
+
+    This scans the main collection and upserts file-level edges into the graph collection.
+    It's idempotent (deterministic IDs) and safe to run continuously in a watcher worker.
+    """
+    from qdrant_client import models as qmodels
+
+    if not base_collection or max_files <= 0:
+        return 0
+
+    graph_coll = ensure_graph_collection(client, base_collection)
+    if not graph_coll:
+        return 0
+
+    must: list[Any] = []
+    if repo_name:
+        must.append(
+            qmodels.FieldCondition(
+                key="metadata.repo", match=qmodels.MatchValue(value=repo_name)
+            )
+        )
+    flt = qmodels.Filter(must=must or None)
+
+    processed_files = 0
+    seen_paths: set[str] = set()
+
+    key = (base_collection, repo_name)
+    next_offset = _BACKFILL_OFFSETS.get(key)
+
+    # We may need to overscan because the main collection is chunked.
+    overscan = max_files * 8
+    while processed_files < max_files:
+        attempts = 0
+        while True:
+            try:
+                points, next_offset = client.scroll(
+                    collection_name=base_collection,
+                    scroll_filter=flt,
+                    limit=min(64, overscan),
+                    with_payload=True,
+                    with_vectors=False,
+                    offset=next_offset,
+                )
+                break
+            except Exception as e:
+                attempts += 1
+                logger.exception(
+                    "Graph edge backfill scroll failed (collection=%s repo=%s offset=%s attempt=%d): %s",
+                    base_collection,
+                    repo_name or "default",
+                    next_offset,
+                    attempts,
+                    e,
+                )
+                # Retry a couple times for transient errors, then raise so failures are not silent.
+                if attempts >= 3:
+                    raise
+                import time
+
+                time.sleep(0.25 * (2 ** (attempts - 1)))
+
+        if not points:
+            break
+
+        for rec in points:
+            if processed_files >= max_files:
+                break
+            payload = getattr(rec, "payload", None) or {}
+            md = payload.get("metadata") or {}
+            path = md.get("path") or ""
+            if not path:
+                continue
+            norm_path = _normalize_path(str(path))
+            if norm_path in seen_paths:
+                continue
+            seen_paths.add(norm_path)
+
+            repo = md.get("repo") or repo_name or "default"
+            calls = md.get("calls") or []
+            imports = md.get("imports") or []
+            if not isinstance(calls, list):
+                calls = []
+            if not isinstance(imports, list):
+                imports = []
+
+            upsert_file_edges(
+                client,
+                base_collection,
+                caller_path=norm_path,
+                repo=str(repo),
+                calls=[str(x) for x in calls if x],
+                imports=[str(x) for x in imports if x],
+            )
+            processed_files += 1
+
+        if next_offset is None:
+            break
+
+    _BACKFILL_OFFSETS[key] = next_offset
+    return processed_files
+
+
+__all__ = [
+    "GRAPH_COLLECTION_SUFFIX",
+    "EDGE_TYPE_CALLS",
+    "EDGE_TYPE_IMPORTS",
+    "get_graph_collection_name",
+    "ensure_graph_collection",
+    "upsert_file_edges",
+    "delete_edges_by_path",
+    "graph_edges_backfill_tick",
+]
diff --git a/scripts/ingest/pipeline.py b/scripts/ingest/pipeline.py
index 45716049..f43c7d3a 100644
--- a/scripts/ingest/pipeline.py
+++ b/scripts/ingest/pipeline.py
@@ -227,6 +227,53 @@ def _normalize_info_for_dense(s: str) -> str:
     return text
 
 
+def _sync_graph_edges_best_effort(
+    client: QdrantClient,
+    collection: str,
+    file_path: str,
+    repo: str | None,
+    calls: list[str] | None,
+    imports: list[str] | None,
+) -> None:
+    """Best-effort sync of file-level graph edges. Safe to skip on failure."""
+    enabled = str(os.environ.get("GRAPH_EDGES_ENABLE", "1") or "").strip().lower() in {
+        "1",
+        "true",
+        "yes",
+        "on",
+    }
+    if not enabled:
+        return
+    try:
+        from scripts.ingest.graph_edges import (
+            delete_edges_by_path,
+            ensure_graph_collection,
+            upsert_file_edges,
+        )
+
+        ensure_graph_collection(client, collection)
+        # Important: delete stale edges for this file before upserting the new set.
+        delete_edges_by_path(
+            client,
+            collection,
+            caller_path=str(file_path),
+            repo=repo,
+        )
+        upsert_file_edges(
+            client,
+            collection,
+            caller_path=str(file_path),
+            repo=repo,
+            calls=calls,
+            imports=imports,
+        )
+    except Exception as exc:
+        try:
+            print(f"[graph_edges] best-effort sync failed for {file_path}: {exc}")
+        except Exception:
+            pass
+
+
 def build_information(
     language: str, path: Path, start: int, end: int, first_line: str
 ) -> str:
@@ -653,6 +700,16 @@ def make_point(pid, dense_vec, lex_vec, payload, lex_text: str = "", code_text:
             for i, v, lx, m, lt, ct in zip(batch_ids, vectors, batch_lex, batch_meta, batch_lex_text, batch_code)
         ]
         upsert_points(client, collection, points)
+        # Optional: materialize file-level graph edges in a companion `<collection>_graph` store.
+        # This is an accelerator for symbol_graph callers/importers and is safe to skip on failure.
+        _sync_graph_edges_best_effort(
+            client,
+            collection,
+            str(file_path),
+            repo_tag,
+            calls,
+            imports,
+        )
         try:
             ws = os.environ.get("WATCH_ROOT") or os.environ.get("WORKSPACE_PATH") or "/work"
             if set_cached_file_hash:
@@ -1367,6 +1424,15 @@ def process_file_with_smart_reindexing(
 
     if all_points:
         _upsert_points_fn(client, current_collection, all_points)
+        # Optional: materialize file-level graph edges (best-effort).
+        _sync_graph_edges_best_effort(
+            client,
+            current_collection,
+            str(file_path),
+            per_file_repo,
+            calls,
+            imports,
+        )
 
     try:
         if set_cached_symbols:
diff --git a/scripts/ingest_code.py b/scripts/ingest_code.py
index 2da24911..574457a5 100644
--- a/scripts/ingest_code.py
+++ b/scripts/ingest_code.py
@@ -212,6 +212,26 @@
     index_repo,
     process_file_with_smart_reindexing,
 )
+
+# ---------------------------------------------------------------------------
+# Graph edges (optional accelerator)
+# ---------------------------------------------------------------------------
+try:
+    from scripts.ingest.graph_edges import (
+        graph_edges_backfill_tick,
+        delete_edges_by_path as delete_graph_edges_by_path,
+        upsert_file_edges as upsert_graph_edges_for_file,
+    )
+except ImportError:
+    # graph_edges_backfill_tick is optional and intentionally left as None to
+    # force callers to explicitly guard long-running backfill behavior.
+    graph_edges_backfill_tick = None  # type: ignore[assignment]
+
+    def delete_graph_edges_by_path(*_args, **_kwargs) -> int:
+        return 0
+
+    def upsert_graph_edges_for_file(*_args, **_kwargs) -> int:
+        return 0
 # ---------------------------------------------------------------------------
 # Re-exports from ingest/cli.py
 # ---------------------------------------------------------------------------
@@ -338,6 +358,10 @@ def main():
     "index_repo",
     "process_file_with_smart_reindexing",
     "pseudo_backfill_tick",
+    # Graph edges (optional)
+    "graph_edges_backfill_tick",
+    "delete_graph_edges_by_path",
+    "upsert_graph_edges_for_file",
     # CLI
     "main",
     # Backward compat
diff --git a/scripts/mcp_impl/search.py b/scripts/mcp_impl/search.py
index e4dc3766..bc310f14 100644
--- a/scripts/mcp_impl/search.py
+++ b/scripts/mcp_impl/search.py
@@ -54,6 +54,46 @@
 )
 
 
+# Fields to strip from results when debug=False (internal/debugging fields)
+_DEBUG_RESULT_FIELDS = {
+    "components",  # Internal scoring breakdown (dense_rrf, lexical, fname_boost, etc.)
+    "doc_id",  # Internal benchmark ID (often null/opaque)
+    "code_id",  # Internal benchmark ID (often null/opaque)
+    "payload",  # Duplicates other fields (information, document, pseudo, tags)
+    "why",  # Often empty []; debugging explanation list
+    "span_budgeted",  # Internal budget flag
+    "relations",  # Call graph info (imports, calls) - useful but often noise
+    "related_paths",  # Optional related file paths
+    "budget_tokens_used",  # Internal token accounting
+    "fname_boost",  # Internal boost value (already applied to score)
+    "host_path",  # Internal dual-path (host side) - use path/client_path instead
+    "container_path",  # Internal dual-path (container side) - use path/client_path instead
+}
+
+# Top-level response fields to strip when debug=False
+_DEBUG_TOP_LEVEL_FIELDS = {
+    "rerank_counters",  # Internal reranking metrics (inproc_hybrid, timeout, etc.)
+    "code_signals",  # Internal code signal detection results
+}
+
+
+def _strip_debug_fields(item: dict, keep_paths: bool = True) -> dict:
+    """Strip internal/debug fields from a result item.
+
+    Args:
+        item: Result dict to strip
+        keep_paths: If True, keep host_path/container_path
+
+    Returns:
+        New dict with debug fields removed
+    """
+    strip_fields = _DEBUG_RESULT_FIELDS
+    if keep_paths:
+        strip_fields = _DEBUG_RESULT_FIELDS - {"host_path", "container_path"}
+    result = {k: v for k, v in item.items() if k not in strip_fields}
+    return result
+
+
 async def _repo_search_impl(
     query: Any = None,
     queries: Any = None,  # Alias for query (many clients use this)
@@ -89,6 +129,7 @@ async def _repo_search_impl(
     repo: Any = None,  # str, list[str], or "*" to search all repos
     # Response shaping
     compact: Any = None,
+    debug: Any = None,  # When True, include verbose internal fields (components, rerank_counters, etc.)
     output_format: Any = None,  # "json" (default) or "toon" for token-efficient format
     args: Any = None,  # Compatibility shim for mcp-remote/Claude wrappers that send args/kwargs
     kwargs: Any = None,
@@ -119,16 +160,21 @@ async def _repo_search_impl(
       Use repo=["frontend","backend"] to search related repos together.
     - Filters (optional): language, under (path prefix), kind, symbol, ext, path_regex,
       path_glob (str or list[str]), not_glob (str or list[str]), not_ (negative text), case.
+    - debug: bool (default false). When true, includes verbose internal fields like
+      components, rerank_counters, code_signals. Default false saves ~60-80% tokens.
 
     Returns:
     - Dict with keys:
-      - results: list of {score, path, symbol, start_line, end_line, why[, components][, relations][, related_paths][, snippet]}
-      - total: int; used_rerank: bool; rerank_counters: dict
+      - results: list of {score, path, symbol, start_line, end_line[, snippet][, tags][, host_path][, container_path]}
+        When debug=true, also includes: components, why, relations, related_paths, doc_id, code_id
+      - total: int; used_rerank: bool
     - If compact=true (and snippets not requested), results contain only {path,start_line,end_line}.
+    - If debug=true, response also includes: rerank_counters, code_signals
 
     Examples:
     - path_glob=["scripts/**","**/*.py"], language="python"
     - symbol="context_answer", under="scripts"
+    - debug=true  # Include internal scoring details for query tuning
     """
     sess = require_auth_session_fn(session) if require_auth_session_fn else session
 
@@ -252,6 +298,8 @@ async def _repo_search_impl(
                 case = _extra.get("case")
             if compact in (None, "") and _extra.get("compact") is not None:
                 compact = _extra.get("compact")
+            if debug in (None, "") and _extra.get("debug") is not None:
+                debug = _extra.get("debug")
             # Optional mode hint: "code_first", "docs_first", "balanced"
             if (
                 mode is None or (isinstance(mode, str) and str(mode).strip() == "")
@@ -446,6 +494,10 @@ def _to_str_list(x):
     if include_snippet:
         compact = False
 
+    # Debug mode: when False (default), strip internal/debug fields from results
+    # to reduce token bloat. Set debug=True to see components, rerank_counters, etc.
+    debug = _to_bool(debug, False)
+
     # Default behavior: exclude commit-history docs (which use path=".git") from
     # generic repo_search calls, unless the caller explicitly asks for git
     # content. This prevents normal code queries from surfacing commit-index
@@ -1491,6 +1543,11 @@ def _read_snip(args):
             }
             for r in results
         ]
+    elif not debug:
+        # Strip debug/internal fields from results to reduce token bloat
+        # Keeps: score, path, host_path, container_path, symbol, snippet,
+        #        start_line, end_line, tags, pseudo
+        results = [_strip_debug_fields(r) for r in results]
 
     response = {
         "args": {
@@ -1518,13 +1575,17 @@ def _read_snip(args):
             "compact": (_to_bool(compact_raw, compact)),
         },
         "used_rerank": bool(used_rerank),
-        "rerank_counters": rerank_counters,
-        "code_signals": code_signals if code_signals.get("has_code_signals") else None,
         "total": len(results),
         "results": results,
         **res,
     }
 
+    # Only include debug fields when explicitly requested
+    if debug:
+        response["rerank_counters"] = rerank_counters
+        if code_signals.get("has_code_signals"):
+            response["code_signals"] = code_signals
+
     # Apply TOON formatting if requested or enabled globally
     # Full mode (compact=False) still saves tokens vs JSON while preserving all fields
     if _should_use_toon(output_format):
diff --git a/scripts/mcp_impl/symbol_graph.py b/scripts/mcp_impl/symbol_graph.py
index da518ac4..6c574dff 100644
--- a/scripts/mcp_impl/symbol_graph.py
+++ b/scripts/mcp_impl/symbol_graph.py
@@ -24,6 +24,14 @@
 
 logger = logging.getLogger(__name__)
 
+try:
+    from scripts.ingest.graph_edges import GRAPH_COLLECTION_SUFFIX as _GRAPH_SUFFIX
+except Exception:
+    _GRAPH_SUFFIX = "_graph"
+
+GRAPH_COLLECTION_SUFFIX = _GRAPH_SUFFIX
+_MISSING_GRAPH_COLLECTIONS: set[str] = set()
+
 __all__ = [
     "_symbol_graph_impl",
     "_format_symbol_graph_toon",
@@ -195,16 +203,28 @@ async def _symbol_graph_impl(
 
     try:
         if query_type == "callers":
-            # Find chunks where metadata.calls array contains the symbol (exact match)
-            results = await _query_array_field(
+            # Prefer graph edges collection when available (fast keyword filters).
+            results = await _query_graph_edges_collection(
                 client=client,
                 collection=coll,
-                field_key="metadata.calls",
-                value=symbol,
+                symbol=symbol,
+                edge_type="calls",
                 limit=limit,
                 language=language,
+                repo_filter=None,
                 under=_norm_under(under),
             )
+            if not results:
+                # Fall back to array field lookup in the main collection.
+                results = await _query_array_field(
+                    client=client,
+                    collection=coll,
+                    field_key="metadata.calls",
+                    value=symbol,
+                    limit=limit,
+                    language=language,
+                    under=_norm_under(under),
+                )
         elif query_type == "definition":
             # Find chunks where symbol_path matches the symbol
             results = await _query_definition(
@@ -216,16 +236,27 @@ async def _symbol_graph_impl(
                 under=_norm_under(under),
             )
         elif query_type == "importers":
-            # Find chunks where metadata.imports array contains the symbol
-            results = await _query_array_field(
+            results = await _query_graph_edges_collection(
                 client=client,
                 collection=coll,
-                field_key="metadata.imports",
-                value=symbol,
+                symbol=symbol,
+                edge_type="imports",
                 limit=limit,
                 language=language,
+                repo_filter=None,
                 under=_norm_under(under),
             )
+            if not results:
+                # Fall back to array field lookup in the main collection.
+                results = await _query_array_field(
+                    client=client,
+                    collection=coll,
+                    field_key="metadata.imports",
+                    value=symbol,
+                    limit=limit,
+                    language=language,
+                    under=_norm_under(under),
+                )
 
         # If no results, fall back to semantic search
         if not results:
@@ -259,6 +290,153 @@ async def _symbol_graph_impl(
     }
 
 
+async def _query_graph_edges_collection(
+    client: Any,
+    collection: str,
+    symbol: str,
+    edge_type: str,
+    limit: int,
+    language: Optional[str] = None,
+    repo_filter: str | None = None,
+    under: str | None = None,
+) -> List[Dict[str, Any]]:
+    """Query `<collection>_graph` and hydrate results from the main collection.
+
+    The graph collection stores file-level edges:
+    - caller_path -> callee_symbol (calls/imports)
+    """
+    from qdrant_client import models as qmodels
+
+    graph_coll = f"{collection}{GRAPH_COLLECTION_SUFFIX}"
+    if graph_coll in _MISSING_GRAPH_COLLECTIONS:
+        return []
+
+    # Build graph filter
+    must: list[Any] = [
+        qmodels.FieldCondition(
+            key="edge_type", match=qmodels.MatchValue(value=str(edge_type))
+        )
+    ]
+    if repo_filter:
+        rf = str(repo_filter).strip()
+        if rf and rf != "*":
+            must.append(
+                qmodels.FieldCondition(key="repo", match=qmodels.MatchValue(value=rf))
+            )
+
+    # Try exact match, then symbol variants.
+    callee_variants = _symbol_variants(symbol) or [symbol]
+    seen_paths: set[str] = set()
+    caller_paths: List[str] = []
+
+    for variant in callee_variants:
+        if len(caller_paths) >= limit:
+            break
+        v = str(variant).strip()
+        if not v:
+            continue
+        flt = qmodels.Filter(
+            must=must
+            + [
+                qmodels.FieldCondition(
+                    key="callee_symbol", match=qmodels.MatchValue(value=v)
+                )
+            ]
+        )
+
+        def _scroll(_flt=flt):
+            return client.scroll(
+                collection_name=graph_coll,
+                scroll_filter=_flt,
+                limit=max(32, limit * 4),
+                with_payload=True,
+                with_vectors=False,
+            )
+
+        try:
+            points, _ = await asyncio.to_thread(_scroll)
+        except Exception as e:
+            err = str(e).lower()
+            if "404" in err or "doesn't exist" in err or "not found" in err:
+                _MISSING_GRAPH_COLLECTIONS.add(graph_coll)
+                return []
+            logger.exception(
+                "_query_graph_edges_collection scroll failed for %s", graph_coll
+            )
+            raise
+
+        for rec in points or []:
+            payload = getattr(rec, "payload", None) or {}
+            p = payload.get("caller_path") or ""
+            if not p:
+                continue
+            path_s = str(p)
+            if under and not str(path_s).startswith(str(under)):
+                continue
+            if path_s in seen_paths:
+                continue
+            seen_paths.add(path_s)
+            caller_paths.append(path_s)
+            if len(caller_paths) >= limit:
+                break
+
+    if not caller_paths:
+        return []
+
+    # Hydrate caller paths back into normal symbol_graph point-shaped results.
+    hydrated: List[Dict[str, Any]] = []
+    for p in caller_paths[:limit]:
+        if len(hydrated) >= limit:
+            break
+
+        def _scroll_main(_p=p, _language=language):
+            must = [
+                qmodels.FieldCondition(
+                    key="metadata.path", match=qmodels.MatchValue(value=_p)
+                )
+            ]
+            if _language:
+                must.append(
+                    qmodels.FieldCondition(
+                        key="metadata.language",
+                        match=qmodels.MatchValue(value=str(_language).lower()),
+                    )
+                )
+            return client.scroll(
+                collection_name=collection,
+                scroll_filter=qmodels.Filter(
+                    must=must
+                ),
+                limit=1,
+                with_payload=True,
+                with_vectors=False,
+            )
+
+        try:
+            pts, _ = await asyncio.to_thread(_scroll_main)
+        except Exception:
+            pts = []
+
+        if pts:
+            hydrated.append(_format_point(pts[0]))
+        else:
+            # If language filtering was requested but no matching main-collection doc
+            # exists (or hydration failed), skip returning a placeholder to avoid
+            # producing language-inconsistent results.
+            if not language:
+                hydrated.append(
+                    {
+                        "path": p,
+                        "symbol": "",
+                        "symbol_path": "",
+                        "start_line": 0,
+                        "end_line": 0,
+                    }
+                )
+
+    return hydrated
+
+
 async def _query_array_field(
     client: Any,
     collection: str,
diff --git a/scripts/mcp_indexer_server.py b/scripts/mcp_indexer_server.py
index d12aee9b..cdd1912b 100644
--- a/scripts/mcp_indexer_server.py
+++ b/scripts/mcp_indexer_server.py
@@ -300,6 +300,23 @@ def _highlight_snippet(snippet, tokens):  # type: ignore
 )
 mcp = FastMCP(APP_NAME, transport_security=_security_settings)
 
+# Minimal resource so MCP clients can verify resource wiring.
+@mcp.resource(
+    "resource://context-engine/indexer/info",
+    name="context-engine-indexer-info",
+    title="Context Engine Indexer Info",
+    description="Basic metadata about the running indexer MCP server.",
+    mime_type="application/json",
+)
+def _indexer_info_resource():
+    return {
+        "app": APP_NAME,
+        "host": HOST,
+        "port": PORT,
+        "qdrant_url": QDRANT_URL,
+        "default_collection": DEFAULT_COLLECTION,
+    }
+
 
 # Capture tool registry automatically by wrapping the decorator once
 _TOOLS_REGISTRY: list[dict] = []
@@ -1082,6 +1099,7 @@ async def repo_search(
     case: Any = None,
     repo: Any = None,
     compact: Any = None,
+    debug: Any = None,
     output_format: Any = None,
     args: Any = None,
     kwargs: Any = None,
@@ -1098,12 +1116,13 @@ async def repo_search(
     - per_path: int (default 2). Max results per file.
     - include_snippet/context_lines: return inline snippets near hits when true.
     - rerank_*: ONNX reranker is ON by default for best relevance; timeouts fall back to hybrid.
+    - debug: bool (default false). Include verbose internal fields (components, rerank_counters, etc).
     - output_format: "json" (default) or "toon" for token-efficient TOON format.
     - collection: str. Target collection; defaults to workspace state or env COLLECTION_NAME.
     - repo: str or list[str]. Filter by repo name(s). Use "*" to search all repos.
 
     Returns:
-    - Dict with keys: results, total, used_rerank, rerank_counters
+    - Dict with keys: results, total, used_rerank, [rerank_counters if debug=true]
     """
     return await _repo_search_impl(
         query=query,
@@ -1134,6 +1153,7 @@ async def repo_search(
         case=case,
         repo=repo,
         compact=compact,
+        debug=debug,
         output_format=output_format,
         args=args,
         kwargs=kwargs,
@@ -1642,6 +1662,9 @@ async def code_search(
     case: Any = None,
     session: Any = None,
     compact: Any = None,
+    debug: Any = None,
+    output_format: Any = None,
+    repo: Any = None,
     kwargs: Any = None,
 ) -> Dict[str, Any]:
     """Exact alias of repo_search (hybrid code search with reranking enabled by default).
@@ -1674,6 +1697,9 @@ async def code_search(
         case=case,
         session=session,
         compact=compact,
+        debug=debug,
+        output_format=output_format,
+        repo=repo,
         kwargs=kwargs,
     )
 
diff --git a/scripts/prune.py b/scripts/prune.py
index 5e2f14fb..d654132a 100755
--- a/scripts/prune.py
+++ b/scripts/prune.py
@@ -39,26 +39,34 @@ def delete_by_path(client: QdrantClient, path_str: str) -> int:
         return 0
 
 
-def delete_graph_edges_by_path(client: QdrantClient, path_str: str) -> int:
+def delete_graph_edges_by_path(client: QdrantClient, path_str: str, repo: str | None = None) -> int:
     """Best-effort deletion for graph-edge collections (if present).
 
     Some deployments store symbol-graph edges in a separate Qdrant collection
-    (commonly `${COLLECTION}_graph`). Those points may reference a file path as
-    either caller or callee; delete both to prevent stale graph results.
+    (commonly `${COLLECTION}_graph`). On this branch, edge docs are file-level and
+    reference a file path as `caller_path`.
     """
     if not path_str:
         return 0
-
-    flt = models.Filter(
-        should=[
-            models.FieldCondition(
-                key="caller_path", match=models.MatchValue(value=path_str)
-            ),
-            models.FieldCondition(
-                key="callee_path", match=models.MatchValue(value=path_str)
-            ),
-        ]
-    )
+    try:
+        path_str = os.path.normpath(str(path_str))
+    except Exception:
+        path_str = str(path_str)
+    path_str = str(path_str).replace("\\", "/")
+
+    must = [
+        models.FieldCondition(key="caller_path", match=models.MatchValue(value=path_str))
+    ]
+    if repo:
+        try:
+            r = str(repo).strip()
+        except Exception:
+            r = ""
+        if r and r != "*":
+            must.append(
+                models.FieldCondition(key="repo", match=models.MatchValue(value=r))
+            )
+    flt = models.Filter(must=must)
     try:
         res = client.delete(
             collection_name=GRAPH_COLLECTION,
@@ -116,13 +124,13 @@ def main():
             )
             if not abs_path.exists():
                 removed_missing += delete_by_path(client, path_str)
-                removed_graph_edges += delete_graph_edges_by_path(client, path_str)
+                removed_graph_edges += delete_graph_edges_by_path(client, path_str, md.get("repo"))
                 print(f"[prune] removed missing file points: {path_str}")
                 continue
             current_hash = sha1_file(abs_path)
             if file_hash and current_hash and current_hash != file_hash:
                 removed_mismatch += delete_by_path(client, path_str)
-                removed_graph_edges += delete_graph_edges_by_path(client, path_str)
+                removed_graph_edges += delete_graph_edges_by_path(client, path_str, md.get("repo"))
                 print(f"[prune] removed outdated points (hash mismatch): {path_str}")
 
         if next_page is None:
diff --git a/scripts/remote_upload_client.py b/scripts/remote_upload_client.py
index fcc7d6ba..bf9b980f 100644
--- a/scripts/remote_upload_client.py
+++ b/scripts/remote_upload_client.py
@@ -520,6 +520,51 @@ def log_mapping_summary(self) -> None:
         logger.info(f"  source_path: {info['source_path']}")
         logger.info(f"  container_path: {info['container_path']}")
 
+    def _excluded_dirnames(self) -> frozenset:
+        # Keep in sync with standalone_upload_client exclusions.
+        # NOTE: This caches the exclusion set per RemoteUploadClient instance.
+        # Runtime changes to DEV_REMOTE_MODE/REMOTE_UPLOAD_MODE won't be reflected
+        # until a new client is created (typically via process restart), which is
+        # acceptable for the upload client use case.
+        cached = getattr(self, "_excluded_dirnames_cache", None)
+        if cached is not None:
+            return cached
+        excluded = {
+            "node_modules", "vendor", "dist", "build", "target", "out",
+            ".git", ".hg", ".svn", ".vscode", ".idea", ".venv", "venv",
+            "__pycache__", ".pytest_cache", ".mypy_cache", ".cache",
+            ".context-engine", ".context-engine-uploader", ".codebase",
+        }
+        dev_remote = os.environ.get("DEV_REMOTE_MODE") == "1" or os.environ.get("REMOTE_UPLOAD_MODE") == "development"
+        if dev_remote:
+            excluded.add("dev-workspace")
+        cached = frozenset(excluded)
+        self._excluded_dirnames_cache = cached
+        return cached
+
+    def _is_ignored_path(self, path: Path) -> bool:
+        """Return True when path is outside workspace or under excluded dirs."""
+        try:
+            workspace_root = Path(self.workspace_path).resolve()
+            rel = path.resolve().relative_to(workspace_root)
+        except Exception:
+            return True
+
+        dir_parts = set(rel.parts[:-1]) if len(rel.parts) > 1 else set()
+        if dir_parts & self._excluded_dirnames():
+            return True
+        # Ignore hidden directories anywhere under the workspace, but allow
+        # extensionless dotfiles like `.gitignore` that we explicitly support.
+        if any(p.startswith(".") for p in rel.parts[:-1]):
+            return True
+        try:
+            extensionless = set((idx.EXTENSIONLESS_FILES or {}).keys())
+        except Exception:
+            extensionless = set()
+        if rel.name.startswith(".") and rel.name.lower() not in extensionless:
+            return True
+        return False
+
     def _get_temp_bundle_dir(self) -> Path:
         """Get or create temporary directory for bundle creation."""
         if not self.temp_dir:
@@ -547,6 +592,8 @@ def detect_file_changes(self, changed_paths: List[Path]) -> Dict[str, List]:
         }
 
         for path in changed_paths:
+            if self._is_ignored_path(path):
+                continue
             # Resolve to an absolute path for stable cache keys
             try:
                 abs_path = str(path.resolve())
@@ -1296,27 +1343,30 @@ def get_all_code_files(self) -> List[Path]:
 
             # Single walk with early pruning similar to standalone client
             ext_suffixes = {str(ext).lower() for ext in idx.CODE_EXTS if str(ext).startswith('.')}
-            name_matches = {str(ext) for ext in idx.CODE_EXTS if not str(ext).startswith('.')}
-            dev_remote = os.environ.get("DEV_REMOTE_MODE") == "1" or os.environ.get("REMOTE_UPLOAD_MODE") == "development"
-            excluded = {
-                "node_modules", "vendor", "dist", "build", "target", "out",
-                ".git", ".hg", ".svn", ".vscode", ".idea", ".venv", "venv",
-                "__pycache__", ".pytest_cache", ".mypy_cache", ".cache",
-                ".context-engine", ".context-engine-uploader", ".codebase"
-            }
-            if dev_remote:
-                excluded.add("dev-workspace")
+            try:
+                extensionless_names = {k.lower() for k in (idx.EXTENSIONLESS_FILES or {}).keys()}
+            except Exception:
+                extensionless_names = set()
+            excluded = self._excluded_dirnames()
 
             seen = set()
             for root, dirnames, filenames in os.walk(workspace_path):
                 dirnames[:] = [d for d in dirnames if d not in excluded and not d.startswith('.')]
 
                 for filename in filenames:
-                    if filename.startswith('.'):
+                    # Allow dotfiles that are in EXTENSIONLESS_FILES (e.g., .gitignore)
+                    fname_lower = filename.lower()
+                    if filename.startswith('.') and fname_lower not in extensionless_names:
                         continue
                     candidate = Path(root) / filename
+                    if self._is_ignored_path(candidate):
+                        continue
                     suffix = candidate.suffix.lower()
-                    if filename in name_matches or suffix in ext_suffixes:
+                    if (
+                        suffix in ext_suffixes
+                        or fname_lower in extensionless_names
+                        or fname_lower.startswith("dockerfile")
+                    ):
                         resolved = candidate.resolve()
                         if resolved not in seen:
                             seen.add(resolved)
@@ -1368,13 +1418,13 @@ def on_any_event(self, event):
 
                 # Always check src_path
                 src_path = Path(event.src_path)
-                if idx.CODE_EXTS.get(src_path.suffix.lower(), "unknown") != "unknown":
+                if not self.client._is_ignored_path(src_path) and idx.CODE_EXTS.get(src_path.suffix.lower(), "unknown") != "unknown":
                     paths_to_process.append(src_path)
 
                 # For FileMovedEvent, also process the destination path
                 if hasattr(event, 'dest_path') and event.dest_path:
                     dest_path = Path(event.dest_path)
-                    if idx.CODE_EXTS.get(dest_path.suffix.lower(), "unknown") != "unknown":
+                    if not self.client._is_ignored_path(dest_path) and idx.CODE_EXTS.get(dest_path.suffix.lower(), "unknown") != "unknown":
                         paths_to_process.append(dest_path)
 
                 if not paths_to_process:
@@ -1413,9 +1463,9 @@ def _process_pending_changes(self):
                             self.client.workspace_path,
                             self.client.repo_name
                         )
-                        all_paths = list(set(pending + [
-                            Path(p) for p in cached_file_hashes.keys()
-                        ]))
+                        cached_paths = [Path(p) for p in cached_file_hashes.keys()]
+                        cached_paths = [p for p in cached_paths if not self.client._is_ignored_path(p)]
+                        all_paths = list(set(pending + cached_paths))
                     else:
                         all_paths = pending
 
diff --git a/scripts/standalone_upload_client.py b/scripts/standalone_upload_client.py
index 7cbd9dd1..62982e91 100644
--- a/scripts/standalone_upload_client.py
+++ b/scripts/standalone_upload_client.py
@@ -730,6 +730,47 @@ def log_mapping_summary(self) -> None:
         logger.info(f"  source_path: {info['source_path']}")
         logger.info(f"  container_path: {info['container_path']}")
 
+    def _excluded_dirnames(self) -> frozenset:
+        # Keep in sync with get_all_code_files exclusions.
+        # NOTE: This caches the exclusion set per client instance.
+        # Runtime changes to DEV_REMOTE_MODE/REMOTE_UPLOAD_MODE won't be reflected
+        # until a new client is created (typically via process restart), which is
+        # acceptable for the standalone upload client use case.
+        cached = getattr(self, "_excluded_dirnames_cache", None)
+        if cached is not None:
+            return cached
+        excluded = {
+            "node_modules", "vendor", "dist", "build", "target", "out",
+            ".git", ".hg", ".svn", ".vscode", ".idea", ".venv", "venv",
+            "__pycache__", ".pytest_cache", ".mypy_cache", ".cache",
+            ".context-engine", ".context-engine-uploader", ".codebase",
+        }
+        dev_remote = os.environ.get("DEV_REMOTE_MODE") == "1" or os.environ.get("REMOTE_UPLOAD_MODE") == "development"
+        if dev_remote:
+            excluded.add("dev-workspace")
+        cached = frozenset(excluded)
+        self._excluded_dirnames_cache = cached
+        return cached
+
+    def _is_ignored_path(self, path: Path) -> bool:
+        """Return True when path is outside workspace or under excluded dirs."""
+        try:
+            workspace_root = Path(self.workspace_path).resolve()
+            rel = path.resolve().relative_to(workspace_root)
+        except Exception:
+            return True
+
+        dir_parts = set(rel.parts[:-1]) if len(rel.parts) > 1 else set()
+        if dir_parts & self._excluded_dirnames():
+            return True
+        # Ignore hidden directories anywhere under the workspace, but allow
+        # extensionless dotfiles like `.gitignore` that we explicitly support.
+        if any(p.startswith(".") for p in rel.parts[:-1]):
+            return True
+        if rel.name.startswith(".") and rel.name.lower() not in EXTENSIONLESS_FILES:
+            return True
+        return False
+
     def _get_temp_bundle_dir(self) -> Path:
         """Get or create temporary directory for bundle creation."""
         if not self.temp_dir:
@@ -757,6 +798,8 @@ def detect_file_changes(self, changed_paths: List[Path]) -> Dict[str, List]:
         }
 
         for path in changed_paths:
+            if self._is_ignored_path(path):
+                continue
             try:
                 abs_path = str(path.resolve())
             except Exception:
@@ -1532,13 +1575,13 @@ def on_any_event(self, event):
 
                 # Always check src_path
                 src_path = Path(event.src_path)
-                if detect_language(src_path) != "unknown":
+                if not self.client._is_ignored_path(src_path) and detect_language(src_path) != "unknown":
                     paths_to_process.append(src_path)
 
                 # For FileMovedEvent, also process the destination path
                 if hasattr(event, 'dest_path') and event.dest_path:
                     dest_path = Path(event.dest_path)
-                    if detect_language(dest_path) != "unknown":
+                    if not self.client._is_ignored_path(dest_path) and detect_language(dest_path) != "unknown":
                         paths_to_process.append(dest_path)
 
                 if not paths_to_process:
@@ -1569,9 +1612,11 @@ def _process_pending_changes(self):
                 try:
                     # Only include cached paths when deletion-related events occurred
                     if check_deletions:
-                        all_paths = list(set(pending + [
+                        cached_paths = [
                             Path(p) for p in get_all_cached_paths(self.client.repo_name)
-                        ]))
+                        ]
+                        cached_paths = [p for p in cached_paths if not self.client._is_ignored_path(p)]
+                        all_paths = list(set(pending + cached_paths))
                     else:
                         all_paths = pending
                     
@@ -1719,16 +1764,10 @@ def get_all_code_files(self) -> List[Path]:
 
             # Single walk with early pruning and set-based matching to reduce IO
             ext_suffixes = {str(ext).lower() for ext in CODE_EXTS if str(ext).startswith('.')}
-            extensionless_names = set(EXTENSIONLESS_FILES.keys())
+            extensionless_names = {k.lower() for k in EXTENSIONLESS_FILES.keys()}
             # Always exclude dev-workspace to prevent recursive upload loops
             # (upload service creates dev-workspace/<collection>/ which would otherwise get re-uploaded)
-            excluded = {
-                "node_modules", "vendor", "dist", "build", "target", "out",
-                ".git", ".hg", ".svn", ".vscode", ".idea", ".venv", "venv",
-                "__pycache__", ".pytest_cache", ".mypy_cache", ".cache",
-                ".context-engine", ".context-engine-uploader", ".codebase",
-                "dev-workspace"
-            }
+            excluded = self._excluded_dirnames()
 
             seen = set()
             for root, dirnames, filenames in os.walk(workspace_path):
@@ -1741,6 +1780,8 @@ def get_all_code_files(self) -> List[Path]:
                     if filename.startswith('.') and fname_lower not in extensionless_names:
                         continue
                     candidate = Path(root) / filename
+                    if self._is_ignored_path(candidate):
+                        continue
                     suffix = candidate.suffix.lower()
                     # Match by extension, extensionless name, or Dockerfile.* prefix
                     if (suffix in ext_suffixes or
diff --git a/scripts/upload_service.py b/scripts/upload_service.py
index 6771d652..9acd9931 100644
--- a/scripts/upload_service.py
+++ b/scripts/upload_service.py
@@ -86,6 +86,10 @@
 except Exception:
     delete_collection_everywhere = None
     copy_collection_qdrant = None
+try:
+    from scripts.qdrant_client_manager import pooled_qdrant_client
+except Exception:
+    pooled_qdrant_client = None
 try:
     from scripts.admin_ui import (
         render_admin_acl,
@@ -935,7 +939,7 @@ async def admin_delete_collection(
         cleanup_fs = False
 
     try:
-        delete_collection_everywhere(
+        out = delete_collection_everywhere(
             collection=name,
             work_dir=WORK_DIR,
             qdrant_url=QDRANT_URL,
@@ -949,7 +953,23 @@ async def admin_delete_collection(
             back_href="/admin/acl",
         )
 
-    return RedirectResponse(url="/admin/acl", status_code=302)
+    graph_deleted: Optional[str] = None
+    try:
+        if isinstance(out, dict) and not name.endswith("_graph"):
+            graph_deleted = "1" if bool(out.get("qdrant_graph_deleted")) else "0"
+    except Exception:
+        graph_deleted = None
+
+    try:
+        from urllib.parse import urlencode
+
+        params = {"deleted": name}
+        if graph_deleted is not None:
+            params["graph_deleted"] = graph_deleted
+        url = "/admin/acl?" + urlencode(params)
+    except Exception:
+        url = "/admin/acl"
+    return RedirectResponse(url=url, status_code=302)
 
 
 @app.post("/admin/staging/start")
@@ -1222,7 +1242,59 @@ async def admin_copy_collection(
             back_href="/admin/acl",
         )
 
-    return RedirectResponse(url="/admin/acl", status_code=302)
+    graph_copied: Optional[str] = None
+    try:
+        if not name.endswith("_graph") and not str(new_name).endswith("_graph"):
+            used_pooled = False
+            if pooled_qdrant_client is not None:
+                used_pooled = True
+                try:
+                    with pooled_qdrant_client(
+                        url=QDRANT_URL,
+                        api_key=os.environ.get("QDRANT_API_KEY"),
+                    ) as cli:
+                        try:
+                            cli.get_collection(collection_name=f"{new_name}_graph")
+                            graph_copied = "1"
+                        except Exception:
+                            graph_copied = "0"
+                except Exception:
+                    # Failed to acquire pooled client; fall back to non-pooled
+                    used_pooled = False
+            if not used_pooled:
+                try:
+                    from qdrant_client import QdrantClient  # type: ignore
+
+                    cli = QdrantClient(
+                        url=QDRANT_URL,
+                        api_key=os.environ.get("QDRANT_API_KEY"),
+                        timeout=float(os.environ.get("QDRANT_TIMEOUT", "5") or 5),
+                    )
+                    try:
+                        cli.get_collection(collection_name=f"{new_name}_graph")
+                        graph_copied = "1"
+                    except Exception:
+                        graph_copied = "0"
+                    finally:
+                        try:
+                            cli.close()
+                        except Exception:
+                            pass
+                except Exception:
+                    graph_copied = "0"
+    except Exception:
+        graph_copied = None
+
+    try:
+        from urllib.parse import urlencode
+
+        params = {"copied": name, "new": new_name}
+        if graph_copied is not None:
+            params["graph_copied"] = graph_copied
+        url = "/admin/acl?" + urlencode(params)
+    except Exception:
+        url = "/admin/acl"
+    return RedirectResponse(url=url, status_code=302)
 
 
 @app.post("/admin/users")
diff --git a/scripts/watch_index_core/processor.py b/scripts/watch_index_core/processor.py
index 45e9db7e..6f529cf7 100644
--- a/scripts/watch_index_core/processor.py
+++ b/scripts/watch_index_core/processor.py
@@ -251,6 +251,15 @@ def _process_paths(
             if client is not None:
                 try:
                     idx.delete_points_by_path(client, collection, str(p))
+                    try:
+                        idx.delete_graph_edges_by_path(
+                            client,
+                            collection,
+                            caller_path=str(p),
+                            repo=repo_name,
+                        )
+                    except Exception:
+                        pass
                     safe_print(f"[deleted] {p} -> {collection}")
                 except Exception:
                     pass
diff --git a/scripts/watch_index_core/pseudo.py b/scripts/watch_index_core/pseudo.py
index dc7bb0a8..33fb6f46 100644
--- a/scripts/watch_index_core/pseudo.py
+++ b/scripts/watch_index_core/pseudo.py
@@ -49,12 +49,21 @@ def _start_pseudo_backfill_worker(
         max_points = 256
     if max_points <= 0:
         max_points = 1
+    try:
+        graph_max_files = int(
+            os.environ.get("GRAPH_EDGES_BACKFILL_MAX_FILES", "128") or 128
+        )
+    except Exception:
+        graph_max_files = 128
+    if graph_max_files <= 0:
+        graph_max_files = 1
 
     shutdown_event = threading.Event()
 
     def _worker() -> None:
         while not shutdown_event.is_set():
             try:
+                graph_backfill_enabled = get_boolean_env("GRAPH_EDGES_BACKFILL")
                 try:
                     mappings = get_collection_mappings(search_root=str(ROOT))
                 except Exception:
@@ -90,6 +99,34 @@ def _worker() -> None:
                                     "[pseudo_backfill] repo=%s collection=%s processed=%d",
                                     repo_name or "default", coll, processed,
                                 )
+                        # Optional: backfill graph edge collection from main points.
+                        # Controlled separately because it may scan large collections over time.
+                        # Run under its own lock to avoid blocking pseudo/tag backfill workers.
+                        if graph_backfill_enabled:
+                            try:
+                                graph_lock_path = state_dir / "graph_edges.lock"
+                                with _cross_process_lock(graph_lock_path):
+                                    files_done = idx.graph_edges_backfill_tick(
+                                        client,
+                                        coll,
+                                        repo_name=repo_name,
+                                        max_files=graph_max_files,
+                                    )
+                                if files_done:
+                                    logger.info(
+                                        "[graph_backfill] repo=%s collection=%s files=%d",
+                                        repo_name or "default",
+                                        coll,
+                                        files_done,
+                                    )
+                            except Exception as exc:
+                                logger.error(
+                                    "[graph_backfill] error repo=%s collection=%s: %s",
+                                    repo_name or "default",
+                                    coll,
+                                    exc,
+                                    exc_info=True,
+                                )
                     except Exception as exc:
                         logger.error(
                             "[pseudo_backfill] error repo=%s collection=%s: %s",
@@ -110,4 +147,3 @@ def _worker() -> None:
 
 
 __all__ = ["_start_pseudo_backfill_worker"]
-
diff --git a/templates/admin/acl.html b/templates/admin/acl.html
index 952a0ce9..98292beb 100644
--- a/templates/admin/acl.html
+++ b/templates/admin/acl.html
@@ -1,6 +1,32 @@
 {% extends "admin/base.html" %}
 
 {% block content %}
+  {% set qp = request.query_params %}
+  {% if qp and ((qp.get("copied") and qp.get("new")) or qp.get("deleted")) %}
+    <div class="card">
+      {% if qp.get("copied") and qp.get("new") %}
+        <div>
+          Copied collection <code>{{ qp.get("copied") }}</code> → <code>{{ qp.get("new") }}</code>.
+          {% if qp.get("graph_copied") == "1" %}
+            <span class="muted">(graph clone copied)</span>
+          {% elif qp.get("graph_copied") == "0" %}
+            <span class="muted">(graph clone not copied; will rebuild/backfill)</span>
+          {% endif %}
+        </div>
+      {% endif %}
+      {% if qp.get("deleted") %}
+        <div>
+          Deleted collection <code>{{ qp.get("deleted") }}</code>.
+          {% if qp.get("graph_deleted") == "1" %}
+            <span class="muted">(graph clone deleted)</span>
+          {% elif qp.get("graph_deleted") == "0" %}
+            <span class="muted">(graph clone not deleted or missing)</span>
+          {% endif %}
+        </div>
+      {% endif %}
+    </div>
+  {% endif %}
+
   <div class="card" data-staging-enabled="{{ '1' if staging_enabled else '0' }}">
     <h2>Users</h2>
 
diff --git a/tests/test_admin_collection_delete.py b/tests/test_admin_collection_delete.py
index ad42807e..c407b5da 100644
--- a/tests/test_admin_collection_delete.py
+++ b/tests/test_admin_collection_delete.py
@@ -52,6 +52,29 @@ def test_admin_role_gate_blocks_non_admin(monkeypatch):
     assert resp.json().get("detail") == "Admin required"
 
 
+@pytest.mark.unit
+def test_delete_redirect_includes_graph_deleted_param(monkeypatch):
+    monkeypatch.setenv("CTXCE_AUTH_ENABLED", "1")
+    monkeypatch.setenv("CTXCE_ADMIN_COLLECTION_DELETE_ENABLED", "1")
+
+    srv = importlib.import_module("scripts.upload_service")
+    srv = importlib.reload(srv)
+
+    monkeypatch.setattr(srv, "_require_admin_session", lambda _req: {"user_id": "admin"})
+
+    def _fake_delete_collection_everywhere(**_kwargs):
+        return {"qdrant_deleted": True, "qdrant_graph_deleted": True}
+
+    monkeypatch.setattr(srv, "delete_collection_everywhere", _fake_delete_collection_everywhere)
+
+    client = TestClient(srv.app)
+    resp = client.post("/admin/collections/delete", data={"collection": "c1", "delete_fs": ""}, follow_redirects=False)
+    assert resp.status_code == 302
+    loc = resp.headers.get("location") or ""
+    assert "deleted=c1" in loc
+    assert "graph_deleted=1" in loc
+
+
 @pytest.mark.unit
 def test_collection_admin_refuses_when_env_disabled(monkeypatch):
     monkeypatch.setenv("CTXCE_ADMIN_COLLECTION_DELETE_ENABLED", "0")
diff --git a/tests/test_ingest_cli.py b/tests/test_ingest_cli.py
new file mode 100644
index 00000000..493d1a68
--- /dev/null
+++ b/tests/test_ingest_cli.py
@@ -0,0 +1,57 @@
+import sys
+from pathlib import Path
+
+import pytest
+
+
+@pytest.mark.unit
+def test_cli_force_collection_disables_multi_repo_enumeration(monkeypatch, tmp_path: Path):
+    from scripts.ingest import cli
+
+    # Create fake repo dirs to prove we are not enumerating them.
+    (tmp_path / "repo_a").mkdir()
+    (tmp_path / "repo_b").mkdir()
+
+    calls = []
+
+    def _fake_index_repo(
+        root,
+        qdrant_url,
+        api_key,
+        collection,
+        model_name,
+        recreate,
+        dedupe,
+        skip_unchanged,
+        pseudo_mode,
+        schema_mode,
+    ):
+        calls.append(
+            {
+                "root": Path(root),
+                "collection": collection,
+                "recreate": recreate,
+                "dedupe": dedupe,
+                "skip_unchanged": skip_unchanged,
+            }
+        )
+
+    monkeypatch.setattr(cli, "index_repo", _fake_index_repo)
+    monkeypatch.setattr(cli, "is_multi_repo_mode", lambda: True)
+    monkeypatch.setattr(cli, "get_collection_name", lambda *_: "should-not-use")
+
+    monkeypatch.setenv("MULTI_REPO_MODE", "1")
+    monkeypatch.setenv("COLLECTION_NAME", "forced-collection")
+    monkeypatch.setenv("CTXCE_FORCE_COLLECTION_NAME", "1")
+
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["ingest_code.py", "--root", str(tmp_path)],
+    )
+
+    cli.main()
+
+    assert len(calls) == 1
+    assert calls[0]["root"] == tmp_path
+    assert calls[0]["collection"] == "forced-collection"
diff --git a/tests/test_staging_lifecycle.py b/tests/test_staging_lifecycle.py
index 01734e32..eb8a93cf 100644
--- a/tests/test_staging_lifecycle.py
+++ b/tests/test_staging_lifecycle.py
@@ -542,6 +542,54 @@ def fake_abort(**kwargs):
     assert calls["abort"] == 1
 
 
+def test_admin_copy_endpoint_reports_graph_clone_in_redirect(monkeypatch: pytest.MonkeyPatch):
+    import sys
+    import types
+    from urllib.parse import parse_qs, urlparse
+
+    from scripts import upload_service
+
+    monkeypatch.setattr(upload_service, "AUTH_ENABLED", True)
+    monkeypatch.setattr(upload_service, "_require_admin_session", lambda request: {"user_id": "admin"})
+    monkeypatch.setattr(upload_service, "WORK_DIR", "/fake/work")
+    monkeypatch.setenv("WORK_DIR", "/fake/work")
+
+    def fake_copy_collection_qdrant(**kwargs):
+        assert kwargs.get("source") == "src"
+        assert kwargs.get("target") == "dst"
+        return "dst"
+
+    monkeypatch.setattr(upload_service, "copy_collection_qdrant", fake_copy_collection_qdrant)
+
+    class _FakeQdrantClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def get_collection(self, collection_name: str):
+            if collection_name == "dst_graph":
+                return {"name": collection_name}
+            raise RuntimeError("not found")
+
+        def close(self):
+            return None
+
+    monkeypatch.setitem(sys.modules, "qdrant_client", types.SimpleNamespace(QdrantClient=_FakeQdrantClient))
+
+    client = TestClient(upload_service.app)
+    resp = client.post(
+        "/admin/staging/copy",
+        data={"collection": "src", "target": "dst", "overwrite": ""},
+        follow_redirects=False,
+    )
+    assert resp.status_code == 302
+    loc = resp.headers.get("location") or ""
+    parsed = urlparse(loc)
+    qs = parse_qs(parsed.query)
+    assert qs.get("copied") == ["src"]
+    assert qs.get("new") == ["dst"]
+    assert qs.get("graph_copied") == ["1"]
+
+
 def test_watcher_collection_resolution_prefers_serving_state_when_staging_enabled(monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
     from scripts.watch_index_core import utils as watch_utils
 
@@ -650,7 +698,9 @@ class _Proc:
     env = captured["env"]
     assert env["BASE_ONLY"] == "system"
     assert env["COLLECTION_NAME"] == "primary-coll"
-    assert "CTXCE_FORCE_COLLECTION_NAME" not in env
+    # Admin-spawned ingests should never enumerate `/work/*` in multi-repo mode;
+    # force exact collection/root handling even when no explicit overrides are provided.
+    assert env.get("CTXCE_FORCE_COLLECTION_NAME") == "1"
 
 
 def test_promote_pending_env_without_pending_config(staging_workspace: dict):
diff --git a/vscode-extension/build/build.sh b/vscode-extension/build/build.sh
index f3e4d9fa..6607ac68 100755
--- a/vscode-extension/build/build.sh
+++ b/vscode-extension/build/build.sh
@@ -76,6 +76,27 @@ if [[ "$BUNDLE_DEPS" == "--bundle-deps" ]]; then
     fi
 fi
 
+# Bundle MCP bridge npm package into the staged extension
+BRIDGE_SRC="$SCRIPT_DIR/../../ctx-mcp-bridge"
+BRIDGE_DIR="ctx-mcp-bridge"
+
+if [[ -d "$BRIDGE_SRC" && -f "$BRIDGE_SRC/package.json" ]]; then
+    echo "Bundling MCP bridge npm package into staged extension..."
+    mkdir -p "$STAGE_DIR/$BRIDGE_DIR"
+    cp -a "$BRIDGE_SRC/bin" "$STAGE_DIR/$BRIDGE_DIR/"
+    cp -a "$BRIDGE_SRC/src" "$STAGE_DIR/$BRIDGE_DIR/"
+    cp "$BRIDGE_SRC/package.json" "$STAGE_DIR/$BRIDGE_DIR/"
+
+    if [[ -d "$BRIDGE_SRC/node_modules" ]]; then
+        cp -a "$BRIDGE_SRC/node_modules" "$STAGE_DIR/$BRIDGE_DIR/"
+    else
+        echo "Warning: Bridge node_modules not found. Run 'npm install' in ctx-mcp-bridge first."
+    fi
+    echo "MCP bridge bundled successfully."
+else
+    echo "Warning: MCP bridge source not found at $BRIDGE_SRC"
+fi
+
 pushd "$STAGE_DIR" >/dev/null
 echo "Packaging extension..."
 npx @vscode/vsce package --no-dependencies --out "$OUT_DIR"
diff --git a/vscode-extension/context-engine-uploader/extension.js b/vscode-extension/context-engine-uploader/extension.js
index 9a387c66..217d14e7 100644
--- a/vscode-extension/context-engine-uploader/extension.js
+++ b/vscode-extension/context-engine-uploader/extension.js
@@ -230,6 +230,7 @@ function activate(context) {
       path,
       fs,
       log,
+      extensionRoot,
       getEffectiveConfig,
       resolveBridgeWorkspacePath: () => configResolver ? configResolver.resolveBridgeWorkspacePath() : undefined,
       attachOutput: (child, label) => processManager ? processManager.attachOutput(child, label) : undefined,
@@ -425,6 +426,7 @@ function activate(context) {
       event.affectsConfiguration('contextEngineUploader.mcpBridgeBinPath') ||
       event.affectsConfiguration('contextEngineUploader.mcpBridgePort') ||
       event.affectsConfiguration('contextEngineUploader.mcpBridgeLocalOnly') ||
+      event.affectsConfiguration('contextEngineUploader.mcpBridgeMode') ||
       event.affectsConfiguration('contextEngineUploader.windsurfMcpPath') ||
       event.affectsConfiguration('contextEngineUploader.augmentMcpPath') ||
       event.affectsConfiguration('contextEngineUploader.antigravityMcpPath') ||
@@ -439,6 +441,7 @@ function activate(context) {
       event.affectsConfiguration('contextEngineUploader.mcpBridgePort') ||
       event.affectsConfiguration('contextEngineUploader.mcpBridgeBinPath') ||
       event.affectsConfiguration('contextEngineUploader.mcpBridgeLocalOnly') ||
+      event.affectsConfiguration('contextEngineUploader.mcpBridgeMode') ||
       event.affectsConfiguration('contextEngineUploader.mcpIndexerUrl') ||
       event.affectsConfiguration('contextEngineUploader.mcpMemoryUrl') ||
       event.affectsConfiguration('contextEngineUploader.mcpServerMode') ||
diff --git a/vscode-extension/context-engine-uploader/mcp_bridge.js b/vscode-extension/context-engine-uploader/mcp_bridge.js
index d9825177..b7c1b441 100644
--- a/vscode-extension/context-engine-uploader/mcp_bridge.js
+++ b/vscode-extension/context-engine-uploader/mcp_bridge.js
@@ -4,6 +4,7 @@ function createBridgeManager(deps) {
   const path = deps.path;
   const fs = deps.fs;
   const log = deps.log;
+  const extensionRoot = deps.extensionRoot;
 
   const getEffectiveConfig = deps.getEffectiveConfig;
   const resolveBridgeWorkspacePath = deps.resolveBridgeWorkspacePath;
@@ -42,7 +43,36 @@ function createBridgeManager(deps) {
     }
   }
 
+  function getBridgeMode() {
+    try {
+      const settings = getEffectiveConfig();
+      return (settings.get('mcpBridgeMode') || 'bundled').trim();
+    } catch (_) {
+      return 'bundled';
+    }
+  }
+
+  function findBundledBridgeBin() {
+    if (!extensionRoot) return undefined;
+    const bundledPath = path.join(extensionRoot, 'ctx-mcp-bridge', 'bin', 'ctxce.js');
+    if (fs.existsSync(bundledPath)) {
+      return path.resolve(bundledPath);
+    }
+    return undefined;
+  }
+
   function findLocalBridgeBin() {
+    // First check for bundled bridge if mode is 'bundled'
+    const mode = getBridgeMode();
+    if (mode === 'bundled') {
+      const bundledBin = findBundledBridgeBin();
+      if (bundledBin) {
+        return bundledBin;
+      }
+      log('Bundled bridge requested but not found; falling back to external resolution');
+    }
+
+    // External mode logic (existing behavior)
     let localOnly = true;
     let configured = '';
     try {
@@ -68,11 +98,12 @@ function createBridgeManager(deps) {
 
   function resolveBridgeCliInvocation() {
     const binPath = findLocalBridgeBin();
+    const mode = getBridgeMode();
     if (binPath) {
       return {
         command: 'node',
         args: [binPath],
-        kind: 'local'
+        kind: mode === 'bundled' ? 'bundled' : 'local'
       };
     }
     const isWindows = process.platform === 'win32';
diff --git a/vscode-extension/context-engine-uploader/package.json b/vscode-extension/context-engine-uploader/package.json
index d5e3584f..77b86261 100644
--- a/vscode-extension/context-engine-uploader/package.json
+++ b/vscode-extension/context-engine-uploader/package.json
@@ -297,7 +297,17 @@
         "contextEngineUploader.mcpBridgeLocalOnly": {
           "type": "boolean",
           "default": false,
-          "description": "Development toggle. When true (default) the extension prefers local bridge binaries resolved from mcpBridgeBinPath or CTXCE_BRIDGE_BIN before falling back to the published npm build via npx."
+          "description": "Development toggle. When true and mcpBridgeMode='external', prefers local bridge binaries resolved from mcpBridgeBinPath or CTXCE_BRIDGE_BIN before falling back to the published npm build via npx. Ignored when mcpBridgeMode='bundled'."
+        },
+        "contextEngineUploader.mcpBridgeMode": {
+          "type": "string",
+          "enum": ["bundled", "external"],
+          "default": "bundled",
+          "description": "Bridge invocation mode. 'bundled' uses the bundled bridge inside the extension (offline, no npx required). 'external' uses external binary path or npx (current behavior).",
+          "enumDescriptions": [
+            "Use the bundled MCP bridge inside the extension (works offline).",
+            "Use external binary path or npx to run the bridge (requires internet for first npx install)."
+          ]
         },
         "contextEngineUploader.mcpServerMode": {
           "type": "string",
diff --git a/vscode-extension/context-engine-uploader/python_env.js b/vscode-extension/context-engine-uploader/python_env.js
index 190f9945..9e5e7fd5 100644
--- a/vscode-extension/context-engine-uploader/python_env.js
+++ b/vscode-extension/context-engine-uploader/python_env.js
@@ -339,10 +339,7 @@ function createPythonEnvManager(deps) {
         }
 
         // As a last resort, offer to create a private venv and install deps via pip
-        if (!allowPrompt) {
-            log('Skipping auto-install prompt; interpreter was auto-detected and missing modules.');
-            return false;
-        }
+        // Always prompt at this point - we've exhausted all other options (initial Python + auto-detected both failed)
         const choice = await vscode.window.showErrorMessage(
             'Context Engine Uploader: missing Python modules. Create isolated environment and auto-install?',
             'Auto-install to private venv',