Merge pull request #311 from hito0512/feat/add-siyuan-cli

feat: add SiYuan CLI (cli-anything-siyuan)
2026-10-30 18:12:56 +08:00 · 2026-06-11 18:19:41 +08:00
parent 7fb741ad9e 59e9f7bc2d
commit a71c672162
20 changed files with 3152 additions and 0 deletions
@@ -100,6 +100,7 @@
 !/3MF/
 !/calibre/
 !/rekordbox/
+!/siyuan/
 # Step 5: Inside each software dir, ignore everything (including dotfiles)
 /gimp/*
 /gimp/.*
@@ -207,6 +208,8 @@
 /calibre/.*
 /rekordbox/*
 /rekordbox/.*
+/siyuan/*
+/siyuan/.*

 # Step 6: ...except agent-harness/
 !/gimp/agent-harness/
@@ -272,6 +275,7 @@
 !/quietshrink/agent-harness/
 !/mailchimp/agent-harness/
 !/rekordbox/agent-harness/
+!/siyuan/agent-harness/

 # Exclude non-gedit demo macros from macrocli (local only)
 /macrocli/agent-harness/cli_anything/macrocli/macro_definitions/demo/flameshot*
@@ -1303,6 +1303,26 @@
          "url": "https://github.com/octo-patch"
        }
      ]
+    },
+    {
+      "name": "siyuan",
+      "display_name": "SiYuan CLI",
+      "version": "1.0.0",
+      "description": "Knowledge management and note-taking via SiYuan HTTP API — manage notebooks, documents, blocks, and search content from the command line",
+      "category": "knowledge-management",
+      "requires": "SiYuan running with HTTP API enabled (default http://127.0.0.1:6806)",
+      "homepage": "https://github.com/siyuan-note/siyuan",
+      "source_url": null,
+      "install_strategy": "pip",
+      "install_cmd": "pip install git+https://github.com/HKUDS/CLI-Anything.git#subdirectory=siyuan/agent-harness",
+      "skill_md": "skills/cli-anything-siyuan/SKILL.md",
+      "entry_point": "cli-anything-siyuan",
+      "contributors": [
+        {
+          "name": "hito0512",
+          "url": "https://github.com/hito0512"
+        }
+      ]
    }
  ]
 }
@@ -0,0 +1,159 @@
+# SiYuan (思源笔记) — Agent CLI Harness Analysis
+
+## Overview
+
+SiYuan is a local-first, privacy-focused knowledge management and note-taking application.
+It uses a Go backend with an Electron/TypeScript frontend. Data is stored as `.sy` files
+(Siyuan JSON format) with a SQLite database for indexing and search.
+
+## Backend
+
+- **Language**: Go
+- **HTTP Server**: Gin framework, runs on `http://127.0.0.1:6806`
+- **Auth**: Token-based (`Authorization: Token xxx`)
+- **Entry point**: `kernel/main.go` → starts HTTP server + background jobs
+
+## Data Model
+
+```
+Notebook (Box)
+  └── Document (tree node, .sy file)
+       └── Blocks (paragraphs, headings, lists, code, etc.)
+            └── Attributes (key-value, custom-* prefix)
+```
+
+- **Notebook**: Top-level container, identified by ID (e.g., `20210817205410-2kvfpfn`)
+- **Document**: Tree node in a notebook, stored as `.sy` file
+- **Block**: Content unit (paragraph, heading, list item, etc.), each has unique ID
+- **SQLite Database**: blocks, refs, attributes, history, asset_content tables
+- **IDs**: Timestamp-based IDs like `20210817205410-2kvfpfn` (datetime + random suffix)
+
+## API Surface
+
+Base URL: `http://127.0.0.1:6806`
+
+### Notebook API (`/api/notebook/*`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `lsNotebooks` | POST | List all notebooks |
+| `openNotebook` | POST | Open a notebook |
+| `closeNotebook` | POST | Close a notebook |
+| `createNotebook` | POST | Create new notebook |
+| `removeNotebook` | POST | Delete notebook |
+| `renameNotebook` | POST | Rename notebook |
+| `getNotebookConf` | POST | Get notebook config |
+| `setNotebookConf` | POST | Set notebook config |
+| `setNotebookIcon` | POST | Set notebook icon |
+
+### Document/Filetree API (`/api/filetree/*`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `createDocWithMd` | POST | Create doc with Markdown content |
+| `createDailyNote` | POST | Create daily note |
+| `renameDoc` | POST | Rename doc by path |
+| `renameDocByID` | POST | Rename doc by ID |
+| `removeDoc` | POST | Delete doc by path |
+| `removeDocByID` | POST | Delete doc by ID |
+| `moveDocs` | POST | Move docs by paths |
+| `moveDocsByID` | POST | Move docs by IDs |
+| `getHPathByID` | POST | Get human-readable path by ID |
+| `getHPathByPath` | POST | Get human-readable path by path |
+| `getIDsByHPath` | POST | Get IDs by human-readable path |
+| `getPathByID` | POST | Get storage path by ID |
+| `listDocsByPath` | POST | List docs at a path |
+| `listDocTree` | POST | List full document tree |
+| `searchDocs` | POST | Search document names |
+
+### Block API (`/api/block/*`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `insertBlock` | POST | Insert block at position |
+| `prependBlock` | POST | Insert as first child |
+| `appendBlock` | POST | Insert as last child |
+| `updateBlock` | POST | Update block content |
+| `deleteBlock` | POST | Delete a block |
+| `moveBlock` | POST | Move a block |
+| `foldBlock` | POST | Fold/collapse a block |
+| `unfoldBlock` | POST | Unfold/expand a block |
+| `getBlockKramdown` | POST | Get block kramdown source |
+| `getChildBlocks` | POST | Get child blocks |
+
+### Attribute API (`/api/attr/*`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `setBlockAttrs` | POST | Set block attributes |
+| `getBlockAttrs` | POST | Get block attributes |
+
+### SQL Query API (`/api/query/sql`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `sql` | POST | Execute SQL query on block database |
+
+### Search API (`/api/search/*`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `fullTextSearchBlock` | POST | Full-text search in blocks |
+| `searchTag` | POST | Search tags |
+| `findReplace` | POST | Find and replace |
+
+### Export API (`/api/export/*`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `exportMdContent` | POST | Export doc as Markdown |
+| `exportResources` | POST | Export files as ZIP |
+
+### Asset API (`/api/asset/*`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `upload` | POST | Upload asset file |
+
+### Tag API (`/api/tag/*`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `getTag` | POST | List tags |
+| `renameTag` | POST | Rename tag |
+| `removeTag` | POST | Remove tag |
+
+### System API (`/api/system/*`)
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `version` | GET/POST | Get system version |
+| `currentTime` | POST | Get current server time |
+
+## Existing CLI Tools
+
+None. SiYuan ships with only the desktop GUI (Electron) and mobile apps.
+The kernel starts automatically with the GUI; there is no standalone CLI mode.
+
+## CLI Harness Strategy
+
+Since SiYuan has no headless CLI backend, the harness will:
+
+1. **Connect to an already-running SiYuan instance** via its HTTP API
+2. **Provide commands for all major API groups** (notebooks, documents, blocks, search, export)
+3. **Store connection state** (host, port, token) for multi-session use
+4. **Support SQL queries** for advanced data access
+5. **Support import/export** of Markdown content
+
+The SiYuan instance must be running with the API server enabled (default: port 6806).
+The API token can be found in Settings → About.
+
+## CLI Usage Notes
+
+### Block insert/update with multi-line content
+
+`block insert` and `block update` accept data via stdin pipe when the argument is `-` or omitted:
+
+```bash
+# Pipe multi-line content
+cat note.md | cli-anything-siyuan block insert --parent <block-id>
+
+# Explicit stdin marker
+echo "hello" | cli-anything-siyuan block update <block-id> -
+
+# Requires one of: --parent <id>, --previous <id>, --next <id>
+```
+
+### Tag list
+
+`tag list` sends `ignoreMaxListHint: true` to ensure all tags are returned regardless of `Conf.FileTree.MaxListCount`. Tag-heavy workspaces return the complete list.
@@ -0,0 +1,143 @@
+# cli-anything-siyuan
+
+A CLI harness for [SiYuan](https://github.com/siyuan-note/siyuan) (思源笔记) —
+interact with your knowledge base from the terminal.
+
+## Prerequisites
+
+- **SiYuan** must be running (version 3.x). Download from [b3log.org/siyuan](https://b3log.org/siyuan).
+- **Python 3.10+**
+
+## Installation
+
+```bash
+cd agent-harness
+pip install -e .
+```
+
+For REPL support (recommended):
+
+```bash
+pip install -e ".[repl]"
+```
+
+## Configuration
+
+Create `~/.siyuan-cli.json`:
+
+```json
+{
+  "host": "127.0.0.1",
+  "port": 6806,
+  "token": "your-api-token-here"
+}
+```
+
+Or use environment variables:
+
+```bash
+export SIYUAN_HOST=127.0.0.1
+export SIYUAN_PORT=6806
+export SIYUAN_TOKEN=your-token
+```
+
+**Finding your API token:** Open SiYuan → Settings → About → API Token.
+
+## Usage
+
+### One-shot commands
+
+```bash
+# List notebooks
+cli-anything-siyuan notebook list
+
+# List notebooks (JSON output)
+cli-anything-siyuan --json notebook list
+
+# Get version
+cli-anything-siyuan version
+
+# Execute SQL query
+cli-anything-siyuan sql "SELECT * FROM blocks LIMIT 5"
+
+# Search blocks
+cli-anything-siyuan search "keyword"
+
+# Export document as Markdown
+cli-anything-siyuan export md <doc-id>
+
+# Show status
+cli-anything-siyuan status
+
+# Insert a block (multi-line content via stdin)
+cat note.md | cli-anything-siyuan block insert --parent <block-id>
+
+# Update a block content via stdin
+echo "new content" | cli-anything-siyuan block update <block-id>
+```
+
+### REPL mode
+
+```bash
+# Just run without arguments to enter REPL
+cli-anything-siyuan
+```
+
+Inside the REPL:
+
+```
+◆  cli-anything · Siyuan                   v1.0.0
+   ◇ Install: npx skills add HKUDS/CLI-Anything --skill cli-anything-siyuan -g -y
+   ◇ Global skill: ~/.agents/skills/cli-anything-siyuan/SKILL.md
+
+   Type help for commands, quit to exit
+
+siyuan ❯ notebook list
+siyuan ❯ doc tree <notebook-id>
+siyuan ❯ search "meeting notes"
+siyuan ❯ help
+siyuan ❯ quit
+```
+
+## Command Groups
+
+| Command | Description |
+|---------|-------------|
+| `notebook list` | List all notebooks |
+| `notebook create <name>` | Create a notebook |
+| `notebook rename <id> <name>` | Rename a notebook |
+| `notebook remove <id>` | Delete a notebook |
+| `notebook open <id>` | Open a notebook |
+| `doc create <notebook-id> <path>` | Create a document |
+| `doc list <notebook-id> [path]` | List documents |
+| `doc tree <notebook-id>` | Show document tree |
+| `doc get <id>` | Get document path by ID |
+| `doc rename <id> <title>` | Rename a document |
+| `doc remove <id>` | Delete a document |
+| `block insert <data>` | Insert a block (use `-` or omit for stdin pipe) |
+| `block update <id> <data>` | Update block content (use `-` or omit for stdin pipe) |
+| `block delete <id>` | Delete a block |
+| `block get <id>` | Get block kramdown source |
+| `block children <id>` | Get child blocks |
+| `sql <stmt>` | Execute SQL query |
+| `search <query>` | Full-text search |
+| `export md <doc-id>` | Export as Markdown |
+| `tag list` | List all tags |
+| `version` | Show SiYuan version |
+| `status` | Show connection status |
+
+## Running Tests
+
+```bash
+# Unit tests (no external dependencies)
+cd agent-harness
+pip install -e ".[test]"
+python -m pytest cli_anything/siyuan/tests/test_core.py -v
+
+# E2E tests (requires running SiYuan)
+python -m pytest cli_anything/siyuan/tests/test_full_e2e.py -v -s
+```
+
+## License
+
+AGPL-3.0 (same as SiYuan)
@@ -0,0 +1 @@
+# cli-anything-siyuan: SiYuan (思源笔记) CLI harness
@@ -0,0 +1,5 @@
+"""Allow running as: python -m cli_anything.siyuan"""
+from cli_anything.siyuan.siyuan_cli import cli
+
+if __name__ == "__main__":
+    cli()
@@ -0,0 +1,307 @@
+"""SiYuan HTTP API client.
+
+Handles connection, authentication, and request/response to the SiYuan kernel.
+"""
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import requests
+
+
+@dataclass(frozen=True)
+class SiYuanConfig:
+    host: str = "127.0.0.1"
+    port: int = 6806
+    token: str = ""
+
+    @property
+    def base_url(self) -> str:
+        return f"http://{self.host}:{self.port}"
+
+
+def load_config(config_path: str | None = None) -> SiYuanConfig:
+    """Load SiYuan connection config from file or environment.
+
+    Priority: explicit path -> env vars -> defaults.
+    Config file is JSON: {"host": "...", "port": 6806, "token": "..."}
+    """
+    if config_path:
+        config_file = Path(config_path)
+    else:
+        config_file = Path.home() / ".siyuan-cli.json"
+
+    if config_file.is_file():
+        try:
+            data = json.loads(config_file.read_text(encoding="utf-8-sig"))
+        except (json.JSONDecodeError, OSError):
+            data = None
+
+        if data is not None:
+            # File values as base, env vars override
+            return SiYuanConfig(
+                host=os.environ.get("SIYUAN_HOST", data.get("host", "127.0.0.1")),
+                port=int(os.environ.get("SIYUAN_PORT", data.get("port", 6806))),
+                token=os.environ.get("SIYUAN_TOKEN", data.get("token", "")),
+            )
+
+    return SiYuanConfig(
+        host=os.environ.get("SIYUAN_HOST", "127.0.0.1"),
+        port=int(os.environ.get("SIYUAN_PORT", "6806")),
+        token=os.environ.get("SIYUAN_TOKEN", ""),
+    )
+
+
+class SiYuanClientError(Exception):
+    """Raised when the SiYuan API returns an error."""
+
+
+class SiYuanClient:
+    """HTTP client for the SiYuan kernel API."""
+
+    def __init__(self, config: SiYuanConfig | None = None):
+        self.config = config or load_config()
+        self._session = requests.Session()
+        self._session.trust_env = False
+        if self.config.token:
+            self._session.headers["Authorization"] = f"Token {self.config.token}"
+        self._session.headers["Content-Type"] = "application/json"
+
+    def _post(self, endpoint: str, data: dict[str, Any] | None = None) -> Any:
+        """Make a POST request to the SiYuan API."""
+        url = f"{self.config.base_url}{endpoint}"
+        try:
+            resp = self._session.post(url, json=data or {}, timeout=30)
+        except requests.ConnectionError as e:
+            raise SiYuanClientError(
+                f"Cannot connect to SiYuan at {self.config.base_url}. "
+                f"Is SiYuan running? ({e})"
+            ) from e
+        except requests.Timeout as e:
+            raise SiYuanClientError(
+                f"Request to SiYuan timed out after 30s ({e})"
+            ) from e
+        except requests.RequestException as e:
+            raise SiYuanClientError(
+                f"Request to SiYuan failed: {e}"
+            ) from e
+
+        if resp.status_code != 200:
+            raise SiYuanClientError(
+                f"API returned status {resp.status_code}: {resp.text[:200]}"
+            )
+
+        body = resp.json()
+        if body.get("code", 0) != 0:
+            raise SiYuanClientError(
+                f"API error: {body.get('msg', 'unknown error')}"
+            )
+        return body.get("data")
+
+    def ping(self) -> bool:
+        """Check if SiYuan kernel is reachable."""
+        try:
+            return self._post("/api/system/version") is not None
+        except SiYuanClientError:
+            return False
+
+    # ── Notebook API ───────────────────────────────────────────────────
+
+    def list_notebooks(self) -> list[dict[str, Any]]:
+        """List all notebooks."""
+        data = self._post("/api/notebook/lsNotebooks")
+        return data.get("notebooks", [])
+
+    def open_notebook(self, notebook_id: str) -> None:
+        self._post("/api/notebook/openNotebook", {"notebook": notebook_id})
+
+    def close_notebook(self, notebook_id: str) -> None:
+        self._post("/api/notebook/closeNotebook", {"notebook": notebook_id})
+
+    def create_notebook(self, name: str) -> dict[str, Any]:
+        data = self._post("/api/notebook/createNotebook", {"name": name})
+        if isinstance(data, dict) and "notebook" in data:
+            return data["notebook"]
+        return data
+
+    def remove_notebook(self, notebook_id: str) -> None:
+        self._post("/api/notebook/removeNotebook", {"notebook": notebook_id})
+
+    def rename_notebook(self, notebook_id: str, name: str) -> None:
+        self._post("/api/notebook/renameNotebook", {"notebook": notebook_id, "name": name})
+
+    def get_notebook_conf(self, notebook_id: str) -> dict[str, Any]:
+        return self._post("/api/notebook/getNotebookConf", {"notebook": notebook_id})
+
+    def set_notebook_conf(self, notebook_id: str, conf: dict[str, Any]) -> dict[str, Any]:
+        return self._post("/api/notebook/setNotebookConf", {"notebook": notebook_id, "conf": conf})
+
+    def set_notebook_icon(self, notebook_id: str, icon: str) -> None:
+        self._post("/api/notebook/setNotebookIcon", {"notebook": notebook_id, "icon": icon})
+
+    def get_notebook_info(self, notebook_id: str) -> dict[str, Any]:
+        return self._post("/api/notebook/getNotebookInfo", {"notebook": notebook_id})
+
+    # ── Document / Filetree API ────────────────────────────────────────
+
+    def create_doc_with_md(self, notebook_id: str, path: str, markdown: str = "") -> str:
+        """Create a document with Markdown content. Returns the doc ID."""
+        return self._post("/api/filetree/createDocWithMd", {
+            "notebook": notebook_id, "path": path, "markdown": markdown,
+        })
+
+    def rename_doc(self, notebook_id: str, path: str, title: str) -> None:
+        self._post("/api/filetree/renameDoc", {"notebook": notebook_id, "path": path, "title": title})
+
+    def rename_doc_by_id(self, doc_id: str, title: str) -> None:
+        self._post("/api/filetree/renameDocByID", {"id": doc_id, "title": title})
+
+    def remove_doc(self, notebook_id: str, path: str) -> None:
+        self._post("/api/filetree/removeDoc", {"notebook": notebook_id, "path": path})
+
+    def remove_doc_by_id(self, doc_id: str) -> None:
+        self._post("/api/filetree/removeDocByID", {"id": doc_id})
+
+    def move_docs(self, from_paths: list[str], to_notebook: str, to_path: str) -> None:
+        self._post("/api/filetree/moveDocs", {
+            "fromPaths": from_paths, "toNotebook": to_notebook, "toPath": to_path,
+        })
+
+    def move_docs_by_id(self, from_ids: list[str], to_id: str) -> None:
+        self._post("/api/filetree/moveDocsByID", {"fromIDs": from_ids, "toID": to_id})
+
+    def get_hpath_by_id(self, block_id: str) -> str:
+        return self._post("/api/filetree/getHPathByID", {"id": block_id})
+
+    def get_path_by_id(self, block_id: str) -> dict[str, str]:
+        return self._post("/api/filetree/getPathByID", {"id": block_id})
+
+    def get_ids_by_hpath(self, notebook_id: str, path: str) -> list[str]:
+        return self._post("/api/filetree/getIDsByHPath", {"notebook": notebook_id, "path": path})
+
+    def list_docs_by_path(self, notebook_id: str, path: str) -> list[dict[str, Any]]:
+        return self._post("/api/filetree/listDocsByPath", {
+            "notebook": notebook_id, "path": path, "maxListCount": 0,
+        })
+
+    def list_doc_tree(self, notebook_id: str, path: str = "/", max_depth: int = -1, sort: int = 0) -> list[dict[str, Any]]:
+        return self._post("/api/filetree/listDocTree", {
+            "notebook": notebook_id, "path": path, "maxDepth": max_depth, "sort": sort,
+        })
+
+    def search_docs(self, keyword: str) -> list[dict[str, Any]]:
+        return self._post("/api/filetree/searchDocs", {"k": keyword})
+
+    def create_daily_note(self, notebook_id: str) -> dict[str, Any]:
+        return self._post("/api/filetree/createDailyNote", {"notebook": notebook_id})
+
+    # ── Block API ──────────────────────────────────────────────────────
+
+    def insert_block(self, data_type: str, data: str,
+                     parent_id: str = "", previous_id: str = "", next_id: str = "") -> list[dict[str, Any]]:
+        params = {"dataType": data_type, "data": data}
+        if parent_id:
+            params["parentID"] = parent_id
+        if previous_id:
+            params["previousID"] = previous_id
+        if next_id:
+            params["nextID"] = next_id
+        return self._post("/api/block/insertBlock", params)
+
+    def prepend_block(self, data_type: str, data: str, parent_id: str) -> list[dict[str, Any]]:
+        return self._post("/api/block/prependBlock", {
+            "dataType": data_type, "data": data, "parentID": parent_id,
+        })
+
+    def append_block(self, data_type: str, data: str, parent_id: str) -> list[dict[str, Any]]:
+        return self._post("/api/block/appendBlock", {
+            "dataType": data_type, "data": data, "parentID": parent_id,
+        })
+
+    def update_block(self, data_type: str, data: str, block_id: str) -> list[dict[str, Any]]:
+        return self._post("/api/block/updateBlock", {
+            "dataType": data_type, "data": data, "id": block_id,
+        })
+
+    def delete_block(self, block_id: str) -> list[dict[str, Any]]:
+        return self._post("/api/block/deleteBlock", {"id": block_id})
+
+    def move_block(self, block_id: str, previous_id: str = "", parent_id: str = "") -> list[dict[str, Any]]:
+        params = {"id": block_id}
+        if previous_id:
+            params["previousID"] = previous_id
+        if parent_id:
+            params["parentID"] = parent_id
+        return self._post("/api/block/moveBlock", params)
+
+    def fold_block(self, block_id: str) -> None:
+        self._post("/api/block/foldBlock", {"id": block_id})
+
+    def unfold_block(self, block_id: str) -> None:
+        self._post("/api/block/unfoldBlock", {"id": block_id})
+
+    def get_block_kramdown(self, block_id: str) -> str:
+        data = self._post("/api/block/getBlockKramdown", {"id": block_id})
+        return data.get("kramdown", "")
+
+    def get_child_blocks(self, block_id: str) -> list[dict[str, Any]]:
+        return self._post("/api/block/getChildBlocks", {"id": block_id})
+
+    # ── Attribute API ──────────────────────────────────────────────────
+
+    def set_block_attrs(self, block_id: str, attrs: dict[str, str]) -> None:
+        self._post("/api/attr/setBlockAttrs", {"id": block_id, "attrs": attrs})
+
+    def get_block_attrs(self, block_id: str) -> dict[str, str]:
+        return self._post("/api/attr/getBlockAttrs", {"id": block_id})
+
+    # ── SQL Query API ──────────────────────────────────────────────────
+
+    def query_sql(self, stmt: str) -> list[dict[str, Any]]:
+        return self._post("/api/query/sql", {"stmt": stmt})
+
+    # ── Search API ─────────────────────────────────────────────────────
+
+    def search_blocks(self, query: str) -> list[dict[str, Any]]:
+        return self._post("/api/search/fullTextSearchBlock", {"query": query})
+
+    def search_tag(self, tag: str = "") -> list[str]:
+        """Search tags. Returns list of tag name strings."""
+        data = self._post("/api/search/searchTag", {"k": tag})
+        if isinstance(data, dict):
+            return data.get("tags", [])
+        return data or []
+
+    def find_replace(self, keyword: str, replacement: str, ids: list[str]) -> None:
+        """Search and replace text in blocks by ID."""
+        self._post("/api/search/findReplace", {
+            "k": keyword, "r": replacement, "ids": ids,
+        })
+
+    # ── Export API ─────────────────────────────────────────────────────
+
+    def export_md_content(self, doc_id: str) -> dict[str, str]:
+        return self._post("/api/export/exportMdContent", {"id": doc_id})
+
+    def export_resources(self, paths: list[str], name: str = "") -> str:
+        params = {"paths": paths}
+        if name:
+            params["name"] = name
+        data = self._post("/api/export/exportResources", params)
+        return data.get("path", "")
+
+    # ── Tag API ────────────────────────────────────────────────────────
+
+    def get_tags(self) -> list[dict[str, Any]]:
+        return self._post("/api/tag/getTag", {"ignoreMaxListHint": True})
+
+    # ── System API ─────────────────────────────────────────────────────
+
+    def get_version(self) -> str:
+        return self._post("/api/system/version")
+
+    def get_current_time(self) -> int:
+        return self._post("/api/system/currentTime")
@@ -0,0 +1,75 @@
+"""Session state management for the SiYuan CLI REPL."""
+
+import json
+import os
+from dataclasses import dataclass, field, asdict
+from pathlib import Path
+from typing import Any
+
+
+@dataclass
+class SessionState:
+    """Mutable session state — represents current notebook/doc context."""
+    current_notebook_id: str = ""
+    current_notebook_name: str = ""
+    current_doc_id: str = ""
+    current_doc_path: str = ""
+    connected: bool = False
+    history: list[str] = field(default_factory=list)
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "SessionState":
+        return cls(
+            current_notebook_id=data.get("current_notebook_id", ""),
+            current_notebook_name=data.get("current_notebook_name", ""),
+            current_doc_id=data.get("current_doc_id", ""),
+            current_doc_path=data.get("current_doc_path", ""),
+            connected=data.get("connected", False),
+            history=data.get("history", []),
+        )
+
+
+class SessionManager:
+    """Manages session persistence for the REPL."""
+
+    def __init__(self, state_dir: str | None = None):
+        if state_dir is None:
+            state_dir = str(Path.home() / ".cli-anything-siyuan")
+        self.state_dir = Path(state_dir)
+        self.state_dir.mkdir(parents=True, exist_ok=True)
+        self.session_path = self.state_dir / "session.json"
+        self.state = SessionState()
+        self._dirty = False
+
+    def load(self) -> SessionState:
+        """Load session state from disk."""
+        if self.session_path.is_file():
+            try:
+                data = json.loads(self.session_path.read_text(encoding="utf-8"))
+                self.state = SessionState.from_dict(data)
+            except (json.JSONDecodeError, KeyError):
+                self.state = SessionState()
+        return self.state
+
+    def save(self) -> None:
+        """Save current session state to disk."""
+        self.session_path.write_text(
+            json.dumps(self.state.to_dict(), ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+
+    def update(self, **kwargs: Any) -> None:
+        """Update session state fields and mark dirty."""
+        for key, value in kwargs.items():
+            if hasattr(self.state, key):
+                setattr(self.state, key, value)
+        self._dirty = True
+
+    def flush(self) -> None:
+        """Write to disk if dirty."""
+        if self._dirty:
+            self.save()
+            self._dirty = False
@@ -0,0 +1,802 @@
+"""cli-anything-siyuan — SiYuan (思源笔记) CLI harness.
+
+Connects to a running SiYuan kernel via its HTTP API and provides
+commands for notebooks, documents, blocks, search, and export.
+"""
+
+import json
+import os
+import shlex
+import sys
+from typing import Any
+
+import click
+
+from cli_anything.siyuan.core.client import (
+    SiYuanClient,
+    SiYuanClientError,
+    SiYuanConfig,
+    load_config,
+)
+from cli_anything.siyuan.core.session import SessionManager
+
+
+def _read_stdin() -> str:
+    """Read stdin as raw bytes and decode as UTF-8 (with BOM handling).
+
+    PowerShell pipes text using the console code page (e.g. GBK on Chinese
+    Windows), which mangles CJK characters before Python sees them.  Reading
+    raw bytes then decoding as UTF-8-sig is the most robust approach.
+    """
+    if sys.stdin.isatty():
+        raise click.UsageError(
+            "stdin pipe expected (e.g. echo 'content' | sy block insert --parent pid -)"
+        )
+    raw = sys.stdin.buffer.read()
+    try:
+        return raw.decode('utf-8-sig')
+    except UnicodeDecodeError:
+        return raw.decode('utf-8-sig', errors='replace')
+
+
+# ── Click context ─────────────────────────────────────────────────────
+
+class SiYuanContext:
+    def __init__(self, client: SiYuanClient | None = None,
+                 session: SessionManager | None = None,
+                 json_output: bool = False):
+        self.json_output = json_output
+        self.client = client
+        self.session = session or SessionManager()
+
+
+class _CatchErrors(click.Group):
+    """Click Group that converts SiYuanClientError to clean CLI errors."""
+
+    def invoke(self, ctx: click.Context) -> object:
+        try:
+            return super().invoke(ctx)
+        except SiYuanClientError as e:
+            click.echo(f"Error: {e}", err=True)
+            sys.exit(1)
+
+
+@click.group(cls=_CatchErrors, invoke_without_command=True)
+@click.option("--json", "json_output", is_flag=True, help="Output in JSON format")
+@click.option("--host", default="", help="SiYuan host (default: 127.0.0.1)")
+@click.option("--port", default=0, type=int, help="SiYuan port (default: 6806)")
+@click.option("--token", default="", help="SiYuan API token")
+@click.option("--config", "config_path", default="", help="Config file path")
+@click.pass_context
+def cli(ctx: click.Context, json_output: bool, host: str, port: int,
+        token: str, config_path: str):
+    # Force UTF-8 output to handle CJK characters on Windows
+    for stream in (sys.stdout, sys.stderr):
+        if hasattr(stream, 'reconfigure'):
+            stream.reconfigure(encoding='utf-8')
+
+    """CLI for SiYuan (思源笔记) — interact with your knowledge base.
+
+    Connects to a running SiYuan instance via its HTTP API.
+    Default: http://127.0.0.1:6806
+
+    Configure connection via ~/.siyuan-cli.json, env vars
+    (SIYUAN_HOST, SIYUAN_PORT, SIYUAN_TOKEN), or CLI flags.
+    """
+    session = SessionManager()
+    session.load()
+    cfg = load_config(config_path or None)
+    if host or port or token:
+        cfg = SiYuanConfig(
+            host=host or cfg.host,
+            port=port or cfg.port,
+            token=token or cfg.token,
+        )
+    client = SiYuanClient(cfg)
+    ctx.obj = SiYuanContext(client=client, session=session, json_output=json_output)
+
+    if ctx.invoked_subcommand is None:
+        if not client.ping():
+            click.echo("Error: Cannot connect to SiYuan. Is it running?", err=True)
+            click.echo("  Configure via: --host --port --token", err=True)
+            click.echo("  Or set: SIYUAN_HOST, SIYUAN_PORT, SIYUAN_TOKEN", err=True)
+            sys.exit(1)
+        ctx.invoke(repl)
+
+
+def _walk_tree(items: list[dict], depth: int = 0) -> list[dict]:
+    """Recursively flatten a doc tree, setting depth on each entry.
+
+    Returns new dicts; does not mutate the originals.
+    """
+    result: list[dict] = []
+    for t in items:
+        entry = {**t, "depth": depth}
+        result.append(entry)
+        children = t.get("children")
+        if children:
+            result.extend(_walk_tree(children, depth + 1))
+    return result
+
+
+# ── REPL ───────────────────────────────────────────────────────────────
+
+def _build_repl_commands() -> dict[str, str]:
+    return {
+        "notebook list": "List all notebooks",
+        "notebook create <name>": "Create a notebook",
+        "notebook rename <id> <name>": "Rename a notebook",
+        "notebook remove <id>": "Remove a notebook",
+        "doc create <notebook> <path>": "Create a document with optional --md content",
+        "doc list <notebook> <path>": "List documents at path",
+        "doc tree <notebook>": "List full document tree",
+        "doc get <id>": "Get document info by ID",
+        "block insert <parent> <data>": "Insert a block (use '-' for stdin)",
+        "block update <id> <data>": "Update a block (use '-' for stdin)",
+        "block delete <id>": "Delete a block",
+        "block get <id>": "Get block kramdown source",
+        "sql <stmt>": "Execute SQL query",
+        "search <query>": "Full-text search",
+        "export md <doc-id>": "Export doc as Markdown",
+        "status": "Show connection and session status",
+        "help": "Show this help",
+        "quit": "Exit REPL",
+    }
+
+
+@cli.command()
+@click.pass_context
+def repl(ctx: click.Context):
+    """Start interactive REPL mode."""
+    from cli_anything.siyuan.utils.repl_skin import ReplSkin
+
+    obj: SiYuanContext = ctx.obj
+
+    try:
+        kernel_version = obj.client.get_version()
+    except SiYuanClientError:
+        kernel_version = "?"
+    skin = ReplSkin("siyuan", version=kernel_version)
+    skin.print_banner()
+
+    pt_session = skin.create_prompt_session()
+    commands = _build_repl_commands()
+
+    state = obj.session.state
+
+    while True:
+        ctx_str = ""
+        if state.current_notebook_name:
+            ctx_str = state.current_notebook_name
+        user_input = skin.get_input(
+            pt_session,
+            project_name=state.current_doc_path or "",
+            modified=False,
+            context=ctx_str,
+        )
+
+        if not user_input:
+            continue
+
+        if user_input in ("quit", "exit", "q"):
+            obj.session.flush()
+            break
+
+        if user_input == "help":
+            skin.help(commands)
+            continue
+
+        if user_input == "status":
+            connected = obj.client.ping()
+            skin.status_block({
+                "Connected": str(connected),
+                "Notebook": state.current_notebook_name or "(none)",
+                "Document": state.current_doc_path or "(none)",
+            }, title="Status")
+            continue
+
+        try:
+            _dispatch_repl(skin, obj, user_input)
+        except SiYuanClientError as e:
+            skin.error(str(e))
+        except Exception as e:
+            skin.error(f"Error: {e}")
+
+    skin.print_goodbye()
+
+
+def _dispatch_repl(skin: Any, ctx: SiYuanContext, cmd: str) -> None:
+    """Parse REPL command and route to the appropriate handler."""
+    parts = shlex.split(cmd.strip())
+    if not parts:
+        return
+
+    json_mode = "--json" in parts
+    parts = [p for p in parts if p != "--json"]
+
+    client = ctx.client
+    session = ctx.session
+
+    command = parts[0]
+    if command == "notebook":
+        _handle_notebook_repl(skin, client, session, parts, json_mode)
+    elif command == "doc":
+        _handle_doc_repl(skin, client, session, parts, json_mode)
+    elif command == "block":
+        _handle_block_repl(skin, client, parts, json_mode)
+    elif command == "sql" and len(parts) >= 2:
+        _handle_sql_repl(skin, client, parts, json_mode)
+    elif command == "search" and len(parts) >= 2:
+        _handle_search_repl(skin, client, parts, json_mode)
+    elif command == "export":
+        _handle_export_repl(skin, client, parts, json_mode)
+    else:
+        skin.error(f"Unknown command: {parts[0]}")
+
+
+# ── REPL sub-handlers ──────────────────────────────────────────────────
+
+def _handle_notebook_repl(skin: Any, client: SiYuanClient,
+                          session: SessionManager, parts: list[str],
+                          json_mode: bool) -> None:
+    if len(parts) < 2:
+        skin.error("Usage: notebook <list|create|rename|remove>")
+        return
+    sub = parts[1]
+    if sub == "list":
+        notebooks = client.list_notebooks()
+        if json_mode:
+            click.echo(json.dumps(notebooks, ensure_ascii=False))
+        else:
+            skin.table(["ID", "Name", "Icon", "Closed"],
+                       [[n["id"], n["name"], n.get("icon", ""), str(n.get("closed", ""))]
+                        for n in notebooks])
+    elif sub == "create" and len(parts) >= 3:
+        name = " ".join(parts[2:])
+        nb = client.create_notebook(name)
+        session.update(current_notebook_id=nb["id"], current_notebook_name=nb["name"])
+        client.open_notebook(nb["id"])
+        skin.success(f'Created notebook: {nb["name"]} ({nb["id"]})')
+    elif sub == "rename" and len(parts) >= 4:
+        client.rename_notebook(parts[2], " ".join(parts[3:]))
+        skin.success("Renamed")
+    elif sub == "remove" and len(parts) >= 3:
+        client.remove_notebook(parts[2])
+        skin.success("Removed")
+    else:
+        skin.error("Invalid notebook command")
+
+
+def _handle_doc_repl(skin: Any, client: SiYuanClient,
+                     session: SessionManager, parts: list[str],
+                     json_mode: bool) -> None:
+    if len(parts) < 2:
+        skin.error("Usage: doc <create|list|tree|get|rename|remove|export>")
+        return
+    sub = parts[1]
+    if sub == "create" and len(parts) >= 4:
+        md = ""
+        if "--md" in parts:
+            idx = parts.index("--md") + 1
+            if idx < len(parts):
+                md = parts[idx]
+            parts = parts[:parts.index("--md")]
+        if md == "-":
+            md = _read_stdin()
+        nb_id = parts[2]
+        doc_path = parts[3]
+        doc_id = client.create_doc_with_md(nb_id, doc_path, md)
+        session.update(current_doc_id=doc_id, current_doc_path=doc_path)
+        if json_mode:
+            click.echo(json.dumps({"id": doc_id}, ensure_ascii=False))
+        else:
+            skin.success(f"Created doc: {doc_id}")
+    elif sub == "list" and len(parts) >= 4:
+        docs = client.list_docs_by_path(parts[2], parts[3])
+        items = docs.get("files", []) if isinstance(docs, dict) else docs
+        if json_mode:
+            click.echo(json.dumps(items, ensure_ascii=False))
+        else:
+            skin.table(["ID", "Name", "Type"],
+                       [[d.get("id", ""), d.get("name", ""), d.get("type", "")]
+                        for d in items])
+    elif sub == "tree" and len(parts) >= 3:
+        tree = client.list_doc_tree(parts[2])
+        if isinstance(tree, dict):
+            items = tree.get("files") or tree.get("tree") or []
+        else:
+            items = tree
+        if json_mode:
+            click.echo(json.dumps(items, ensure_ascii=False))
+        else:
+            skin.table(["ID", "Name", "Path"],
+                       [[t.get("id", ""), "  " * t.get("depth", 0) + t.get("name", ""), t.get("path", "")]
+                        for t in _walk_tree(items)])
+    elif sub == "get" and len(parts) >= 3:
+        hpath = client.get_hpath_by_id(parts[2])
+        if json_mode:
+            click.echo(json.dumps({"hpath": hpath}, ensure_ascii=False))
+        else:
+            skin.success(f"Path: {hpath}")
+    elif sub == "rename" and len(parts) >= 4:
+        client.rename_doc_by_id(parts[2], " ".join(parts[3:]))
+        skin.success("Renamed")
+    elif sub == "remove" and len(parts) >= 3:
+        client.remove_doc_by_id(parts[2])
+        skin.success("Removed")
+    elif sub == "export" and len(parts) >= 3:
+        md = client.export_md_content(parts[2])
+        if json_mode:
+            click.echo(json.dumps(md, ensure_ascii=False))
+        else:
+            skin.section(md.get("hPath", ""))
+            click.echo(md.get("content", ""))
+    else:
+        skin.error("Invalid doc command")
+
+
+def _handle_block_repl(skin: Any, client: SiYuanClient,
+                       parts: list[str], json_mode: bool) -> None:
+    if len(parts) < 2:
+        skin.error("Usage: block <insert|prepend|append|update|delete|get|child>")
+        return
+    sub = parts[1]
+    if sub == "insert":
+        if len(parts) < 4:
+            skin.error("Usage: block insert <parent_id> <data>")
+            return
+        data = parts[3]
+        if data == "-":
+            data = _read_stdin()
+        result = client.insert_block("markdown", data, parent_id=parts[2])
+        if json_mode:
+            click.echo(json.dumps(result, ensure_ascii=False))
+        else:
+            skin.success("Block inserted")
+    elif sub == "prepend" and len(parts) >= 4:
+        data = parts[3]
+        if data == "-":
+            data = _read_stdin()
+        client.prepend_block("markdown", data, parts[2])
+        skin.success("Block prepended")
+    elif sub == "append" and len(parts) >= 4:
+        data = parts[3]
+        if data == "-":
+            data = _read_stdin()
+        client.append_block("markdown", data, parts[2])
+        skin.success("Block appended")
+    elif sub == "update" and len(parts) >= 4:
+        data = parts[3]
+        if data == "-":
+            data = _read_stdin()
+        client.update_block("markdown", data, parts[2])
+        skin.success("Block updated")
+    elif sub == "delete" and len(parts) >= 3:
+        client.delete_block(parts[2])
+        skin.success("Block deleted")
+    elif sub == "get" and len(parts) >= 3:
+        kramdown = client.get_block_kramdown(parts[2])
+        if json_mode:
+            click.echo(json.dumps({"kramdown": kramdown}, ensure_ascii=False))
+        else:
+            click.echo(kramdown)
+    elif sub == "child" and len(parts) >= 3:
+        children = client.get_child_blocks(parts[2])
+        if json_mode:
+            click.echo(json.dumps(children, ensure_ascii=False))
+        else:
+            skin.table(["ID", "Type", "SubType"],
+                       [[c.get("id", ""), c.get("type", ""), c.get("subType", "")]
+                        for c in children])
+    else:
+        skin.error("Invalid block command")
+
+
+def _handle_sql_repl(skin: Any, client: SiYuanClient,
+                     parts: list[str], json_mode: bool) -> None:
+    stmt = " ".join(parts[1:])
+    results = client.query_sql(stmt)
+    if json_mode:
+        click.echo(json.dumps(results, ensure_ascii=False))
+    elif not results:
+        skin.info("No results")
+    else:
+        headers = list(results[0].keys())
+        rows = [[str(r.get(h, "")) for h in headers] for r in results]
+        skin.table(headers, rows)
+
+
+def _handle_search_repl(skin: Any, client: SiYuanClient,
+                        parts: list[str], json_mode: bool) -> None:
+    query = " ".join(parts[1:])
+    data = client.search_blocks(query)
+    blocks = data.get("blocks", []) if isinstance(data, dict) else data
+    if json_mode:
+        click.echo(json.dumps(blocks, ensure_ascii=False))
+    elif not blocks:
+        skin.info("No results")
+    else:
+        skin.table(["ID", "Content"],
+                   [[r.get("id", ""), r.get("content", "")[:80]] for r in blocks])
+
+
+def _handle_export_repl(skin: Any, client: SiYuanClient,
+                        parts: list[str], json_mode: bool) -> None:
+    if parts[1] == "md" and len(parts) >= 3:
+        md = client.export_md_content(parts[2])
+        if json_mode:
+            click.echo(json.dumps(md, ensure_ascii=False))
+        else:
+            skin.section(md.get("hPath", ""))
+            click.echo(md.get("content", ""))
+    else:
+        skin.error("Usage: export md <doc-id>")
+
+
+# ── Notebook commands ──────────────────────────────────────────────────
+
+@cli.group()
+def notebook():
+    """Manage notebooks (笔记本)."""
+
+
+@notebook.command("list")
+@click.pass_obj
+def notebook_list(ctx: SiYuanContext):
+    """List all notebooks."""
+    notebooks = ctx.client.list_notebooks()
+    if ctx.json_output:
+        click.echo(json.dumps(notebooks, ensure_ascii=False))
+    else:
+        click.echo(f"{'ID':<30} {'Name':<30} {'Closed':<8}")
+        click.echo("-" * 70)
+        for nb in notebooks:
+            click.echo(f"{nb['id']:<30} {nb['name']:<30} {str(nb.get('closed', '')):<8}")
+
+
+@notebook.command("create")
+@click.argument("name")
+@click.pass_obj
+def notebook_create(ctx: SiYuanContext, name: str):
+    """Create a new notebook."""
+    nb = ctx.client.create_notebook(name)
+    if ctx.json_output:
+        click.echo(json.dumps(nb, ensure_ascii=False))
+    else:
+        click.echo(f"Created notebook: {nb['name']} ({nb['id']})")
+
+
+@notebook.command("remove")
+@click.argument("notebook_id")
+@click.pass_obj
+def notebook_remove(ctx: SiYuanContext, notebook_id: str):
+    """Remove a notebook by ID."""
+    ctx.client.remove_notebook(notebook_id)
+    click.echo(f"Removed notebook: {notebook_id}")
+
+
+@notebook.command("rename")
+@click.argument("notebook_id")
+@click.argument("name")
+@click.pass_obj
+def notebook_rename(ctx: SiYuanContext, notebook_id: str, name: str):
+    """Rename a notebook."""
+    ctx.client.rename_notebook(notebook_id, name)
+    click.echo(f"Renamed notebook {notebook_id} to: {name}")
+
+
+@notebook.command("open")
+@click.argument("notebook_id")
+@click.pass_obj
+def notebook_open(ctx: SiYuanContext, notebook_id: str):
+    """Open a notebook."""
+    ctx.client.open_notebook(notebook_id)
+    name = notebook_id
+    for nb in ctx.client.list_notebooks():
+        if nb.get("id") == notebook_id:
+            name = nb.get("name", notebook_id)
+            break
+    ctx.session.update(current_notebook_id=notebook_id, current_notebook_name=name)
+    ctx.session.flush()
+    click.echo(f"Opened notebook: {name} ({notebook_id})")
+
+
+# ── Document commands ──────────────────────────────────────────────────
+
+@cli.group()
+def doc():
+    """Manage documents (文档)."""
+
+
+@doc.command("create")
+@click.argument("notebook_id")
+@click.argument("path")
+@click.option("--md", default="", help="Markdown content. Use '-' to read from stdin.")
+@click.pass_obj
+def doc_create(ctx: SiYuanContext, notebook_id: str, path: str, md: str):
+    """Create a document with optional Markdown content.
+
+    To avoid shell escaping issues with special characters
+    (backticks, quotes, parentheses), pipe Markdown via stdin:
+      echo "## Title" | sy doc create nb1 /test --md -
+    """
+    if md == "-":
+        md = _read_stdin()
+    doc_id = ctx.client.create_doc_with_md(notebook_id, path, md)
+    if ctx.json_output:
+        click.echo(json.dumps({"id": doc_id}, ensure_ascii=False))
+    else:
+        click.echo(f"Created doc: {doc_id}")
+
+
+@doc.command("list")
+@click.argument("notebook_id")
+@click.argument("path", default="/")
+@click.pass_obj
+def doc_list(ctx: SiYuanContext, notebook_id: str, path: str):
+    """List documents at a path."""
+    docs = ctx.client.list_docs_by_path(notebook_id, path)
+    items = docs.get("files", []) if isinstance(docs, dict) else docs
+    if ctx.json_output:
+        click.echo(json.dumps(items, ensure_ascii=False))
+    else:
+        click.echo(f"{'ID':<30} {'Name':<30} {'Type':<10}")
+        click.echo("-" * 70)
+        for d in items:
+            click.echo(f"{d.get('id', ''):<30} {d.get('name', ''):<30} {d.get('type', ''):<10}")
+
+
+@doc.command("tree")
+@click.argument("notebook_id")
+@click.option("--path", default="/", help="Root path")
+@click.option("--depth", default=-1, type=int, help="Max depth")
+@click.pass_obj
+def doc_tree(ctx: SiYuanContext, notebook_id: str, path: str, depth: int):
+    """List document tree."""
+    tree = ctx.client.list_doc_tree(notebook_id, path=path, max_depth=depth)
+    if isinstance(tree, dict):
+        items = tree.get("files") or tree.get("tree") or []
+    else:
+        items = tree
+    if ctx.json_output:
+        click.echo(json.dumps(items, ensure_ascii=False))
+    else:
+        for t in _walk_tree(items):
+            indent = "  " * t.get("depth", 0)
+            click.echo(f"{indent}{t.get('name', '')}  ({t.get('id', '')})")
+
+
+@doc.command("get")
+@click.argument("doc_id")
+@click.pass_obj
+def doc_get(ctx: SiYuanContext, doc_id: str):
+    """Get document info by ID."""
+    hpath = ctx.client.get_hpath_by_id(doc_id)
+    if ctx.json_output:
+        click.echo(json.dumps({"hpath": hpath}, ensure_ascii=False))
+    else:
+        click.echo(f"Path: {hpath}")
+
+
+@doc.command("rename")
+@click.argument("doc_id")
+@click.argument("title")
+@click.pass_obj
+def doc_rename(ctx: SiYuanContext, doc_id: str, title: str):
+    """Rename a document."""
+    ctx.client.rename_doc_by_id(doc_id, title)
+    click.echo(f"Renamed {doc_id} to: {title}")
+
+
+@doc.command("remove")
+@click.argument("doc_id")
+@click.pass_obj
+def doc_remove(ctx: SiYuanContext, doc_id: str):
+    """Remove a document."""
+    ctx.client.remove_doc_by_id(doc_id)
+    click.echo(f"Removed: {doc_id}")
+
+
+# ── Block commands ─────────────────────────────────────────────────────
+
+@cli.group()
+def block():
+    """Manage blocks (内容块)."""
+
+
+@block.command("insert")
+@click.argument("data", required=False)
+@click.option("--previous", default="", help="Previous block ID")
+@click.option("--parent", default="", help="Parent block ID")
+@click.option("--next", "next_", default="", help="Next block ID")
+@click.option("--data-type", default="markdown", help="Data type (markdown/dom)")
+@click.pass_obj
+def block_insert(ctx: SiYuanContext, data: str | None, previous: str, parent: str, next_: str, data_type: str):
+    """Insert a block. Data reads from stdin when '-' or omitted."""
+    if not parent and not previous and not next_:
+        raise click.UsageError("An anchor is required: --parent, --previous, or --next")
+    if not data or data == "-":
+        data = _read_stdin()
+    result = ctx.client.insert_block(data_type, data, parent_id=parent, previous_id=previous, next_id=next_)
+    if ctx.json_output:
+        click.echo(json.dumps(result, ensure_ascii=False))
+    else:
+        click.echo("Block inserted")
+
+
+@block.command("update")
+@click.argument("block_id")
+@click.argument("data", required=False)
+@click.option("--data-type", default="markdown", help="Data type")
+@click.pass_obj
+def block_update(ctx: SiYuanContext, block_id: str, data: str | None, data_type: str):
+    """Update a block's content. Data reads from stdin when '-' or omitted."""
+    if not data or data == "-":
+        data = _read_stdin()
+    ctx.client.update_block(data_type, data, block_id)
+    click.echo(f"Updated block: {block_id}")
+
+
+@block.command("delete")
+@click.argument("block_id")
+@click.pass_obj
+def block_delete(ctx: SiYuanContext, block_id: str):
+    """Delete a block."""
+    ctx.client.delete_block(block_id)
+    click.echo(f"Deleted block: {block_id}")
+
+
+@block.command("get")
+@click.argument("block_id")
+@click.pass_obj
+def block_get(ctx: SiYuanContext, block_id: str):
+    """Get block kramdown source."""
+    kramdown = ctx.client.get_block_kramdown(block_id)
+    if ctx.json_output:
+        click.echo(json.dumps({"kramdown": kramdown}, ensure_ascii=False))
+    else:
+        click.echo(kramdown)
+
+
+@block.command("children")
+@click.argument("block_id")
+@click.pass_obj
+def block_children(ctx: SiYuanContext, block_id: str):
+    """Get child blocks."""
+    children = ctx.client.get_child_blocks(block_id)
+    if ctx.json_output:
+        click.echo(json.dumps(children, ensure_ascii=False))
+    else:
+        for c in children:
+            click.echo(f"{c.get('id', ''):<30} {c.get('type', ''):<8} {c.get('subType', '')}")
+
+
+# ── SQL command ────────────────────────────────────────────────────────
+
+@cli.command()
+@click.argument("stmt")
+@click.pass_obj
+def sql(ctx: SiYuanContext, stmt: str):
+    """Execute a SQL query on the block database.
+
+    Example: sql "SELECT * FROM blocks WHERE content LIKE '%keyword%' LIMIT 10"
+    """
+    results = ctx.client.query_sql(stmt)
+    if ctx.json_output:
+        click.echo(json.dumps(results, ensure_ascii=False))
+    elif not results:
+        click.echo("No results")
+    else:
+        for r in results:
+            click.echo(json.dumps(r, ensure_ascii=False))
+
+
+# ── Search commands ────────────────────────────────────────────────────
+
+@cli.command()
+@click.argument("query")
+@click.pass_obj
+def search(ctx: SiYuanContext, query: str):
+    """Full-text search across all blocks."""
+    data = ctx.client.search_blocks(query)
+    blocks = data.get("blocks", []) if isinstance(data, dict) else data
+    if ctx.json_output:
+        click.echo(json.dumps(blocks, ensure_ascii=False))
+    elif not blocks:
+        click.echo("No results")
+    else:
+        for r in blocks[:20]:
+            click.echo(f"- {r.get('id', '')}: {r.get('content', '')[:120]}")
+
+
+# ── Export commands ────────────────────────────────────────────────────
+
+@cli.group()
+def export():
+    """Export content from SiYuan."""
+
+
+@export.command("md")
+@click.argument("doc_id")
+@click.pass_obj
+def export_md(ctx: SiYuanContext, doc_id: str):
+    """Export a document as Markdown."""
+    md = ctx.client.export_md_content(doc_id)
+    if ctx.json_output:
+        click.echo(json.dumps(md, ensure_ascii=False))
+    else:
+        click.echo(f"# {md.get('hPath', '')}")
+        click.echo("")
+        click.echo(md.get("content", ""))
+
+
+# ── System commands ────────────────────────────────────────────────────
+
+@cli.command()
+@click.pass_obj
+def version(ctx: SiYuanContext):
+    """Show SiYuan kernel version."""
+    ver = ctx.client.get_version()
+    if ctx.json_output:
+        click.echo(json.dumps({"version": ver}, ensure_ascii=False))
+    else:
+        click.echo(f"SiYuan version: {ver}")
+
+
+@cli.command()
+@click.pass_obj
+def status(ctx: SiYuanContext):
+    """Show connection and session status."""
+    connected = ctx.client.ping()
+    siyuan_ver = ctx.client.get_version() if connected else ""
+    session = ctx.session
+    state = session.state
+
+    info = {
+        "connected": connected,
+        "siyuan_version": siyuan_ver,
+        "host": ctx.client.config.host,
+        "port": ctx.client.config.port,
+        "current_notebook": state.current_notebook_name or "",
+        "current_doc": state.current_doc_path or "",
+    }
+
+    if ctx.json_output:
+        click.echo(json.dumps(info, ensure_ascii=False))
+    else:
+        click.echo(f"  Connected:     {info['connected']}")
+        click.echo(f"  SiYuan:        v{info['siyuan_version']}")
+        click.echo(f"  Host:          {info['host']}:{info['port']}")
+        click.echo(f"  Notebook:      {info['current_notebook']}")
+        click.echo(f"  Document:      {info['current_doc']}")
+
+
+# ── Tag commands ───────────────────────────────────────────────────────
+
+@cli.group()
+def tag():
+    """Manage tags (标签)."""
+
+
+def _print_tags(tags: list[dict], indent: int = 0) -> None:
+    """Recursively print tags with hierarchical indentation."""
+    pad = "  " * indent
+    for t in tags:
+        if indent == 0:
+            click.echo(f"{t.get('name', ''):<30} ({t.get('count', 0)})")
+        else:
+            click.echo(f"{pad}{t.get('name', ''):<{30 - 2 * indent}} ({t.get('count', 0)})")
+        children = t.get("children")
+        if children:
+            _print_tags(children, indent + 1)
+
+
+@tag.command("list")
+@click.pass_obj
+def tag_list(ctx: SiYuanContext):
+    """List all tags (including nested tags)."""
+    tags = ctx.client.get_tags()
+    if ctx.json_output:
+        click.echo(json.dumps(tags, ensure_ascii=False))
+    else:
+        _print_tags(tags)
@@ -0,0 +1,106 @@
+---
+name: cli-anything-siyuan
+description: SiYuan (思源笔记) CLI — manage notebooks, documents, blocks, and search your knowledge base from the terminal.
+---
+
+# cli-anything-siyuan
+
+CLI harness for [SiYuan](https://github.com/siyuan-note/siyuan) (思源笔记),
+a local-first knowledge management and note-taking application.
+
+This CLI connects to a running SiYuan kernel via its HTTP API
+(`http://127.0.0.1:6806`) and provides structured access to notebooks,
+documents, blocks, search, and export.
+
+## Prerequisites
+
+- SiYuan must be running (Settings → About shows the API token)
+- Python 3.10+
+- Install: `pip install cli-anything-siyuan` or `pip install -e .` from agent-harness/
+
+## Command Groups
+
+### notebook — Notebook management
+| Subcommand | Description |
+|------------|-------------|
+| `list` | List all notebooks |
+| `create <name>` | Create a new notebook |
+| `rename <id> <name>` | Rename a notebook |
+| `remove <id>` | Delete a notebook |
+| `open <id>` | Open a notebook |
+
+### doc — Document management
+| Subcommand | Description |
+|------------|-------------|
+| `create <notebook> <path> [--md "content"]` | Create a document |
+| `list <notebook> [path]` | List documents |
+| `tree <notebook> [--path / --depth]` | Show doc tree |
+| `get <id>` | Get document path by ID |
+| `rename <id> <title>` | Rename a document |
+| `remove <id>` | Delete a document |
+
+### block — Block operations
+| Subcommand | Description |
+|------------|-------------|
+| `insert <data> [--previous / --parent]` | Insert a block |
+| `update <id> <data>` | Update a block |
+| `delete <id>` | Delete a block |
+| `get <id>` | Get block kramdown source |
+| `children <id>` | Get child blocks |
+
+### Other commands
+| Command | Description |
+|---------|-------------|
+| `sql <stmt>` | Execute SQL on the block database |
+| `search <query>` | Full-text search across blocks |
+| `export md <doc-id>` | Export document as Markdown |
+| `tag list` | List all tags |
+| `version` | Show SiYuan kernel version |
+| `status` | Show connection and session status |
+| `repl` | Start interactive REPL |
+
+## Agent Guidance
+
+- Always use `--json` for machine-readable output
+- Power search: use `sql "SELECT * FROM blocks WHERE content LIKE '%keyword%'"` for SQL-level access
+- Document IDs look like `20210817205410-2kvfpfn` (timestamp-based)
+- API token can be found in SiYuan Settings → About
+- Connection defaults: `http://127.0.0.1:6806`
+
+## Examples
+
+```bash
+# List notebooks (JSON)
+cli-anything-siyuan --json notebook list
+
+# Create a document with Markdown (--md flag, watch shell escaping)
+cli-anything-siyuan doc create nb1 /projects/new --md "## Title\n\nContent"
+
+# Create a document — pipe Markdown via stdin (avoids shell escaping)
+# PowerShell here-string (literal, no escaping needed):
+@'
+## Title
+Content with `backticks` and (parentheses) and "quotes"
+'@ | cli-anything-siyuan doc create nb1 /projects/new --md -
+
+# Bash heredoc:
+# cat <<'EOF' | cli-anything-siyuan doc create nb1 /projects/new --md -
+# ## Title
+# Content with `backticks` and (parentheses)
+# EOF
+
+# SQL search
+cli-anything-siyuan sql "SELECT id, content FROM blocks WHERE content LIKE '%meeting%' LIMIT 5"
+
+# Export
+cli-anything-siyuan export md doc123
+
+# Enter REPL
+cli-anything-siyuan
+```
+
+## Error Handling
+
+- Connection errors: check that SiYuan is running and the API token is correct
+- API errors: returned as `{"code": N, "msg": "..."}` — check the message field
+- Auth errors: verify the token in `~/.siyuan-cli.json` or `SIYUAN_TOKEN` env var
@@ -0,0 +1,408 @@
+"""Unit tests for cli-anything-siyuan CLI commands.
+
+Tests CLI output formatting using Click's CliRunner with mocked client.
+No external dependencies or running SiYuan instance required.
+"""
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+from click.testing import CliRunner
+
+from cli_anything.siyuan.siyuan_cli import cli
+
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+@pytest.fixture
+def mock_ctx():
+    """Create a mock SiYuanContext with a mock client."""
+    ctx = MagicMock()
+    ctx.json_output = False
+    return ctx
+
+
+# ── Search command ─────────────────────────────────────────────────────
+
+
+class TestSearchCommand:
+    def test_search_returns_blocks_list(self, runner, mock_ctx):
+        """search handles API returning list of blocks directly."""
+        mock_ctx.client.search_blocks.return_value = [
+            {"id": "b1", "content": "Dit模型训练结果如何"},
+            {"id": "b2", "content": "训练完成，准确率90%"},
+        ]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["search", "Dit模型"])
+            assert result.exit_code == 0
+            assert "b1" in result.output
+            assert "Dit模型训练结果如何" in result.output
+            assert "b2" in result.output
+
+    def test_search_returns_dict_with_blocks_key(self, runner, mock_ctx):
+        """search handles API returning dict with 'blocks' key (real SiYuan format).
+
+        This is the fix for KeyError: slice — real SiYuan API returns
+        {"blocks": [...], "rootBlocks": {...}} not a flat list.
+        """
+        mock_ctx.client.search_blocks.return_value = {
+            "blocks": [
+                {"id": "b1", "content": "Dit模型训练完成"},
+                {"id": "b2", "content": "loss=0.02"},
+            ],
+            "rootBlocks": {},
+        }
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["search", "Dit模型"])
+            assert result.exit_code == 0
+            assert "b1" in result.output
+            assert "Dit模型训练完成" in result.output
+
+    def test_search_json_output(self, runner, mock_ctx):
+        """--json search returns raw data."""
+        mock_ctx.json_output = True
+        mock_ctx.client.search_blocks.return_value = {
+            "blocks": [{"id": "b1", "content": "test"}],
+        }
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["--json", "search", "test"])
+            assert result.exit_code == 0
+            data = json.loads(result.output)
+            assert data[0]["id"] == "b1"
+
+    def test_search_no_results(self, runner, mock_ctx):
+        """search shows 'No results' when empty list."""
+        mock_ctx.client.search_blocks.return_value = []
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["search", "nonexistent"])
+            assert result.exit_code == 0
+            assert "No results" in result.output
+
+    def test_search_no_results_dict(self, runner, mock_ctx):
+        """search shows 'No results' when dict with empty blocks list."""
+        mock_ctx.client.search_blocks.return_value = {"blocks": [], "rootBlocks": {}}
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["search", "nonexistent"])
+            assert result.exit_code == 0
+            assert "No results" in result.output
+
+
+# ── Doc list command ───────────────────────────────────────────────────
+
+
+class TestDocListCommand:
+    def test_doc_list_returns_files(self, runner, mock_ctx):
+        """doc list handles API returning dict with 'files' key."""
+        mock_ctx.client.list_docs_by_path.return_value = {
+            "box": "nb1",
+            "files": [
+                {"id": "doc1", "name": "测试文档", "type": "d"},
+                {"id": "doc2", "name": "笔记", "type": "d"},
+            ],
+            "path": "/",
+        }
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["doc", "list", "nb1"])
+            assert result.exit_code == 0
+            assert "doc1" in result.output
+            assert "测试文档" in result.output
+            assert "doc2" in result.output
+
+    def test_doc_list_empty(self, runner, mock_ctx):
+        """doc list handles empty files list."""
+        mock_ctx.client.list_docs_by_path.return_value = {
+            "box": "nb1", "files": [], "path": "/",
+        }
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["doc", "list", "nb1"])
+            assert result.exit_code == 0
+            # Should show header but no data rows
+            assert "ID" in result.output
+            assert "Name" in result.output
+
+    def test_doc_list_json(self, runner, mock_ctx):
+        """--json doc list returns raw files array."""
+        mock_ctx.json_output = True
+        mock_ctx.client.list_docs_by_path.return_value = {
+            "box": "nb1", "files": [{"id": "doc1", "name": "Test"}], "path": "/",
+        }
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["--json", "doc", "list", "nb1"])
+            assert result.exit_code == 0
+            data = json.loads(result.output)
+            assert data[0]["id"] == "doc1"
+
+
+# ── Doc tree command ───────────────────────────────────────────────────
+
+
+class TestDocTreeCommand:
+    def test_doc_tree_returns_files(self, runner, mock_ctx):
+        """doc tree handles API returning dict with 'files' key."""
+        mock_ctx.client.list_doc_tree.return_value = {
+            "files": [
+                {"id": "doc1", "name": "根文档", "depth": 0},
+                {"id": "doc2", "name": "子文档", "depth": 1},
+            ],
+        }
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["doc", "tree", "nb1"])
+            assert result.exit_code == 0
+            assert "根文档" in result.output
+            assert "子文档" in result.output
+
+    def test_doc_tree_json(self, runner, mock_ctx):
+        """--json doc tree returns raw files array."""
+        mock_ctx.json_output = True
+        mock_ctx.client.list_doc_tree.return_value = {
+            "files": [{"id": "doc1", "name": "Root"}],
+        }
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["--json", "doc", "tree", "nb1"])
+            assert result.exit_code == 0
+            data = json.loads(result.output)
+            assert data[0]["id"] == "doc1"
+
+
+# ── Notebook list command ──────────────────────────────────────────────
+
+
+class TestNotebookListCommand:
+    def test_notebook_list(self, runner, mock_ctx):
+        """notebook list returns formatted table."""
+        mock_ctx.client.list_notebooks.return_value = [
+            {"id": "nb1", "name": "工作笔记", "closed": False},
+            {"id": "nb2", "name": "归档", "closed": True},
+        ]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["notebook", "list"])
+            assert result.exit_code == 0
+            assert "工作笔记" in result.output
+            assert "nb1" in result.output
+
+
+# ── Tag list command ───────────────────────────────────────────────────
+
+
+class TestTagListCommand:
+    def test_tag_list(self, runner, mock_ctx):
+        """tag list shows tag names with counts."""
+        mock_ctx.client.get_tags.return_value = [
+            {"name": "AI", "count": 5},
+            {"name": "Python", "count": 12},
+            {"name": "笔记", "count": 3},
+        ]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["tag", "list"])
+            assert result.exit_code == 0
+            assert "AI" in result.output
+            assert "5" in result.output
+            assert "Python" in result.output
+            assert "12" in result.output
+
+
+# ── SQL command ────────────────────────────────────────────────────────
+
+
+class TestSQLCommand:
+    def test_sql_query(self, runner, mock_ctx):
+        """sql returns query results."""
+        mock_ctx.client.query_sql.return_value = [
+            {"id": "b1", "content": "hello"},
+            {"id": "b2", "content": "world"},
+        ]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["sql", "SELECT * FROM blocks"])
+            assert result.exit_code == 0
+            assert "b1" in result.output
+            assert "hello" in result.output
+
+
+# ── Status command ─────────────────────────────────────────────────────
+
+
+class TestStatusCommand:
+    def test_status(self, runner, mock_ctx):
+        """status shows connection info."""
+        mock_ctx.client.status.return_value = {
+            "connected": True,
+            "version": "3.6.5",
+        }
+        mock_ctx.client.get_version.return_value = "3.6.5"
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            mock_session = MagicMock()
+            mock_session.state.connected = True
+            mock_session.state.current_notebook_id = "nb1"
+            mock_session.state.current_notebook_name = "工作"
+            mock_session.state.current_doc_id = "doc1"
+            mock_ctx.client.get_version.return_value = "3.6.5"
+            mock_ctx.session = mock_session
+            mock_ctx.current_notebook_id = "nb1"
+            mock_ctx.current_notebook_name = "工作"
+            mock_ctx.current_doc_id = "doc1"
+
+            result = runner.invoke(cli, ["status"])
+            assert result.exit_code == 0
+            assert "Connected" in result.output or "connected" in result.output.lower()
+
+
+# ── Block insert command ────────────────────────────────────────────────
+
+
+class TestBlockInsertCommand:
+    def test_block_insert_with_parent(self, runner, mock_ctx):
+        """block insert with --parent succeeds."""
+        mock_ctx.client.insert_block.return_value = [{"id": "new-block"}]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["block", "insert", "hello", "--parent", "pid"])
+            assert result.exit_code == 0
+            assert "Block inserted" in result.output
+            mock_ctx.client.insert_block.assert_called_once_with(
+                "markdown", "hello", parent_id="pid", previous_id="", next_id=""
+            )
+
+    def test_block_insert_with_previous(self, runner, mock_ctx):
+        """block insert with --previous succeeds."""
+        mock_ctx.client.insert_block.return_value = [{"id": "new-block"}]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["block", "insert", "hello", "--previous", "prev"])
+            assert result.exit_code == 0
+            assert "Block inserted" in result.output
+            mock_ctx.client.insert_block.assert_called_once_with(
+                "markdown", "hello", parent_id="", previous_id="prev", next_id=""
+            )
+
+    def test_block_insert_with_next(self, runner, mock_ctx):
+        """block insert with --next succeeds."""
+        mock_ctx.client.insert_block.return_value = [{"id": "new-block"}]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["block", "insert", "hello", "--next", "nid"])
+            assert result.exit_code == 0
+            assert "Block inserted" in result.output
+            mock_ctx.client.insert_block.assert_called_once_with(
+                "markdown", "hello", parent_id="", previous_id="", next_id="nid"
+            )
+
+    def test_block_insert_without_anchor_errors(self, runner, mock_ctx):
+        """block insert without any anchor raises UsageError."""
+        mock_ctx.client.insert_block.return_value = [{"id": "new-block"}]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["block", "insert", "hello"])
+            assert result.exit_code == 2
+            assert "anchor" in result.output.lower() or "Error" in result.output
+
+    def test_block_insert_json_output(self, runner, mock_ctx):
+        """--json block insert returns raw data."""
+        mock_ctx.json_output = True
+        mock_ctx.client.insert_block.return_value = [{"id": "new-block"}]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["--json", "block", "insert", "hello", "--parent", "pid"])
+            assert result.exit_code == 0
+            data = json.loads(result.output)
+            assert data[0]["id"] == "new-block"
+
+
+# ── Doc get command ────────────────────────────────────────────────────
+
+
+class TestDocGetCommand:
+    def test_doc_get(self, runner, mock_ctx):
+        """doc get shows document path."""
+        mock_ctx.client.get_hpath_by_id.return_value = "/我的文档/测试"
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["doc", "get", "doc123"])
+            assert result.exit_code == 0
+            assert "/我的文档/测试" in result.output
+
+
+# ── Doc create command ──────────────────────────────────────────────────
+
+
+class TestDocCreateCommand:
+    def test_doc_create_without_md(self, runner, mock_ctx):
+        """doc create without --md does not read stdin (passes empty string)."""
+        mock_ctx.client.create_doc_with_md.return_value = "doc123"
+        with (
+            patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx),
+        ):
+            result = runner.invoke(cli, ["doc", "create", "nb1", "/test"])
+            assert result.exit_code == 0
+            assert "doc123" in result.output
+
+    def test_doc_create_with_md(self, runner, mock_ctx):
+        """doc create with --md passes the markdown content."""
+        mock_ctx.client.create_doc_with_md.return_value = "doc123"
+        with (
+            patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx),
+        ):
+            result = runner.invoke(cli, ["doc", "create", "nb1", "/test", "--md", "# Hello"])
+            assert result.exit_code == 0
+            mock_ctx.client.create_doc_with_md.assert_called_with("nb1", "/test", "# Hello")
+            assert "doc123" in result.output
+
+    def test_doc_create_json_output(self, runner, mock_ctx):
+        """--json doc create returns doc ID."""
+        mock_ctx.json_output = True
+        mock_ctx.client.create_doc_with_md.return_value = "doc123"
+        with (
+            patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx),
+        ):
+            result = runner.invoke(cli, ["--json", "doc", "create", "nb1", "/test"])
+            assert result.exit_code == 0
+            data = json.loads(result.output)
+            assert data["id"] == "doc123"
+
+
+# ── Doc tree recursive rendering ──────────────────────────────────────
+
+
+class TestDocTreeRecursiveCommand:
+    def test_doc_tree_handle_nested_children(self, runner, mock_ctx):
+        """doc tree recursively renders nested children."""
+        mock_ctx.client.list_doc_tree.return_value = {
+            "files": [
+                {
+                    "id": "doc1", "name": "Root", "depth": 0,
+                    "children": [
+                        {"id": "doc2", "name": "Child", "depth": 1, "children": []},
+                    ],
+                },
+                {"id": "doc3", "name": "Sibling", "depth": 0, "children": []},
+            ],
+        }
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["doc", "tree", "nb1"])
+            assert result.exit_code == 0
+            assert "Root" in result.output
+            assert "Child" in result.output
+            assert "Sibling" in result.output
+            assert "doc1" in result.output
+            assert "doc2" in result.output
+
+    def test_doc_tree_flat_items(self, runner, mock_ctx):
+        """doc tree also works with flat items (no children key)."""
+        mock_ctx.client.list_doc_tree.return_value = [
+            {"id": "doc1", "name": "Flat1", "depth": 0},
+            {"id": "doc2", "name": "Flat2", "depth": 1},
+        ]
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["doc", "tree", "nb1"])
+            assert result.exit_code == 0
+            assert "Flat1" in result.output
+            assert "Flat2" in result.output
+
+    def test_doc_tree_nested_json_output(self, runner, mock_ctx):
+        """--json doc tree with nested children returns raw data."""
+        mock_ctx.json_output = True
+        mock_ctx.client.list_doc_tree.return_value = {
+            "files": [{"id": "doc1", "name": "Root", "children": []}],
+        }
+        with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx):
+            result = runner.invoke(cli, ["--json", "doc", "tree", "nb1"])
+            assert result.exit_code == 0
+            data = json.loads(result.output)
+            assert data[0]["id"] == "doc1"
@@ -0,0 +1,204 @@
+"""Unit tests for cli-anything-siyuan core modules.
+
+Tests use synthetic data and mocking — no external dependencies.
+"""
+
+import json
+import os
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from cli_anything.siyuan.core.client import SiYuanClient, SiYuanConfig, load_config
+from cli_anything.siyuan.core.session import SessionManager, SessionState
+
+
+# ── Client config tests ────────────────────────────────────────────────
+
+class TestClientConfig:
+    def test_default_config(self):
+        """Default config has correct default values."""
+        config = SiYuanConfig()
+        assert config.host == "127.0.0.1"
+        assert config.port == 6806
+        assert config.token == ""
+        assert config.base_url == "http://127.0.0.1:6806"
+
+    def test_custom_config(self):
+        """Custom config values are used."""
+        config = SiYuanConfig(host="localhost", port=6807, token="abc123")
+        assert config.host == "localhost"
+        assert config.port == 6807
+        assert config.token == "abc123"
+        assert config.base_url == "http://localhost:6807"
+
+    @patch.dict(os.environ, {"SIYUAN_HOST": "192.168.1.100", "SIYUAN_PORT": "6808", "SIYUAN_TOKEN": "envtoken"}, clear=True)
+    @patch("cli_anything.siyuan.core.client.Path.home")
+    def test_load_config_from_env(self, mock_home):
+        """load_config reads from environment variables."""
+        mock_home.return_value = Path(tempfile.mkdtemp())
+        config = load_config()
+        assert config.host == "192.168.1.100"
+        assert config.port == 6808
+        assert config.token == "envtoken"
+
+    def test_load_config_from_file(self):
+        """load_config reads from JSON config file."""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False, encoding="utf-8") as f:
+            json.dump({"host": "10.0.0.1", "port": 6809, "token": "filetoken"}, f)
+            fname = f.name
+        try:
+            config = load_config(fname)
+            assert config.host == "10.0.0.1"
+            assert config.port == 6809
+            assert config.token == "filetoken"
+        finally:
+            os.unlink(fname)
+
+
+# ── Client API tests ───────────────────────────────────────────────────
+
+class TestSiYuanClient:
+    @pytest.fixture
+    def client(self):
+        return SiYuanClient(SiYuanConfig(token="test-token"))
+
+    def test_ping_success(self, client):
+        """ping returns True when server responds."""
+        mock_session = MagicMock()
+        mock_session.post.return_value.status_code = 200
+        mock_session.post.return_value.json.return_value = {"code": 0, "data": "3.6.5"}
+        client._session = mock_session
+        assert client.ping() is True
+
+    def test_ping_failure(self, client):
+        """ping returns False when server is unreachable."""
+        mock_session = MagicMock()
+        mock_session.post.side_effect = __import__("requests").ConnectionError("refused")
+        client._session = mock_session
+        assert client.ping() is False
+
+    def test_list_notebooks(self, client):
+        """list_notebooks parses API response correctly."""
+        mock_data = {
+            "notebooks": [
+                {"id": "nb1", "name": "Test Notebook", "icon": "1f4d4", "closed": False},
+                {"id": "nb2", "name": "Daily Notes", "icon": "1f4cb", "closed": True},
+            ]
+        }
+        mock_session = MagicMock()
+        mock_session.post.return_value.status_code = 200
+        mock_session.post.return_value.json.return_value = {"code": 0, "data": mock_data}
+        client._session = mock_session
+
+        notebooks = client.list_notebooks()
+        assert len(notebooks) == 2
+        assert notebooks[0]["name"] == "Test Notebook"
+        assert notebooks[1]["id"] == "nb2"
+
+    def test_create_doc_with_md(self, client):
+        """create_doc_with_md returns the doc ID."""
+        mock_session = MagicMock()
+        mock_session.post.return_value.status_code = 200
+        mock_session.post.return_value.json.return_value = {"code": 0, "data": "doc123"}
+        client._session = mock_session
+
+        doc_id = client.create_doc_with_md("nb1", "/test/path", "# Hello")
+        assert doc_id == "doc123"
+        # Verify correct API call (requests uses json= kwarg, not data=)
+        call_kwargs = mock_session.post.call_args[1]
+        body = call_kwargs["json"]
+        assert body["notebook"] == "nb1"
+        assert body["path"] == "/test/path"
+        assert body["markdown"] == "# Hello"
+
+    def test_query_sql(self, client):
+        """query_sql returns list of rows."""
+        mock_session = MagicMock()
+        mock_session.post.return_value.status_code = 200
+        mock_session.post.return_value.json.return_value = {
+            "code": 0, "data": [{"id": "b1", "content": "hello"}, {"id": "b2", "content": "world"}],
+        }
+        client._session = mock_session
+
+        results = client.query_sql("SELECT * FROM blocks")
+        assert len(results) == 2
+        assert results[0]["content"] == "hello"
+
+
+# ── Session tests ──────────────────────────────────────────────────────
+
+class TestSessionManager:
+    @pytest.fixture
+    def mgr(self):
+        tmp_dir = tempfile.mkdtemp()
+        return SessionManager(state_dir=tmp_dir)
+
+    def test_default_state(self, mgr):
+        """Fresh session has correct defaults."""
+        assert mgr.state.connected is False
+        assert mgr.state.current_notebook_id == ""
+        assert mgr.state.current_notebook_name == ""
+        assert mgr.state.current_doc_id == ""
+
+    def test_save_load_roundtrip(self, mgr):
+        """Save then load preserves all state."""
+        mgr.update(current_notebook_id="nb1", current_notebook_name="Test", connected=True)
+        mgr.save()
+
+        mgr2 = SessionManager(state_dir=str(mgr.state_dir))
+        mgr2.load()
+        assert mgr2.state.current_notebook_id == "nb1"
+        assert mgr2.state.current_notebook_name == "Test"
+        assert mgr2.state.connected is True
+
+    def test_partial_update(self, mgr):
+        """Update only changes specified fields."""
+        mgr.update(current_notebook_id="nb1")
+        assert mgr.state.current_notebook_id == "nb1"
+        assert mgr.state.connected is False  # unchanged
+
+
+class TestFindReplace:
+    """Tests for find_replace using SiYuan API k/r/ids payload."""
+
+    @pytest.fixture
+    def client(self):
+        return SiYuanClient(SiYuanConfig(token="test-token"))
+
+    def test_find_replace_uses_k_r_ids_payload(self, client):
+        """find_replace sends k/r/ids payload keys."""
+        mock_session = MagicMock()
+        mock_session.post.return_value.status_code = 200
+        mock_session.post.return_value.json.return_value = {"code": 0, "data": {"count": 3}}
+        client._session = mock_session
+        client.find_replace("old", "new", ["id1", "id2"])
+        call_kwargs = mock_session.post.call_args[1]
+        body = call_kwargs["json"]
+        assert "k" in body
+        assert "r" in body
+        assert "ids" in body
+        assert body["k"] == "old"
+        assert body["r"] == "new"
+        assert body["ids"] == ["id1", "id2"]
+
+class TestUncappedListings:
+    """Tests for list_docs_by_path with maxListCount."""
+
+    @pytest.fixture
+    def client(self):
+        return SiYuanClient(SiYuanConfig(token="test-token"))
+
+    def test_list_docs_by_path_sends_max_list_count(self, client):
+        """list_docs_by_path sends maxListCount: 0 for uncapped listings."""
+        mock_session = MagicMock()
+        mock_session.post.return_value.status_code = 200
+        mock_session.post.return_value.json.return_value = {"code": 0, "data": {"files": [{"id": "doc1", "name": "test"}]}}
+        client._session = mock_session
+        client.list_docs_by_path("nb1", "/")
+        call_kwargs = mock_session.post.call_args[1]
+        body = call_kwargs["json"]
+        assert "maxListCount" in body
+        assert body["maxListCount"] == 0
@@ -0,0 +1,122 @@
+"""E2E tests for cli-anything-siyuan.
+
+Requires a running SiYuan instance at http://127.0.0.1:6806.
+Set SIYUAN_TOKEN env var if authentication is enabled.
+"""
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from cli_anything.siyuan.core.client import SiYuanClient, SiYuanClientError, load_config
+
+
+def _resolve_cli(name: str) -> list[str]:
+    """Resolve installed CLI command; falls back to python -m for dev."""
+    import shutil
+    force = os.environ.get("CLI_ANYTHING_FORCE_INSTALLED", "").strip() == "1"
+    path = shutil.which(name)
+    if path:
+        print(f"[_resolve_cli] Using installed command: {path}")
+        return [path]
+    if force:
+        raise RuntimeError(f"{name} not found in PATH. Install with: pip install -e .")
+    module = name.replace("cli-anything-", "cli_anything.")  # e.g. cli_anything.siyuan (has __main__.py)
+    print(f"[_resolve_cli] Falling back to: {sys.executable} -m {module}")
+    return [sys.executable, "-m", module]
+
+
+CLI_BASE = _resolve_cli("cli-anything-siyuan")
+
+
+# ── Fixtures ───────────────────────────────────────────────────────────
+
+@pytest.fixture(scope="session")
+def client() -> SiYuanClient:
+    """Create a client connected to the real SiYuan instance."""
+    config = load_config()
+    c = SiYuanClient(config)
+    if not c.ping():
+        pytest.skip("SiYuan is not running — E2E tests require a running instance")
+    return c
+
+
+@pytest.fixture
+def tmp_dir() -> Path:
+    """Create a temporary directory for test artifacts."""
+    import tempfile
+    path = Path(tempfile.mkdtemp(prefix="siyuan_e2e_"))
+    yield path
+    import shutil
+    shutil.rmtree(path, ignore_errors=True)
+
+
+class TestCLISubprocess:
+    CLI_BASE = CLI_BASE
+
+    def _run(self, args: list[str], check: bool = True) -> subprocess.CompletedProcess:
+        return subprocess.run(
+            self.CLI_BASE + args,
+            capture_output=True, text=True, check=check,
+        )
+
+    def test_help(self):
+        """--help returns 0 and shows usage."""
+        result = self._run(["--help"])
+        assert result.returncode == 0
+        assert "Usage:" in result.stdout
+        print(f"\n  --help: OK ({len(result.stdout)} chars)")
+
+    def test_version_json(self, client: SiYuanClient):
+        """--json version returns valid JSON with version string."""
+        result = self._run(["--json", "version"])
+        assert result.returncode == 0
+        data = json.loads(result.stdout)
+        assert "version" in data
+        assert data["version"]
+        print(f"\n  SiYuan version: {data['version']}")
+
+    def test_status_json(self, client: SiYuanClient):
+        """--json status returns valid JSON with connection info."""
+        result = self._run(["--json", "status"])
+        assert result.returncode == 0
+        data = json.loads(result.stdout)
+        assert "siyuan_version" in data
+        assert data["siyuan_version"]
+        print(f"\n  Status: connected to SiYuan v{data['siyuan_version']}")
+
+
+class TestBackendE2E:
+    """Tests that hit the real SiYuan API via Python client."""
+
+    def test_get_version(self, client: SiYuanClient):
+        """Get the SiYuan kernel version from a running instance."""
+        ver = client.get_version()
+        assert ver
+        assert isinstance(ver, str)
+        assert len(ver) > 0
+        print(f"\n  SiYuan kernel version: {ver}")
+
+    def test_list_notebooks(self, client: SiYuanClient):
+        """List all notebooks from the running instance."""
+        notebooks = client.list_notebooks()
+        assert isinstance(notebooks, list)
+        if notebooks:
+            nb = notebooks[0]
+            assert "id" in nb
+            assert "name" in nb
+            print(f"\n  Notebooks found: {len(notebooks)}")
+            print(f"  First: {nb['name']} ({nb['id']})")
+        else:
+            print("\n  No notebooks found (empty instance)")
+
+    def test_get_current_time(self, client: SiYuanClient):
+        """Get current time from server."""
+        t = client.get_current_time()
+        assert t > 0
+        assert isinstance(t, int)
+        print(f"\n  Server time (ts): {t}")
@@ -0,0 +1,567 @@
+"""cli-anything REPL Skin — Unified terminal interface for all CLI harnesses.
+
+Copy this file into your CLI package at:
+    cli_anything/<software>/utils/repl_skin.py
+
+Usage:
+    from cli_anything.<software>.utils.repl_skin import ReplSkin
+
+    skin = ReplSkin("shotcut", version="1.0.0")
+    skin.print_banner()  # auto-detects repo-root or packaged SKILL.md
+    prompt_text = skin.prompt(project_name="my_video.mlt", modified=True)
+    skin.success("Project saved")
+    skin.error("File not found")
+    skin.warning("Unsaved changes")
+    skin.info("Processing 24 clips...")
+    skin.status("Track 1", "3 clips, 00:02:30")
+    skin.table(headers, rows)
+    skin.print_goodbye()
+"""
+
+import os
+import sys
+from pathlib import Path
+
+# ── ANSI color codes (no external deps for core styling) ──────────────
+
+_RESET = "\033[0m"
+_BOLD = "\033[1m"
+_DIM = "\033[2m"
+_ITALIC = "\033[3m"
+_UNDERLINE = "\033[4m"
+
+# Brand colors
+_CYAN = "\033[38;5;80m"       # cli-anything brand cyan
+_CYAN_BG = "\033[48;5;80m"
+_WHITE = "\033[97m"
+_GRAY = "\033[38;5;245m"
+_DARK_GRAY = "\033[38;5;240m"
+_LIGHT_GRAY = "\033[38;5;250m"
+
+# Software accent colors — each software gets a unique accent
+_ACCENT_COLORS = {
+    "gimp":        "\033[38;5;214m",   # warm orange
+    "blender":     "\033[38;5;208m",   # deep orange
+    "inkscape":    "\033[38;5;39m",    # bright blue
+    "audacity":    "\033[38;5;33m",    # navy blue
+    "libreoffice": "\033[38;5;40m",    # green
+    "obs_studio":  "\033[38;5;55m",    # purple
+    "kdenlive":    "\033[38;5;69m",    # slate blue
+    "shotcut":     "\033[38;5;35m",    # teal green
+}
+_DEFAULT_ACCENT = "\033[38;5;75m"      # default sky blue
+
+# Status colors
+_GREEN = "\033[38;5;78m"
+_YELLOW = "\033[38;5;220m"
+_RED = "\033[38;5;196m"
+_BLUE = "\033[38;5;75m"
+_MAGENTA = "\033[38;5;176m"
+
+_SKILL_SOURCE_REPO = os.environ.get("CLI_ANYTHING_SKILL_REPO", "HKUDS/CLI-Anything")
+
+# ── Brand icon ────────────────────────────────────────────────────────
+
+# The cli-anything icon: a small colored diamond/chevron mark
+_ICON = f"{_CYAN}{_BOLD}◆{_RESET}"
+_ICON_SMALL = f"{_CYAN}▸{_RESET}"
+
+# ── Box drawing characters ────────────────────────────────────────────
+
+_H_LINE = "─"
+_V_LINE = "│"
+_TL = "╭"
+_TR = "╮"
+_BL = "╰"
+_BR = "╯"
+_T_DOWN = "┬"
+_T_UP = "┴"
+_T_RIGHT = "├"
+_T_LEFT = "┤"
+_CROSS = "┼"
+
+
+def _strip_ansi(text: str) -> str:
+    """Remove ANSI escape codes for length calculation."""
+    import re
+    return re.sub(r"\033\[[^m]*m", "", text)
+
+
+def _visible_len(text: str) -> int:
+    """Get visible length of text (excluding ANSI codes)."""
+    return len(_strip_ansi(text))
+
+
+def _display_home_path(path: str) -> str:
+    """Display a path relative to the home directory when possible."""
+    expanded = Path(path).expanduser().resolve()
+    home = Path.home().resolve()
+    try:
+        relative = expanded.relative_to(home)
+        return f"~/{relative.as_posix()}"
+    except ValueError:
+        return str(expanded)
+
+
+class ReplSkin:
+    """Unified REPL skin for cli-anything CLIs.
+
+    Provides consistent branding, prompts, and message formatting
+    across all CLI harnesses built with the cli-anything methodology.
+    """
+
+    def __init__(self, software: str, version: str = "1.0.0",
+                 history_file: str | None = None, skill_path: str | None = None):
+        """Initialize the REPL skin.
+
+        Args:
+            software: Software name (e.g., "gimp", "shotcut", "blender").
+            version: CLI version string.
+            history_file: Path for persistent command history.
+                         Defaults to ~/.cli-anything-<software>/history
+            skill_path: Path to the SKILL.md file for agent discovery.
+                        Auto-detected from the repo-root skills/ tree when present,
+                        otherwise from the package's skills/ directory.
+                        Displayed in banner for AI agents to know where to read skill info.
+        """
+        self.software = software.lower().replace("-", "_")
+        self.display_name = software.replace("_", " ").title()
+        self.version = version
+        software_aliases = {"iterm2_ctl": "iterm2"}
+        self.skill_slug = software_aliases.get(self.software, self.software).replace("_", "-")
+        self.skill_id = f"cli-anything-{self.skill_slug}"
+        self.skill_install_cmd = (
+            f"npx skills add {_SKILL_SOURCE_REPO} --skill {self.skill_id} -g -y"
+        )
+        global_skill_root = Path(
+            os.environ.get("CLI_ANYTHING_GLOBAL_SKILLS_DIR", str(Path.home() / ".agents" / "skills"))
+        ).expanduser()
+        self.global_skill_path = str(global_skill_root / self.skill_id / "SKILL.md")
+
+        # Prefer repo-root canonical skills/<skill-id>/SKILL.md when running
+        # inside the CLI-Anything monorepo. Fall back to the packaged
+        # cli_anything/<software>/skills/SKILL.md for installed harnesses.
+        if skill_path is None:
+            package_skill = Path(__file__).resolve().parent.parent / "skills" / "SKILL.md"
+            repo_skill = None
+            for parent in Path(__file__).resolve().parents:
+                candidate = parent / "skills" / self.skill_id / "SKILL.md"
+                if candidate.is_file():
+                    repo_skill = candidate
+                    break
+            if repo_skill and repo_skill.is_file():
+                skill_path = str(repo_skill)
+            elif package_skill.is_file():
+                skill_path = str(package_skill)
+        self.skill_path = skill_path
+        self.accent = _ACCENT_COLORS.get(self.software, _DEFAULT_ACCENT)
+
+        # History file
+        if history_file is None:
+            hist_dir = Path.home() / f".cli-anything-{self.software}"
+            hist_dir.mkdir(parents=True, exist_ok=True)
+            self.history_file = str(hist_dir / "history")
+        else:
+            self.history_file = history_file
+
+        # Detect terminal capabilities
+        self._color = self._detect_color_support()
+
+    def _detect_color_support(self) -> bool:
+        """Check if terminal supports color."""
+        if os.environ.get("NO_COLOR"):
+            return False
+        if os.environ.get("CLI_ANYTHING_NO_COLOR"):
+            return False
+        if not hasattr(sys.stdout, "isatty"):
+            return False
+        return sys.stdout.isatty()
+
+    def _c(self, code: str, text: str) -> str:
+        """Apply color code if colors are supported."""
+        if not self._color:
+            return text
+        return f"{code}{text}{_RESET}"
+
+    # ── Banner ────────────────────────────────────────────────────────
+
+    def print_banner(self):
+        """Print the startup banner with branding."""
+        import textwrap
+
+        inner = 72
+
+        def _box_line(content: str) -> str:
+            """Wrap content in box drawing, padding to inner width."""
+            pad = inner - _visible_len(content)
+            vl = self._c(_DARK_GRAY, _V_LINE)
+            return f"{vl}{content}{' ' * max(0, pad)}{vl}"
+
+        def _meta_lines(label: str, value: str) -> list[str]:
+            """Wrap a metadata line for the banner box."""
+            icon = self._c(_MAGENTA, "◇")
+            label_text = self._c(_DARK_GRAY, label)
+            prefix = f" {icon} {label_text} "
+            available = max(12, inner - _visible_len(prefix))
+            wrapped = textwrap.wrap(
+                value,
+                width=available,
+                break_long_words=True,
+                break_on_hyphens=False,
+            ) or [""]
+            lines = [f"{prefix}{self._c(_LIGHT_GRAY, wrapped[0])}"]
+            continuation_prefix = " " * _visible_len(prefix)
+            for chunk in wrapped[1:]:
+                lines.append(f"{continuation_prefix}{self._c(_LIGHT_GRAY, chunk)}")
+            return lines
+
+        top = self._c(_DARK_GRAY, f"{_TL}{_H_LINE * inner}{_TR}")
+        bot = self._c(_DARK_GRAY, f"{_BL}{_H_LINE * inner}{_BR}")
+
+        # Title:  ◆  cli-anything · Shotcut
+        icon = self._c(_CYAN + _BOLD, "◆")
+        brand = self._c(_CYAN + _BOLD, "cli-anything")
+        dot = self._c(_DARK_GRAY, "·")
+        name = self._c(self.accent + _BOLD, self.display_name)
+        title = f" {icon}  {brand} {dot} {name}"
+
+        ver = f" {self._c(_DARK_GRAY, f'   v{self.version}')}"
+        tip = f" {self._c(_DARK_GRAY, '   Type help for commands, quit to exit')}"
+        empty = ""
+
+        meta_lines: list[str] = []
+        meta_lines.extend(_meta_lines("Install:", self.skill_install_cmd))
+        meta_lines.extend(_meta_lines("Global skill:", _display_home_path(self.global_skill_path)))
+        print(top)
+        print(_box_line(title))
+        print(_box_line(ver))
+        for line in meta_lines:
+            print(_box_line(line))
+        print(_box_line(empty))
+        print(_box_line(tip))
+        print(bot)
+        print()
+
+    # ── Prompt ────────────────────────────────────────────────────────
+
+    def prompt(self, project_name: str = "", modified: bool = False,
+               context: str = "") -> str:
+        """Build a styled prompt string for prompt_toolkit or input().
+
+        Args:
+            project_name: Current project name (empty if none open).
+            modified: Whether the project has unsaved changes.
+            context: Optional extra context to show in prompt.
+
+        Returns:
+            Formatted prompt string.
+        """
+        parts = []
+
+        # Icon
+        if self._color:
+            parts.append(f"{_CYAN}◆{_RESET} ")
+        else:
+            parts.append("> ")
+
+        # Software name
+        parts.append(self._c(self.accent + _BOLD, self.software))
+
+        # Project context
+        if project_name or context:
+            ctx = context or project_name
+            mod = "*" if modified else ""
+            parts.append(f" {self._c(_DARK_GRAY, '[')}")
+            parts.append(self._c(_LIGHT_GRAY, f"{ctx}{mod}"))
+            parts.append(self._c(_DARK_GRAY, ']'))
+
+        parts.append(self._c(_GRAY, " ❯ "))
+
+        return "".join(parts)
+
+    def prompt_tokens(self, project_name: str = "", modified: bool = False,
+                      context: str = ""):
+        """Build prompt_toolkit formatted text tokens for the prompt.
+
+        Use with prompt_toolkit's FormattedText for proper ANSI handling.
+
+        Returns:
+            list of (style, text) tuples for prompt_toolkit.
+        """
+        accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff")
+        tokens = []
+
+        tokens.append(("class:icon", "◆ "))
+        tokens.append(("class:software", self.software))
+
+        if project_name or context:
+            ctx = context or project_name
+            mod = "*" if modified else ""
+            tokens.append(("class:bracket", " ["))
+            tokens.append(("class:context", f"{ctx}{mod}"))
+            tokens.append(("class:bracket", "]"))
+
+        tokens.append(("class:arrow", " ❯ "))
+
+        return tokens
+
+    def get_prompt_style(self):
+        """Get a prompt_toolkit Style object matching the skin.
+
+        Returns:
+            prompt_toolkit.styles.Style
+        """
+        try:
+            from prompt_toolkit.styles import Style
+        except ImportError:
+            return None
+
+        accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff")
+
+        return Style.from_dict({
+            "icon": "#5fdfdf bold",     # cyan brand color
+            "software": f"{accent_hex} bold",
+            "bracket": "#585858",
+            "context": "#bcbcbc",
+            "arrow": "#808080",
+            # Completion menu
+            "completion-menu.completion": "bg:#303030 #bcbcbc",
+            "completion-menu.completion.current": f"bg:{accent_hex} #000000",
+            "completion-menu.meta.completion": "bg:#303030 #808080",
+            "completion-menu.meta.completion.current": f"bg:{accent_hex} #000000",
+            # Auto-suggest
+            "auto-suggest": "#585858",
+            # Bottom toolbar
+            "bottom-toolbar": "bg:#1c1c1c #808080",
+            "bottom-toolbar.text": "#808080",
+        })
+
+    # ── Messages ──────────────────────────────────────────────────────
+
+    def success(self, message: str):
+        """Print a success message with green checkmark."""
+        icon = self._c(_GREEN + _BOLD, "✓")
+        print(f"  {icon} {self._c(_GREEN, message)}")
+
+    def error(self, message: str):
+        """Print an error message with red cross."""
+        icon = self._c(_RED + _BOLD, "✗")
+        print(f"  {icon} {self._c(_RED, message)}", file=sys.stderr)
+
+    def warning(self, message: str):
+        """Print a warning message with yellow triangle."""
+        icon = self._c(_YELLOW + _BOLD, "⚠")
+        print(f"  {icon} {self._c(_YELLOW, message)}")
+
+    def info(self, message: str):
+        """Print an info message with blue dot."""
+        icon = self._c(_BLUE, "●")
+        print(f"  {icon} {self._c(_LIGHT_GRAY, message)}")
+
+    def hint(self, message: str):
+        """Print a subtle hint message."""
+        print(f"  {self._c(_DARK_GRAY, message)}")
+
+    def section(self, title: str):
+        """Print a section header."""
+        print()
+        print(f"  {self._c(self.accent + _BOLD, title)}")
+        print(f"  {self._c(_DARK_GRAY, _H_LINE * len(title))}")
+
+    # ── Status display ────────────────────────────────────────────────
+
+    def status(self, label: str, value: str):
+        """Print a key-value status line."""
+        lbl = self._c(_GRAY, f"  {label}:")
+        val = self._c(_WHITE, f" {value}")
+        print(f"{lbl}{val}")
+
+    def status_block(self, items: dict[str, str], title: str = ""):
+        """Print a block of status key-value pairs.
+
+        Args:
+            items: Dict of label -> value pairs.
+            title: Optional title for the block.
+        """
+        if title:
+            self.section(title)
+
+        max_key = max(len(k) for k in items) if items else 0
+        for label, value in items.items():
+            lbl = self._c(_GRAY, f"  {label:<{max_key}}")
+            val = self._c(_WHITE, f"  {value}")
+            print(f"{lbl}{val}")
+
+    def progress(self, current: int, total: int, label: str = ""):
+        """Print a simple progress indicator.
+
+        Args:
+            current: Current step number.
+            total: Total number of steps.
+            label: Optional label for the progress.
+        """
+        pct = int(current / total * 100) if total > 0 else 0
+        bar_width = 20
+        filled = int(bar_width * current / total) if total > 0 else 0
+        bar = "█" * filled + "░" * (bar_width - filled)
+        text = f"  {self._c(_CYAN, bar)} {self._c(_GRAY, f'{pct:3d}%')}"
+        if label:
+            text += f" {self._c(_LIGHT_GRAY, label)}"
+        print(text)
+
+    # ── Table display ─────────────────────────────────────────────────
+
+    def table(self, headers: list[str], rows: list[list[str]],
+              max_col_width: int = 40):
+        """Print a formatted table with box-drawing characters.
+
+        Args:
+            headers: Column header strings.
+            rows: List of rows, each a list of cell strings.
+            max_col_width: Maximum column width before truncation.
+        """
+        if not headers:
+            return
+
+        # Calculate column widths
+        col_widths = [min(len(h), max_col_width) for h in headers]
+        for row in rows:
+            for i, cell in enumerate(row):
+                if i < len(col_widths):
+                    col_widths[i] = min(
+                        max(col_widths[i], len(str(cell))), max_col_width
+                    )
+
+        def pad(text: str, width: int) -> str:
+            t = str(text)[:width]
+            return t + " " * (width - len(t))
+
+        # Header
+        header_cells = [
+            self._c(_CYAN + _BOLD, pad(h, col_widths[i]))
+            for i, h in enumerate(headers)
+        ]
+        sep = self._c(_DARK_GRAY, f" {_V_LINE} ")
+        header_line = f"  {sep.join(header_cells)}"
+        print(header_line)
+
+        # Separator
+        sep_parts = [self._c(_DARK_GRAY, _H_LINE * w) for w in col_widths]
+        sep_line = self._c(_DARK_GRAY, f"  {'───'.join([_H_LINE * w for w in col_widths])}")
+        print(sep_line)
+
+        # Rows
+        for row in rows:
+            cells = []
+            for i, cell in enumerate(row):
+                if i < len(col_widths):
+                    cells.append(self._c(_LIGHT_GRAY, pad(str(cell), col_widths[i])))
+            row_sep = self._c(_DARK_GRAY, f" {_V_LINE} ")
+            print(f"  {row_sep.join(cells)}")
+
+    # ── Help display ──────────────────────────────────────────────────
+
+    def help(self, commands: dict[str, str]):
+        """Print a formatted help listing.
+
+        Args:
+            commands: Dict of command -> description pairs.
+        """
+        self.section("Commands")
+        max_cmd = max(len(c) for c in commands) if commands else 0
+        for cmd, desc in commands.items():
+            cmd_styled = self._c(self.accent, f"  {cmd:<{max_cmd}}")
+            desc_styled = self._c(_GRAY, f"  {desc}")
+            print(f"{cmd_styled}{desc_styled}")
+        print()
+
+    # ── Goodbye ───────────────────────────────────────────────────────
+
+    def print_goodbye(self):
+        """Print a styled goodbye message."""
+        print(f"\n  {_ICON_SMALL} {self._c(_GRAY, 'Goodbye!')}\n")
+
+    # ── Prompt toolkit session factory ────────────────────────────────
+
+    def create_prompt_session(self):
+        """Create a prompt_toolkit PromptSession with skin styling.
+
+        Returns:
+            A configured PromptSession, or None if prompt_toolkit unavailable.
+        """
+        try:
+            from prompt_toolkit import PromptSession
+            from prompt_toolkit.history import FileHistory
+            from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
+            from prompt_toolkit.formatted_text import FormattedText
+
+            style = self.get_prompt_style()
+
+            session = PromptSession(
+                history=FileHistory(self.history_file),
+                auto_suggest=AutoSuggestFromHistory(),
+                style=style,
+                enable_history_search=True,
+            )
+            return session
+        except ImportError:
+            return None
+
+    def get_input(self, pt_session, project_name: str = "",
+                  modified: bool = False, context: str = "") -> str:
+        """Get input from user using prompt_toolkit or fallback.
+
+        Args:
+            pt_session: A prompt_toolkit PromptSession (or None).
+            project_name: Current project name.
+            modified: Whether project has unsaved changes.
+            context: Optional context string.
+
+        Returns:
+            User input string (stripped).
+        """
+        if pt_session is not None:
+            from prompt_toolkit.formatted_text import FormattedText
+            tokens = self.prompt_tokens(project_name, modified, context)
+            return pt_session.prompt(FormattedText(tokens)).strip()
+        else:
+            raw_prompt = self.prompt(project_name, modified, context)
+            return input(raw_prompt).strip()
+
+    # ── Toolbar builder ───────────────────────────────────────────────
+
+    def bottom_toolbar(self, items: dict[str, str]):
+        """Create a bottom toolbar callback for prompt_toolkit.
+
+        Args:
+            items: Dict of label -> value pairs to show in toolbar.
+
+        Returns:
+            A callable that returns FormattedText for the toolbar.
+        """
+        def toolbar():
+            from prompt_toolkit.formatted_text import FormattedText
+            parts = []
+            for i, (k, v) in enumerate(items.items()):
+                if i > 0:
+                    parts.append(("class:bottom-toolbar.text", "  │  "))
+                parts.append(("class:bottom-toolbar.text", f" {k}: "))
+                parts.append(("class:bottom-toolbar", v))
+            return FormattedText(parts)
+        return toolbar
+
+
+# ── ANSI 256-color to hex mapping (for prompt_toolkit styles) ─────────
+
+_ANSI_256_TO_HEX = {
+    "\033[38;5;33m":  "#0087ff",  # audacity navy blue
+    "\033[38;5;35m":  "#00af5f",  # shotcut teal
+    "\033[38;5;39m":  "#00afff",  # inkscape bright blue
+    "\033[38;5;40m":  "#00d700",  # libreoffice green
+    "\033[38;5;55m":  "#5f00af",  # obs purple
+    "\033[38;5;69m":  "#5f87ff",  # kdenlive slate blue
+    "\033[38;5;75m":  "#5fafff",  # default sky blue
+    "\033[38;5;80m":  "#5fd7d7",  # brand cyan
+    "\033[38;5;208m": "#ff8700",  # blender deep orange
+    "\033[38;5;214m": "#ffaf00",  # gimp warm orange
+}
@@ -0,0 +1,82 @@
+"""SiYuan backend integration.
+
+Provides helper functions to find a running SiYuan instance and verify
+connectivity. Since SiYuan has no headless CLI mode, this module handles
+the connection check and configuration discovery.
+"""
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+from cli_anything.siyuan.core.client import SiYuanClient, SiYuanClientError, SiYuanConfig
+
+
+def find_siyuan_data_dir() -> str | None:
+    """Try to locate the SiYuan data directory.
+
+    Checks platform-specific default locations:
+    - Windows: %USERPROFILE%\\SiYuan\\data\\
+    - Linux: ~/.config/siyuan/data/
+    - macOS: ~/.config/siyuan/data/
+    """
+    home = Path.home()
+    candidates = []
+
+    if sys.platform == "win32":
+        candidates.append(home / "SiYuan" / "data")
+    else:
+        candidates.append(home / ".config" / "siyuan" / "data")
+
+    for candidate in candidates:
+        if candidate.is_dir():
+            return str(candidate.resolve())
+
+    return None
+
+
+def check_siyuan_running(config: SiYuanConfig | None = None) -> dict[str, Any]:
+    """Check if a SiYuan instance is reachable and return its status.
+
+    Returns:
+        dict with keys: connected (bool), version (str), error (str)
+    """
+    if config is None:
+        from cli_anything.siyuan.core.client import load_config
+        config = load_config()
+
+    client = SiYuanClient(config)
+    try:
+        version = client.get_version()
+        return {"connected": True, "version": version, "error": ""}
+    except SiYuanClientError as e:
+        return {"connected": False, "version": "", "error": str(e)}
+
+
+def get_api_token_from_conf() -> str:
+    """Try to read the API token from SiYuan config files.
+
+    Returns empty string if not found.
+    """
+    home = Path.home()
+    candidates = []
+
+    if sys.platform == "win32":
+        candidates.append(home / "SiYuan" / "conf" / "conf.json")
+    else:
+        candidates.append(home / ".config" / "siyuan" / "conf" / "conf.json")
+
+    import json
+    for conf_path in candidates:
+        if conf_path.is_file():
+            try:
+                data = json.loads(conf_path.read_text(encoding="utf-8"))
+                token = data.get("api", {}).get("token", "")
+                if token:
+                    return token
+            except (json.JSONDecodeError, KeyError, OSError):
+                continue
+
+    return ""
@@ -0,0 +1,41 @@
+from setuptools import setup, find_namespace_packages
+
+with open("cli_anything/siyuan/README.md", encoding="utf-8") as f:
+    long_description = f.read()
+
+setup(
+    name="cli-anything-siyuan",
+    version="1.0.0",
+    description="SiYuan (思源笔记) CLI — manage notebooks, documents, blocks from the terminal",
+    long_description=long_description,
+    long_description_content_type="text/markdown",
+    author="cli-anything",
+    packages=find_namespace_packages(include=["cli_anything.*"]),
+    package_data={
+        "cli_anything.siyuan": ["skills/*.md"],
+    },
+    include_package_data=True,
+    python_requires=">=3.10",
+    install_requires=[
+        "click>=8.0",
+        "requests>=2.28",
+    ],
+    extras_require={
+        "repl": ["prompt_toolkit>=3.0"],
+        "test": ["pytest>=7.0"],
+    },
+    entry_points={
+        "console_scripts": [
+            "cli-anything-siyuan=cli_anything.siyuan.siyuan_cli:cli",
+        ],
+    },
+    classifiers=[
+        "Development Status :: 4 - Beta",
+        "Intended Audience :: Developers",
+        "License :: OSI Approved :: MIT License",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
+        "Programming Language :: Python :: 3.12",
+    ],
+)
@@ -0,0 +1,106 @@
+---
+name: "cli-anything-siyuan"
+description: SiYuan (思源笔记) CLI — manage notebooks, documents, blocks, and search your knowledge base from the terminal.
+---
+
+# cli-anything-siyuan
+
+CLI harness for [SiYuan](https://github.com/siyuan-note/siyuan) (思源笔记),
+a local-first knowledge management and note-taking application.
+
+This CLI connects to a running SiYuan kernel via its HTTP API
+(`http://127.0.0.1:6806`) and provides structured access to notebooks,
+documents, blocks, search, and export.
+
+## Prerequisites
+
+- SiYuan must be running (Settings → About shows the API token)
+- Python 3.10+
+- Install: `pip install cli-anything-siyuan` or `pip install -e .` from agent-harness/
+
+## Command Groups
+
+### notebook — Notebook management
+| Subcommand | Description |
+|------------|-------------|
+| `list` | List all notebooks |
+| `create <name>` | Create a new notebook |
+| `rename <id> <name>` | Rename a notebook |
+| `remove <id>` | Delete a notebook |
+| `open <id>` | Open a notebook |
+
+### doc — Document management
+| Subcommand | Description |
+|------------|-------------|
+| `create <notebook> <path> [--md "content"]` | Create a document |
+| `list <notebook> [path]` | List documents |
+| `tree <notebook> [--path / --depth]` | Show doc tree |
+| `get <id>` | Get document path by ID |
+| `rename <id> <title>` | Rename a document |
+| `remove <id>` | Delete a document |
+
+### block — Block operations
+| Subcommand | Description |
+|------------|-------------|
+| `insert <data> [--previous / --parent]` | Insert a block |
+| `update <id> <data>` | Update a block |
+| `delete <id>` | Delete a block |
+| `get <id>` | Get block kramdown source |
+| `children <id>` | Get child blocks |
+
+### Other commands
+| Command | Description |
+|---------|-------------|
+| `sql <stmt>` | Execute SQL on the block database |
+| `search <query>` | Full-text search across blocks |
+| `export md <doc-id>` | Export document as Markdown |
+| `tag list` | List all tags |
+| `version` | Show SiYuan kernel version |
+| `status` | Show connection and session status |
+| `repl` | Start interactive REPL |
+
+## Agent Guidance
+
+- Always use `--json` for machine-readable output
+- Power search: use `sql "SELECT * FROM blocks WHERE content LIKE '%keyword%'"` for SQL-level access
+- Document IDs look like `20210817205410-2kvfpfn` (timestamp-based)
+- API token can be found in SiYuan Settings → About
+- Connection defaults: `http://127.0.0.1:6806`
+
+## Examples
+
+```bash
+# List notebooks (JSON)
+cli-anything-siyuan --json notebook list
+
+# Create a document with Markdown (--md flag, watch shell escaping)
+cli-anything-siyuan doc create nb1 /projects/new --md "## Title\n\nContent"
+
+# Create a document — pipe Markdown via stdin (avoids shell escaping)
+# PowerShell here-string (literal, no escaping needed):
+@'
+## Title
+Content with `backticks` and (parentheses) and "quotes"
+'@ | cli-anything-siyuan doc create nb1 /projects/new --md -
+
+# Bash heredoc:
+# cat <<'EOF' | cli-anything-siyuan doc create nb1 /projects/new --md -
+# ## Title
+# Content with `backticks` and (parentheses)
+# EOF
+
+# SQL search
+cli-anything-siyuan sql "SELECT id, content FROM blocks WHERE content LIKE '%meeting%' LIMIT 5"
+
+# Export
+cli-anything-siyuan export md doc123
+
+# Enter REPL
+cli-anything-siyuan
+```
+
+## Error Handling
+
+- Connection errors: check that SiYuan is running and the API token is correct
+- API errors: returned as `{"code": N, "msg": "..."}` — check the message field
+- Auth errors: verify the token in `~/.siyuan-cli.json` or `SIYUAN_TOKEN` env var
				`@@ -0,0 +1 @@`
				`# cli-anything-siyuan: SiYuan (思源笔记) CLI harness`