数据源不可用时的 db_schema 替身也收 keyword；重读范围内读不到结构的表不记成按需加入；起草、剖析排在探查后面的测试

- tools/datasource._unavailable_tool：db_schema__ 多了可选的 keyword，固定的版本不在了时的替身也要收这个参数（调用工具节点按
  参数模型的全部字段传参，不收就报「不接受名为 keyword 的参数」，全量测试里 test_run_data_versions 抓到的）。
- source_objects.add_to_scope：本来就在范围里、只是上次没读到结构的表（pinned、规则命中的）这次重读一遍，不改原因、不记留痕，
  不算进「每次运行最多按需加入 8 张」，也不出现在面板的「按需加入的表」里。
- tests/test_introspect_jobs.py：起草、剖析任务排在同源的探查任务后面时 waiting_for 是 introspect。

backend/app/data/source_objects.py

@@ -939,8 +939,11 @@ async def add_to_scope(source_id: str, names: list[str], *, trail: Trail, reason
             for o in todo:
                 was_in = bool(o.in_scope)
                 o.in_scope = True
-                if not was_in or reason == "explicit":
-                    o.scope_reason = reason if not was_in or o.scope_reason in (None, "on_demand") else o.scope_reason
+                if not was_in or (reason == "explicit" and o.scope_reason in (None, "on_demand")):
+                    o.scope_reason = reason
+                elif was_in:
+                    # 本来就在范围里、只是上次没读到结构（pinned、规则命中的）：这次重读一遍，不算「加入」，不记留痕
+                    continue
                 o.added_via, o.added_by, o.added_at = trail.via, (trail.actor or None), now
                 o.added_run_id, o.added_conversation_id = trail.run_id, trail.conversation_id
                 o.added_ref = (trail.ref or None) and str(trail.ref)[:64]

backend/app/tools/datasource.py

@@ -581,7 +581,7 @@ def _unavailable_tool(row: DataSource, prefix: str, reason: str) -> StructuredTo
             metadata={"dangerous_if": lambda args: False, "timeout_s": query_timeout(row)},
         )
 
-    async def _schema(table: str | list[str] | None = None) -> str:
+    async def _schema(table: str | list[str] | None = None, keyword: str | None = None) -> str:
         return f"{DATA_UNAVAILABLE}{reason}"
 
     return StructuredTool(

backend/tests/test_introspect_jobs.py

@@ -416,3 +416,36 @@ def test_index_sql_golden_parses_and_passes_the_guard(kind, schema):
     check(sql, readonly=True, source_name="x", dialect=kind)
     assert so.index_sql("mariadb", "shop") == GOLDEN[("mysql", "shop")]
     assert so.index_sql("db2", None) is None
+
+
+async def test_draft_and_profile_jobs_say_they_wait_for_introspect(small):
+    """起草、剖析排在同源的探查任务后面时，waiting_for 照实说是探查（界面写「探查结束后自动开始」）。"""
+    from datetime import datetime, timezone
+
+    from app.data import catalog_jobs, catalog_profile_jobs
+    from app.db.base import new_id
+    from app.db.models import CatalogDraftJob, CatalogProfileJob
+
+    sid, _ = small
+    now = datetime.now(timezone.utc)
+    async with SessionLocal() as session:
+        profile = CatalogProfileJob(id=new_id(), source_id=sid, status="queued", scope="needed", created_at=now)
+        draft = CatalogDraftJob(id=new_id(), source_id=sid, status="queued")
+        session.add_all([profile, draft])
+        await session.commit()
+    try:
+        for kind, job, waiting in (("profile", profile, catalog_profile_jobs.waiting_for),
+                                   ("draft", draft, catalog_jobs.waiting_for)):
+            held = runtime.lanes.enter(sid, "introspect", "an-introspect-job")
+            mine = runtime.lanes.enter(sid, kind, job.id)
+            try:
+                async with SessionLocal() as session:
+                    assert await waiting(session, job) == "introspect", kind
+            finally:
+                held.release()
+                mine.release()
+    finally:
+        async with SessionLocal() as session:
+            for model, job_id in ((CatalogProfileJob, profile.id), (CatalogDraftJob, draft.id)):
+                await session.delete(await session.get(model, job_id))
+            await session.commit()