feat(engine): allow [[objects]] to override per-object task priority

docs/connectors.md

@@ -151,6 +151,21 @@ Gmail, Discord, Feishu) ship presets, so `[[objects]]` is optional there — use
 only to override the default fields. See [Search and browse](search-and-browse.md#locate-exact-hits)
 for how these become `content`, `locator`, and `metadata.fields` in a result.
 
+### Processing order within a connector
+
+Add `priority` to any `[[objects]]` block to change which of *that connector's own*
+objects get indexed first during a sync (lower runs earlier; unmatched objects keep the
+connector's own default order):
+
+```toml
+[[objects]]
+match    = "archive/**"
+priority = 900   # push a low-value subtree to the back of the queue
+```
+
+This only reorders work inside one connector's sync job — it does not affect scheduling
+across different connectors running at the same time.
+
 ## After ingest
 
 Every scheme answers the same commands:

docs/ingest-pipeline.md

@@ -74,7 +74,11 @@ Notes:
 
 - **Queue ① is durable.** `ObjectTask` rows sit in the metadata database, so a
   worker crash never loses pending work. Workers claim by `(priority ASC,
-  started_at ASC)`.
+  started_at ASC)`. `priority` defaults to the connector's own bucketing (e.g.
+  the file connector's entrypoint/src/tests/generated heuristic) but a user
+  `[[objects]] priority` match overrides it — see
+  [Connectors](connectors.md#processing-order-within-a-connector). This orders
+  work only within one connector's own sync job, not across connectors.
 - **The Producer pool is okind-dispatched.** `select_producer(okind, ctx)` maps
   the eight object kinds (`document`, `code`, `text_blob`, `image`,
   `message_stream`, `table_rows`, `record_collection`, `table_schema`) onto five

server/python/src/mfs_server/connectors/base.py

@@ -231,6 +231,11 @@ class ObjectConfig:
     indexable: bool = True
     chunk_max: int = 1_000_000
     group_by: Optional[str] = None  # message_stream: override the auto-detected thread key
+    # Task processing order within THIS connector's own sync job — lower runs earlier
+    # (matches the file connector's built-in task_priority() bucket scale, e.g. -350 for
+    # entrypoints). None (default) means "no user override, use the connector's own
+    # task_priority()". Does not affect ordering across different connectors/jobs.
+    priority: Optional[int] = None
     # Optional Python str.format template applied per record before chunking.
     # Lets message presets render `alice: 部署炸了, 503 spike` instead of the
     # default labeled `text: 部署炸了, 503 spike`, which embeds better and

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

@@ -705,14 +705,17 @@ async def _drain_job(
                         # 'explicit_only' (yielded events, e.g. upload) honor them
                         continue  # never-delete connectors (slack/gmail) keep the index
                     tid = uuid.uuid4().hex
+                    # a user [[objects]] `priority =` match overrides the connector's own
+                    # task_priority() default (e.g. file's README/tests/vendor buckets).
+                    user_priority = plugin.ctx.object_config_for(ch.uri).priority
                     await self.objects.insert_task(
                         tid,
                         job_id,
                         cid,
                         ch.uri,
                         ch.old_uri,
                         ch.kind,
-                        plugin.task_priority(ch),
+                        user_priority if user_priority is not None else plugin.task_priority(ch),
                     )
                     if ch.kind != "deleted":
                         # Accumulate the dir tree (okind passed in — no extra DB hit, §6.4.1).

server/python/tests/test_engine_object_priority_e2e.py

@@ -0,0 +1,99 @@
+"""End-to-end: a user `[[objects]] priority=` override changes the processing order of a
+real `add()` on a real file connector — drives the actual `plugin.sync()` enumeration +
+`_drain_job` task insertion, no fakes on that path. Verifies both that the override lands
+on the right objects and that untouched objects still fall back to the connector's own
+`task_priority()` buckets (§6.3), and that the DB claim order (priority ASC, existing
+behavior) reflects it end to end."""
+
+from __future__ import annotations
+
+from mfs_server.config import ServerConfig
+from mfs_server.connectors.registry import load_builtin
+from mfs_server.engine.engine import Engine
+
+
+class _NoopMilvus:
+    def delete_by_connector(self, *args, **kwargs):
+        return None
+
+
+class _NoopJobLane:
+    def register_job(self, *args, **kwargs):
+        return None
+
+    def on_sync_done(self, *args, **kwargs):
+        return None
+
+    def on_yield_object_change(self, *args, **kwargs):
+        return None
+
+    def evict_job(self, *args, **kwargs):
+        return None
+
+
+async def _build_engine(tmp_path) -> Engine:
+    load_builtin()
+    cfg = ServerConfig()
+    cfg.metadata.backend = "sqlite"
+    cfg.metadata.path = str(tmp_path / "metadata.db")
+    cfg.transformation_cache.backend = "sqlite"
+    cfg.transformation_cache.db_path = str(tmp_path / "tx.db")
+    cfg.artifact_cache.root = str(tmp_path / "artifacts")
+    cfg.milvus.uri = str(tmp_path / "milvus.db")
+    eng = Engine(cfg)
+    eng.milvus = _NoopMilvus()
+    eng._job_lane = _NoopJobLane()
+    await eng.meta.connect()
+    await eng.meta.init_schema()
+    return eng
+
+
+async def test_objects_priority_override_wins_and_falls_back_end_to_end(tmp_path):
+    eng = await _build_engine(tmp_path)
+    try:
+        root = tmp_path / "repo"
+        (root / "src").mkdir(parents=True)
+        (root / "archive").mkdir()
+        (root / "README.md").write_text("entrypoint")
+        (root / "notes.md").write_text("scratch notes")
+        (root / "src" / "main.py").write_text("print(1)")
+        (root / "archive" / "old.txt").write_text("stale")
+
+        root_uri = f"file://local{root}"
+        await eng.add(
+            str(root),
+            config={
+                "root": str(root),
+                "client_id": "local",
+                "objects": [
+                    # user override: bump archive/* to the back of the line
+                    {"match": "archive/*", "priority": 900},
+                    # user override: pull notes.md ahead of everything, including
+                    # the file connector's own -350 entrypoint bucket
+                    {"match": "notes.md", "priority": -500},
+                ],
+            },
+            process=False,  # enumerate + insert_task only; no embedding/Milvus needed
+        )
+
+        cid = await eng.objects.get_connector_id_by_uri(root_uri)
+        assert cid is not None
+        rows = await eng.meta.fetchall(
+            "SELECT object_uri, priority FROM object_tasks WHERE connector_id=?", (cid,)
+        )
+        by_uri = {r["object_uri"].lstrip("/"): r["priority"] for r in rows}
+
+        assert by_uri["archive/old.txt"] == 900  # user override
+        assert by_uri["notes.md"] == -500  # user override, beats the built-in -350 bucket
+        assert by_uri["README.md"] == -350  # no override -> falls back to file's own bucket
+        assert by_uri["src/main.py"] == -220  # no override -> falls back to file's own bucket
+
+        # end-to-end: the claim order (existing priority ASC, started_at ASC behavior)
+        # actually reflects the override, not just the raw column value.
+        claimed = await eng.objects.claim_tasks(cid, limit=10)
+        order = [r["object_uri"].lstrip("/") for r in claimed]
+        assert order.index("notes.md") < order.index("README.md")
+        assert order.index("README.md") < order.index("src/main.py")
+        assert order.index("src/main.py") < order.index("archive/old.txt")
+    finally:
+        await eng.meta.close()

skills/mfs-ingest/SKILL.md

@@ -271,6 +271,7 @@ User wants to change a registered connector — new token, different
    | Cap on per-object chunks | `chunk_max` |
    | Which columns to embed (DB / SaaS) | `[[objects]] text_fields` |
    | Make a previously-indexed object stop indexing | `[[objects]] indexable = false` |
+   | Process some objects before others in this sync | `[[objects]] priority` (lower = earlier; doesn't affect ordering across different connectors) |
 
    See `reference/connectors/<scheme>.md` for the full field list.
 
@@ -293,6 +294,7 @@ User wants to change a registered connector — new token, different
 | `text_fields` (which columns become content) | yes, with `--force-index` (see below) |
 | `embedding.*` in server config | yes (and affects ALL connectors) |
 | `[[objects]] indexable = false` | drops that object from index |
+| `[[objects]] priority` | no — only changes processing order for objects enumerated in a future sync, not already-succeeded ones |
 
 **Applying a `text_fields` change to existing objects needs
 `mfs add <uri> --force-index`.** A plain `mfs connector update` saves the new