feat(sync): add incremental Obsidian DataPlug

src/contextseek/cli/main.py

@@ -1151,6 +1151,7 @@ def run_cli(
         _p = _pl.Path(args.path).expanduser()
         fmt = detect_format(_p)
         format_label = {
+            "obsidian_vault": "Obsidian vault",
             "auto_dir": "auto-detected directory",
             "markdown_file": "Markdown/text file",
             "code_file": "code file",
@@ -1181,11 +1182,14 @@ def run_cli(
 
         if args.dry_run:
             print_success(
-                f"dry-run: would import {report.added} items "
-                f"({report.skipped} already exist)"
+                f"dry-run: would add {report.added}, update {report.updated}, "
+                f"delete {report.deleted} items ({report.skipped} unchanged)"
             )
         else:
-            print_success(f"done: added {report.added}  skipped {report.skipped}")
+            print_success(
+                f"done: added {report.added}  updated {report.updated}  "
+                f"deleted {report.deleted}  skipped {report.skipped}"
+            )
         return 0
 
     return 1

src/contextseek/client/contextseek.py

@@ -25,7 +25,7 @@
     from contextseek.scope import ScopeStats, ScopeTree
 
 from contextseek.storage.protocol import SeekVFSAdapter
-from contextseek.plugs.core.protocols import DataPlug, PlugMeta
+from contextseek.plugs.core.protocols import DataPlug, PlugMeta, RawEvent
 from contextseek.domain.context_item import ContextItem
 from contextseek.domain.conflicts import ConflictType
 from contextseek.domain.inference import (
@@ -436,9 +436,11 @@ def plug(self, source: DataPlug, *, scope: str | None = None) -> None:
         """Register and consume a DataPlug.
 
         The plug's ``stream()`` iterator is consumed immediately, adding
-        each RawEvent as a ContextItem. Scope is taken from the explicit
-        ``scope`` argument first, then ``event.metadata["scope"]``, then the
-        plug name as a local fallback. ``event.source`` remains provenance.
+        each RawEvent as a ContextItem. Incremental events with a stable
+        ``item_id`` are updated in place, while delete events are soft-deleted.
+        Scope is taken from the explicit ``scope`` argument first, then
+        ``event.metadata["scope"]``, then the plug name as a local fallback.
+        ``event.source`` remains provenance.
 
         Args:
             source: A DataPlug implementation to register.
@@ -452,6 +454,29 @@ def plug(self, source: DataPlug, *, scope: str | None = None) -> None:
         for event in source.stream():
             event_scope = scope or str(event.metadata.get("scope") or meta.name)
             source_id = event.source or meta.name
+            operation = getattr(event, "operation", "add")
+            item_id = getattr(event, "item_id", None)
+            links = list(getattr(event, "links", []))
+
+            if operation not in {"add", "update", "delete", "noop"}:
+                raise ValueError(f"unsupported plug operation: {operation}")
+            if operation == "noop":
+                continue
+            if operation == "update" and not item_id:
+                raise ValueError("incremental plug update events require item_id")
+            if operation == "delete":
+                if not item_id:
+                    raise ValueError("incremental plug delete events require item_id")
+                ref = self.resolver.ref_for(event_scope, item_id)
+                existing = self._read_item(ref)
+                if existing is not None and not existing.is_deleted:
+                    existing.soft_delete(f"plug_delete:{meta.name}:{source_id}")
+                    self._write_item_with_audit(
+                        existing,
+                        action="plug_delete",
+                        detail={"source": source_id},
+                    )
+                continue
 
             # Allow importers/plugs to override stage/stability (e.g. skill import)
             event_stage: Stage | None = None
@@ -467,15 +492,26 @@ def plug(self, source: DataPlug, *, scope: str | None = None) -> None:
                 except ValueError:
                     pass
 
-            item = self.add(
-                content=event.content,
-                scope=event_scope,
-                source=source_id,
-                source_type=source_type,
-                tags=event.tags,
-                stage=event_stage,
-                stability=event_stability,
-            )
+            if item_id:
+                item = self._upsert_raw_event(
+                    event,
+                    scope=event_scope,
+                    source=source_id,
+                    source_type=source_type,
+                    stage=event_stage,
+                    stability=event_stability,
+                )
+            else:
+                item = self.add(
+                    content=event.content,
+                    scope=event_scope,
+                    source=source_id,
+                    source_type=source_type,
+                    tags=event.tags,
+                    stage=event_stage,
+                    stability=event_stability,
+                    links=links,
+                )
             if "embedding" in event.metadata and self.embedder is None:
                 item.embedding = event.metadata["embedding"]
             if "importance" in event.metadata:
@@ -487,6 +523,63 @@ def plug(self, source: DataPlug, *, scope: str | None = None) -> None:
             ):
                 self._write_item(item)
 
+    def _upsert_raw_event(
+        self,
+        event: RawEvent,
+        *,
+        scope: str,
+        source: str,
+        source_type: str,
+        stage: Stage | None,
+        stability: Stability | None,
+    ) -> ContextItem:
+        """Materialize an incremental RawEvent at its stable item identity."""
+        item_id = getattr(event, "item_id", None)
+        if item_id is None:
+            raise ValueError("incremental plug events require item_id")
+
+        content = self._apply_write_policy(
+            event.content,
+            scope=scope,
+            source=source,
+            source_type=source_type,
+        )
+        resolved_stage, resolved_stability = self._resolve_stage_stability(
+            stage=stage,
+            stability=stability,
+            content=content,
+            source_type=source_type,
+        )
+        ref = self.resolver.ref_for(scope, item_id)
+        existing = self._read_item(ref)
+        item = ContextItem(
+            id=item_id,
+            content=content,
+            scope=scope,
+            provenance=self._build_provenance(source, source_type, None),
+            tags=list(event.tags or []),
+            stage=resolved_stage,
+            stability=resolved_stability,
+            links=list(getattr(event, "links", [])),
+        )
+        action = "plug_add"
+        if existing is not None:
+            action = "plug_update"
+            item.created_at = existing.created_at
+            item.updated_at = datetime.now(timezone.utc)
+            item.relevance_boost = existing.relevance_boost
+            item.access_count = existing.access_count
+            item.lineage_access_count = existing.lineage_access_count
+            item.last_accessed_at = existing.last_accessed_at
+
+        self._prepare_item(item, check_conflicts=False)
+        self._write_item_with_audit(
+            item,
+            action=action,
+            detail={"source": source},
+        )
+        return item
+
     def add(
         self,
         content: str | dict[str, Any],

src/contextseek/daemon/sync_cmd.py

@@ -85,6 +85,8 @@ def _iter_files(root: pathlib.Path) -> list[pathlib.Path]:
 @dataclass
 class SyncReport:
     added: int = 0
+    updated: int = 0
+    deleted: int = 0
     skipped: int = 0
     format_detected: str = "unknown"
     errors: list[str] = field(default_factory=list)
@@ -98,12 +100,14 @@ class SyncReport:
 def detect_format(path: str | pathlib.Path) -> str:
     """Detect the source format from path structure and file content.
 
-    Returns one of: auto_dir, markdown_file, code_file, chatgpt_json,
+    Returns one of: obsidian_vault, auto_dir, markdown_file, code_file, chatgpt_json,
     claude_json, bookmarks_html, plaintext.
     """
     p = pathlib.Path(path).expanduser()
 
     if p.is_dir():
+        if (p / ".obsidian").is_dir():
+            return "obsidian_vault"
         return "auto_dir"
 
     if p.is_file():
@@ -530,6 +534,25 @@ def sync_path(
     fmt = detect_format(p)
     report = SyncReport(format_detected=fmt)
 
+    if fmt == "obsidian_vault":
+        from contextseek.plugs.obsidian import ObsidianVaultPlug
+
+        plug = ObsidianVaultPlug(
+            p,
+            sync_key=scope,
+            persist_state=not dry_run,
+            on_progress=on_progress,
+        )
+        if dry_run:
+            list(plug.stream())
+        else:
+            ctx.plug(plug, scope=scope)
+        report.added = plug.stats.added
+        report.updated = plug.stats.updated
+        report.deleted = plug.stats.deleted
+        report.skipped = plug.stats.skipped
+        return report
+
     sync_backend = _resolve_seekdb_backend(ctx)
     if sync_backend is not None:
         if on_progress is not None:

src/contextseek/plugs/__init__.py

@@ -4,6 +4,7 @@
     "HermesSkillImporter",
     "MCPToolImporter",
     "OpenAIFunctionImporter",
+    "ObsidianVaultPlug",
     "PowerMemPlug",
     "PowerMemProxyPlug",
     "RAGPlug",
@@ -23,6 +24,10 @@ def __getattr__(name: str):
         from contextseek.plugs.rag import RAGPlug
 
         return RAGPlug
+    if name == "ObsidianVaultPlug":
+        from contextseek.plugs.obsidian import ObsidianVaultPlug
+
+        return ObsidianVaultPlug
     if name in {"HermesSkillImporter", "MCPToolImporter", "OpenAIFunctionImporter"}:
         from contextseek.plugs.skills import (
             HermesSkillImporter,

src/contextseek/plugs/core/protocols.py

@@ -14,6 +14,8 @@
 from datetime import datetime, timezone
 from typing import Any, Iterator, Literal, Protocol, runtime_checkable
 
+from contextseek.domain.links import Link
+
 
 PlugOperation = Literal["add", "update", "delete", "noop"]
 MaterializationStatus = Literal["applied", "skipped", "failed"]
@@ -64,7 +66,20 @@ class RawEvent:
     """Optional tags to attach to the resulting ContextItem."""
 
     metadata: dict = field(default_factory=dict)
-    """Extra key-value pairs the plug can supply (passed to provenance context)."""
+    """Optional materialization hints supplied by the plug.
+
+    ContextSeek recognizes ``scope``, ``stage``, ``stability``, ``embedding``,
+    ``importance``, and ``summary``. Other keys are not persisted automatically.
+    """
+
+    operation: PlugOperation = "add"
+    """Requested materialization operation for incremental plugs."""
+
+    item_id: str | None = None
+    """Stable ContextItem identity used by incremental plugs for upserts/deletes."""
+
+    links: list[Link] = field(default_factory=list)
+    """Typed relationships to attach to the resulting ContextItem."""
 
 
 @runtime_checkable

src/contextseek/plugs/obsidian/__init__.py

@@ -0,0 +1,5 @@
+"""Incremental Obsidian vault DataPlug."""
+
+from contextseek.plugs.obsidian.plug import ObsidianSyncStats, ObsidianVaultPlug
+
+__all__ = ["ObsidianSyncStats", "ObsidianVaultPlug"]