完善 RAG、Skill 与 Memory 契约测试并修复边界问题

.gitattributes

@@ -0,0 +1,8 @@
+# Keep launcher and installer scripts executable in WSL and readable by
+# PowerShell.  Without this, a Windows checkout can pass CRLF into the bash
+# shebang and WSL reports `/usr/bin/env: bash\r: No such file or directory`.
+*.sh text eol=lf
+*.ps1 text eol=crlf
+*.py text eol=lf
+*.json text eol=lf
+*.toml text eol=lf

datamind/agent/options.py

@@ -13,24 +13,21 @@
 from pathlib import Path
 from typing import Any
 
-from datamind.capabilities.db import DBService, build_db_service, build_db_tools
+from datamind.capabilities.db import DBService, build_db_service
 from datamind.capabilities.embedding import build_embedding
-from datamind.capabilities.graph import GraphService, build_graph_service, build_graph_tools
+from datamind.capabilities.graph import GraphService, build_graph_service
 from datamind.capabilities.hooks import AuditLogHook, DestructiveSqlHook, PathAllowlistHook
 from datamind.capabilities.ingest import (
     IngestLedger,
     IngestService,
     build_ingest_service,
-    build_ingest_tools,
     with_receipts,
 )
-from datamind.capabilities.kb import KBService, build_kb_service, build_kb_tools
-from datamind.capabilities.memory import MemoryService, build_memory_service, build_memory_tools
+from datamind.capabilities.kb import KBService, build_kb_service
+from datamind.capabilities.memory import MemoryService, build_memory_service
 from datamind.capabilities.skills import (
     SkillsService,
     build_skills_service,
-    build_skills_store_tools,
-    build_skills_tools,
 )
 from datamind.config import Settings
 from datamind.core.contracts import ToolAccess
@@ -40,7 +37,7 @@
 from datamind.core.context import RequestContext
 from datamind.core.model_clients import build_model_client
 from datamind.core.protocols import EmbeddingProvider, TextModelClient, ToolCallingModelClient
-from datamind.core.tools import ToolRegistry
+from datamind.core.tools import ToolRegistry, tool_provider_registry
 
 from .base import AgentLoopConfig, AgentLoopProtocol
 from .loop_native import NativeAgentLoop
@@ -63,6 +60,29 @@ class AgentServices:
     skills: SkillsService | None = None
     memory: MemoryService | None = None
     ingest: IngestService | None = None
+    _closed: bool = False
+
+    async def aclose(self) -> None:
+        """Close shared runtime resources once, regardless of facade owner."""
+        if self._closed:
+            return
+        self._closed = True
+        resources = [
+            self.db,
+            self.graph,
+            self.kb,
+            self.embedding,
+            self.fallback_client,
+            self.client,
+        ]
+        seen: set[int] = set()
+        for resource in resources:
+            if resource is None or id(resource) in seen:
+                continue
+            seen.add(id(resource))
+            close = getattr(resource, "aclose", None)
+            if callable(close):
+                await close()
 
 
 @dataclass
@@ -140,6 +160,9 @@ async def query(
             user_message=message, history=history, final_contract=final_contract,
         )
 
+    async def aclose(self) -> None:
+        await self.services.aclose()
+
 
 @dataclass
 class StoreAgent:
@@ -167,6 +190,9 @@ async def store(
     ) -> dict[str, Any]:
         return await self.loop.run_turn(user_message=message, history=history)
 
+    async def aclose(self) -> None:
+        await self.services.aclose()
+
 
 @dataclass
 class DataMind:
@@ -176,7 +202,6 @@ class DataMind:
     retrieve_agent: RetrieveAgent
     services: AgentServices
     profile: str = "default"
-    _closed: bool = False
 
     @property
     def store(self) -> StoreAgent:
@@ -224,25 +249,7 @@ async def query(
 
     async def aclose(self) -> None:
         """Idempotently close clients, engines, stores, and providers."""
-        if self._closed:
-            return
-        self._closed = True
-        resources = [
-            self.services.db,
-            self.services.graph,
-            self.services.kb,
-            self.services.embedding,
-            self.services.fallback_client,
-            self.services.client,
-        ]
-        seen: set[int] = set()
-        for resource in resources:
-            if resource is None or id(resource) in seen:
-                continue
-            seen.add(id(resource))
-            close = getattr(resource, "aclose", None)
-            if callable(close):
-                await close()
+        await self.services.aclose()
 
 
 def _build_hook_chain(settings: Settings) -> HookChain | None:
@@ -367,27 +374,22 @@ async def build_datamind(
     )
 
     catalogue = ToolRegistry()
-    ingest_tools = build_ingest_tools(ingest) if ingest is not None else []
-    catalogue.extend([t for t in ingest_tools if t.surface is None])
-    if "kb" in active:
-        assert kb is not None
-        catalogue.extend(build_kb_tools(kb))
-        catalogue.extend([t for t in ingest_tools if t.surface and t.surface.value == "kb"])
-    if "db" in active:
-        assert db is not None
-        catalogue.extend(build_db_tools(db))
-        catalogue.extend([t for t in ingest_tools if t.surface and t.surface.value == "db"])
-    if "graph" in active:
-        assert graph is not None
-        catalogue.extend(build_graph_tools(graph))
-        catalogue.extend([t for t in ingest_tools if t.surface and t.surface.value == "graph"])
-    if "skills" in active:
-        assert skills is not None
-        catalogue.extend(build_skills_tools(skills))
-        catalogue.extend(build_skills_store_tools(skills))
-    if "memory" in active:
-        assert memory is not None
-        catalogue.extend(build_memory_tools(memory))
+    provider_services = {
+        "kb_service": kb,
+        "db_service": db,
+        "graph_service": graph,
+        "skills_service": skills,
+        "memory_service": memory,
+        "ingest_service": ingest,
+    }
+    if ingest is not None:
+        ingest_provider = tool_provider_registry.create("ingest")
+        for spec in ingest_provider.build(**provider_services):
+            if spec.surface is None or spec.surface.value in active:
+                catalogue.add(spec)
+    for provider_name in sorted(active):
+        provider = tool_provider_registry.create(provider_name)
+        catalogue.extend(provider.build(**provider_services))
 
     retrieve_tools = catalogue.select(access={ToolAccess.READ, ToolAccess.UTILITY})
     raw_store_tools = catalogue.select(access={ToolAccess.WRITE})

datamind/capabilities/ingest/ledger.py

@@ -5,6 +5,8 @@
 import hashlib
 import inspect
 import json
+import os
+import time
 from pathlib import Path
 from typing import Any, Awaitable, Callable
 
@@ -116,9 +118,46 @@ def __init__(self, *, storage_dir: Path, profile: str) -> None:
         self._profile = profile
         self._state_path = self._storage_dir / "ingest_state.json"
         self._receipts_path = self._storage_dir / "ingest_receipts.jsonl"
+        self._process_lock_path = self._storage_dir / "ingest.lock"
         self._lock = asyncio.Lock()
         self._storage_dir.mkdir(parents=True, exist_ok=True)
 
+    async def _acquire_process_lock(self) -> None:
+        """Serialize ledger read/execute/write across MCP processes.
+
+        The in-memory asyncio lock only protects one DataMind instance.  An
+        atomic lock-file claim extends the same idempotency boundary to a
+        second MCP process or a separately-created Ledger object.
+        """
+        while True:
+            try:
+                fd = os.open(
+                    self._process_lock_path,
+                    os.O_CREAT | os.O_EXCL | os.O_WRONLY,
+                )
+                try:
+                    os.write(fd, f"{os.getpid()} {time.time():.6f}\n".encode())
+                finally:
+                    os.close(fd)
+                return
+            except FileExistsError:
+                try:
+                    age = time.time() - self._process_lock_path.stat().st_mtime
+                    if age > 3600:
+                        self._process_lock_path.unlink(missing_ok=True)
+                        continue
+                except FileNotFoundError:
+                    continue
+                await asyncio.sleep(0.02)
+
+    async def _release_process_lock(self) -> None:
+        try:
+            self._process_lock_path.unlink(missing_ok=True)
+        except OSError:
+            # The lock may have been recovered as stale by another process;
+            # the receipt/state writes remain the source of truth.
+            pass
+
     def _load_state(self) -> dict[str, Any]:
         if not self._state_path.is_file():
             return {"revision": 0, "successful": {}}
@@ -150,7 +189,21 @@ async def execute(
         args: dict[str, Any],
         invoke: Callable[..., Awaitable[Any]],
     ) -> dict[str, Any]:
-        """Deduplicate retryable imports, but always apply explicit DB replacement."""
+        """Deduplicate retryable imports, including across MCP processes."""
+        await self._acquire_process_lock()
+        try:
+            return await self._execute_locked(spec=spec, args=args, invoke=invoke)
+        finally:
+            await self._release_process_lock()
+
+    async def _execute_locked(
+        self,
+        *,
+        spec: ToolSpec,
+        args: dict[str, Any],
+        invoke: Callable[..., Awaitable[Any]],
+    ) -> dict[str, Any]:
+        """Run one call while both ledger locks are held."""
         source = _source_from_call(spec, args)
         fingerprint = _hash_text(
             _canonical(

datamind/capabilities/ingest/service.py

@@ -820,6 +820,18 @@ async def _upsert_chunks(
         store = self._kb.vector_store
         texts = [c.text for c in chunks]
         vectors = await provider.embed_texts(texts)
+        # Keep track of IDs that existed before this call.  If a third-party
+        # store fails after partially accepting a batch, only remove newly
+        # created IDs during rollback; an idempotent re-write must not erase
+        # the previous revision.
+        existing_ids: set[str] = set()
+        try:
+            existing_ids = {
+                str(chunk_id)
+                for chunk_id, _text, _metadata in await store.get_all_texts()
+            }
+        except Exception:  # noqa: BLE001 - rollback is best effort
+            pass
         stale_ids: list[str] = []
         if replace_sources:
             new_ids = {chunk.id for chunk in chunks}
@@ -830,15 +842,28 @@ async def _upsert_chunks(
                 }
                 if recorded_sources & replace_sources and chunk_id not in new_ids:
                     stale_ids.append(chunk_id)
-        await store.add(
-            ids=[c.id for c in chunks],
-            texts=texts,
-            embeddings=vectors,
-            metadatas=[
-                {**(c.metadata or {}), "source": c.source or ""}
-                for c in chunks
-            ],
-        )
+        try:
+            await store.add(
+                ids=[c.id for c in chunks],
+                texts=texts,
+                embeddings=vectors,
+                metadatas=[
+                    {**(c.metadata or {}), "source": c.source or ""}
+                    for c in chunks
+                ],
+            )
+        except Exception:
+            rollback_ids = [c.id for c in chunks if c.id not in existing_ids]
+            delete = getattr(store, "delete", None)
+            if rollback_ids and callable(delete):
+                try:
+                    await delete(rollback_ids)
+                except Exception:  # noqa: BLE001 - preserve original write error
+                    _log.warning(
+                        "kb_partial_write_rollback_failed",
+                        extra={"chunk_ids": rollback_ids},
+                    )
+            raise
         if stale_ids:
             await store.delete(stale_ids)
         # Incremental writes must leave the same compatibility metadata that

datamind/capabilities/ingest/tools.py

@@ -15,7 +15,7 @@
 """
 from __future__ import annotations
 
-from datamind.core.tools import ToolSpec
+from datamind.core.tools import ToolSpec, tool_provider_registry
 
 from .service import IngestService
 
@@ -448,4 +448,13 @@ async def _graph_add_path(
     ]
 
 
+@tool_provider_registry.register("ingest")
+class _IngestToolProvider:
+    def build(self, **services: object) -> list[ToolSpec]:
+        ingest = services.get("ingest_service")
+        if not isinstance(ingest, IngestService):
+            raise ValueError("ingest tool provider requires 'ingest_service'")
+        return build_ingest_tools(ingest)
+
+
 __all__ = ["build_ingest_tools"]

datamind/capabilities/kb/indexer.py

@@ -74,6 +74,10 @@ def _split_text(
     they fall inside the size window. Not fancy; deterministic; good
     enough for the default.
     """
+    if chunk_size <= 0:
+        raise ValueError("chunk_size must be > 0")
+    if chunk_overlap < 0:
+        raise ValueError("chunk_overlap must be >= 0")
     if chunk_overlap >= chunk_size:
         raise ValueError("chunk_overlap must be < chunk_size")
     if len(text) <= chunk_size: