[CodyTonn]

resolve #166

Background

server/python/src/mfs_server/engine/engine.py was a 1504-line god-class bearing 9 responsibilities (infra assembly, pipeline singleton, connector factory, ingest write orchestration, worker scheduling, upload protocol, artifact cache, read-path retrieval, connector management). Prior PRs extracted InfraStack / PipelineSupervisor / IngestOrchestrator / ConnectorFactory / ObjectRepository / ArtifactCacheService. This PR extracts the remaining 4 business components, completes the stage-6 strategy tables, and pulls out value objects, reducing Engine to a pure Facade.

Overview

Engine shrinks from 1504 to 227 lines (-85%), now a pure Facade (thin delegates + lifecycle + _write_object_row test delegate). 9 commits, 18 files, +2283/-1378.

Commit	Component	engine.py	Notes
ca9c65f	ReadService	1504->887	8 read methods + pure helpers + locators
c10db97	UploadService	887->635	tar + manifest-diff upload
fec96bb	WorkerScheduler	635->511	queue layer + concurrent workers + reclaim
6ba2450	ConnectorManager	511->247	wire probe/estimate/inspect/remove
7fbf50d	_await_worker_drained	247	extract remove_connector's while-loop to a method
1246532	GrepStrategy chain	-	grep 3-branch -> Pushdown->BM25->Linear chain
6b5e7ea	CatRouter	-	cat 6-branch -> 7-strategy first-match table
f8b96a9	BundleValidator + StagingLocator	-	upload validation + staging classes + shared drain tail
894b9b4	CircuitBreaker/BackoffPolicy/ErrorClassifier	-	retry/circuit-breaker/error value objects

Architecture

Engine composes 10 collaborating components (6 business + 2 infra + repository + process singleton); dependencies are unidirectional and acyclic.

HTTP/CLI -> Engine (Facade, 227 lines)
            ├─ reads: ReadService (+ CatRouter + GrepChain)
            ├─ upload: UploadService (+ BundleValidator + StagingLocator)
            ├─ worker: WorkerScheduler
            ├─ connector_manager: ConnectorManager
            ├─ ingest: IngestOrchestrator (+ policies: CircuitBreaker/BackoffPolicy/ErrorClassifier)
            ├─ pipeline: PipelineSupervisor
            ├─ factory: ConnectorFactory
            ├─ objects: ObjectRepository
            ├─ artifacts: ArtifactCacheService
            └─ infra: InfraStack

Design patterns:

• ReadService locators (open_path/match_connector) are public methods connecting directly to ConnectorFactory+ObjectRepository, with no back-reference to Engine (decision D1: direct-connect + public method, not delayed-lambda)
• cat uses a CatRouter first-match strategy table (7 strategies: Meta/LocatorLines/LocatorRecord/StructuredStream/ArtifactText/ImageDescription/PlainText); the D3 fallthrough (image+vlm failure) is expressed as read() -> None so the chain continues
• grep uses a GrepStrategy chain (Pushdown->BM25->Linear), match() -> list|None (None=continue, list=terminal); Pushdown is terminal on hit, BM25 appends and continues, Linear appends and is terminal
• UploadService's BundleValidator/StagingLocator are standalone classes; _drain_after_upload is a shared tail (partial Template Method)
• IngestOrchestrator's retry/circuit-breaker logic is extracted into engine/policies.py value objects

Behavior Equivalence

Zero external behavior change: the 20 call sites in api/app.py and server/__main__.py keep their signatures + the infra attribute.

Each component's method bodies are migrated verbatim; only self.<dep> resolution targets change (e.g. self.infra.milvus -> self._infra.milvus, self.objects -> self._objects). Key invariants are preserved explicitly:

• WorkerScheduler two-layer try/except: the outer except Exception: jid = None (coroutine-immortal, sqlite single-worker semantics) must stay in run_forever._loop; the inner except (business-level: drain the queue) stays in run_worker_once
• cat density only applies to ArtifactText/PlainText (the original D1/D2/E paths), via an _apply_density helper; meta/locator/structured/image strategies return before density (matching the original early returns)
• grep D3 fallthrough: ImageDescriptionStrategy returns None on vlm failure, the chain falls to PlainTextStrategy
• ConnectorManager.remove_connector's "wait for worker to leave running / stale heartbeat before taking over" invariant is extracted as _await_worker_drained
• The chunk exists iff a committed objects row points at it invariant is not touched by the paths in this PR (read/upload/worker don't touch pipeline finalize)

Tests

• 447 passed, 9 skipped (9 skip = missing optional connector SDKs)
• 4 new independent unit-test files (constructed without an Engine/Milvus):
  • test_engine_read_service.py (16 cases: text_views pure helpers + locators + search short-circuit + resolve_connector_uri)
  • test_engine_upload_service.py (9 cases: _norm_rel + _validate_upload_member zip-slip + staging locator)
  • test_engine_worker.py (5 cases: _resolve_concurrency + queue-empty short-circuit)
  • cat/grep strategies are covered by existing test_cat_locator / test_grep_not_found / e2e tests
• 4 test monkeypatch targets updated (eng._open_path -> eng.reads.open_path, the D1 direct-connect strategy); test_engine_forwards_to_ingest's bind_remover assertion changed to connector_manager.remove_connector (design change)
• ruff format --check + ruff check clean (CI gate)

Files

File	Change
engine/components/reads/{__init__,text_views,read_service}.py	new: ReadService + pure helpers + read-path constants
engine/components/reads/cat_strategies.py	new: CatRouter + 7 CatStrategy + _apply_density
engine/components/reads/grep_strategies.py	new: GrepContext + GrepStrategy chain (Pushdown/BM25/Linear)
engine/components/upload.py	new: UploadService + BundleValidator + StagingLocator + _drain tail
engine/worker.py	new: WorkerScheduler + two worker constants
engine/manage.py	new: ConnectorManager (probe/estimate/inspect/remove + _await_worker_drained)
engine/policies.py	new: ErrorClassifier + BackoffPolicy + CircuitBreaker
engine/engine.py	1504->227: Facade forwarding + lifecycle
engine/ingest.py	process_with_retry/run_job_loop use value objects
4 tests	monkeypatch target / _Cfg fake / bind_remover assertion updates

Risk & Rollback

• Low risk: method bodies migrated verbatim, only dependency resolution targets change; invariants preserved explicitly with comments
• Rollback granularity: each of the 9 commits is independently green and revertible; a full rollback is recommended (components have DI dependencies on each other)
• Not pushed: local commits, no release triggered
• Remaining follow-up (not in this PR): UploadService staging-step full Template Method (extractall vs renames/bytes/deletions differ enough that a shared template would risk drift; left as a long-term follow-up with no behavioral benefit)

Verification

cd server/python
uv run --extra dev pytest                                      # 447 passed, 9 skipped
uv run --extra dev ruff format --check src/ tests/             # CI gate
uv run --extra dev ruff check src/mfs_server/engine/           # All checks passed