Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion REQUIREMENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

> **Note:** This document is automatically generated and verified against the live test suite by `scripts/generate_requirements.py` and `tests/backend/test_requirements_sync.py`.

**Test Verification Baseline:** **1216 Automated Tests** (852 Pytest Backend + 314 Vitest Frontend + 50 Playwright E2E).
**Test Verification Baseline:** **1224 Automated Tests** (860 Pytest Backend + 314 Vitest Frontend + 50 Playwright E2E).

---

Expand Down Expand Up @@ -1398,6 +1398,16 @@ over the PostgresConnectionAdapter using an initialized test engine._
- `test_editor_role_permissions` - _Editor key can perform content mutations and syncs, but is denied administrative settings (403)._
- `test_admin_role_permissions` - _Admin key has full permissions across configuration, editor, and viewer endpoints._

#### `tests/test_route_specificity.py` (8 tests)
- `test_literal_segment_beats_parameter`
- `test_match_route_and_call_still_matches_the_general_route`
- `test_call_goes_to_most_specific_route`
- `test_method_and_repository_are_resolved_separately`
- `test_find_api_callers_general_route_skips_calls_of_specific_routes`
- `test_find_api_callers_specific_route_keeps_its_calls`
- `test_find_api_callers_without_known_route_keeps_every_match`
- `test_trace_path_external_callers_and_handlers_use_most_specific_route`

#### `tests/test_storage_api_routes.py` (4 tests)
- `test_storage_upload_and_get_file`
- `test_storage_upload_multipart_and_put`
Expand Down
45 changes: 36 additions & 9 deletions app/mcp/handlers/route_handlers.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,17 @@
import json
import logging
import sys
from typing import List, Dict, Any, Optional, Annotated
from typing import List, Dict, Any, Optional, Annotated, Callable
from pydantic import Field

from app.services.database import get_db_connection
from app.services.git_manager import format_git_permalink
from app.services.chunking import match_route_and_call
from app.services.chunking.route_specificity import (
is_test_route_file as _is_test_route_file,
most_specific_routes,
route_key,
)
from app.services.source_paths import LocalSourceRoots

logger = logging.getLogger("contextcortex")
Expand All @@ -17,14 +22,32 @@ def _get_tools_attr(name, default):
return getattr(t_mod, name, default) if t_mod else default


def _is_test_route_file(rel_path: str) -> bool:
path_parts = [p.lower() for p in rel_path.strip("/").split("/") if p]
bname = path_parts[-1] if path_parts else ""
return (
any(p in ("tests", "test", "__tests__") for p in path_parts[:-1])
or bname.startswith("test_")
or bname.endswith(("_test.py", ".test.ts", ".spec.ts", "tests.cs"))
)
def _target_ownership(target_path: str, target_method: Optional[str], route_rows) -> Callable[[str, str], bool]:
"""
Builds a check that tells if a client call belongs to the target path, or to a more specific known route.
The target path is resolved to its own route(s) first: in each repository, the most specific route that matches
it. A call that matches the target is then kept only when one of those routes is also the most specific route
for the call in its repository (so `assistants/{id}` does not take the calls of `assistants/details`).
When no known route matches the target, every matching call is kept.
"""
routes = []
for r in route_rows:
rel_path = (r["filepath"].split("://", 1)[1] if "://" in r["filepath"] else r["filepath"]).replace("\\", "/")
if not _is_test_route_file(rel_path):
routes.append({"repo": r["repo"], "http_method": r["http_method"], "path_pattern": r["path_pattern"]})

target_routes = most_specific_routes(target_path, target_method, routes)
if not target_routes:
return lambda _url, _method: True
target_keys = {route_key(r) for r in target_routes}
target_repos = {r["repo"] for r in target_routes}
competing = [r for r in routes if r["repo"] in target_repos]

def belongs(call_url: str, call_method: str) -> bool:
owners = most_specific_routes(call_url, call_method or None, competing)
return not owners or any(route_key(o) in target_keys for o in owners)

return belongs


async def handle_find_routes(
Expand Down Expand Up @@ -148,9 +171,11 @@ async def handle_find_api_callers(

sql += " ORDER BY c.repo ASC, c.filepath ASC, c.line_number ASC"
rows = conn.execute(sql, params).fetchall()
route_rows = conn.execute("SELECT repo, filepath, http_method, path_pattern FROM api_routes").fetchall()
local_roots = LocalSourceRoots.load(conn)

target_method_clean = method.strip().upper() if method and method.strip() else None
belongs_to_target = _target_ownership(target_path, target_method_clean, route_rows)
all_callers = []
seen_callers = set()

Expand All @@ -161,6 +186,8 @@ async def handle_find_api_callers(

if not match_route_and_call(target_path, c["url_pattern"]):
continue
if not belongs_to_target(c["url_pattern"], call_method):
continue

rel_path = local_roots.relative(c["filepath"], c["repo"])
dedup_key = (c["repo"], rel_path, c["http_method"], c["url_pattern"], c["caller_symbol"], c["line_number"])
Expand Down
115 changes: 115 additions & 0 deletions app/services/chunking/route_specificity.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
"""
Route specificity: give each client call to the most specific route that matches it.

A route parameter matches any segment, so `GET api/v{version}/assistants/{id}` also matches the call
`GET /api/v2/Assistants/details`. When the route `GET api/v{version}/assistants/details` exists in the same
repository, ASP.NET (and most other frameworks) send the call to that route: a literal segment beats a parameter.
These functions apply the same rule. `match_route_and_call` stays a plain yes/no matcher.
"""

import re
from functools import lru_cache
from typing import Any, Dict, Iterable, List, Optional, Tuple

from app.services.chunking.api_route_extractor import match_route_and_call, normalize_path_pattern

_PARAM_ONLY = re.compile(r'^\{[^{}/]+\}$')

# Segment ranks: a literal segment beats a mixed segment (`v{version}`), which beats a parameter (`{id}`).
_RANK_PARAM, _RANK_MIXED, _RANK_LITERAL = 0, 1, 2


@lru_cache(maxsize=4096)
def segment_ranks(path: str) -> Tuple[int, ...]:
"""Returns the rank of each segment of a route path, from left to right."""
norm = normalize_path_pattern(re.sub(r'^https?://[^/]+', '', (path or '').strip("'\"` ")) or "/")
segs = [s for s in norm.split("?", 1)[0].strip("/").split("/") if s]
return tuple(
_RANK_PARAM if _PARAM_ONLY.match(s) else _RANK_MIXED if "{" in s else _RANK_LITERAL
for s in segs
)


def compare_specificity(path_a: str, path_b: str) -> int:
"""
Returns 1 when path_a is more specific than path_b, -1 when it is less specific, and 0 for a tie.
Paths with the same number of segments are compared segment by segment from the left (the first segment where
one path has a literal and the other a parameter decides). Paths of different length (a base-relative call
can match both) are compared by their total literal weight; equal weight is a tie.
"""
ranks_a, ranks_b = segment_ranks(path_a), segment_ranks(path_b)
if len(ranks_a) == len(ranks_b):
return (ranks_a > ranks_b) - (ranks_a < ranks_b)
weight_a, weight_b = sum(ranks_a), sum(ranks_b)
return (weight_a > weight_b) - (weight_a < weight_b)


def is_test_route_file(rel_path: str) -> bool:
"""True for a route declared in a test file or test folder (such routes never take calls from real routes)."""
path_parts = [p.lower() for p in (rel_path or "").replace("\\", "/").strip("/").split("/") if p]
bname = path_parts[-1] if path_parts else ""
return (
any(p in ("tests", "test", "__tests__") for p in path_parts[:-1])
or bname.startswith("test_")
or bname.endswith(("_test.py", ".test.ts", ".spec.ts", "tests.cs"))
)


def methods_compatible(m1: Optional[str], m2: Optional[str]) -> bool:
"""An empty method or `ALL` is compatible with every method."""
if not m1 or not m2:
return True
u1, u2 = m1.upper(), m2.upper()
return u1 == "ALL" or u2 == "ALL" or u1 == u2


def route_key(route: Dict[str, Any]) -> Tuple[str, str, str]:
"""Identity of a route: repository, HTTP method, and normalized path."""
return (
route.get("repo") or "",
(route.get("http_method") or "").upper(),
normalize_path_pattern(route.get("path_pattern") or "").lower(),
)


def most_specific_routes(
call_url: str, call_method: Optional[str], routes: Iterable[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""
Returns the routes that own a call: the routes that match the call (path and method) and that no other matching
route of the same repository beats on specificity. Each repository is resolved on its own, so a route in an
unrelated repository never takes a call away from another repository's route. Ties keep all tied routes.
"""
matches: Dict[str, List[Dict[str, Any]]] = {}
for r in routes:
if methods_compatible(r.get("http_method"), call_method) and match_route_and_call(r.get("path_pattern") or "", call_url):
matches.setdefault(r.get("repo") or "", []).append(r)
owners: List[Dict[str, Any]] = []
for repo_matches in matches.values():
for r in repo_matches:
path = r.get("path_pattern") or ""
if not any(compare_specificity(o.get("path_pattern") or "", path) > 0 for o in repo_matches):
owners.append(r)
return owners


def call_belongs_to_route(
route: Dict[str, Any], call_url: str, call_method: Optional[str], routes: Iterable[Dict[str, Any]]
) -> bool:
"""
True when the call matches `route` and no other route of the route's repository (with a compatible method)
matches the call with more specificity. `routes` is the known route set; it can include `route` itself.
"""
if not methods_compatible(route.get("http_method"), call_method):
return False
path = route.get("path_pattern") or ""
if not match_route_and_call(path, call_url):
return False
repo = route.get("repo") or ""
for other in routes:
if (other.get("repo") or "") != repo or not methods_compatible(other.get("http_method"), call_method):
continue
other_path = other.get("path_pattern") or ""
if compare_specificity(other_path, path) > 0 and match_route_and_call(other_path, call_url):
return False
return True
24 changes: 18 additions & 6 deletions app/services/graph_traversal.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from collections import deque
from typing import Dict, List, Optional, Set, Tuple

from app.services.chunking import match_route_and_call
from app.services.chunking.route_specificity import call_belongs_to_route, is_test_route_file
from app.services.database import get_db_connection
from app.services.git_manager import format_git_permalink
from app.services.source_paths import LocalSourceRoots
Expand Down Expand Up @@ -272,6 +272,15 @@ def _resolve_enclosing_symbol(r_repo: str, r_filepath: str, line_num: Optional[i
return (s.get("full_symbol") or s.get("name") or fallback_sym), s
return fallback_sym, None

# Routes that compete for a call, per repository (routes in test files never take a call from a real route).
competing_routes: Dict[str, List[dict]] = {}
for r in all_routes:
if not is_test_route_file(_rel_path(r.get("filepath"))):
competing_routes.setdefault(r.get("repo") or "", []).append(r)

def _owns_call(route: dict, call_url: str, call_method: Optional[str]) -> bool:
return call_belongs_to_route(route, call_url, call_method, competing_routes.get(route.get("repo") or "", []))

def resolve_http_boundaries(sym_name: str, sym_repo: str):
route_meta, ext_callers, http_calls = None, [], []
short_s = sym_name.split(".")[-1]
Expand All @@ -283,8 +292,9 @@ def resolve_http_boundaries(sym_name: str, sym_repo: str):
if matched_routes:
route_meta = {"method": matched_routes[0]["http_method"].upper(), "path": matched_routes[0]["path_pattern"]}
for r_item in matched_routes:
# A call belongs to this route only when no more specific route of the same repository matches it.
for c in all_calls:
if c["repo"] != sym_repo and _methods_compatible(r_item["http_method"], c.get("http_method")) and match_route_and_call(r_item["path_pattern"], c.get("url_pattern", "")):
if c["repo"] != sym_repo and _owns_call(r_item, c.get("url_pattern", ""), c.get("http_method")):
entry = {"repo": c["repo"], "symbol": c.get("caller_symbol") or "", "file": _rel_path(c["filepath"]), "line": c.get("line_number"), "method": c.get("http_method", "").upper(), "url": c.get("url_pattern", "")}
if entry not in ext_callers:
ext_callers.append(entry)
Expand All @@ -298,10 +308,12 @@ def resolve_http_boundaries(sym_name: str, sym_repo: str):
c_method, c_url = c.get("http_method", "").upper(), c.get("url_pattern", "")
ext_handlers = []
for r in all_routes:
if r["repo"] != sym_repo and _methods_compatible(r.get("http_method"), c_method) and match_route_and_call(r.get("path_pattern", ""), c_url):
h_entry = {"repo": r["repo"], "symbol": r.get("handler_symbol") or "", "file": _rel_path(r["filepath"]), "line": r.get("start_line"), "path": r.get("path_pattern")}
if h_entry not in ext_handlers:
ext_handlers.append(h_entry)
# Per repository, only the most specific matching route handles the call.
if r["repo"] == sym_repo or not _owns_call(r, c_url, c_method):
continue
h_entry = {"repo": r["repo"], "symbol": r.get("handler_symbol") or "", "file": _rel_path(r["filepath"]), "line": r.get("start_line"), "path": r.get("path_pattern")}
if h_entry not in ext_handlers:
ext_handlers.append(h_entry)
http_calls.append({"method": c_method, "url": c_url, "line": c.get("line_number"), "external_handlers": ext_handlers})
return route_meta, ext_callers, http_calls

Expand Down
2 changes: 1 addition & 1 deletion docs/mcp-tools-and-resources.md
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ Paginated MCP resource templates (`files/{page}` and `routes/{page}`) use 1-inde
### HTTP Route and Client Discovery Tools

- **`find_routes`**: Searches declared backend endpoints (`api_routes`) by `query`, `method`, and `repo` while excluding unit test fixtures. Clamps `limit` to `1..200` (default `50`) with `offset`.
- **`find_api_callers`**: Matches frontend and service HTTP client calls (`api_client_calls`) against `path` (including `{id}` / `:id` parameter templates), `method`, and `repo`. Client calls are recognized for `fetch`, `axios`, Python `requests`/`httpx`, Go `http`, C# `HttpClient` (also `new HttpRequestMessage(HttpMethod.Post, url)` and a URL in a variable that an earlier line sets, for example `string url = cond ? $"api/a/{id}" : $"api/b/{id}";`; each branch is one call), and JS/TS request helpers whose name suggests HTTP. URLs built from same-file string constants (`const BASE = ...`, C# `const string Base = ...`), URL variables (`const url = ...; sendHttpRequest(url, ...)`), and C# wrappers (`SendAsync(HttpMethod.Post, url, ...)`) are resolved. A URL whose base cannot be resolved (`${baseUrl}/api/v2/items/${id}`) matches a route when the rest of its path matches the end of the route with at least two literal segments (for example ``getJson(`/api/items/${id}`)`` or `apiClient.post('/api/items', ...)`); the method comes from the helper name or a `method:` option. Clamps `limit` to `1..200` (default `50`) with `offset`.
- **`find_api_callers`**: Matches frontend and service HTTP client calls (`api_client_calls`) against `path` (including `{id}` / `:id` parameter templates), `method`, and `repo`. Client calls are recognized for `fetch`, `axios`, Python `requests`/`httpx`, Go `http`, C# `HttpClient` (also `new HttpRequestMessage(HttpMethod.Post, url)` and a URL in a variable that an earlier line sets, for example `string url = cond ? $"api/a/{id}" : $"api/b/{id}";`; each branch is one call), and JS/TS request helpers whose name suggests HTTP. URLs built from same-file string constants (`const BASE = ...`, C# `const string Base = ...`), URL variables (`const url = ...; sendHttpRequest(url, ...)`), and C# wrappers (`SendAsync(HttpMethod.Post, url, ...)`) are resolved. A URL whose base cannot be resolved (`${baseUrl}/api/v2/items/${id}`) matches a route when the rest of its path matches the end of the route with at least two literal segments (for example ``getJson(`/api/items/${id}`)`` or `apiClient.post('/api/items', ...)`); the method comes from the helper name or a `method:` option. A client call belongs only to its most specific route. When more than one known route (`api_routes`) of the same repository and a compatible method matches a call, the most specific route gets the call: the routes are compared segment by segment from the left, and a literal segment (`details`) beats a mixed segment (`v{version}`), which beats a parameter (`{id}`). Routes of different length are compared by their number of literal segments (a mixed segment counts as half). Thus `find_api_callers(path='api/v{version}/assistants/{id}')` does not list a call to `/api/v2/Assistants/details` when the route `api/v{version}/assistants/details` exists. The tool first resolves `path` to its own route in each repository, and it compares only routes of those repositories. Routes in test files do not take part. When no known route matches `path`, every matching call is listed. `trace_path` applies the same rule to `external_callers` and `external_handlers`. Clamps `limit` to `1..200` (default `50`) with `offset`.

### File Inspection and Pattern Matching Tools

Expand Down
Loading
Loading