From 374dcff1cd0ace5dc2bdf9c0b45fbaf2317d7152 Mon Sep 17 00:00:00 2001 From: code2t Date: Tue, 21 Jul 2026 12:00:10 +0800 Subject: [PATCH 01/14] refactor(engine): extract ReadService for search and content retrieval MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move search/ls/cat/head/tail/grep/export/resolve_connector_uri out of the Engine god-class into a dedicated ReadService component (engine-redesign §4.6 stage 3). Engine delegates to self.reads; the 8 public read methods become thin forwards. Method bodies are unchanged; only self. resolution targets moved. - New engine/components/reads/{__init__,text_views,read_service}.py: ReadService + pure helpers (_density_view/_locator_matches) + read-path constants. Locator pair (open_path/match_connector) is a ReadService public method connecting directly to ConnectorFactory+ObjectRepository, no reverse reference to Engine (D1: direct connect, not delayed-lambda). - engine.py: ~1504 -> 887 lines. Assemble self.reads; 8 read methods thin delegates; drop _read_full/_warn_if_huge_export/_has_registered_search_scope /_locator_matches/_density_view/_CODE_SYMBOL/_open_path + read-path constants + now-orphan imports; drop dead-code _now(). - Tests: patch target eng._open_path -> eng.reads.open_path (4 sites: test_cat_locator x2, test_engine_tail, test_grep_not_found). - New tests/test_engine_read_service.py: 16 cases, constructs ReadService without an Engine/Milvus, covers text_views pure helpers + the locator pair + search short-circuits + resolve_connector_uri. Verified: 433 passed, 9 skipped; ruff format --check + ruff check clean. --- .../engine/components/reads/__init__.py | 12 + .../engine/components/reads/read_service.py | 644 +++++++++++++++++ .../engine/components/reads/text_views.py | 62 ++ server/python/src/mfs_server/engine/engine.py | 678 +----------------- server/python/tests/test_cat_locator.py | 4 +- .../python/tests/test_engine_read_service.py | 182 +++++ server/python/tests/test_engine_tail.py | 2 +- server/python/tests/test_grep_not_found.py | 2 +- 8 files changed, 934 insertions(+), 652 deletions(-) create mode 100644 server/python/src/mfs_server/engine/components/reads/__init__.py create mode 100644 server/python/src/mfs_server/engine/components/reads/read_service.py create mode 100644 server/python/src/mfs_server/engine/components/reads/text_views.py create mode 100644 server/python/tests/test_engine_read_service.py diff --git a/server/python/src/mfs_server/engine/components/reads/__init__.py b/server/python/src/mfs_server/engine/components/reads/__init__.py new file mode 100644 index 00000000..cdd4d312 --- /dev/null +++ b/server/python/src/mfs_server/engine/components/reads/__init__.py @@ -0,0 +1,12 @@ +"""ReadService: search/ls/cat/head/tail/grep/export/resolve_connector_uri. + +Extracted from the Engine god-class (engine-redesign §4.6 stage 3). Engine +delegates to self.reads; the public read methods become thin forwards. The +locator pair (open_path / match_connector) is a ReadService public method that +connects directly to ConnectorFactory + ObjectRepository - no reverse reference +to Engine (D1: direct connect + public method, not delayed-lambda). +""" + +from .read_service import ReadService + +__all__ = ["ReadService"] diff --git a/server/python/src/mfs_server/engine/components/reads/read_service.py b/server/python/src/mfs_server/engine/components/reads/read_service.py new file mode 100644 index 00000000..25da4e73 --- /dev/null +++ b/server/python/src/mfs_server/engine/components/reads/read_service.py @@ -0,0 +1,644 @@ +"""ReadService: search / ls / cat / head / tail / grep / export / resolve_connector_uri. + +Extracted verbatim from the Engine god-class (engine-redesign §4.6 stage 3). +Method bodies are unchanged; only `self.` resolution targets moved (see +the dependency-rewire table in docs-dev/engine-redesign-read-upload.md §3.2). +The locator pair (open_path / match_connector) is a public method connecting +directly to ConnectorFactory + ObjectRepository - no reverse reference to +Engine (D1: direct connect + public method, not delayed-lambda). +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import re + +from ....common.converter import CONVERT_EXTS +from ....common.retrieval import build_filter, collapse_results, to_envelope +from ....storage.milvus import MILVUS_MAX_RESULT_WINDOW +from ..connector_factory import ConnectorLocator +from .text_views import ( + _BARE_CAT_MAX_BYTES, + _GREP_LINEAR_SCAN_MAX, + _HEAD_CACHE_N, + _density_view, + _locator_matches, +) + +logger = logging.getLogger(__name__) + + +class ReadService: + """Read path over registered connectors: search the vector index, and + ls/cat/head/tail/grep/export object content. Holds no mutable state.""" + + def __init__(self, cfg, infra, factory, objects, artifacts): + self._cfg = cfg + self._infra = infra + self._factory = factory + self._objects = objects + self._artifacts = artifacts + self._ns = cfg.namespace + + # --- locators: direct connect to factory + objects, no Engine back-reference (D1) --- + async def open_path(self, path: str): + """(connector_id, connector_uri, relpath, plugin) for the registered connector + whose root is the longest prefix of `path`. Thin delegate to + ConnectorFactory.open_path.""" + rows = await self._objects.list_connectors_all() + resolved = await self._factory.open_path(rows, path) + return resolved.cid, resolved.connector_uri, resolved.relpath, resolved.built.plugin + + async def match_connector(self, path: str) -> tuple[dict, str] | None: + """Find the registered connector whose root is the longest prefix of `path`; + return (connector_row, relpath) or None. Thin delegate to + ConnectorLocator.match; rows are fetched from ObjectRepository so the factory + stays SQL-free.""" + rows = await self._objects.list_connectors_all() + return ConnectorLocator.match(rows, path) + + # --- search --- + async def _has_registered_search_scope(self, connector_uri: str | None) -> bool: + """Return whether a search scope can match registered connector-owned chunks. + + Searching an empty namespace, or a path that resolves to an unregistered connector + URI, cannot produce hits. Fast-pathing that case avoids cold-starting the query + embedder for a guaranteed-empty result. + """ + if connector_uri is None: + return await self._objects.has_any_connector() + return await self._objects.has_connector_uri(connector_uri) + + async def search( + self, + query: str, + connector_uri: str | None = None, + object_prefix: str | None = None, + mode: str = "hybrid", + top_k: int = 10, + chunk_kinds: list[str] | None = None, + collapse: bool = False, + ) -> list[dict]: + if top_k <= 0 or not query or not query.strip(): + return [] # nothing to ask for: skip the embed call and Milvus' limit>0 rule + # Coarse fast-path: reject an absurd top_k before embedding/querying. Hybrid + # over-fetches each sub-search by over_fetch_ratio, so the request Milvus sees is + # top_k * ratio; other modes send top_k directly. This only catches values above the + # hard window - the backend's real per-search cap is lower and backend-specific + # (Milvus Lite tops out far below Zilliz), so MilvusStore translates the actual + # MilvusException into the same `top_k_too_large` error as the authoritative guard. + effective = top_k * self._cfg.search.over_fetch_ratio if mode == "hybrid" else top_k + if effective > MILVUS_MAX_RESULT_WINDOW: + raise ValueError("top_k_too_large") + if not await self._has_registered_search_scope(connector_uri): + return [] + expr = build_filter(self._ns, connector_uri, object_prefix, chunk_kinds) + if mode == "keyword": + hits = await asyncio.to_thread( + self._infra.milvus.sparse_search, self._ns, query, top_k, expr + ) + else: + qvec = (await self._infra.embed.batch_embed([query]))[0] + if mode == "semantic": + hits = await asyncio.to_thread( + self._infra.milvus.search_dense, self._ns, qvec, top_k, expr + ) + else: # hybrid + hits = await asyncio.to_thread( + self._infra.milvus.hybrid_search, + self._ns, + qvec, + query, + top_k, + expr, + None, + self._cfg.search.over_fetch_ratio, + ) + envs = [to_envelope(h) for h in hits] + return collapse_results(envs) if collapse else envs + + async def resolve_connector_uri(self, target: str) -> tuple[str, str | None]: + """Map a user path/URI to (connector_uri, object_prefix) for search/grep scope. + Matches the registered connector whose root is the longest prefix of `target`, + so `search q /repo/src` (after `add /repo`) scopes to the /src subtree instead + of fabricating a brand-new connector_uri that would match no indexed chunks.""" + match = await self.match_connector(target) + if match is None: + # not under any registered connector: fall back to literal resolution so an + # exact connector root still scopes correctly (object_prefix unknown -> None). + r = self._factory.resolve_target(target) + return r.connector_uri, None + row, rel = match + connector_uri = row["root_uri"] + # stored chunk object_uri == connector_uri + relpath, so prefix on the full URI + object_prefix = (connector_uri + rel) if rel not in ("", "/") else None + return connector_uri, object_prefix + + async def ls(self, path: str) -> dict: + """List children, each enriched with its full path + index state from the + objects table, plus the connector's capabilities.""" + cid, curi, rel, plugin = await self.open_path(path) + try: + entries = await plugin.list(rel) + caps = plugin.CAPABILITIES.to_dict() + finally: + await plugin.close() + out = [] + base = rel.rstrip("/") + for e in entries: + child_rel = f"{base}/{e.name}" if base else "/" + e.name + row = await self._objects.get_object_search_status(cid, child_rel) + out.append( + { + "name": e.name, + "type": e.type, + "media_type": e.media_type, + "size_hint": e.size_hint, + "path": curi + child_rel, + "search_status": row["search_status"] if row else None, + "indexable": ( + bool(row["indexable"]) if row and row["indexable"] is not None else None + ), + "real_id": e.extra.get("real_id"), + } + ) + return {"entries": out, "capabilities": caps} + + async def cat( + self, + path: str, + range: tuple[int, int] | None = None, + meta: bool = False, + density: str | None = None, + locator: dict | None = None, + ): + import json as _json + from contextlib import aclosing + + from ....connectors.base import Range + + cid, curi, rel, plugin = await self.open_path(path) + try: + st = await plugin.stat(rel) + if st.type == "dir": + raise IsADirectoryError(path) + if meta: + return { + "source": curi + rel, + "media_type": st.media_type, + "size_hint": st.size_hint, + "fingerprint": st.fingerprint, + } + okind = plugin.object_kind_of(rel) + structured = okind in ("table_rows", "record_collection", "message_stream") + # Binary objects have no line-based view - reading them with --range + # would return mojibake (UTF-8 errors="replace") under the guise of a + # text slice. Refuse cleanly so the caller falls back to `export`. + if range is not None and okind == "binary": + raise ValueError("range_unsupported") + + # --- locator with reserved "lines" key: route to the range path --- + # Body / code / document chunks store identity as {"lines":[s,e]}; + # reopening one means slicing the file by line range, not iterating + # structured records. locator.lines is 1-based half-open (matches + # how cat --range is exposed); plugin.read takes 0-based half-open. + if ( + locator is not None + and isinstance(locator, dict) + and "lines" in locator + and len(locator) == 1 + and not structured + ): + s, e = int(locator["lines"][0]), int(locator["lines"][1]) + rg = Range(max(0, s - 1), max(0, e - 1)) + buf = bytearray() + async for ch in plugin.read(rel, rg): + buf += ch + return bytes(buf).decode("utf-8", errors="replace") + + # --- locator: reopen a single structured record --- + if locator is not None: + records = plugin.read_records(rel) + if records is None: + raise ValueError("range_unsupported") # not a structured object + ocfg = plugin.ctx.object_config_for(rel) + i = 0 + # aclosing: a match returns mid-iteration, so the record generator must be + # closed deterministically - else a connector holding a cursor/transaction + # (e.g. asyncpg) leaks the connection and pool.close() later blocks ~60s. + async with aclosing(records): + async for rec in records: + if _locator_matches(rec, ocfg, i, locator): + return { + "source": curi + rel, + "locator": locator, + "content": _json.dumps(rec, default=str, ensure_ascii=False), + } + i += 1 + raise ValueError("locator_not_found") + + # --- structured object: range pushdown over records (lazy, not materialized) --- + if structured: + if range is None: + # Bare cat of a structured object: stream the records as JSONL + # into a buffer up to _BARE_CAT_MAX_BYTES, then return. Small + # objects (Slack users.jsonl, small GitHub issue feeds, dozens- + # of-row tables) fit comfortably and round-trip as JSONL. Large + # ones (a postgres table with 1M rows) blow the budget mid- + # stream and raise the same object_too_large_for_cat so the + # caller still falls back to head / cat --range / export. + records = plugin.read_records(rel) + if records is None: + raise ValueError("object_too_large_for_cat") + budget = _BARE_CAT_MAX_BYTES + out: list[str] = [] + size = 0 + async with aclosing(records): + async for rec in records: + line = _json.dumps(rec, default=str, ensure_ascii=False) + size += len(line.encode("utf-8")) + 1 # +1 newline + if size > budget: + raise ValueError("object_too_large_for_cat") + out.append(line) + return "\n".join(out) + start, end = range[0], range[1] + # Pass Range(0, end) - a LIMIT-only hint - and slice [start, end) HERE, in one + # place. Connectors disagree on whether they honor the Range: the DB ones + # (mysql/postgres/mongo/bigquery) push OFFSET start + LIMIT down, while the SaaS + # ones (jira/slack/notion/…) ignore it and return from row 0 - yet ALL declare + # paged_cat=True, so the engine can't tell them apart. Pushing OFFSET start AND + # then re-slicing `i >= start` double-applied the offset on the DB connectors, so + # `cat --range 100:200` returned an empty/wrong page. With offset=0 every + # connector returns rows from 0 and the single `i >= start` slice is correct for + # both. (Trade-off: the DB connectors lose OFFSET pushdown and read `end` rows for + # a deep page - still LIMIT-bounded; restoring true offset-pushdown needs an + # explicit "range honored" capability - see human_todo [dborder/D65].) + records = plugin.read_records(rel, Range(0, end)) + if records is not None: + out, i = [], 0 + async with aclosing(records): # break-early must close the generator + async for rec in records: + if i >= end: + break + if i >= start: + out.append(_json.dumps(rec, default=str, ensure_ascii=False)) + i += 1 + return "\n".join(out) + + ext = os.path.splitext(rel)[1].lower() + text: str | None = None + # converted markdown artifact: pdf/docx/html (CONVERT_EXTS) AND web/github pages, + # whose .md is generated at ingest - read it from the artifact store so cat works + # across restarts / fresh plugin instances, not just in-memory. + if ext in CONVERT_EXTS: + # On-read freshness: stat() above already fetched the live source fingerprint, + # so comparing it to the one recorded at ingest is free. If it changed, the + # cached markdown is stale -> re-convert from the current source. + if await self._artifacts.converted_md_stale(cid, rel, st.fingerprint): + raw = bytearray() + async for ch in plugin.read(rel): + raw += ch + text = await self._infra.converter.convert(bytes(raw), ext) + else: + art = await self._artifacts.read_artifact(self._ns, curi + rel, "converted_md") + if art is not None: + text = art.decode("utf-8", errors="replace") + elif curi.startswith(("web://", "github://")): + art = await self._artifacts.read_artifact(self._ns, curi + rel, "converted_md") + if art is not None: + text = art.decode("utf-8", errors="replace") + if text is None and okind == "image" and self._cfg.description.enabled: + # An image's description is a model output served through the transformation + # cache: describe() returns the description memoized at ingest, or computes it + # on a miss. (Re-reading the bytes here is the source round-trip - cheap for + # local files; see TODO §10.9 for the snapshot path on remote connectors.) + raw = bytearray() + async for ch in plugin.read(rel): + raw += ch + try: + return await self._infra.vlm.describe(bytes(raw), ext) + except Exception: # noqa: BLE001 - fall through to the raw read on provider error + pass + if text is None: + if range is None and st.size_hint and st.size_hint > _BARE_CAT_MAX_BYTES: + raise ValueError("object_too_large_for_cat") + rg = Range(range[0], range[1]) if range else None + buf = bytearray() + async for ch in plugin.read(rel, rg): + buf += ch + text = bytes(buf).decode("utf-8", errors="replace") + if density and density != "deep": + okind = plugin.object_kind_of(rel) + if okind not in ("document", "code"): + raise ValueError("density_unsupported") + return _density_view(text, ext, density) + return text + finally: + await plugin.close() + + async def _read_full(self, path: str) -> tuple[str, bool]: + """Whole object content for export / tail: returns (text, partial). + partial=True when the connector capped the read (structured objects + above max_read_rows). The bare-cat size guard is not applied. Backs + export and tail; tail discards the partial flag (just wants the last + N lines), export surfaces it.""" + import json as _json + + cid, curi, rel, plugin = await self.open_path(path) + try: + st = await plugin.stat(rel) + if st.type == "dir": + raise IsADirectoryError(path) + okind = plugin.object_kind_of(rel) + if okind in ("table_rows", "record_collection", "message_stream"): + records = plugin.read_records(rel) + if records is not None: + out = [] + async for rec in records: + out.append(_json.dumps(rec, default=str, ensure_ascii=False)) + text = "\n".join(out) + # ctx.declare_partial is the channel structured connectors use + # to flag "we capped at max_read_rows". Read it back here so + # export tells the truth instead of silently returning a slice. + partial = bool(getattr(plugin.ctx, "was_partial", lambda _r: False)(rel)) + self._warn_if_huge_export(curi + rel, text) + return text, partial + ext = os.path.splitext(rel)[1].lower() + if ext in CONVERT_EXTS: + # On-read freshness (same as cat): a changed source fingerprint means the + # cached markdown is stale, so re-convert from the current source. + if await self._artifacts.converted_md_stale(cid, rel, st.fingerprint): + raw = bytearray() + async for ch in plugin.read(rel): + raw += ch + text = await self._infra.converter.convert(bytes(raw), ext) + self._warn_if_huge_export(curi + rel, text) + return text, False + art = await self._artifacts.read_artifact(self._ns, curi + rel, "converted_md") + if art is not None: + text = art.decode("utf-8", errors="replace") + self._warn_if_huge_export(curi + rel, text) + return text, False + elif curi.startswith(("web://", "github://")): + art = await self._artifacts.read_artifact(self._ns, curi + rel, "converted_md") + if art is not None: + text = art.decode("utf-8", errors="replace") + self._warn_if_huge_export(curi + rel, text) + return text, False + if okind == "image" and self._cfg.description.enabled: + raw = bytearray() + async for ch in plugin.read(rel): + raw += ch + try: + text = await self._infra.vlm.describe(bytes(raw), ext) + self._warn_if_huge_export(curi + rel, text) + return text, False + except Exception: # noqa: BLE001 - fall through to the raw read on provider error + pass + buf = bytearray() + async for ch in plugin.read(rel): + buf += ch + text = bytes(buf).decode("utf-8", errors="replace") + self._warn_if_huge_export(curi + rel, text) + return text, False + finally: + await plugin.close() + + def _warn_if_huge_export(self, uri: str, text: str) -> None: + """Single-host export materializes the whole object in memory; warn on anything over + 100 MB so the operator sees the cost before the next OOM rather than after. A streaming + export path is the proper fix but is deferred - objects this large are rare on the + single-host deployment this guard covers, and the warning makes the cost explicit.""" + size = len(text.encode("utf-8", errors="ignore")) if text else 0 + if size > 100 * 1024 * 1024: + logger.warning( + "export %s materialized %d MB in memory (streaming export not yet implemented)", + uri, + size // (1024 * 1024), + ) + + async def export(self, path: str) -> tuple[str, bool]: + """Full content for `mfs export`: returns (text, partial). Honest + boundary - structured connectors with more rows than max_read_rows + return partial=True; the caller (API layer) surfaces it in the + CatResponse. The bare-cat size guard does not apply, but each + connector's own row cap does (true streaming export is deferred).""" + return await self._read_full(path) + + async def head(self, path: str, n: int = 20) -> str: + cid, curi, rel, plugin = await self.open_path(path) + try: + okind = plugin.object_kind_of(rel) + structured = okind in ("table_rows", "record_collection", "message_stream") + if structured: + # fast path: pre-cached first rows. The cache is capped at _HEAD_CACHE_N, so + # it's authoritative ONLY when it holds the whole object (< the cap) OR n fits + # within it; for a larger n on a capped cache, fall through to the live bounded + # query below - otherwise `head -n 200` would silently return just the 100 + # cached rows instead of 200 (head must give min(n, total), not min(n, cache)). + art = await self._artifacts.read_artifact(self._ns, curi + rel, "head_cache") + if art is not None: + cached = art.decode("utf-8", errors="replace").splitlines() + if len(cached) < _HEAD_CACHE_N or n <= len(cached): + return "\n".join(cached[:n]) + else: + ext = os.path.splitext(rel)[1].lower() + # plain text / code / logs: stream just the first n lines so a large file + # never materializes and never trips bare-cat's size guard - head is exactly + # the escape hatch for big objects. Artifact-backed + # objects (pdf/docx/html, web/github pages, images) have bounded cached text, + # so they fall through to cat below. + if not ( + okind == "image" + or ext in CONVERT_EXTS + or curi.startswith(("web://", "github://")) + ): + lines: list[str] = [] + buf = b"" + async for chunk in plugin.read(rel): + buf += chunk + while len(lines) < n: + nl = buf.find(b"\n") + if nl < 0: + break + lines.append(buf[:nl].decode("utf-8", errors="replace")) + buf = buf[nl + 1 :] + if len(lines) >= n: + break + if len(lines) < n and buf: + lines.append(buf.decode("utf-8", errors="replace")) + return "\n".join(lines[:n]) + finally: + await plugin.close() + if structured: + return await self.cat(path, range=(0, n)) # bounded page, not the whole table + text = await self.cat(path) # artifact-backed text, bounded + return "\n".join(text.splitlines()[:n]) + + async def tail(self, path: str, n: int = 20) -> str: + if n <= 0: + return "" + # plain-text real local file: read the last n lines straight off disk (native + # accelerator / bounded reverse-read), so a huge log isn't fully materialized. + # Artifact-backed (pdf/docx/html, web/github) and structured objects fall back. + from ....common import accel + + cid, curi, rel, plugin = await self.open_path(path) + try: + okind = plugin.object_kind_of(rel) + if okind in ("table_rows", "record_collection", "message_stream"): + raise ValueError("tail_unsupported") + ext = os.path.splitext(rel)[1].lower() + plain_local = ( + curi.startswith("file://local") + and okind + not in ( + "image", + "table_rows", + "record_collection", + "message_stream", + "table_schema", + ) + and ext not in CONVERT_EXTS + ) + abs_file = (curi.replace("file://local", "", 1) + rel) if plain_local else None + finally: + await plugin.close() + if abs_file and os.path.isfile(abs_file): + return "\n".join(await asyncio.to_thread(accel.tail_lines, abs_file, n)) + text, _partial = await self._read_full( + path + ) # artifact-backed / structured / non-local; tail ignores the partial flag + return "\n".join(text.splitlines()[-n:]) + + async def grep( + self, pattern: str, path: str, top_k: int = 100, regex: bool = False + ) -> list[dict]: + """Dispatch: pushdown (file: none) -> BM25 (indexed scope) -> linear scan + (not_indexed objects in scope). The linear scan uses the native + accelerator (mfs_server_rs) when the object is a real local file, else falls + back to reading bytes + pure-Python regex.""" + from ....common import accel + from ....connectors.base import GrepOptions + + cid, curi, rel, plugin = await self.open_path(path) + scope_prefix = (curi + rel) if rel != "/" else None + try: + # open_path only resolves which connector owns the prefix, not whether + # `rel` exists under it -- unlike ls/cat, nothing downstream fails loudly + # for a missing path (pushdown/BM25/linear-scan all just yield zero + # matches). Stat it explicitly so a bad path 404s like ls/cat instead of + # looking like a real, empty search. + await plugin.stat(rel) + results: list[dict] = [] + # 2a connector grep pushdown: exact, source-side (e.g. + # SQL ILIKE for structured connectors). Returns None when unsupported. + ocfg = plugin.ctx.object_config_for(rel) + try: + gen = await plugin.grep( + pattern, + rel, + GrepOptions( + pattern=pattern, + text_fields=ocfg.text_fields, + metadata_fields=ocfg.metadata_fields, + ), + ) + except Exception: # noqa: BLE001 - pushdown failure shouldn't kill grep + gen = None + if gen is not None: + async for gm in gen: + # Structured pushdown carries gm.locator (PK dict); text/code + # pushdown carries gm.line_no. locator.lines is 1-based + # half-open [s,e), so a single line n is [n, n+1] - not + # [n, n], which would round-trip as an empty slice. + loc = ( + gm.locator + if gm.locator is not None + else ({"lines": [gm.line_no, gm.line_no + 1]} if gm.line_no else None) + ) + results.append( + { + "source": curi + gm.path, + "locator": loc, + "content": gm.content, + "via": "pushdown", + } + ) + return results + # 2b BM25 over indexed objects in scope + expr = build_filter(self._ns, curi, scope_prefix) + hits = await asyncio.to_thread( + self._infra.milvus.sparse_search, self._ns, pattern, top_k, expr + ) + for h in hits: + e = h.get("entity", h) + results.append( + { + "source": e.get("object_uri"), + "locator": e.get("locator"), + "content": e.get("content"), + "via": "bm25", + } + ) + # 2c linear scan over not_indexed objects in scope (file connector) + root_abs = ( + curi.replace("file://local", "", 1) if curi.startswith("file://local") else None + ) + # Path-component boundary, same fix as build_filter: scope `/src` must match the + # object itself OR `/src/...`, NOT a sibling `/src-old`. Escape SQL LIKE wildcards + # ('_'/'%') in the literal prefix so a path with '_' doesn't over-match either. + not_idx = await self._objects.list_not_indexed_in_scope(cid, rel) + if len(not_idx) > _GREP_LINEAR_SCAN_MAX: + # don't silently scan a subset and imply it was exhaustive - tell the agent + # so it can narrow the path or index first. + results.append( + { + "source": None, + "locator": None, + "via": "notice", + "content": f"(grep linear scan capped at {_GREP_LINEAR_SCAN_MAX} of " + f"{len(not_idx)} not-indexed files in scope; narrow the path " + f"or run `mfs add` to index them for complete results)", + } + ) + for o in not_idx[:_GREP_LINEAR_SCAN_MAX]: + relp = o["object_uri"] + try: + abs_file = (root_abs + relp) if root_abs else None + if abs_file and os.path.isfile(abs_file): + # native (or pure-Python) streaming grep straight off disk + for ln, line in await asyncio.to_thread( + accel.linear_grep_file, abs_file, pattern, False, regex, 200 + ): + results.append( + { + "source": curi + relp, + "locator": {"lines": [ln, ln + 1]}, + "content": line, + "via": "linear", + } + ) + else: + rx = re.compile(pattern if regex else re.escape(pattern)) + buf = bytearray() + async for ch in plugin.read(relp): + buf += ch + text = bytes(buf).decode("utf-8", errors="replace") + for i, line in enumerate(text.splitlines(), 1): + if rx.search(line): + results.append( + { + "source": curi + relp, + "locator": {"lines": [i, i + 1]}, + "content": line, + "via": "linear", + } + ) + except Exception: # noqa: BLE001 + pass + return results + finally: + await plugin.close() diff --git a/server/python/src/mfs_server/engine/components/reads/text_views.py b/server/python/src/mfs_server/engine/components/reads/text_views.py new file mode 100644 index 00000000..bd94bbb2 --- /dev/null +++ b/server/python/src/mfs_server/engine/components/reads/text_views.py @@ -0,0 +1,62 @@ +"""ReadService pure helpers: density view skeleton + locator match guard + +read-path size caps. Extracted from Engine so ReadService and its tests can use +them without a connector stack.""" + +from __future__ import annotations + +import re + +from ...producers.render import resolve_path + +_HEAD_CACHE_N = 100 # rows pre-cached per structured object to speed `head` +_BARE_CAT_MAX_BYTES = 5 * 1024 * 1024 # bare `cat` (no range) rejects objects larger than this +_GREP_LINEAR_SCAN_MAX = 200 # cap on not-indexed files a single grep scans linearly + +_CODE_SYMBOL = re.compile(r"^\s*(def |class |func |fn |public |private |func\(|type )") + + +def _density_view(text: str, ext: str, density: str) -> str: + """Skeleton view of a document/code object: + peek = headings (markdown #) or code symbol lines only; + skim = peek + the first non-blank line of prose under each heading. + """ + lines = text.splitlines() + is_md = ext in (".md", ".markdown", ".rst", ".txt", "") + out: list[str] = [] + if is_md: + for i, ln in enumerate(lines): + if ln.lstrip().startswith("#"): + out.append(ln.rstrip()) + if density == "skim": + for nxt in lines[i + 1 :]: + if nxt.strip(): + out.append(" " + nxt.strip()[:120]) + break + else: + for ln in lines: + if _CODE_SYMBOL.match(ln): + out.append(ln.rstrip() if density == "skim" else ln.split("(")[0].rstrip()) + if not out: + # nothing structural found -> first lines as a fallback peek + out = [ln.rstrip() for ln in lines[:15]] + return "\n".join(out) + + +def _locator_matches(rec: dict, ocfg, idx: int, locator: dict) -> bool: + if "_row" in locator: + return idx == int(locator["_row"]) + # "lines" is the framework-reserved key for body/code chunks and is never a + # structured-record PK - never compare it against the row. The cat router + # dispatches body-chunk reads through plugin.read(range=...) before reaching + # this helper, so seeing it here is a misconfiguration we just ignore. + keys = [k for k in (ocfg.locator_fields or list(locator.keys())) if k != "lines"] + present = [k for k in keys if k in locator] + # Require at least one recognized locator key: a locator that's empty or whose keys + # don't correspond to this object's locator_fields matches nothing. Without this guard + # `all([])` is True, so a bogus/typo'd locator silently returns record #0 instead of + # the documented locator_not_found. + if not present: + return False + # resolve with the SAME JSONPath-lite used to WRITE the locator (engine indexing: + # {f: resolve_path(rec, f)}); plain rec.get() couldn't reopen a nested locator key. + return all(str(resolve_path(rec, k)) == str(locator.get(k)) for k in present) diff --git a/server/python/src/mfs_server/engine/engine.py b/server/python/src/mfs_server/engine/engine.py index e483c8ac..6fe62d95 100644 --- a/server/python/src/mfs_server/engine/engine.py +++ b/server/python/src/mfs_server/engine/engine.py @@ -12,39 +12,29 @@ import json import logging import os -import re import uuid from datetime import datetime, timedelta, timezone -from ..common.converter import CONVERT_EXTS -from ..common.retrieval import build_filter, collapse_results, to_envelope from ..config import ServerConfig from ..connectors.base import SyncOptions from ..storage.file_state import FileStateStore -from ..storage.milvus import MILVUS_MAX_RESULT_WINDOW from .components import ConnectorFactory, ConnectorLocator, CredentialService from .components.artifact_cache import ArtifactCacheService from .components.object_repository import ObjectRepository +from .components.reads import ReadService from .infra import InfraStack from .ingest import IngestOrchestrator from .pipeline_supervisor import PipelineSupervisor -from .producers.render import render_record, resolve_path +from .producers.render import render_record from .state import ConnectorStateStore logger = logging.getLogger(__name__) -_HEAD_CACHE_N = 100 # rows pre-cached per structured object to speed `head` -_BARE_CAT_MAX_BYTES = 5 * 1024 * 1024 # bare `cat` (no range) rejects objects larger than this -_GREP_LINEAR_SCAN_MAX = 200 # cap on not-indexed files a single grep scans linearly _JOB_STALE_AFTER_S = 120 # no heartbeat for this long => worker presumed dead _WORKER_CONNECT_TIMEOUT_S = 30 # bound plugin.connect() in the worker so a hanging/unreachable # connector fails its job cleanly instead of blocking the single in-process worker forever -def _now() -> str: - return datetime.now(timezone.utc).isoformat() - - def _norm_rel(p: str) -> str: """Connector-relative path with a single leading '/' (file_state / object_uri convention).""" return "/" + p.lstrip("/") @@ -65,36 +55,6 @@ def _validate_upload_member(m) -> None: raise ValueError(f"unsafe path in archive: {rel}") -_CODE_SYMBOL = re.compile(r"^\s*(def |class |func |fn |public |private |func\(|type )") - - -def _density_view(text: str, ext: str, density: str) -> str: - """Skeleton view of a document/code object: - peek = headings (markdown #) or code symbol lines only; - skim = peek + the first non-blank line of prose under each heading. - """ - lines = text.splitlines() - is_md = ext in (".md", ".markdown", ".rst", ".txt", "") - out: list[str] = [] - if is_md: - for i, ln in enumerate(lines): - if ln.lstrip().startswith("#"): - out.append(ln.rstrip()) - if density == "skim": - for nxt in lines[i + 1 :]: - if nxt.strip(): - out.append(" " + nxt.strip()[:120]) - break - else: - for ln in lines: - if _CODE_SYMBOL.match(ln): - out.append(ln.rstrip() if density == "skim" else ln.split("(")[0].rstrip()) - if not out: - # nothing structural found -> first lines as a fallback peek - out = [ln.rstrip() for ln in lines[:15]] - return "\n".join(out) - - class Engine: def __init__(self, cfg: ServerConfig): self.cfg = cfg @@ -123,6 +83,12 @@ def __init__(self, cfg: ServerConfig): cfg, self.infra, self.connector_factory, self.pipeline, self.objects, self.artifacts ) self.ingest.bind_remover(self.remove_connector) + # ReadService owns the read path (search/ls/cat/head/tail/grep/export/resolve_connector_uri). + # Locators (open_path/match_connector) are ReadService public methods connecting directly + # to ConnectorFactory + ObjectRepository - no reverse reference to Engine (D1). + self.reads = ReadService( + cfg, self.infra, self.connector_factory, self.objects, self.artifacts + ) async def startup(self, *, preload_local_models: bool = False) -> None: # Infra connect + schema + optional model preload, then the pipeline process singletons @@ -618,82 +584,13 @@ async def _evict_artifacts_if_needed(self, ns: str) -> int: # Thin delegate to ArtifactCacheService. return await self.artifacts.evict_if_needed(ns) - # --- search --- - async def _has_registered_search_scope(self, connector_uri: str | None) -> bool: - """Return whether a search scope can match registered connector-owned chunks. + # --- search / read commands: thin delegates to ReadService (stage 3) --- - Searching an empty namespace, or a path that resolves to an unregistered connector - URI, cannot produce hits. Fast-pathing that case avoids cold-starting the query - embedder for a guaranteed-empty result. - """ - if connector_uri is None: - return await self.objects.has_any_connector() - return await self.objects.has_connector_uri(connector_uri) + async def search(self, *a, **kw): + return await self.reads.search(*a, **kw) - async def search( - self, - query: str, - connector_uri: str | None = None, - object_prefix: str | None = None, - mode: str = "hybrid", - top_k: int = 10, - chunk_kinds: list[str] | None = None, - collapse: bool = False, - ) -> list[dict]: - if top_k <= 0 or not query or not query.strip(): - return [] # nothing to ask for: skip the embed call and Milvus' limit>0 rule - # Coarse fast-path: reject an absurd top_k before embedding/querying. Hybrid - # over-fetches each sub-search by over_fetch_ratio, so the request Milvus sees is - # top_k * ratio; other modes send top_k directly. This only catches values above the - # hard window — the backend's real per-search cap is lower and backend-specific - # (Milvus Lite tops out far below Zilliz), so MilvusStore translates the actual - # MilvusException into the same `top_k_too_large` error as the authoritative guard. - effective = top_k * self.cfg.search.over_fetch_ratio if mode == "hybrid" else top_k - if effective > MILVUS_MAX_RESULT_WINDOW: - raise ValueError("top_k_too_large") - if not await self._has_registered_search_scope(connector_uri): - return [] - expr = build_filter(self.ns, connector_uri, object_prefix, chunk_kinds) - if mode == "keyword": - hits = await asyncio.to_thread( - self.infra.milvus.sparse_search, self.ns, query, top_k, expr - ) - else: - qvec = (await self.infra.embed.batch_embed([query]))[0] - if mode == "semantic": - hits = await asyncio.to_thread( - self.infra.milvus.search_dense, self.ns, qvec, top_k, expr - ) - else: # hybrid - hits = await asyncio.to_thread( - self.infra.milvus.hybrid_search, - self.ns, - qvec, - query, - top_k, - expr, - None, - self.cfg.search.over_fetch_ratio, - ) - envs = [to_envelope(h) for h in hits] - return collapse_results(envs) if collapse else envs - - async def resolve_connector_uri(self, target: str) -> tuple[str, str | None]: - """Map a user path/URI to (connector_uri, object_prefix) for search/grep scope. - Matches the registered connector whose root is the longest prefix of `target`, - so `search q /repo/src` (after `add /repo`) scopes to the /src subtree instead - of fabricating a brand-new connector_uri that would match no indexed chunks.""" - match = await self._match_connector(target) - if match is None: - # not under any registered connector: fall back to literal resolution so an - # exact connector root still scopes correctly (object_prefix unknown -> None). - _, connector_uri, _, _ = self._resolve_target(target) - return connector_uri, None - row, rel = match - connector_uri = row["root_uri"] - # stored chunk object_uri == connector_uri + relpath, so prefix on the full URI - object_prefix = (connector_uri + rel) if rel not in ("", "/") else None - return connector_uri, object_prefix + async def resolve_connector_uri(self, *a, **kw): + return await self.reads.resolve_connector_uri(*a, **kw) async def _resolve_readonly_config( self, ctype: str, connector_uri: str, config: dict | None, default_config: dict @@ -958,547 +855,32 @@ async def remove_connector(self, target: str) -> bool: await self.objects.delete_connector(cid) return True - # --- read commands — any connector --- + # --- connector locator: kept on Engine for inspect/remove_connector until + # ConnectorManager wiring (stage 5). ReadService has its own open_path / + # match_connector (direct factory+objects connect, D1). --- async def _match_connector(self, path: str) -> tuple[dict, str] | None: """Find the registered connector whose root is the longest prefix of `path`; - return (connector_row, relpath) or None. Shared by _open_path (read commands) - and resolve_connector_uri (search/grep scope). Thin delegate to + return (connector_row, relpath) or None. Used by inspect/remove_connector + (connector management). Thin delegate to ConnectorLocator.match; rows are fetched from ObjectRepository so the factory stays SQL-free.""" rows = await self.objects.list_connectors_all() return ConnectorLocator.match(rows, path) - async def _open_path(self, path: str): - """(connector_id, connector_uri, relpath, plugin) for the registered connector - whose root is the longest prefix of `path`. Thin delegate to - ConnectorFactory.open_path.""" - rows = await self.objects.list_connectors_all() - resolved = await self.connector_factory.open_path(rows, path) - return resolved.cid, resolved.connector_uri, resolved.relpath, resolved.built.plugin - - async def ls(self, path: str) -> dict: - """List children, each enriched with its full path + index state from the - objects table, plus the connector's capabilities.""" - cid, curi, rel, plugin = await self._open_path(path) - try: - entries = await plugin.list(rel) - caps = plugin.CAPABILITIES.to_dict() - finally: - await plugin.close() - out = [] - base = rel.rstrip("/") - for e in entries: - child_rel = f"{base}/{e.name}" if base else "/" + e.name - row = await self.objects.get_object_search_status(cid, child_rel) - out.append( - { - "name": e.name, - "type": e.type, - "media_type": e.media_type, - "size_hint": e.size_hint, - "path": curi + child_rel, - "search_status": row["search_status"] if row else None, - "indexable": ( - bool(row["indexable"]) if row and row["indexable"] is not None else None - ), - "real_id": e.extra.get("real_id"), - } - ) - return {"entries": out, "capabilities": caps} + async def ls(self, *a, **kw): + return await self.reads.ls(*a, **kw) - @staticmethod - def _locator_matches(rec: dict, ocfg, idx: int, locator: dict) -> bool: - if "_row" in locator: - return idx == int(locator["_row"]) - # "lines" is the framework-reserved key for body/code chunks and is never a - # structured-record PK — never compare it against the row. The cat router - # dispatches body-chunk reads through plugin.read(range=...) before reaching - # this helper, so seeing it here is a misconfiguration we just ignore. - keys = [k for k in (ocfg.locator_fields or list(locator.keys())) if k != "lines"] - present = [k for k in keys if k in locator] - # Require at least one recognized locator key: a locator that's empty or whose keys - # don't correspond to this object's locator_fields matches nothing. Without this guard - # `all([])` is True, so a bogus/typo'd locator silently returns record #0 instead of - # the documented locator_not_found. - if not present: - return False - # resolve with the SAME JSONPath-lite used to WRITE the locator (engine indexing: - # {f: resolve_path(rec, f)}); plain rec.get() couldn't reopen a nested locator key. - return all(str(resolve_path(rec, k)) == str(locator.get(k)) for k in present) - - async def cat( - self, - path: str, - range: tuple[int, int] | None = None, - meta: bool = False, - density: str | None = None, - locator: dict | None = None, - ): - import json as _json - from contextlib import aclosing + async def cat(self, *a, **kw): + return await self.reads.cat(*a, **kw) - from ..connectors.base import Range - - cid, curi, rel, plugin = await self._open_path(path) - try: - st = await plugin.stat(rel) - if st.type == "dir": - raise IsADirectoryError(path) - if meta: - return { - "source": curi + rel, - "media_type": st.media_type, - "size_hint": st.size_hint, - "fingerprint": st.fingerprint, - } - okind = plugin.object_kind_of(rel) - structured = okind in ("table_rows", "record_collection", "message_stream") - # Binary objects have no line-based view — reading them with --range - # would return mojibake (UTF-8 errors="replace") under the guise of a - # text slice. Refuse cleanly so the caller falls back to `export`. - if range is not None and okind == "binary": - raise ValueError("range_unsupported") - - # --- locator with reserved "lines" key: route to the range path --- - # Body / code / document chunks store identity as {"lines":[s,e]}; - # reopening one means slicing the file by line range, not iterating - # structured records. locator.lines is 1-based half-open (matches - # how cat --range is exposed); plugin.read takes 0-based half-open. - if ( - locator is not None - and isinstance(locator, dict) - and "lines" in locator - and len(locator) == 1 - and not structured - ): - s, e = int(locator["lines"][0]), int(locator["lines"][1]) - rg = Range(max(0, s - 1), max(0, e - 1)) - buf = bytearray() - async for ch in plugin.read(rel, rg): - buf += ch - return bytes(buf).decode("utf-8", errors="replace") - - # --- locator: reopen a single structured record --- - if locator is not None: - records = plugin.read_records(rel) - if records is None: - raise ValueError("range_unsupported") # not a structured object - ocfg = plugin.ctx.object_config_for(rel) - i = 0 - # aclosing: a match returns mid-iteration, so the record generator must be - # closed deterministically — else a connector holding a cursor/transaction - # (e.g. asyncpg) leaks the connection and pool.close() later blocks ~60s. - async with aclosing(records): - async for rec in records: - if self._locator_matches(rec, ocfg, i, locator): - return { - "source": curi + rel, - "locator": locator, - "content": _json.dumps(rec, default=str, ensure_ascii=False), - } - i += 1 - raise ValueError("locator_not_found") - - # --- structured object: range pushdown over records (lazy, not materialized) --- - if structured: - if range is None: - # Bare cat of a structured object: stream the records as JSONL - # into a buffer up to _BARE_CAT_MAX_BYTES, then return. Small - # objects (Slack users.jsonl, small GitHub issue feeds, dozens- - # of-row tables) fit comfortably and round-trip as JSONL. Large - # ones (a postgres table with 1M rows) blow the budget mid- - # stream and raise the same object_too_large_for_cat so the - # caller still falls back to head / cat --range / export. - records = plugin.read_records(rel) - if records is None: - raise ValueError("object_too_large_for_cat") - budget = _BARE_CAT_MAX_BYTES - out: list[str] = [] - size = 0 - async with aclosing(records): - async for rec in records: - line = _json.dumps(rec, default=str, ensure_ascii=False) - size += len(line.encode("utf-8")) + 1 # +1 newline - if size > budget: - raise ValueError("object_too_large_for_cat") - out.append(line) - return "\n".join(out) - start, end = range[0], range[1] - # Pass Range(0, end) — a LIMIT-only hint — and slice [start, end) HERE, in one - # place. Connectors disagree on whether they honor the Range: the DB ones - # (mysql/postgres/mongo/bigquery) push OFFSET start + LIMIT down, while the SaaS - # ones (jira/slack/notion/…) ignore it and return from row 0 — yet ALL declare - # paged_cat=True, so the engine can't tell them apart. Pushing OFFSET start AND - # then re-slicing `i >= start` double-applied the offset on the DB connectors, so - # `cat --range 100:200` returned an empty/wrong page. With offset=0 every - # connector returns rows from 0 and the single `i >= start` slice is correct for - # both. (Trade-off: the DB connectors lose OFFSET pushdown and read `end` rows for - # a deep page — still LIMIT-bounded; restoring true offset-pushdown needs an - # explicit "range honored" capability — see human_todo [dborder/D65].) - records = plugin.read_records(rel, Range(0, end)) - if records is not None: - out, i = [], 0 - async with aclosing(records): # break-early must close the generator - async for rec in records: - if i >= end: - break - if i >= start: - out.append(_json.dumps(rec, default=str, ensure_ascii=False)) - i += 1 - return "\n".join(out) - - ext = os.path.splitext(rel)[1].lower() - text: str | None = None - # converted markdown artifact: pdf/docx/html (CONVERT_EXTS) AND web/github pages, - # whose .md is generated at ingest — read it from the artifact store so cat works - # across restarts / fresh plugin instances, not just in-memory. - if ext in CONVERT_EXTS: - # On-read freshness: stat() above already fetched the live source fingerprint, - # so comparing it to the one recorded at ingest is free. If it changed, the - # cached markdown is stale -> re-convert from the current source. - if await self._converted_md_stale(cid, rel, st.fingerprint): - raw = bytearray() - async for ch in plugin.read(rel): - raw += ch - text = await self.infra.converter.convert(bytes(raw), ext) - else: - art = await self._read_artifact(self.ns, curi + rel, "converted_md") - if art is not None: - text = art.decode("utf-8", errors="replace") - elif curi.startswith(("web://", "github://")): - art = await self._read_artifact(self.ns, curi + rel, "converted_md") - if art is not None: - text = art.decode("utf-8", errors="replace") - if text is None and okind == "image" and self.cfg.description.enabled: - # An image's description is a model output served through the transformation - # cache: describe() returns the description memoized at ingest, or computes it - # on a miss. (Re-reading the bytes here is the source round-trip — cheap for - # local files; see TODO §10.9 for the snapshot path on remote connectors.) - raw = bytearray() - async for ch in plugin.read(rel): - raw += ch - try: - return await self.infra.vlm.describe(bytes(raw), ext) - except Exception: # noqa: BLE001 — fall through to the raw read on provider error - pass - if text is None: - if range is None and st.size_hint and st.size_hint > _BARE_CAT_MAX_BYTES: - raise ValueError("object_too_large_for_cat") - rg = Range(range[0], range[1]) if range else None - buf = bytearray() - async for ch in plugin.read(rel, rg): - buf += ch - text = bytes(buf).decode("utf-8", errors="replace") - if density and density != "deep": - okind = plugin.object_kind_of(rel) - if okind not in ("document", "code"): - raise ValueError("density_unsupported") - return _density_view(text, ext, density) - return text - finally: - await plugin.close() - - async def _read_full(self, path: str) -> tuple[str, bool]: - """Whole object content for export / tail: returns (text, partial). - partial=True when the connector capped the read (structured objects - above max_read_rows). The bare-cat size guard is not applied. Backs - export and tail; tail discards the partial flag (just wants the last - N lines), export surfaces it.""" - import json as _json + async def export(self, *a, **kw): + return await self.reads.export(*a, **kw) - cid, curi, rel, plugin = await self._open_path(path) - try: - st = await plugin.stat(rel) - if st.type == "dir": - raise IsADirectoryError(path) - okind = plugin.object_kind_of(rel) - if okind in ("table_rows", "record_collection", "message_stream"): - records = plugin.read_records(rel) - if records is not None: - out = [] - async for rec in records: - out.append(_json.dumps(rec, default=str, ensure_ascii=False)) - text = "\n".join(out) - # ctx.declare_partial is the channel structured connectors use - # to flag "we capped at max_read_rows". Read it back here so - # export tells the truth instead of silently returning a slice. - partial = bool(getattr(plugin.ctx, "was_partial", lambda _r: False)(rel)) - self._warn_if_huge_export(curi + rel, text) - return text, partial - ext = os.path.splitext(rel)[1].lower() - if ext in CONVERT_EXTS: - # On-read freshness (same as cat): a changed source fingerprint means the - # cached markdown is stale, so re-convert from the current source. - if await self._converted_md_stale(cid, rel, st.fingerprint): - raw = bytearray() - async for ch in plugin.read(rel): - raw += ch - text = await self.infra.converter.convert(bytes(raw), ext) - self._warn_if_huge_export(curi + rel, text) - return text, False - art = await self._read_artifact(self.ns, curi + rel, "converted_md") - if art is not None: - text = art.decode("utf-8", errors="replace") - self._warn_if_huge_export(curi + rel, text) - return text, False - elif curi.startswith(("web://", "github://")): - art = await self._read_artifact(self.ns, curi + rel, "converted_md") - if art is not None: - text = art.decode("utf-8", errors="replace") - self._warn_if_huge_export(curi + rel, text) - return text, False - if okind == "image" and self.cfg.description.enabled: - raw = bytearray() - async for ch in plugin.read(rel): - raw += ch - try: - text = await self.infra.vlm.describe(bytes(raw), ext) - self._warn_if_huge_export(curi + rel, text) - return text, False - except Exception: # noqa: BLE001 — fall through to the raw read on provider error - pass - buf = bytearray() - async for ch in plugin.read(rel): - buf += ch - text = bytes(buf).decode("utf-8", errors="replace") - self._warn_if_huge_export(curi + rel, text) - return text, False - finally: - await plugin.close() - - def _warn_if_huge_export(self, uri: str, text: str) -> None: - """Single-host export materializes the whole object in memory; warn on anything over - 100 MB so the operator sees the cost before the next OOM rather than after. A streaming - export path is the proper fix but is deferred — objects this large are rare on the - single-host deployment this guard covers, and the warning makes the cost explicit.""" - size = len(text.encode("utf-8", errors="ignore")) if text else 0 - if size > 100 * 1024 * 1024: - logger.warning( - "export %s materialized %d MB in memory (streaming export not yet implemented)", - uri, - size // (1024 * 1024), - ) + async def head(self, *a, **kw): + return await self.reads.head(*a, **kw) - async def export(self, path: str) -> tuple[str, bool]: - """Full content for `mfs export`: returns (text, partial). Honest - boundary — structured connectors with more rows than max_read_rows - return partial=True; the caller (API layer) surfaces it in the - CatResponse. The bare-cat size guard does not apply, but each - connector's own row cap does (true streaming export is deferred).""" - return await self._read_full(path) + async def tail(self, *a, **kw): + return await self.reads.tail(*a, **kw) - async def head(self, path: str, n: int = 20) -> str: - cid, curi, rel, plugin = await self._open_path(path) - try: - okind = plugin.object_kind_of(rel) - structured = okind in ("table_rows", "record_collection", "message_stream") - if structured: - # fast path: pre-cached first rows. The cache is capped at _HEAD_CACHE_N, so - # it's authoritative ONLY when it holds the whole object (< the cap) OR n fits - # within it; for a larger n on a capped cache, fall through to the live bounded - # query below — otherwise `head -n 200` would silently return just the 100 - # cached rows instead of 200 (head must give min(n, total), not min(n, cache)). - art = await self._read_artifact(self.ns, curi + rel, "head_cache") - if art is not None: - cached = art.decode("utf-8", errors="replace").splitlines() - if len(cached) < _HEAD_CACHE_N or n <= len(cached): - return "\n".join(cached[:n]) - else: - ext = os.path.splitext(rel)[1].lower() - # plain text / code / logs: stream just the first n lines so a large file - # never materializes and never trips bare-cat's size guard — head is exactly - # the escape hatch for big objects. Artifact-backed - # objects (pdf/docx/html, web/github pages, images) have bounded cached text, - # so they fall through to cat below. - if not ( - okind == "image" - or ext in CONVERT_EXTS - or curi.startswith(("web://", "github://")) - ): - lines: list[str] = [] - buf = b"" - async for chunk in plugin.read(rel): - buf += chunk - while len(lines) < n: - nl = buf.find(b"\n") - if nl < 0: - break - lines.append(buf[:nl].decode("utf-8", errors="replace")) - buf = buf[nl + 1 :] - if len(lines) >= n: - break - if len(lines) < n and buf: - lines.append(buf.decode("utf-8", errors="replace")) - return "\n".join(lines[:n]) - finally: - await plugin.close() - if structured: - return await self.cat(path, range=(0, n)) # bounded page, not the whole table - text = await self.cat(path) # artifact-backed text, bounded - return "\n".join(text.splitlines()[:n]) - - async def tail(self, path: str, n: int = 20) -> str: - if n <= 0: - return "" - # plain-text real local file: read the last n lines straight off disk (native - # accelerator / bounded reverse-read), so a huge log isn't fully materialized. - # Artifact-backed (pdf/docx/html, web/github) and structured objects fall back. - from ..common import accel - - cid, curi, rel, plugin = await self._open_path(path) - try: - okind = plugin.object_kind_of(rel) - if okind in ("table_rows", "record_collection", "message_stream"): - raise ValueError("tail_unsupported") - ext = os.path.splitext(rel)[1].lower() - plain_local = ( - curi.startswith("file://local") - and okind - not in ( - "image", - "table_rows", - "record_collection", - "message_stream", - "table_schema", - ) - and ext not in CONVERT_EXTS - ) - abs_file = (curi.replace("file://local", "", 1) + rel) if plain_local else None - finally: - await plugin.close() - if abs_file and os.path.isfile(abs_file): - return "\n".join(await asyncio.to_thread(accel.tail_lines, abs_file, n)) - text, _partial = await self._read_full( - path - ) # artifact-backed / structured / non-local; tail ignores the partial flag - return "\n".join(text.splitlines()[-n:]) - - async def grep( - self, pattern: str, path: str, top_k: int = 100, regex: bool = False - ) -> list[dict]: - """Dispatch: pushdown (file: none) -> BM25 (indexed scope) -> linear scan - (not_indexed objects in scope). The linear scan uses the native - accelerator (mfs_server_rs) when the object is a real local file, else falls - back to reading bytes + pure-Python regex.""" - from ..common import accel - from ..connectors.base import GrepOptions - - cid, curi, rel, plugin = await self._open_path(path) - scope_prefix = (curi + rel) if rel != "/" else None - try: - # _open_path only resolves which connector owns the prefix, not whether - # `rel` exists under it -- unlike ls/cat, nothing downstream fails loudly - # for a missing path (pushdown/BM25/linear-scan all just yield zero - # matches). Stat it explicitly so a bad path 404s like ls/cat instead of - # looking like a real, empty search. - await plugin.stat(rel) - results: list[dict] = [] - # 2a connector grep pushdown: exact, source-side (e.g. - # SQL ILIKE for structured connectors). Returns None when unsupported. - ocfg = plugin.ctx.object_config_for(rel) - try: - gen = await plugin.grep( - pattern, - rel, - GrepOptions( - pattern=pattern, - text_fields=ocfg.text_fields, - metadata_fields=ocfg.metadata_fields, - ), - ) - except Exception: # noqa: BLE001 - pushdown failure shouldn't kill grep - gen = None - if gen is not None: - async for gm in gen: - # Structured pushdown carries gm.locator (PK dict); text/code - # pushdown carries gm.line_no. locator.lines is 1-based - # half-open [s,e), so a single line n is [n, n+1] — not - # [n, n], which would round-trip as an empty slice. - loc = ( - gm.locator - if gm.locator is not None - else ({"lines": [gm.line_no, gm.line_no + 1]} if gm.line_no else None) - ) - results.append( - { - "source": curi + gm.path, - "locator": loc, - "content": gm.content, - "via": "pushdown", - } - ) - return results - # 2b BM25 over indexed objects in scope - expr = build_filter(self.ns, curi, scope_prefix) - hits = await asyncio.to_thread( - self.infra.milvus.sparse_search, self.ns, pattern, top_k, expr - ) - for h in hits: - e = h.get("entity", h) - results.append( - { - "source": e.get("object_uri"), - "locator": e.get("locator"), - "content": e.get("content"), - "via": "bm25", - } - ) - # 2c linear scan over not_indexed objects in scope (file connector) - root_abs = ( - curi.replace("file://local", "", 1) if curi.startswith("file://local") else None - ) - # Path-component boundary, same fix as build_filter: scope `/src` must match the - # object itself OR `/src/...`, NOT a sibling `/src-old`. Escape SQL LIKE wildcards - # ('_'/'%') in the literal prefix so a path with '_' doesn't over-match either. - not_idx = await self.objects.list_not_indexed_in_scope(cid, rel) - if len(not_idx) > _GREP_LINEAR_SCAN_MAX: - # don't silently scan a subset and imply it was exhaustive — tell the agent - # so it can narrow the path or index first. - results.append( - { - "source": None, - "locator": None, - "via": "notice", - "content": f"(grep linear scan capped at {_GREP_LINEAR_SCAN_MAX} of " - f"{len(not_idx)} not-indexed files in scope; narrow the path " - f"or run `mfs add` to index them for complete results)", - } - ) - for o in not_idx[:_GREP_LINEAR_SCAN_MAX]: - relp = o["object_uri"] - try: - abs_file = (root_abs + relp) if root_abs else None - if abs_file and os.path.isfile(abs_file): - # native (or pure-Python) streaming grep straight off disk - for ln, line in await asyncio.to_thread( - accel.linear_grep_file, abs_file, pattern, False, regex, 200 - ): - results.append( - { - "source": curi + relp, - "locator": {"lines": [ln, ln + 1]}, - "content": line, - "via": "linear", - } - ) - else: - rx = re.compile(pattern if regex else re.escape(pattern)) - buf = bytearray() - async for ch in plugin.read(relp): - buf += ch - text = bytes(buf).decode("utf-8", errors="replace") - for i, line in enumerate(text.splitlines(), 1): - if rx.search(line): - results.append( - { - "source": curi + relp, - "locator": {"lines": [i, i + 1]}, - "content": line, - "via": "linear", - } - ) - except Exception: # noqa: BLE001 - pass - return results - finally: - await plugin.close() + async def grep(self, *a, **kw): + return await self.reads.grep(*a, **kw) diff --git a/server/python/tests/test_cat_locator.py b/server/python/tests/test_cat_locator.py index f7f829e5..2a60cf1e 100644 --- a/server/python/tests/test_cat_locator.py +++ b/server/python/tests/test_cat_locator.py @@ -115,7 +115,7 @@ async def test_cat_with_matching_dict_locator_returns_the_record(tmp_path) -> No async def fake_open_path(path: str): return "cid", "postgres://db", "/orders", plugin - eng._open_path = fake_open_path # type: ignore[method-assign] + eng.reads.open_path = fake_open_path # type: ignore[method-assign] out = await eng.cat("postgres://db/orders", locator={"id": "ord_1001"}) @@ -131,7 +131,7 @@ async def test_cat_with_dict_locator_no_match_raises_locator_not_found(tmp_path) async def fake_open_path(path: str): return "cid", "postgres://db", "/orders", plugin - eng._open_path = fake_open_path # type: ignore[method-assign] + eng.reads.open_path = fake_open_path # type: ignore[method-assign] with pytest.raises(ValueError, match="locator_not_found"): await eng.cat("postgres://db/orders", locator={"id": "does_not_exist"}) diff --git a/server/python/tests/test_engine_read_service.py b/server/python/tests/test_engine_read_service.py new file mode 100644 index 00000000..f76e6be4 --- /dev/null +++ b/server/python/tests/test_engine_read_service.py @@ -0,0 +1,182 @@ +"""Independent unit tests for ReadService + text_views: constructed without an +Engine, without real Milvus/meta, injecting fakes. Covers the pure helpers and +the locator pair (open_path / match_connector) that ReadService owns directly +(D1: no Engine back-reference). +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from mfs_server.config import ServerConfig +from mfs_server.engine.components.reads import ReadService +from mfs_server.engine.components.reads.text_views import _density_view, _locator_matches + + +# --- pure helpers (text_views) --- + + +def test_density_view_peek_markdown_headings(): + text = "# Title\n\nsome prose\n## Sub\nmore" + assert _density_view(text, ".md", "peek") == "# Title\n## Sub" + + +def test_density_view_skim_includes_first_prose_line(): + out = _density_view("# Title\n\nfirst prose line\nsecond", ".md", "skim") + assert "# Title" in out + assert "first prose line" in out + + +def test_density_view_code_symbols(): + out = _density_view("def foo():\n pass\nclass Bar:\n pass\nrandom", ".py", "peek") + assert "def foo" in out + assert "class Bar" in out + + +def test_density_view_fallback_first_15_lines(): + text = "\n".join(f"line {i}" for i in range(20)) + assert _density_view(text, ".txt", "peek") == "\n".join(f"line {i}" for i in range(15)) + + +class _OCFG: + locator_fields = ["id"] + + +def test_locator_matches_dict_key(): + rec = {"id": "ord_1", "name": "x"} + assert _locator_matches(rec, _OCFG(), 0, {"id": "ord_1"}) is True + assert _locator_matches(rec, _OCFG(), 0, {"id": "other"}) is False + + +def test_locator_matches_row(): + assert _locator_matches({}, _OCFG(), 3, {"_row": 3}) is True + assert _locator_matches({}, _OCFG(), 3, {"_row": 1}) is False + + +def test_locator_matches_lines_key_ignored(): + # "lines" is framework-reserved, never a structured PK - must not match record #0. + assert _locator_matches({"id": "ord_1"}, _OCFG(), 0, {"lines": [1, 2]}) is False + + +def test_locator_matches_empty_locator_no_match(): + # empty locator must not silently return record #0 (the all([]) == True trap). + assert _locator_matches({"id": "x"}, _OCFG(), 0, {}) is False + + +# --- ReadService construction + locator pair (fakes, no Engine/Milvus) --- + + +class _FakeFactory: + def __init__(self): + self.open_path_calls = [] + self.resolve_target_calls = [] + + async def open_path(self, rows, path): + self.open_path_calls.append((rows, path)) + return SimpleNamespace( + cid="cid-fake", + connector_uri="postgres://db", + relpath="/orders", + built=SimpleNamespace(plugin="PLUGIN"), + ) + + def resolve_target(self, target): + self.resolve_target_calls.append(target) + return SimpleNamespace( + connector_uri="postgres://db", ctype="postgres", scheme="postgres", config={} + ) + + +class _FakeObjects: + def __init__(self, rows=None, has_any=False): + self._rows = rows or [] + self._has_any = has_any + + async def list_connectors_all(self): + return self._rows + + async def has_any_connector(self): + return self._has_any + + async def has_connector_uri(self, uri): + return any(r["root_uri"] == uri for r in self._rows) + + +class _FakeArtifacts: + async def read_artifact(self, ns, uri, kind): + return None + + async def converted_md_stale(self, cid, uri, fp): + return False + + +def _build_reads(factory=None, objects=None): + cfg = ServerConfig() + return ReadService( + cfg, + SimpleNamespace(), + factory or _FakeFactory(), + objects or _FakeObjects(), + _FakeArtifacts(), + ) + + +async def test_open_path_delegates_to_factory(): + factory = _FakeFactory() + objs = _FakeObjects(rows=[{"root_uri": "postgres://db"}]) + reads = _build_reads(factory=factory, objects=objs) + out = await reads.open_path("postgres://db/orders") + assert out == ("cid-fake", "postgres://db", "/orders", "PLUGIN") + assert factory.open_path_calls == [([{"root_uri": "postgres://db"}], "postgres://db/orders")] + + +async def test_match_connector_uses_locator(): + rows = [{"root_uri": "postgres://db", "id": "c1"}] + reads = _build_reads(objects=_FakeObjects(rows=rows)) + out = await reads.match_connector("postgres://db/orders") + assert out is not None + row, rel = out + assert row["root_uri"] == "postgres://db" + + +async def test_match_connector_none_when_no_rows(): + reads = _build_reads(objects=_FakeObjects(rows=[])) + assert await reads.match_connector("postgres://db/orders") is None + + +async def test_search_empty_namespace_short_circuits(): + reads = _build_reads(objects=_FakeObjects(has_any=False)) + assert await reads.search("query") == [] + + +async def test_search_top_k_too_large(): + reads = _build_reads(objects=_FakeObjects(has_any=True)) + with pytest.raises(ValueError, match="top_k_too_large"): + await reads.search("q", top_k=10**9) + + +async def test_resolve_connector_uri_matches_registered(): + rows = [{"root_uri": "postgres://db", "id": "c1"}] + reads = _build_reads(objects=_FakeObjects(rows=rows)) + curi, prefix = await reads.resolve_connector_uri("postgres://db/orders") + assert curi == "postgres://db" + assert prefix == "postgres://db/orders" + + +async def test_resolve_connector_uri_root_rel_none(): + rows = [{"root_uri": "postgres://db", "id": "c1"}] + reads = _build_reads(objects=_FakeObjects(rows=rows)) + curi, prefix = await reads.resolve_connector_uri("postgres://db") + assert curi == "postgres://db" + assert prefix is None + + +async def test_resolve_connector_uri_fallback_to_factory(): + factory = _FakeFactory() + reads = _build_reads(factory=factory, objects=_FakeObjects(rows=[])) + curi, prefix = await reads.resolve_connector_uri("postgres://other") + assert curi == "postgres://db" + assert prefix is None + assert factory.resolve_target_calls == ["postgres://other"] diff --git a/server/python/tests/test_engine_tail.py b/server/python/tests/test_engine_tail.py index 0a4e41d9..9a8b053c 100644 --- a/server/python/tests/test_engine_tail.py +++ b/server/python/tests/test_engine_tail.py @@ -32,7 +32,7 @@ async def test_tail_rejects_unstable_structured_objects(tmp_path, okind: str) -> async def fake_open_path(path: str): return "cid", "postgres://db", "/rows.jsonl", plugin - eng._open_path = fake_open_path # type: ignore[method-assign] + eng.reads.open_path = fake_open_path # type: ignore[method-assign] with pytest.raises(ValueError, match="tail_unsupported"): await eng.tail("postgres://db/rows.jsonl", 5) diff --git a/server/python/tests/test_grep_not_found.py b/server/python/tests/test_grep_not_found.py index 42629950..c5761798 100644 --- a/server/python/tests/test_grep_not_found.py +++ b/server/python/tests/test_grep_not_found.py @@ -53,7 +53,7 @@ async def test_grep_missing_path_raises_file_not_found(tmp_path) -> None: async def fake_open_path(path: str): return "cid", "file://local/root", "/does/not/exist", plugin - eng._open_path = fake_open_path # type: ignore[method-assign] + eng.reads.open_path = fake_open_path # type: ignore[method-assign] with pytest.raises(FileNotFoundError): await eng.grep("needle", "file://local/root/does/not/exist") From c56a1fe55f4ac73da3d70eeaf8ec96b2010d4264 Mon Sep 17 00:00:00 2001 From: code2t Date: Tue, 21 Jul 2026 13:35:23 +0800 Subject: [PATCH 02/14] refactor(engine): extract UploadService for tar + manifest-diff upload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move ingest_upload/files_manifest/files_upload out of the Engine god-class into a dedicated UploadService component (engine-redesign §4.7 stage 5). Engine delegates to self.upload; the 3 public upload methods become thin forwards. Method bodies are unchanged; only self. resolution targets moved. Reuses the already-public IngestOrchestrator entrypoints (register_or_get_connector / open_sync_job / drain_job). - New engine/components/upload.py: UploadService + module-level _norm_rel / _validate_upload_member (zip-slip guard). DI is precise: (cfg, infra, objects, ingest) - no unused factory/artifacts (UploadService touches infra.artifact_cache.files_root + meta, not the cache service). - engine.py: 887 -> 635 lines. Assemble self.upload; 3 upload methods thin delegates; drop _staging_root/_staging_connector + module-level _norm_rel/_validate_upload_member + now-orphan FileStateStore import. - BundleValidator / StagingLocator class extraction + shared Template Method are follow-up (plan §2.5) - this round is pure relocation. - New tests/test_engine_upload_service.py: 9 cases, constructs UploadService without an Engine/meta, covers _norm_rel / _validate_upload_member (symlink/hardlink/absolute/traversal) + the staging locator (_staging_root sha1 sub / _staging_connector stable URI). Verified: 442 passed, 9 skipped; ruff format --check + ruff check clean. --- .../mfs_server/engine/components/upload.py | 296 ++++++++++++++++++ server/python/src/mfs_server/engine/engine.py | 275 +--------------- .../tests/test_engine_upload_service.py | 114 +++++++ 3 files changed, 421 insertions(+), 264 deletions(-) create mode 100644 server/python/src/mfs_server/engine/components/upload.py create mode 100644 server/python/tests/test_engine_upload_service.py diff --git a/server/python/src/mfs_server/engine/components/upload.py b/server/python/src/mfs_server/engine/components/upload.py new file mode 100644 index 00000000..bfedd41e --- /dev/null +++ b/server/python/src/mfs_server/engine/components/upload.py @@ -0,0 +1,296 @@ +"""UploadService: tar + manifest-diff upload protocol. + +Extracted verbatim from the Engine god-class (engine-redesign §4.7 stage 5). +Method bodies are unchanged; only `self.` resolution targets moved (see +docs-dev/engine-redesign-read-upload.md §3.3). BundleValidator / StagingLocator +extraction and the shared five-step Template Method are follow-up (§2.5) - +this round is pure relocation to preserve behavior. +""" + +from __future__ import annotations + +import json +import os + +from ...storage.file_state import FileStateStore + + +def _norm_rel(p: str) -> str: + """Connector-relative path with a single leading '/' (file_state / object_uri convention).""" + return "/" + p.lstrip("/") + + +def _validate_upload_member(m) -> None: + """Reject archive members that tarfile could materialize outside the staging tree.""" + import posixpath as _posixpath + + if m.issym() or m.islnk(): + raise ValueError(f"links not allowed in upload: {m.name}") + if not (m.isfile() or m.isdir()): + raise ValueError(f"unsupported member in upload: {m.name}") + rel = str(m.name or "") + if not rel or _posixpath.isabs(rel) or any(part == ".." for part in rel.split("/")): + raise ValueError(f"unsafe path in archive: {rel}") + if _posixpath.normpath(rel) in ("", ".") and not m.isdir(): + raise ValueError(f"unsafe path in archive: {rel}") + + +class UploadService: + """CS upload (tar snapshot) + manifest-diff upload (byte-diff + index-diff). + Reuses the already-public IngestOrchestrator entrypoints (register_or_get_connector / + open_sync_job / drain_job) to stage + sync. Holds no mutable state.""" + + def __init__(self, cfg, infra, objects, ingest): + self._cfg = cfg + self._infra = infra + self._objects = objects + self._ingest = ingest + self._ns = cfg.namespace + + async def ingest_upload( + self, name: str, data: bytes, fmt: str = "tar", process: bool = True + ) -> dict: + """CS upload flow: client/server don't share a fs, so the client + ships a tar(.gz) of the tree (?name=