Refactor(engine): Engine God-Class Refactor - Extract 4 Business Components + Stage 6 Strategy Tables + Value Objects

server/python/src/mfs_server/engine/components/artifact_cache.py

@@ -1,13 +1,8 @@
 """ArtifactCacheService — artifact bytes + metadata row + LRU + freshness.
 
-Consolidates the ``artifact_cache`` table SQL and LRU eviction previously inline in
-``engine.py`` (six ``_*_artifact`` / ``_evict`` / ``_converted_md_stale`` methods plus
-the rename row-rewrite in ``_index_object``). The storage-layer
-``LocalArtifactCache`` (bytes CRUD + path) is injected; this service owns the
-metadata-row bookkeeping and the LRU throttle counter. Mirrors the
-``ObjectRepository`` pattern (component-local ``_now()``, public methods, SQL migrated
-verbatim with zero behavior change). ``Engine`` keeps thin delegate methods so call
-sites + ``ArtifactStoreAdapter`` wiring stay unchanged.
+Owns the ``artifact_cache`` table SQL + LRU eviction + freshness; the
+storage-layer ``LocalArtifactCache`` (bytes CRUD + path) is injected. Mirrors the
+``ObjectRepository`` pattern (component-local ``_now()``, public methods).
 """
 
 from __future__ import annotations
@@ -32,11 +27,9 @@ class ArtifactCacheService:
     image text, head snapshot, message_stream raw_records) per object, with LRU size
     eviction and a ``source_key`` freshness check.
 
-    ``Engine`` delegates every ``self.meta.execute/fetchone/fetchall`` against this
-    table to ``self.artifacts.xxx(...)``; SQL is migrated verbatim with zero behavior
-    change. The storage-layer ``LocalArtifactCache`` (bytes + paths) stays on
-    ``Engine`` (it has non-cache call sites — uploads staging / raw_records GC) and is
-    injected here for the bytes CRUD this service needs.
+    The storage-layer ``LocalArtifactCache`` (bytes + paths) lives on InfraStack
+    (it has non-cache call sites: uploads staging / raw_records GC) and is injected
+    here for the bytes CRUD this service needs.
     """
 
     def __init__(self, cfg, meta, artifact_cache, objects):
@@ -124,7 +117,7 @@ async def converted_md_stale(self, cid: str, object_uri: str, live_fp: str | Non
         the cached converted_md no longer reflects the source. `live_fp` comes from the stat()
         the read already did, so this costs one local metadata lookup. A connector that yields
         no fingerprint (live_fp falsy) can't be cheaply checked -> serve the cached copy (the
-        deferred snapshot/recheck path for those is TODO §10.9)."""
+        deferred snapshot/recheck path for those is TODO)."""
         if not live_fp:
             return False
         stored = await self._obj.get_object_fingerprint(cid, object_uri)

server/python/src/mfs_server/engine/components/connector_factory.py

@@ -123,7 +123,7 @@ class CredentialService:
                env:/file: references)
     - resolve: turn env:/file: references into real values before plugin build
 
-    Security invariants (engine-redesign.md §5):
+    Security invariants:
     1. Credentials are resolved ONLY through resolve(); business code never reads os.environ[...] directly.
     2. config_json is persisted ONLY after redact().
     3. resolve() raises explicitly on unimplemented schemes (secret:/vault:), never silently using them as literal tokens.
@@ -387,11 +387,9 @@ def match(rows: list[dict], path: str) -> tuple[dict, str] | None:
 class ConnectorFactory:
     """Factory facade: target resolution / credential redact+resolve / plugin build.
 
-    Does NOT touch connectors-table SQL (that's ``ObjectRepository``). To keep this
-    change merge-safe, ingest orchestration and the read path stay on ``Engine``;
-    the factory is invoked only through thin Engine delegates. ``cfg`` + ``meta``
-    are held solely to build ``ConnectorStateStore`` / ``FileStateStore`` (plugin
-    dependencies).
+    Does NOT touch connectors-table SQL (that's ``ObjectRepository``). ``cfg`` +
+    ``meta`` are held solely to build ``ConnectorStateStore`` / ``FileStateStore``
+    (plugin dependencies).
     """
 
     def __init__(self, cfg: ServerConfig, meta: MetadataStore):

server/python/src/mfs_server/engine/components/object_repository.py

@@ -1,10 +1,9 @@
 """
 ``ObjectRepository`` + task/job/connector state machine.
 
-Consolidates all SQL for the ``objects`` / ``object_tasks`` / ``connector_jobs`` /
-``connectors`` tables and the status-transition guard (``advance_task``) so the
-invariants previously expressed only in ``WHERE status=?`` guards and inline
-comments live as a single queryable transition table.
+Owns all SQL for the ``objects`` / ``object_tasks`` / ``connector_jobs`` /
+``connectors`` tables and the status-transition guard (``advance_task``); the
+``WHERE status=?`` invariants live as a single queryable transition table.
 """
 
 from __future__ import annotations
@@ -93,11 +92,6 @@ class ConnectorStatus(str, Enum):
 class ObjectRepository:
     """SQL repository + state machine for the ``objects`` / ``object_tasks`` /
     ``connector_jobs`` / ``connectors`` tables.
-
-    ``Engine`` delegates every ``self.meta.execute/fetchone/fetchall`` against these
-    four tables to ``self.objects.xxx(...)``; SQL is migrated verbatim with zero
-    behavior change. ``Engine`` keeps the non-SQL orchestration and the thin
-    delegate methods called directly by tests.
     """
 
     def __init__(self, meta, cfg):
@@ -123,7 +117,7 @@ async def advance_task(
         on ``status=from_status`` and returns the rowcount (``won``). ``won == 0``
         means a concurrent cancel/remove beat us — the caller reconciles (e.g. delete
         orphan chunks in ``_on_pipeline_object_indexed``). Preserves the original
-        ``WHERE id=? AND status='running'`` guard + ``won`` semantics (§4.4 / §5).
+        ``WHERE id=? AND status='running'`` guard + ``won`` semantics.
 
         Only used on the paths that originally carried a ``status='running'`` guard
         (pipeline/inline succeeded, pipeline failed). Unguarded terminal writes
@@ -455,8 +449,8 @@ async def summarize_jobs_by_status(self, cid: str) -> list[dict]:
     async def write_object_row(
         self, cid: str, relpath: str, st, indexable: bool, search_status: str, chunk_count: int
     ) -> None:
-        """UPSERT the ``objects`` registry row. Shared by the inline _index_object tail,
-        the rename branch, and the pipeline success hook."""
+        """UPSERT the ``objects`` registry row. Shared by the inline index path, the
+        rename branch, and the pipeline success hook."""
         import os
 
         await self._meta.execute(

server/python/src/mfs_server/engine/components/reads/__init__.py

@@ -0,0 +1,10 @@
+"""ReadService: search/ls/cat/head/tail/grep/export/resolve_connector_uri.
+
+Engine forwards to self.reads. The locator pair (open_path / match_connector)
+connects directly to ConnectorFactory + ObjectRepository with no reverse
+reference to Engine.
+"""
+
+from .read_service import ReadService
+
+__all__ = ["ReadService"]

server/python/src/mfs_server/engine/components/reads/cat_strategies.py

@@ -0,0 +1,257 @@
+"""CatRouter + CatStrategy: first-match dispatch over cat's branch matrix.
+
+Each strategy `applies(ctx)` + `read(ctx) -> Any | None`: None means
+"not applicable / try next", a non-None value is terminal. Density only applies
+to the artifact-backed / plain-text paths, so it lives in those strategies via
+_apply_density; the locator / structured / meta / image paths return before
+density.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+from ....common.converter import CONVERT_EXTS
+from .text_views import _BARE_CAT_MAX_BYTES, _density_view, _locator_matches
+
+
+@dataclass
+class CatContext:
+    path: str
+    range: tuple[int, int] | None
+    meta: bool
+    density: str | None
+    locator: dict | None
+    cid: str
+    curi: str
+    rel: str
+    plugin: object
+    st: object
+    okind: str
+    structured: bool
+    ext: str
+
+
+class CatStrategy(Protocol):
+    def applies(self, ctx: CatContext) -> bool: ...
+
+    async def read(self, ctx: CatContext) -> Any | None: ...
+
+
+def _apply_density(text: str, ctx: CatContext) -> str:
+    """density view only over document/code; matches cat's tail check."""
+    if ctx.density and ctx.density != "deep":
+        if ctx.okind not in ("document", "code"):
+            raise ValueError("density_unsupported")
+        return _density_view(text, ctx.ext, ctx.density)
+    return text
+
+
+class MetaStrategy:
+    def applies(self, ctx: CatContext) -> bool:
+        return ctx.meta
+
+    async def read(self, ctx: CatContext) -> Any | None:
+        return {
+            "source": ctx.curi + ctx.rel,
+            "media_type": ctx.st.media_type,
+            "size_hint": ctx.st.size_hint,
+            "fingerprint": ctx.st.fingerprint,
+        }
+
+
+class LocatorLinesStrategy:
+    """locator with reserved "lines" key (1-based half-open [s,e)) on a
+    non-structured object -> slice by line range. plugin.read takes 0-based."""
+
+    def applies(self, ctx: CatContext) -> bool:
+        return (
+            ctx.locator is not None
+            and isinstance(ctx.locator, dict)
+            and "lines" in ctx.locator
+            and len(ctx.locator) == 1
+            and not ctx.structured
+        )
+
+    async def read(self, ctx: CatContext) -> Any | None:
+        from ....connectors.base import Range
+
+        s, e = int(ctx.locator["lines"][0]), int(ctx.locator["lines"][1])
+        rg = Range(max(0, s - 1), max(0, e - 1))
+        buf = bytearray()
+        async for ch in ctx.plugin.read(ctx.rel, rg):
+            buf += ch
+        return bytes(buf).decode("utf-8", errors="replace")
+
+
+class LocatorRecordStrategy:
+    """locator (PK dict) -> reopen a single structured record."""
+
+    def applies(self, ctx: CatContext) -> bool:
+        return ctx.locator is not None
+
+    async def read(self, ctx: CatContext) -> Any | None:
+        import json as _json
+        from contextlib import aclosing
+
+        records = ctx.plugin.read_records(ctx.rel)
+        if records is None:
+            raise ValueError("range_unsupported")  # not a structured object
+        ocfg = ctx.plugin.ctx.object_config_for(ctx.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, ctx.locator):
+                    return {
+                        "source": ctx.curi + ctx.rel,
+                        "locator": ctx.locator,
+                        "content": _json.dumps(rec, default=str, ensure_ascii=False),
+                    }
+                i += 1
+        raise ValueError("locator_not_found")
+
+
+class StructuredStreamStrategy:
+    """structured object, no locator -> JSONL stream (range=None) or paged slice."""
+
+    def applies(self, ctx: CatContext) -> bool:
+        return ctx.structured
+
+    async def read(self, ctx: CatContext) -> Any | None:
+        import json as _json
+        from contextlib import aclosing
+
+        from ....connectors.base import Range
+
+        if ctx.range is None:
+            # Bare cat of a structured object: stream records as JSONL up to
+            # _BARE_CAT_MAX_BYTES; large ones raise object_too_large_for_cat so the
+            # caller falls back to head / cat --range / export.
+            records = ctx.plugin.read_records(ctx.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 = ctx.range[0], ctx.range[1]
+        # Pass Range(0, end) - LIMIT-only - and slice [start, end) here. Connectors
+        # disagree on honoring Range (DB push OFFSET+LIMIT, SaaS ignore it), yet all
+        # declare paged_cat=True. offset=0 + single `i >= start` slice is correct for
+        # both (see human_todo [dborder/D65] for restoring true offset pushdown).
+        records = ctx.plugin.read_records(ctx.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)
+        return None
+
+
+class ArtifactTextStrategy:
+    """converted-markdown artifact: pdf/docx/html (CONVERT_EXTS) AND web/github
+    pages whose .md is generated at ingest. On-read freshness via fingerprint.
+    Returns None if no artifact (chain falls through to image / plain text)."""
+
+    def __init__(self, artifacts, infra, ns):
+        self._artifacts = artifacts
+        self._infra = infra
+        self._ns = ns
+
+    def applies(self, ctx: CatContext) -> bool:
+        return ctx.ext in CONVERT_EXTS or ctx.curi.startswith(("web://", "github://"))
+
+    async def read(self, ctx: CatContext) -> Any | None:
+        text: str | None = None
+        if ctx.ext in CONVERT_EXTS:
+            # stat() already fetched the live fingerprint; if it changed, the cached
+            # markdown is stale -> re-convert from the current source.
+            if await self._artifacts.converted_md_stale(ctx.cid, ctx.rel, ctx.st.fingerprint):
+                raw = bytearray()
+                async for ch in ctx.plugin.read(ctx.rel):
+                    raw += ch
+                text = await self._infra.converter.convert(bytes(raw), ctx.ext)
+            else:
+                art = await self._artifacts.read_artifact(
+                    self._ns, ctx.curi + ctx.rel, "converted_md"
+                )
+                if art is not None:
+                    text = art.decode("utf-8", errors="replace")
+        else:  # web:// / github:// - generated markdown, read from artifact store
+            art = await self._artifacts.read_artifact(self._ns, ctx.curi + ctx.rel, "converted_md")
+            if art is not None:
+                text = art.decode("utf-8", errors="replace")
+        if text is None:
+            return None  # no artifact -> fall through to image / plain text
+        return _apply_density(text, ctx)
+
+
+class ImageDescriptionStrategy:
+    """image + description.enabled -> VLM describe. Returns None on provider
+    failure so the chain falls through to the raw read (original try/except pass)."""
+
+    def __init__(self, infra, cfg):
+        self._infra = infra
+        self._cfg = cfg
+
+    def applies(self, ctx: CatContext) -> bool:
+        return ctx.okind == "image" and self._cfg.description.enabled
+
+    async def read(self, ctx: CatContext) -> Any | None:
+        raw = bytearray()
+        async for ch in ctx.plugin.read(ctx.rel):
+            raw += ch
+        try:
+            return await self._infra.vlm.describe(bytes(raw), ctx.ext)
+        except Exception:  # noqa: BLE001 - fall through to the raw read on provider error
+            return None
+
+
+class PlainTextStrategy:
+    """fallback: raw text read (with the bare-cat size guard). Density applies."""
+
+    def applies(self, ctx: CatContext) -> bool:
+        return True
+
+    async def read(self, ctx: CatContext) -> Any | None:
+        from ....connectors.base import Range
+
+        if ctx.range is None and ctx.st.size_hint and ctx.st.size_hint > _BARE_CAT_MAX_BYTES:
+            raise ValueError("object_too_large_for_cat")
+        rg = Range(ctx.range[0], ctx.range[1]) if ctx.range else None
+        buf = bytearray()
+        async for ch in ctx.plugin.read(ctx.rel, rg):
+            buf += ch
+        text = bytes(buf).decode("utf-8", errors="replace")
+        return _apply_density(text, ctx)
+
+
+class CatRouter:
+    """Ordered first-match strategy table."""
+
+    def __init__(self, strategies: list[CatStrategy]):
+        self._strategies = strategies
+
+    async def read(self, ctx: CatContext) -> Any:
+        for s in self._strategies:
+            if s.applies(ctx):
+                result = await s.read(ctx)
+                if result is not None:
+                    return result
+        raise ValueError("cat_unsupported")