Implement graph edges collection support with admin and ingest updates

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 && template.resourceTemplate && typeof template.resourceTemplate.uriTemplate === "string"
+        ? template.resourceTemplate.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;
 }
-

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