Merge pull request #122 from open-webui/main

0.5.0

CHANGELOG.md

@@ -4,6 +4,39 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
+## [0.5.0] - 2026-09-16
+
+### Added
+
+- **Multiple sources in one Knowledge Base**: combine sources such as a Confluence space and a local handbook without one sync deleting the other's files. Sources listed together in the same configuration are synced as a group, including scheduled runs and updates triggered by source changes. You can choose a destination folder for each source to keep matching filenames separate.
+- **Confluence Server and Data Center**: sync spaces from self-hosted Confluence, including installations under an address such as `/confluence`. Sign in with a personal access token, and use Cloud and Data Center sources together with separate connection settings.
+- **Confluence page folders**: optionally preserve parent and child pages as folders, so you can include or exclude a whole section, such as Engineering runbooks.
+- **SharePoint sites**: sync document libraries from team sites and nested subsites, in addition to the main site.
+- **SharePoint government clouds**: connect to Microsoft GCC High and Department of Defense environments.
+- **Multiple Zotero libraries**: sync more than one personal or group library from one configuration, with separate access keys, collection choices, notes, annotations, and PDF download settings for each library.
+- **Separate source sign-ins**: supply credentials for GitHub, GitLab, Bitbucket, SharePoint, and Confluence individually in the configuration or when running a sync or preview. This lets one setup access sources that require different accounts.
+
+### Fixed
+
+- **Google Drive subfolders**: files inside nested folders are now included instead of being silently skipped.
+- **Large Outline collections**: syncing continues past the first 100 documents instead of leaving the rest out.
+- **Zotero collection selection**: selecting a named collection or subcollection now finds it correctly when its top-level collection has no parent.
+- **SharePoint filenames**: files and folders containing characters such as `#` or `%` can now be found and downloaded.
+- **SharePoint certificate sign-in**: valid certificates no longer fail with a "key was not found" error caused by how oikb identifies the certificate.
+- **GitHub and GitLab access**: syncs now use the credentials supplied for each source instead of ignoring them. Configuration checks also use the supplied source credentials.
+- **Confluence space selection**: spaces can be selected by their short names, such as `ENG`, without looking up a numeric space ID. Cloud addresses that already end in `/wiki` are accepted.
+- **Confluence page content**: text inside macros and code blocks is retained, including code indentation and line breaks. Blank pages and pages containing only an automatic index are skipped with a warning instead of repeatedly failing to upload.
+- **Confluence matching titles**: pages whose titles produce the same filename now receive distinct names, so one page does not replace another.
+- **Confluence large spaces**: page lists continue correctly when Confluence limits the number returned at once. If the list cannot be completed, the sync stops before changing the Knowledge Base.
+- **Knowledge Base file listings**: file listing and status commands now read the dedicated file list, avoiding failures when the Knowledge Base summary contains no file details. Connection checks read the reported total file count.
+- **Empty files**: empty content is skipped with a warning instead of being sent for upload.
+- **Upload errors**: failures now include the reason returned by Open WebUI, such as a text extraction failure.
+
+### Changed
+
+- **Shared Knowledge Base setup**: list all sources for the same Knowledge Base in one configuration. Selecting any one of those sources syncs the whole group. Running a standalone sync for just one source does not protect files from the others. Preview changes before combining existing sources, since destination folders can move files.
+- **Contributor credits**: added explicit credit for thiswillbeyourgithub's Zotero work and hexsm's BookStack export work already included in 0.4.0.
+
 ## [0.4.0] - 2026-07-17
 
 ### Added

README.md

@@ -99,7 +99,7 @@ docker run -d \
   -e OPEN_WEBUI_API_KEY=sk-... \
   -e OIKB_API_KEY=your-daemon-key \
   -e LOG_FORMAT=json \
-  -v ./.oikb.yaml:/app/.oikb.yaml:ro \
+  -v ./.oikb.yaml:/data/.oikb.yaml:ro \
   -p 8080:8080 \
   ghcr.io/open-webui/oikb:latest daemon
 ```
@@ -121,7 +121,7 @@ services:
       - OIKB_API_KEY=${OIKB_API_KEY}
       - LOG_FORMAT=json
     volumes:
-      - ./.oikb.yaml:/app/.oikb.yaml:ro
+      - ./.oikb.yaml:/data/.oikb.yaml:ro
     command: daemon
     ports:
       - "8080:8080"
@@ -168,8 +168,9 @@ Some connectors need an optional extra: `pip install oikb[gdrive]`, `pip install
 export ZOTERO_LIBRARY_ID=123456
 export ZOTERO_API_KEY=...
 
-oikb sync "zotero:" --kb-id your-kb-id                 # all top-level collections plus _unfiled
-oikb sync "zotero:Research%%Machine Learning" --kb-id your-kb-id
+oikb sync "zotero:" --kb-id your-kb-id # syncs all top-level collections plus _unfiled
+oikb sync "zotero:Research" --kb-id your-kb-id # syncs only the 'Research' collection
+oikb sync "zotero:Research%%Machine Learning" --kb-id your-kb-id # syncs only the 'Machine Learning' subcollection
 ```
 
 Options:

docker-compose.yaml

@@ -15,7 +15,7 @@ services:
       - OIKB_API_KEY=${OIKB_API_KEY}
       - LOG_FORMAT=json
     volumes:
-      - ./.oikb.yaml:/app/.oikb.yaml:ro
+      - ./.oikb.yaml:/data/.oikb.yaml:ro
     command: daemon
     ports:
       - "8080:8080"

docs/guide.md

@@ -156,6 +156,39 @@ Then sync all sources at once:
 oikb sync
 ```
 
+### Multiple sources in one Knowledge Base
+
+Entries with the same `kb-id` are scanned together and sent as one manifest.
+This prevents a sync of one source from deleting files belonging to another.
+`oikb sync --name NAME`, daemon schedules, and webhooks all sync the entire
+configured group. Each entry keeps its schedule; whenever one is due, the whole
+group runs under the existing KB lock. Use the same `url` and `token` settings
+for every entry in a group.
+
+```yaml
+sources:
+  - name: engineering
+    source: confluence:ENG
+    kb-id: shared-kb
+    target-path: ENG
+  - name: handbook
+    source: ./handbook
+    kb-id: shared-kb
+    target-path: handbook
+```
+
+`target-path` is an optional destination directory. Filters still match paths
+inside each source, before this prefix is applied. In a group with multiple
+sources, Confluence entries default to their space key as the destination
+directory; other connectors keep their original paths. Explicit `target-path`
+values keep paths stable when sources are added or removed. Duplicate destination
+paths or a failed source scan abort the group before any KB changes.
+
+Existing single-source paths remain unchanged. Combining previously separate
+sources can relocate files; preview with `oikb sync --dry-run`. Use YAML mode
+for a shared KB: a standalone `oikb sync SOURCE --kb-id ID` only knows about that
+one source.
+
 ### Global Defaults
 
 Avoid repeating the same config across entries:
@@ -247,9 +280,39 @@ Requires `GITLAB_TOKEN` or `BITBUCKET_TOKEN` respectively.
 
 ```bash
 oikb sync confluence:SPACE_KEY --kb-id your-kb-id
+
+# Preserve the Confluence page hierarchy in manifest paths.
+oikb sync 'confluence:SPACE_KEY?structure=hierarchical' --kb-id your-kb-id
 ```
 
-Requires `CONFLUENCE_URL`, `CONFLUENCE_USERNAME`, and `CONFLUENCE_API_TOKEN`.
+With hierarchical structure enabled, existing `filter.include` and
+`filter.exclude` patterns can select page trees, for example
+`Engineering/Runbooks*`.
+
+Cloud uses REST API v2 by default. Set `CONFLUENCE_URL` to your instance URL,
+`CONFLUENCE_USER` to your email, and `CONFLUENCE_TOKEN` to your API token.
+
+For Server/Data Center, select v1 and omit the user to use a personal access token:
+
+```bash
+export CONFLUENCE_API_VERSION=v1
+export CONFLUENCE_URL=https://wiki.example.com/confluence
+export CONFLUENCE_TOKEN=your-personal-access-token
+unset CONFLUENCE_USER
+oikb sync confluence:ENG --kb-id your-kb-id
+```
+
+Include your installation's context path (such as `/confluence`) in the URL.
+Both versions support flat and hierarchical paths. A user plus token selects
+Basic authentication; a token alone selects Bearer authentication, as described
+in [Atlassian's PAT documentation](https://confluence.atlassian.com/enterprise/using-personal-access-tokens-1026032365.html).
+Per-source `auth` keys `base_url`, `user`, `token`, and `api_version` override
+environment settings, so Cloud and Data Center can coexist in one config. Set
+`user: ""` to override an inherited user when using a PAT.
+
+Page titles that map to the same filename receive page-ID suffixes. Macro text
+and code blocks retain the existing extraction behavior; blank or index-only
+pages are skipped with a warning.
 
 ### BookStack
 
@@ -390,8 +453,10 @@ pip install oikb[zotero]
 export ZOTERO_LIBRARY_ID=123456
 export ZOTERO_API_KEY=...
 
-oikb sync "zotero:" --kb-id your-kb-id
-oikb sync "zotero:Research%%Machine Learning" --kb-id your-kb-id
+oikb sync "zotero:" --kb-id your-kb-id # syncs all top-level collections plus _unfiled
+oikb sync "zotero:Research" --kb-id your-kb-id # syncs only the 'Research' collection
+oikb sync "zotero:Research%%Machine Learning" --kb-id your-kb-id # syncs only the 'Machine Learning' subcollection
+
 ```
 
 Optional settings:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "oikb"
-version = "0.4.0"
+version = "0.5.0"
 description = "Sync anything to Open WebUI Knowledge Bases"
 readme = "README.md"
 authors = [

src/oikb/__init__.py

@@ -1,3 +1,3 @@
 """oikb — Open WebUI Knowledge Base CLI."""
 
-__version__ = "0.4.0"
+__version__ = "0.5.0"

src/oikb/cli.py

@@ -43,12 +43,29 @@ def _make_client(url: str | None, token: str | None):
     )
 
 
-def _resolve_connector(source: str, branch: str | None = None, path: str | None = None):
-    """Resolve a source string to the appropriate connector."""
+def _resolve_connector(
+    source: str,
+    branch: str | None = None,
+    path: str | None = None,
+    auth: dict[str, Any] | None = None,
+):
+    """Resolve a source string to the appropriate connector.
+
+    `auth` carries source-specific credentials — token, client_id,
+    client_secret, tenant_id, base_url, whatever a given connector's
+    constructor accepts — sourced from an .oikb.yaml `auth:` block or
+    the CLI's --auth flag. This function stays agnostic about what any
+    one connector needs: it just forwards the dict as **auth and lets
+    the connector's own __init__ validate it (and fall back to its env
+    vars when a key is missing, same as today). An unsupported key
+    surfaces as a normal TypeError naming the bad field.
+    """
+    auth = auth or {}
+
     if source.startswith("github:"):
         from oikb.connectors.github import GitHubConnector, parse_github_source
         parsed = parse_github_source(source)
-        return GitHubConnector(owner=parsed["owner"], repo=parsed["repo"], branch=branch, path=path or parsed.get("path"))
+        return GitHubConnector(owner=parsed["owner"], repo=parsed["repo"], branch=branch, path=path or parsed.get("path"), **auth)
 
     if source.startswith("gitlab:"):
         from oikb.connectors.gitlab import GitLabConnector, parse_gitlab_source
@@ -59,6 +76,7 @@ def _resolve_connector(source: str, branch: str | None = None, path: str | None
             branch=branch,
             path=path or parsed.get("path"),
             is_wiki=bool(parsed.get("wiki")),
+            **auth,
         )
 
     if source.startswith("s3://"):
@@ -89,7 +107,14 @@ def _resolve_connector(source: str, branch: str | None = None, path: str | None
     if source.startswith("confluence:"):
         from oikb.connectors.confluence import ConfluenceConnector, parse_confluence_source
         parsed = parse_confluence_source(source)
-        return ConfluenceConnector(space_key=parsed["space_key"], base_url=parsed.get("base_url"))
+        return ConfluenceConnector(
+            space_key=parsed["space_key"],
+            base_url=auth.get("base_url") or parsed.get("base_url"),
+            user=auth.get("user"),
+            token=auth.get("token"),
+            api_version=auth.get("api_version"),
+            structure=parsed.get("structure", "flat"),
+        )
 
     if source.startswith("notion:"):
         from oikb.connectors.notion import NotionConnector, parse_notion_source
@@ -116,7 +141,7 @@ def _resolve_connector(source: str, branch: str | None = None, path: str | None
     if source.startswith("sharepoint:"):
         from oikb.connectors.sharepoint import SharePointConnector, parse_sharepoint_source
         parsed = parse_sharepoint_source(source)
-        return SharePointConnector(site=parsed["site"], library=parsed.get("library", "Documents"))
+        return SharePointConnector(site=parsed["site"], site_path=parsed["site_path"], library=parsed.get("library", "Documents"), **auth)
 
     if source.startswith("nextcloud:"):
         from oikb.connectors.nextcloud import NextcloudConnector, parse_nextcloud_source
@@ -131,7 +156,7 @@ def _resolve_connector(source: str, branch: str | None = None, path: str | None
     if source.startswith("bitbucket:"):
         from oikb.connectors.bitbucket import BitbucketConnector, parse_bitbucket_source
         parsed = parse_bitbucket_source(source)
-        return BitbucketConnector(owner=parsed["owner"], repo=parsed["repo"], branch=branch, path=path or parsed.get("path"))
+        return BitbucketConnector(owner=parsed["owner"], repo=parsed["repo"], branch=branch, path=path or parsed.get("path"), **auth)
 
     if source.startswith("discord:"):
         from oikb.connectors.discord import DiscordConnector, parse_discord_source
@@ -290,7 +315,7 @@ def _resolve_connector(source: str, branch: str | None = None, path: str | None
     if source.startswith("zotero:"):
         from oikb.connectors.zotero import ZoteroConnector, parse_zotero_source
         parsed = parse_zotero_source(source)
-        return ZoteroConnector(hierarchy=parsed.get("hierarchy"))
+        return ZoteroConnector(hierarchy=parsed.get("hierarchy"), library_id=parsed.get("library_id"))
 
     # Default: local filesystem.
     from oikb.connectors.filesystem import FilesystemConnector
@@ -368,6 +393,27 @@ def _build_cli_filter(max_file_size: str | None):
     return build_manifest_filter(max_size=parse_size(max_file_size))
 
 
+def _parse_auth_option(pairs: tuple[str, ...]) -> dict[str, str]:
+    """Turn repeated --auth key=value flags into a credentials dict.
+
+    Mirrors the shape of an .oikb.yaml `auth:` block so single-source
+    CLI invocations (no .oikb.yaml) can also authenticate connectors
+    like gitlab/github/sharepoint/bitbucket without env vars.
+    """
+    auth: dict[str, str] = {}
+    for pair in pairs:
+        if "=" not in pair:
+            raise click.BadOptionUsage(
+                "auth", f"--auth expects key=value, got: {pair!r}"
+            )
+        key, _, value = pair.partition("=")
+        key = key.strip()
+        if not key:
+            raise click.BadOptionUsage("auth", f"--auth expects key=value, got: {pair!r}")
+        auth[key] = value
+    return auth
+
+
 # ── sync ────────────────────────────────────────────────────────
 
 @cli.command()
@@ -380,6 +426,14 @@ def _build_cli_filter(max_file_size: str | None):
 @click.option("--name", default=None, help="Target a specific entry in .oikb.yaml by name/kb-id.")
 @click.option("--concurrency", default=1, type=int, help="Parallel upload workers (default: 1, sequential).")
 @click.option("--max-file-size", default=None, help="Skip files larger than this (e.g. 50mb, 1gb).")
+@click.option(
+    "--auth",
+    "auth_pairs",
+    multiple=True,
+    metavar="KEY=VALUE",
+    help="Source credential, repeatable (e.g. --auth token=... --auth base_url=...). "
+    "Ignored in .oikb.yaml mode — use each entry's own auth: block instead.",
+)
 @click.pass_context
 def sync(
     ctx: click.Context,
@@ -394,6 +448,7 @@ def sync(
     name: str | None,
     concurrency: int,
     max_file_size: str | None,
+    auth_pairs: tuple[str, ...],
 ):
     """Incremental sync from a source to a Knowledge Base.
 
@@ -413,54 +468,33 @@ def sync(
             )
             sys.exit(1)
 
-        # Filter by --name if specified.
-        if name:
-            entries = [e for e in entries if e.get("name") == name or e.get("kb-id") == name]
-            if not entries:
-                click.echo(click.style(f"No entry matching '{name}' in .oikb.yaml", fg="red"), err=True)
-                sys.exit(1)
-
-        has_errors = False
-        for entry in entries:
-            entry_source = entry.get("source")
-            entry_kb = entry.get("kb-id")
-            entry_branch = entry.get("branch")
-            entry_path = entry.get("path")
-            entry_filter = entry.get("filter", {})
-
-            if not entry_source or not entry_kb:
-                click.echo(click.style(f"Skipping invalid entry (needs source + kb-id): {entry}", fg="yellow"), err=True)
-                continue
+        from oikb.kb_sync import group_entries_by_kb, run_entries_sync
 
         try:
-                connector = _resolve_connector(entry_source, entry_branch, entry_path)
-                client = _make_client(url, token)
+            groups = group_entries_by_kb(entries)
+        except ValueError as e:
+            raise click.ClickException(str(e)) from e
+        if name:
+            groups = [group for group in groups if any(
+                e.get("name") == name or e["kb-id"] == name for e in group
+            )]
+            if not groups:
+                raise click.ClickException(f"No entry matching '{name}' in .oikb.yaml")
 
+        has_errors = False
+        for group in groups:
+            entry = group[0]
+            client = None
+            try:
+                client = _make_client(url or entry.get("url"), token or entry.get("token"))
                 if not quiet:
                     click.echo(f"\n{'─' * 40}")
-                    click.echo(f"Syncing: {entry_source} → {entry_kb}")
+                    click.echo(f"Syncing: {', '.join(e['source'] for e in group)} → {entry['kb-id']}")
 
-                mf = None
-                inc = entry_filter.get("include")
-                exc = entry_filter.get("exclude")
-                ms = entry_filter.get("max-size") or max_file_size
-                if inc or exc or ms:
-                    from oikb.sync import build_manifest_filter, parse_size
-                    mf = build_manifest_filter(
-                        include=inc,
-                        exclude=exc,
-                        max_size=parse_size(ms),
-                    )
-
-                result = run_sync(
-                    client=client,
-                    connector=connector,
-                    kb_id=entry_kb,
-                    dry_run=dry_run,
-                    verbose=verbose,
-                    quiet=quiet,
-                    manifest_filter=mf,
-                    concurrency=entry.get("concurrency", concurrency),
+                result = run_entries_sync(
+                    client, group, resolve_connector=_resolve_connector,
+                    dry_run=dry_run, verbose=verbose, quiet=quiet,
+                    concurrency=concurrency, max_file_size=max_file_size,
                 )
 
                 if not quiet:
@@ -485,11 +519,12 @@ def sync(
                 if result.errors:
                     has_errors = True
 
-                client.close()
-
             except Exception as e:
                 click.echo(click.style(f"  Failed: {e}", fg="red"), err=True)
                 has_errors = True
+            finally:
+                if client:
+                    client.close()
 
         if has_errors:
             sys.exit(1)
@@ -501,7 +536,7 @@ def sync(
         sys.exit(1)
 
     try:
-        connector = _resolve_connector(source, branch, source_path)
+        connector = _resolve_connector(source, branch, source_path, auth=_parse_auth_option(auth_pairs))
     except (FileNotFoundError, ImportError, ValueError) as e:
         click.echo(click.style(f"Error: {e}", fg="red"), err=True)
         sys.exit(1)
@@ -554,6 +589,13 @@ def sync(
 @click.option("--branch", default=None, help="Branch for GitHub sources.")
 @click.option("--path", "source_path", default=None, help="Subdirectory within the source.")
 @click.option("-v", "--verbose", is_flag=True, help="Show detailed output.")
+@click.option(
+    "--auth",
+    "auth_pairs",
+    multiple=True,
+    metavar="KEY=VALUE",
+    help="Source credential, repeatable (e.g. --auth token=... --auth base_url=...).",
+)
 @click.pass_context
 def diff(
     ctx: click.Context,
@@ -564,6 +606,7 @@ def diff(
     branch: str | None,
     source_path: str | None,
     verbose: bool,
+    auth_pairs: tuple[str, ...],
 ):
     """Preview what a sync would do (alias for sync --dry-run)."""
     if not kb:
@@ -573,7 +616,7 @@ def diff(
     from oikb.sync import run_sync
 
     try:
-        connector = _resolve_connector(source, branch, source_path)
+        connector = _resolve_connector(source, branch, source_path, auth=_parse_auth_option(auth_pairs))
     except (FileNotFoundError, ImportError, ValueError) as e:
         click.echo(click.style(f"Error: {e}", fg="red"), err=True)
         sys.exit(1)
@@ -758,7 +801,7 @@ def status(url: str | None, token: str | None, kb: str | None):
 
     try:
         info = client.get_kb(kb)
-        files = info.get("files", [])
+        files = client.list_kb_files(kb)
     except Exception as e:
         click.echo(click.style(f"Failed: {e}", fg="red"), err=True)
         sys.exit(1)
@@ -924,9 +967,10 @@ def validate(config_file: str | None, deep: bool):
             has_errors = True
             continue
 
-        # Syntax check: resolve the connector.
+        # Syntax check: resolve the connector (using its configured auth,
+        # since some connectors — e.g. SharePoint — authenticate eagerly).
         try:
-            _resolve_connector(source)
+            _resolve_connector(source, auth=entry.get("auth", {}))
         except Exception as e:
             click.echo(click.style(f"  ✗ {entry_name}: {e}", fg="red"))
             has_errors = True
@@ -937,7 +981,7 @@ def validate(config_file: str | None, deep: bool):
             try:
                 kb = client.get_kb(kb_id)
                 kb_name = kb.get("name", "?")
-                file_count = len(kb.get("files", []))
+                file_count = client.count_kb_files(kb_id)
                 click.echo(
                     click.style(f"  ✓ {entry_name}", fg="green")
                     + f"  {source} → {kb_name} ({file_count} files)"
@@ -985,7 +1029,7 @@ def daemon(port: int, no_server: bool, config_file: str | None, log_format: str
         entries = _load_oikb_yaml()
 
     if not entries:
-        click.echo(click.style("No sync entries found. Create a .oikb.yaml file.", fg="red"), err=True)
+        click.echo(click.style("No sync entries found. Create a .oikb.yaml file at {yaml_path}.", fg="red"), err=True)
         sys.exit(1)
 
     # Validate entries.

src/oikb/client.py

@@ -125,14 +125,25 @@ class OikbClient:
         return resp.json()
 
     def get_kb(self, kb_id: str) -> dict[str, Any]:
-        """GET /knowledge/{id} — get KB info."""
+        """GET /knowledge/{id} — get KB metadata.
+
+        Note: this endpoint returns metadata only. Its ``files`` field is a
+        server-hydrated convenience that some Open WebUI versions return as
+        null; use ``list_kb_files``/``count_kb_files`` for the file list.
+        """
         resp = self._http.get(f"/knowledge/{kb_id}")
         resp.raise_for_status()
         return resp.json()
 
     def list_kb_files(self, kb_id: str) -> list[dict[str, Any]]:
         """GET /knowledge/{id}/files — list files in a KB."""
-        resp = self._http.get(f"/knowledge/{kb_id}")
+        resp = self._http.get(f"/knowledge/{kb_id}/files")
         resp.raise_for_status()
         data = resp.json()
-        return data.get("files", [])
+        return data.get("items", [])
+
+    def count_kb_files(self, kb_id: str) -> int:
+        """GET /knowledge/{id}/files — total file count for a KB."""
+        resp = self._http.get(f"/knowledge/{kb_id}/files")
+        resp.raise_for_status()
+        return resp.json().get("total", 0)

src/oikb/connectors/bookstack.py

@@ -1,4 +1,10 @@
-"""BookStack connector — sync pages from a BookStack instance.
+# ---
+# co_authored_by:
+#   - @hexsm
+#     pull_request: https://github.com/open-webui/oikb/pull/84
+# ---
+
+"""BookStack connector — sync pages, chapters, or books from a BookStack instance.
 
 Auth via BOOKSTACK_URL, BOOKSTACK_TOKEN_ID, BOOKSTACK_TOKEN_SECRET env vars.
 """

src/oikb/connectors/confluence.py

@@ -1,7 +1,8 @@
 """Confluence connector — sync a Confluence space to a Knowledge Base.
 
-Uses the Confluence Cloud REST API v2. Pages are exported as plain text.
-Auth via CONFLUENCE_URL, CONFLUENCE_USER, CONFLUENCE_TOKEN env vars.
+Supports Cloud REST API v2 (default) and Server/Data Center REST API v1.
+Set CONFLUENCE_API_VERSION=v1 for Server/Data Center. Authentication uses
+Basic auth with CONFLUENCE_USER, or Bearer auth when only a token is supplied.
 """
 
 from __future__ import annotations
@@ -10,31 +11,75 @@ import hashlib
 import html
 import os
 import re
+from collections import Counter
+from dataclasses import replace
 from typing import Any
+from urllib.parse import parse_qsl, urlsplit
 
 import httpx
 
-from oikb.connectors import BaseConnector, ManifestEntry
+from oikb.connectors import BaseConnector, ManifestEntry, SourceFileUnavailable
+
+# A link's visible label is the title of another page in the space, which syncs
+# as its own file. Dropped with the link so a section index does not become a
+# document that is nothing but titles already in the KB.
+_LINK_BODY = re.compile(
+    r"<ac:plain-text-link-body>.*?</ac:plain-text-link-body>", re.S | re.I
+)
+_CDATA = re.compile(r"<!\[CDATA\[(.*?)\]\]>", re.S)
+# Where one block ends the next begins on its own line: a code block run
+# together with the sentence before it reads as one thought.
+_BREAK = re.compile(
+    r"<br\s*/?>"
+    r"|</(?:p|div|h[1-6]|li|ul|ol|tr|td|th|blockquote|pre|table|section)\s*>"
+    r"|</ac:[\w.-]+\s*>",
+    re.I,
+)
+_TAG = re.compile(r"<[^>]+>")
+# Marks where a parked macro body goes back. Storage format cannot contain NUL.
+_PARKED = re.compile("\x00(\\d+)\x00")
+_INLINE_SPACE = re.compile(r"[^\S\n]+")
 
 
 def _storage_to_text(storage_html: str) -> str:
     """Convert Confluence storage format (XHTML) to plain text."""
-    # Strip all HTML tags.
-    text = re.sub(r"<[^>]+>", " ", storage_html)
+    if not storage_html:
+        return ""
+
+    text = _LINK_BODY.sub(" ", storage_html)
+
+    # Macro bodies are literal text, so they are parked before the markup passes
+    # run: `<[^>]+>` would otherwise treat `<![CDATA[...]]>` as one tag and
+    # delete a code block or panel whole, and their entities are not escaped.
+    parked: list[str] = []
+
+    def _park(match: re.Match) -> str:
+        parked.append(match.group(1))
+        return f"\n\x00{len(parked) - 1}\x00\n"
+
+    text = _CDATA.sub(_park, text)
+    text = _BREAK.sub("\n", text)
+    text = _TAG.sub(" ", text)
     text = html.unescape(text)
-    # Collapse whitespace.
-    text = re.sub(r"\s+", " ", text).strip()
-    return text
+
+    lines = (_INLINE_SPACE.sub(" ", line).strip() for line in text.split("\n"))
+    text = "\n".join(line for line in lines if line)
+
+    # Restored after the whitespace pass so a code block keeps its own line
+    # breaks and indentation.
+    return _PARKED.sub(lambda m: parked[int(m.group(1))].strip(), text).strip()
 
 
 class ConfluenceConnector(BaseConnector):
-    """Sync pages from a Confluence Cloud space.
+    """Sync pages from a Confluence space.
 
     Args:
         space_key: Confluence space key (e.g. "ENG").
         base_url:  Confluence instance URL (or CONFLUENCE_URL env var).
         user:      Confluence user email (or CONFLUENCE_USER env var).
         token:     Confluence API token (or CONFLUENCE_TOKEN env var).
+        structure: "flat" or "hierarchical" manifest paths.
+        api_version: "v1" or "v2" (or CONFLUENCE_API_VERSION, default "v2").
     """
 
     def __init__(
@@ -43,11 +88,19 @@ class ConfluenceConnector(BaseConnector):
         base_url: str | None = None,
         user: str | None = None,
         token: str | None = None,
+        structure: str = "flat",
+        api_version: str | None = None,
     ):
+        if structure not in {"flat", "hierarchical"}:
+            raise ValueError("structure must be 'flat' or 'hierarchical'")
         self.space_key = space_key
+        self.structure = structure
+        self._api_version = (api_version or os.environ.get("CONFLUENCE_API_VERSION", "v2")).lower()
+        if self._api_version not in {"v1", "v2"}:
+            raise ValueError("CONFLUENCE_API_VERSION must be 'v1' or 'v2'")
 
         self._base_url = (base_url or os.environ.get("CONFLUENCE_URL", "")).rstrip("/")
-        self._user = user or os.environ.get("CONFLUENCE_USER", "")
+        self._user = user if user is not None else os.environ.get("CONFLUENCE_USER", "")
         self._token = token or os.environ.get("CONFLUENCE_TOKEN", "")
 
         if not self._base_url:
@@ -61,86 +114,163 @@ class ConfluenceConnector(BaseConnector):
                 "  export CONFLUENCE_TOKEN=<api_token>"
             )
 
+        headers = {"Accept": "application/json"}
+        if not self._user:
+            headers["Authorization"] = f"Bearer {self._token}"
+        api_base = self._base_url
+        if self._api_version == "v2" and not api_base.endswith("/wiki"):
+            api_base += "/wiki"
         self._http = httpx.Client(
-            base_url=f"{self._base_url}/wiki",
+            base_url=api_base,
             auth=(self._user, self._token) if self._user else None,
-            headers={"Accept": "application/json"},
+            headers=headers,
             timeout=60.0,
         )
 
+        # Resolve space key to numeric ID (v2 API requires ID).
+        if self._api_version == "v2" and not self.space_key.isdecimal():
+            try:
+                resp = self._http.get(
+                    "/api/v2/spaces", params={"keys": [self.space_key]}
+                )
+                resp.raise_for_status()
+                results = resp.json().get("results", [])
+                matches = [
+                    space
+                    for space in results
+                    if space.get("key", "").casefold() == self.space_key.casefold()
+                ]
+                if len(matches) != 1:
+                    raise ValueError(f"Confluence space '{self.space_key}' not found")
+                self.space_key = str(matches[0]["id"])
+            except Exception:
+                self._http.close()
+                raise
+
         # Cache page content for read_file.
         self._page_cache: dict[str, str] = {}
 
     def build_manifest(self) -> list[ManifestEntry]:
         """List all pages in the space and build a manifest."""
-        entries: list[ManifestEntry] = []
-        cursor = None
+        self._page_cache.clear()
+        pages: list[dict[str, Any]] = []
+        params: dict[str, Any] = {"limit": 250}
+        endpoint = f"/api/v2/spaces/{self.space_key}/pages"
+        if self._api_version == "v1":
+            endpoint = "/rest/api/content"
+            params.update(spaceKey=self.space_key, type="page", start=0, expand="ancestors,version")
+        seen_pages: set[str] = set()
 
         while True:
-            params: dict[str, Any] = {"limit": 250}
-            if cursor:
-                params["cursor"] = cursor
-
-            resp = self._http.get(
-                f"/api/v2/spaces/{self.space_key}/pages",
-                params=params,
-            )
+            resp = self._http.get(endpoint, params=params)
             resp.raise_for_status()
             data = resp.json()
 
-            for page in data.get("results", []):
-                page_id = page["id"]
-                title = page["title"]
-                version = page.get("version", {}).get("number", 0)
-
-                # Use version number as part of checksum.
-                checksum = hashlib.sha256(
-                    f"{page_id}:v{version}".encode()
-                ).hexdigest()[:16]
-
-                # Sanitize title for filename.
-                filename = re.sub(r'[<>:"/\\|?*]', "_", title) + ".txt"
-
-                entries.append(
-                    ManifestEntry(
-                        filename=filename,
-                        path="",
-                        checksum=checksum,
-                        size=0,
-                    )
-                )
-
-                # Store page ID for later retrieval.
-                self._page_cache[filename] = page_id
+            results = data["results"]
+            for page in results:
+                page_id = str(page["id"])
+                if page_id in seen_pages:
+                    raise ValueError(f"Repeated Confluence page in pagination: {page_id}")
+                seen_pages.add(page_id)
+            pages.extend(results)
 
             # Handle pagination.
             next_link = data.get("_links", {}).get("next")
             if not next_link:
                 break
-            # Extract cursor from next link.
-            cursor_match = re.search(r"cursor=([^&]+)", next_link)
-            cursor = cursor_match.group(1) if cursor_match else None
-            if not cursor:
-                break
+            # Use only pagination parameters, never a server-provided host/path.
+            parameter = "start" if self._api_version == "v1" else "cursor"
+            next_value = dict(parse_qsl(urlsplit(next_link).query)).get(parameter)
+            if not results or not next_value or str(params.get(parameter)) == next_value:
+                raise ValueError("Invalid Confluence pagination link")
+            if parameter == "start" and int(next_value) <= int(params["start"]):
+                raise ValueError("Confluence pagination did not advance")
+            params[parameter] = next_value
 
+        pages_by_id = {str(page["id"]): page for page in pages}
+        entries = [self._page_entry(page, pages_by_id) for page in pages]
+        counts = Counter(entry.display_path for entry in entries)
+        reserved = set(counts)
+        for index, (page, entry) in enumerate(zip(pages, entries)):
+            if counts[entry.display_path] > 1:
+                # Rename every colliding title so API ordering cannot change identity.
+                stem = entry.filename.removesuffix(".txt") + f"_{page['id']}"
+                candidate = replace(entry, filename=f"{stem}.txt")
+                while candidate.display_path in reserved:
+                    stem += "_"
+                    candidate = replace(entry, filename=f"{stem}.txt")
+                entry = entries[index] = candidate
+                reserved.add(entry.display_path)
+            self._page_cache[entry.display_path] = str(page["id"])
         entries.sort(key=lambda e: e.display_path)
         return entries
 
+    def _page_entry(
+        self, page: dict[str, Any], pages_by_id: dict[str, dict[str, Any]]
+    ) -> ManifestEntry:
+        page_id = str(page["id"])
+        title = page["title"]
+        version = page.get("version", {}).get("number", 0)
+        checksum = hashlib.sha256(f"{page_id}:v{version}".encode()).hexdigest()[:16]
+        filename = self._safe_name(title) + ".txt"
+        path = self._page_path(page, pages_by_id)
+        return ManifestEntry(filename=filename, path=path, checksum=checksum, size=0)
+
+    def _page_path(
+        self, page: dict[str, Any], pages_by_id: dict[str, dict[str, Any]]
+    ) -> str:
+        if self.structure != "hierarchical":
+            return ""
+        if self._api_version == "v1":
+            return "/".join(self._safe_name(a.get("title")) for a in page.get("ancestors", []))
+
+        ancestors: list[str] = []
+        parent_id = page.get("parentId")
+        seen: set[str] = set()
+        while parent_id:
+            parent_id = str(parent_id)
+            if parent_id in seen:
+                raise ValueError(f"Circular Confluence page hierarchy at page {page['id']}")
+            seen.add(parent_id)
+            parent = pages_by_id.get(parent_id)
+            if not parent:
+                break
+            ancestors.append(self._safe_name(parent.get("title")))
+            parent_id = parent.get("parentId")
+        return "/".join(reversed(ancestors))
+
+    @staticmethod
+    def _safe_name(name: str | None) -> str:
+        safe = re.sub(r'[<>:"/\\|?*]', "_", name or "Untitled").strip()
+        return safe or "Untitled"
+
+    @staticmethod
+    def _entry_key(path: str, filename: str) -> str:
+        return f"{path}/{filename}" if path else filename
+
     def read_file(self, path: str, filename: str) -> bytes:
         """Fetch a page's content and return as text."""
-        page_id = self._page_cache.get(filename)
+        page_id = self._page_cache.get(self._entry_key(path, filename))
         if not page_id:
             raise FileNotFoundError(f"Page not found: {filename}")
 
-        resp = self._http.get(
-            f"/api/v2/pages/{page_id}",
-            params={"body-format": "storage"},
-        )
+        if self._api_version == "v1":
+            resp = self._http.get(f"/rest/api/content/{page_id}", params={"expand": "body.storage"})
+        else:
+            resp = self._http.get(f"/api/v2/pages/{page_id}", params={"body-format": "storage"})
         resp.raise_for_status()
         data = resp.json()
 
         storage = data.get("body", {}).get("storage", {}).get("value", "")
         text = _storage_to_text(storage)
+        if not text:
+            # Open WebUI extracts text as part of POST /files/ and answers 400
+            # for a file it can get nothing out of, so uploading this would fail
+            # the page on every run for as long as it exists.
+            raise SourceFileUnavailable(
+                "page has no text to sync (blank, or only a macro such as a "
+                "children index)"
+            )
         return text.encode("utf-8")
 
     def close(self) -> None:
@@ -153,14 +283,28 @@ def parse_confluence_source(source: str) -> dict[str, str | None]:
     Examples:
         confluence:ENG
         confluence:https://company.atlassian.net/ENG
+        confluence:ENG?structure=hierarchical
     """
     source = source.removeprefix("confluence:")
-
-    # Check if it includes a URL.
-    if source.startswith("https://"):
-        parts = source.rsplit("/", 1)
-        if len(parts) == 2:
-            return {"base_url": parts[0], "space_key": parts[1]}
+    is_url = source.startswith(("http://", "https://"))
+    parsed = urlsplit(source if is_url else f"confluence://{source}")
+    base_path = ""
+    space_key = parsed.netloc
+    if is_url:
+        base_path, _, space_key = parsed.path.rstrip("/").rpartition("/")
+    if not space_key or (not is_url and parsed.path):
         raise ValueError("Invalid Confluence source. Expected: confluence:SPACEKEY")
 
-    return {"space_key": source, "base_url": None}
+    params = dict(parse_qsl(parsed.query, keep_blank_values=True))
+    unknown = set(params) - {"structure"}
+    if unknown:
+        raise ValueError(f"Invalid Confluence source. Unknown parameter: {min(unknown)}")
+    structure = params.get("structure", "flat")
+    if structure not in {"flat", "hierarchical"}:
+        raise ValueError("Invalid Confluence source. Expected structure=flat or structure=hierarchical")
+
+    return {
+        "base_url": f"{parsed.scheme}://{parsed.netloc}{base_path}" if is_url else None,
+        "space_key": space_key,
+        "structure": structure,
+    }

src/oikb/connectors/gdrive.py

@@ -19,9 +19,8 @@ _EXPORT_MIMES: dict[str, tuple[str, str]] = {
     "application/vnd.google-apps.presentation": ("text/plain", ".txt"),
 }
 
-# MIME types to skip (folders, forms, maps, etc.).
+# Non-file Google app MIME types to skip.
 _SKIP_MIMES = frozenset({
-    "application/vnd.google-apps.folder",
     "application/vnd.google-apps.form",
     "application/vnd.google-apps.map",
     "application/vnd.google-apps.site",

src/oikb/connectors/outline.py

@@ -27,18 +27,38 @@ class OutlineConnector(BaseConnector):
         self._cache: dict[str, str] = {}
 
     def build_manifest(self) -> list[ManifestEntry]:
-        params: dict = {"limit": 100}
+        # Resolve collection ID once before starting the loop
+        collection_id = None
         if self._collection:
-            # Resolve collection by name.
-            cols = self._http.post("/api/collections.list", json={}).json().get("data", [])
+            cols_resp = self._http.post("/api/collections.list", json={})
+            cols_resp.raise_for_status()
+            cols = cols_resp.json().get("data", [])
             col = next((c for c in cols if c.get("name") == self._collection or c.get("id") == self._collection), None)
             if col:
-                params["collectionId"] = col["id"]
+                collection_id = col["id"]
+
+        entries: list[ManifestEntry] = []
+        offset = 0
+        limit = 100  # The maximum allowed by the server per request
+
+        while True:
+            # Use 'offset' instead of 'page' as per the API spec
+            params: dict = {
+                "offset": offset,
+                "limit": limit,
+                "sort": "updatedAt",
+                "direction": "DESC"
+            }
+            if collection_id:
+                params["collectionId"] = collection_id
 
             resp = self._http.post("/api/documents.list", json=params)
             resp.raise_for_status()
             docs = resp.json().get("data", [])
-        entries: list[ManifestEntry] = []
+
+            if not docs:
+                break
+
             for doc in docs:
                 title = doc.get("title", "untitled")
                 text = doc.get("text", "")
@@ -47,6 +67,14 @@ class OutlineConnector(BaseConnector):
                 checksum = hashlib.sha256(content.encode()).hexdigest()[:16]
                 entries.append(ManifestEntry(filename=filename, path="", checksum=checksum, size=len(content.encode())))
                 self._cache[filename] = content
+
+            # If we received fewer than the limit, we've reached the end of the list
+            if len(docs) < limit:
+                break
+
+            # Increment offset by the number of items retrieved to get the next batch
+            offset += limit
+
         return entries
 
     def read_file(self, path: str, filename: str) -> bytes:

src/oikb/connectors/sharepoint.py

@@ -8,6 +8,12 @@ and one of:
     encrypted PEM keys.
 
 The two auth methods are mutually exclusive.
+
+Select Microsoft Endpoints. Use SHAREPOINT_CLOUD set to
+one of:
+  - SHAREPOINT_CLOUD=commercial // default
+  - SHAREPOINT_CLOUD=gcc_high
+  - SHAREPOINT_CLOUD=dod
 """
 
 from __future__ import annotations
@@ -18,11 +24,27 @@ import os
 import time
 import uuid
 from typing import Any
+from urllib.parse import quote
 
 import httpx
 
 from oikb.connectors import BaseConnector, ManifestEntry
 
+_CLOUD_ENDPOINTS: dict[str, dict[str, str]] = {
+    "commercial": {
+        "authority": "https://login.microsoftonline.com",
+        "graph": "https://graph.microsoft.com/v1.0",
+    },
+    "gcc_high": {
+        "authority": "https://login.microsoftonline.us",
+        "graph": "https://graph.microsoft.us/v1.0",
+    },
+    "dod": {
+        "authority": "https://login.microsoftonline.us",
+        "graph": "https://dod-graph.microsoft.us/v1.0",
+    },
+}
+
 
 class SharePointConnector(BaseConnector):
     """Sync files from a SharePoint document library."""
@@ -30,6 +52,7 @@ class SharePointConnector(BaseConnector):
     def __init__(
         self,
         site: str,
+        site_path: str = "",
         library: str = "Documents",
         tenant_id: str | None = None,
         client_id: str | None = None,
@@ -38,6 +61,7 @@ class SharePointConnector(BaseConnector):
         certificate_password: str | None = None,
     ):
         self.site = site
+        self.site_path = site_path.strip("/")
         self.library = library
 
         tid = tenant_id or os.environ.get("SHAREPOINT_TENANT_ID", "")
@@ -46,6 +70,11 @@ class SharePointConnector(BaseConnector):
         cert_path = certificate_path or os.environ.get("SHAREPOINT_CERTIFICATE_PATH", "")
         cert_password = certificate_password or os.environ.get("SHAREPOINT_CERTIFICATE_PASSWORD", "")
 
+        cloud = os.environ.get("SHAREPOINT_CLOUD", "commercial")
+        if cloud not in _CLOUD_ENDPOINTS:
+            raise ValueError(f"SHAREPOINT_CLOUD must be one of {list(_CLOUD_ENDPOINTS)}, got '{cloud}'")
+        endpoints = _CLOUD_ENDPOINTS[cloud]
+
         if not tid or not cid:
             raise ValueError(
                 "SharePoint credentials required. Set env vars:\n"
@@ -66,7 +95,7 @@ class SharePointConnector(BaseConnector):
                 "  SHAREPOINT_CERTIFICATE_PATH  (certificate)"
             )
 
-        token_url = f"https://login.microsoftonline.com/{tid}/oauth2/v2.0/token"
+        token_url = f"{endpoints['authority']}/{tid}/oauth2/v2.0/token"
 
         if cert_path:
             access_token = _get_token_via_certificate(
@@ -74,23 +103,26 @@ class SharePointConnector(BaseConnector):
                 client_id=cid,
                 certificate_path=cert_path,
                 certificate_password=cert_password or None,
+                graph_base=endpoints["graph"]
             )
         else:
             access_token = _get_token_via_secret(
                 token_url=token_url,
                 client_id=cid,
                 client_secret=secret,
+                graph_base=endpoints["graph"]
             )
 
         self._http = httpx.Client(
-            base_url="https://graph.microsoft.com/v1.0",
+            base_url=endpoints["graph"],
             headers={"Authorization": f"Bearer {access_token}"},
             timeout=60.0,
             follow_redirects=True,
         )
 
         # Resolve site ID.
-        site_resp = self._http.get(f"/sites/{self.site}")
+        site_identifier = f"{self.site}:/{self.site_path}" if self.site_path else self.site
+        site_resp = self._http.get(f"/sites/{site_identifier}")
         site_resp.raise_for_status()
         self._site_id = site_resp.json()["id"]
 
@@ -113,7 +145,7 @@ class SharePointConnector(BaseConnector):
         return entries
 
     def _walk_folder(self, folder_path: str, prefix: str, entries: list[ManifestEntry]) -> None:
-        url = f"/drives/{self._drive_id}/root/children" if folder_path == "/" else f"/drives/{self._drive_id}/root:/{folder_path}:/children"
+        url = f"/drives/{self._drive_id}/root/children" if folder_path == "/" else f"/drives/{self._drive_id}/root:/{quote(folder_path)}:/children"
         resp = self._http.get(url)
         resp.raise_for_status()
 
@@ -133,7 +165,7 @@ class SharePointConnector(BaseConnector):
 
     def read_file(self, path: str, filename: str) -> bytes:
         file_path = f"{path}/{filename}" if path else filename
-        resp = self._http.get(f"/drives/{self._drive_id}/root:/{file_path}:/content")
+        resp = self._http.get(f"/drives/{self._drive_id}/root:/{quote(file_path)}:/content", follow_redirects=True)
         resp.raise_for_status()
         return resp.content
 
@@ -144,7 +176,7 @@ class SharePointConnector(BaseConnector):
 # ── Auth helpers ────────────────────────────────────────────────
 
 
-def _get_token_via_secret(token_url: str, client_id: str, client_secret: str) -> str:
+def _get_token_via_secret(token_url: str, client_id: str, client_secret: str, graph_base: str) -> str:
     """Obtain an access token using client ID + client secret."""
     token_resp = httpx.post(
         token_url,
@@ -152,7 +184,7 @@ def _get_token_via_secret(token_url: str, client_id: str, client_secret: str) ->
             "grant_type": "client_credentials",
             "client_id": client_id,
             "client_secret": client_secret,
-            "scope": "https://graph.microsoft.com/.default",
+            "scope": f"{graph_base.rsplit('/v1.0', 1)[0]}/.default",
         },
     )
     token_resp.raise_for_status()
@@ -163,6 +195,7 @@ def _get_token_via_certificate(
     token_url: str,
     client_id: str,
     certificate_path: str,
+    graph_base: str,
     certificate_password: str | None = None,
 ) -> str:
     """Obtain an access token using client ID + certificate (JWT assertion).
@@ -173,7 +206,7 @@ def _get_token_via_certificate(
     """
     try:
         from cryptography import x509
-        from cryptography.hazmat.primitives import serialization
+        from cryptography.hazmat.primitives import hashes, serialization
     except ImportError:
         raise ImportError(
             "Certificate auth requires the 'cryptography' package.\n"
@@ -201,9 +234,11 @@ def _get_token_via_certificate(
     # Load private key.
     private_key = serialization.load_pem_private_key(pem_data, password=password_bytes)
 
-    # Load certificate to extract thumbprint.
+    # Load certificate to extract thumbprint. The x5t header identifies the registered
+    # key by its SHA-1 thumbprint, whatever algorithm the certificate itself is signed
+    # with — a SHA-256 digest here is rejected with AADSTS700027 ("key was not found").
     cert = x509.load_pem_x509_certificate(pem_data)
-    thumbprint = cert.fingerprint(cert.signature_hash_algorithm or x509.hashes.SHA256())
+    thumbprint = cert.fingerprint(hashes.SHA1())
     x5t = base64.urlsafe_b64encode(thumbprint).rstrip(b"=").decode("ascii")
 
     # Build JWT assertion.
@@ -231,7 +266,7 @@ def _get_token_via_certificate(
             "client_id": client_id,
             "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
             "client_assertion": assertion,
-            "scope": "https://graph.microsoft.com/.default",
+            "scope": f"{graph_base.rsplit('/v1.0', 1)[0]}/.default",
         },
     )
     token_resp.raise_for_status()
@@ -240,13 +275,50 @@ def _get_token_via_certificate(
 
 # ── Source parser ───────────────────────────────────────────────
 
+_SITE_PATH_PREFIXES = ("sites", "teams")
+_SEPARATOR = "::"
 
-def parse_sharepoint_source(source: str) -> dict[str, str | None]:
-    """Parse sharepoint:site/library or sharepoint:site."""
+def parse_sharepoint_source(source: str) -> dict[str, str]:
+    """Parse a SharePoint source string. Supports:
+      sharepoint:<hostname>/<library>
+      sharepoint:<hostname>/sites/<site_name>/<library>
+      sharepoint:<hostname>/sites/<site_name>/<subsite>::<library>
+    """
     source = source.removeprefix("sharepoint:")
-    parts = source.split("/", 1)
-    site = parts[0]
-    library = parts[1] if len(parts) > 1 else "Documents"
-    if not site:
-        raise ValueError("Invalid SharePoint source. Expected: sharepoint:<site>[/library]")
-    return {"site": site, "library": library}
+    host, _, rest = source.partition("/")
+    if not host:
+        raise ValueError(
+            "Invalid SharePoint source. Expected one of:\n"
+            "  sharepoint:<hostname>/<library>\n"
+            "  sharepoint:<hostname>/sites/<site_name>/<library>\n"
+            "  sharepoint:<hostname>/sites/<site_name>/<subsite>::<library>"
+        )
+
+    if _SEPARATOR in rest:
+        site_path_str, _, library = rest.partition(_SEPARATOR)
+        site_path = site_path_str.strip("/")
+        if not library:
+            raise ValueError(f"Invalid SharePoint source: '{_SEPARATOR}' must be followed by a library name.")
+        return {"site": host, "site_path": site_path, "library": library}
+
+    segments = [s for s in rest.split("/") if s]
+
+    if segments and segments[0] in _SITE_PATH_PREFIXES:
+        if len(segments) < 3:
+            raise ValueError(
+                f"Invalid SharePoint source. '{segments[0]}/...' requires a site name and "
+                f"library, e.g. sharepoint:{host}/{segments[0]}/TeamSite/Documents"
+            )
+        if len(segments) > 3:
+            raise ValueError(
+                "Ambiguous SharePoint source with a subsite path — separate the site path "
+                f"from the library explicitly with '{_SEPARATOR}', e.g.\n"
+                f"  sharepoint:{host}/{'/'.join(segments[:-1])}{_SEPARATOR}{segments[-1]}"
+            )
+        site_path = "/".join(segments[:2])   # e.g. "sites/TeamSite"
+        library = segments[2]
+    else:
+        site_path = ""
+        library = "/".join(segments) if segments else "Documents"
+
+    return {"site": host, "site_path": site_path, "library": library}

src/oikb/connectors/zotero.py

@@ -1,8 +1,36 @@
+# ---
+# co_authored_by:
+#   - @thiswillbeyourgithub
+#     pull_request: https://github.com/open-webui/oikb/pull/61
+# ---
+
 """Zotero connector - sync PDF attachment text from a Zotero library.
 
 Auth via ZOTERO_LIBRARY_ID, ZOTERO_API_KEY, and optional ZOTERO_LIBRARY_TYPE.
 Source syntax: zotero:<collection>%%<subcollection>. Use bare zotero: for all
-top-level collections.
+top-level collections and unfiled library items.
+
+Multiple libraries in a single daemon
+--------------------------------------
+Embed a numeric library ID before the collection path to target a specific
+library and resolve its credentials from per-library env vars:
+
+    source: "zotero:123456:"              # all collections in library 123456
+    source: "zotero:123456:Research%%ML"  # specific collection
+
+Every ZOTERO_* env var supports a per-library override by appending the
+library ID (e.g. ZOTERO_API_KEY_123456). The global var is the fallback:
+
+    ZOTERO_API_KEY_123456      overrides  ZOTERO_API_KEY
+    ZOTERO_LIBRARY_TYPE_123456 overrides  ZOTERO_LIBRARY_TYPE
+    ZOTERO_CHECKSUM_123456     overrides  ZOTERO_CHECKSUM
+    ZOTERO_INCLUDE_NOTES_123456           ZOTERO_INCLUDE_NOTES
+    ZOTERO_INCLUDE_ANNOTATIONS_123456     ZOTERO_INCLUDE_ANNOTATIONS
+    ZOTERO_EXCLUDE_123456                 ZOTERO_EXCLUDE
+    ZOTERO_UNFILED_DIR_123456             ZOTERO_UNFILED_DIR
+    ZOTERO_WEBDAV_URL_123456              ZOTERO_WEBDAV_URL
+    ZOTERO_WEBDAV_USER_123456             ZOTERO_WEBDAV_USER
+    ZOTERO_WEBDAV_PASSWORD_123456         ZOTERO_WEBDAV_PASSWORD
 """
 
 from __future__ import annotations
@@ -24,6 +52,19 @@ from oikb.connectors import BaseConnector, ManifestEntry, SourceFileUnavailable
 HIERARCHY_SEP = "%%"
 
 
+def _zotero_env(key: str, library_id: str | None, default: str = "") -> str:
+    """Return the per-library env var if set, otherwise the global one.
+
+    Allows each Zotero source entry to be configured independently when
+    multiple libraries are synced from a single daemon instance.
+    """
+    if library_id:
+        value = os.environ.get(f"{key}_{library_id}")
+        if value is not None:
+            return value
+    return os.environ.get(key, default)
+
+
 @dataclass(frozen=True, slots=True)
 class _ZoteroFile:
     item: dict[str, Any]
@@ -51,7 +92,7 @@ class ZoteroConnector(BaseConnector):
             ) from exc
 
         library_id = library_id or os.environ.get("ZOTERO_LIBRARY_ID", "")
-        api_key = api_key or os.environ.get("ZOTERO_API_KEY", "")
+        api_key = api_key or _zotero_env("ZOTERO_API_KEY", library_id)
         if not library_id or not api_key:
             raise ValueError(
                 "Zotero credentials required. Set ZOTERO_LIBRARY_ID and ZOTERO_API_KEY."
@@ -60,27 +101,28 @@ class ZoteroConnector(BaseConnector):
         self.hierarchy = hierarchy
         self._zot = zotero.Zotero(
             library_id,
-            library_type or os.environ.get("ZOTERO_LIBRARY_TYPE", "user"),
+            library_type or _zotero_env("ZOTERO_LIBRARY_TYPE", library_id, "user"),
             api_key,
         )
-        self.include_notes = os.environ.get("ZOTERO_INCLUDE_NOTES", "").strip().lower() in {
+        self.include_notes = _zotero_env("ZOTERO_INCLUDE_NOTES", library_id).strip().lower() in {
             "1", "true", "yes", "on",
         }
-        self.include_annotations = os.environ.get(
-            "ZOTERO_INCLUDE_ANNOTATIONS", ""
-        ).strip().lower() in {"1", "true", "yes", "on"}
-        self.checksum_mode = os.environ.get("ZOTERO_CHECKSUM", "version").strip().lower()
+        self.include_annotations = (
+            _zotero_env("ZOTERO_INCLUDE_ANNOTATIONS", library_id).strip().lower()
+            in {"1", "true", "yes", "on"}
+        )
+        self.checksum_mode = _zotero_env("ZOTERO_CHECKSUM", library_id, "version").strip().lower()
         if self.checksum_mode not in {"version", "content"}:
             raise ValueError("ZOTERO_CHECKSUM must be 'version' or 'content'.")
         self.exclude = [
             part.strip().strip("/")
-            for part in os.environ.get("ZOTERO_EXCLUDE", "").split(",")
+            for part in _zotero_env("ZOTERO_EXCLUDE", library_id).split(",")
             if part.strip().strip("/")
         ]
-        self.unfiled_dir = os.environ.get("ZOTERO_UNFILED_DIR", "_unfiled").strip("/")
-        self.webdav_url = os.environ.get("ZOTERO_WEBDAV_URL", "").rstrip("/")
-        self.webdav_user = os.environ.get("ZOTERO_WEBDAV_USER", "")
-        self.webdav_password = os.environ.get("ZOTERO_WEBDAV_PASSWORD", "")
+        self.unfiled_dir = _zotero_env("ZOTERO_UNFILED_DIR", library_id, "_unfiled").strip("/")
+        self.webdav_url = _zotero_env("ZOTERO_WEBDAV_URL", library_id).rstrip("/")
+        self.webdav_user = _zotero_env("ZOTERO_WEBDAV_USER", library_id)
+        self.webdav_password = _zotero_env("ZOTERO_WEBDAV_PASSWORD", library_id)
         self._files: dict[tuple[str, str], _ZoteroFile] = {}
         self._text_cache: dict[str, str] = {}
 
@@ -216,7 +258,7 @@ class ZoteroConnector(BaseConnector):
             return None
         for item in collections:
             data = item.get("data", {})
-            parent = data.get("parentCollection")
+            parent = data.get("parentCollection") or None
             if parent != parent_key or data.get("name") != parts[0]:
                 continue
             if len(parts) == 1:
@@ -381,5 +423,16 @@ class ZoteroConnector(BaseConnector):
 
 
 def parse_zotero_source(source: str) -> dict[str, str | None]:
-    hierarchy = source.removeprefix("zotero:")
-    return {"hierarchy": None if hierarchy in {"", "*"} else hierarchy}
+    rest = source.removeprefix("zotero:")
+    library_id: str | None = None
+    # Optional library ID prefix: "zotero:123456:Collection%%Sub". A numeric
+    # segment before the first colon is treated as the library ID, allowing
+    # multiple Zotero libraries in a single .oikb.yaml. Plain collection paths
+    # (e.g. "zotero:Research%%ML") are unchanged — collection names are not
+    # purely numeric, so there is no ambiguity.
+    if ":" in rest:
+        head, tail = rest.split(":", 1)
+        if head.isdigit():
+            library_id = head
+            rest = tail
+    return {"hierarchy": None if rest in {"", "*"} else rest, "library_id": library_id}

src/oikb/daemon.py

@@ -240,7 +240,7 @@ async def _send_notification(entry: dict, payload: dict) -> None:
 
 
 async def _run_entry(entry: dict, dry_run: bool = False) -> dict | None:
-    """Run a single sync for an entry.
+    """Sync every configured source sharing this entry's KB.
 
     Uses a per-KB lock to prevent overlapping syncs to the same
     Knowledge Base (e.g. webhook fires while a scheduled sync is running).
@@ -263,53 +263,30 @@ async def _run_entry(entry: dict, dry_run: bool = False) -> dict | None:
 async def _run_entry_locked(entry: dict, dry_run: bool = False) -> dict | None:
     """Inner sync logic, called under the per-KB lock."""
     from oikb.cli import _make_client, _resolve_connector
-    from oikb.sync import SyncCancelled, run_sync
+    from oikb.kb_sync import run_entries_sync
+    from oikb.sync import SyncCancelled
 
-    source = entry["source"]
+    entries = [e for e in _entries if e.get("kb-id") == entry["kb-id"]] or [entry]
+    source = ", ".join(e["source"] for e in entries)
     kb_id = entry["kb-id"]
     started_at = time.time()
     client = None
 
-    _scheduler_state[source] = {
-        **_scheduler_state.get(source, {}),
-        "name": entry.get("name", source),
-        "status": "running",
-        "started_at": started_at,
+    def set_state(**state):
+        for member in entries:
+            _scheduler_state[member["source"]] = {
+                "name": member.get("name", member["source"]), **state,
             }
 
+    if not dry_run:
+        set_state(status="running", started_at=started_at)
+
     try:
-        connector = _resolve_connector(
-            source,
-            branch=entry.get("branch"),
-            path=entry.get("path"),
-        )
-        client = _make_client(
-            url=entry.get("url"),
-            token=entry.get("token"),
-        )
-
-        mf = None
-        entry_filter = entry.get("filter", {})
-        inc = entry_filter.get("include")
-        exc = entry_filter.get("exclude")
-        ms = entry_filter.get("max-size")
-        if inc or exc or ms:
-            from oikb.sync import build_manifest_filter, parse_size
-            mf = build_manifest_filter(
-                include=inc,
-                exclude=exc,
-                max_size=parse_size(ms),
-            )
-
+        client = _make_client(url=entry.get("url"), token=entry.get("token"))
         result = await asyncio.to_thread(
-            run_sync,
-            client=client,
-            connector=connector,
-            kb_id=kb_id,
-            dry_run=dry_run,
-            quiet=True,
-            manifest_filter=mf,
-            concurrency=entry.get("concurrency", 1),
+            run_entries_sync,
+            client, entries, resolve_connector=_resolve_connector,
+            dry_run=dry_run, quiet=True,
             cancel_requested=_shutdown_event.is_set if _shutdown_event else None,
         )
 
@@ -328,18 +305,12 @@ async def _run_entry_locked(entry: dict, dry_run: bool = False) -> dict | None:
         duration_ms = int(duration_s * 1000)
         status = "success" if not result.errors else "partial"
 
-        _scheduler_state[source] = {
-            "name": entry.get("name", source),
-            "status": status,
-            "last_sync": time.time(),
-            "duration_ms": duration_ms,
-            "files_added": result.added,
-            "files_modified": result.modified,
-            "files_deleted": result.deleted,
-            "unmodified": result.unmodified,
-            "warnings": result.warnings or [],
-            "errors": result.errors or [],
-        }
+        set_state(
+            status=status, last_sync=time.time(), duration_ms=duration_ms,
+            files_added=result.added, files_modified=result.modified,
+            files_deleted=result.deleted, unmodified=result.unmodified,
+            warnings=result.warnings or [], errors=result.errors or [],
+        )
 
         record_sync(
             source=source,
@@ -368,7 +339,8 @@ async def _run_entry_locked(entry: dict, dry_run: bool = False) -> dict | None:
             f"Synced {source} -> {kb_id}: {result.summary()} ({duration_ms}ms)"
         )
 
-        await _send_notification(entry, {
+        for member in entries:
+            await _send_notification(member, {
                 "source": source,
                 "kb_id": kb_id,
                 "status": status,
@@ -382,20 +354,11 @@ async def _run_entry_locked(entry: dict, dry_run: bool = False) -> dict | None:
             })
 
     except SyncCancelled:
-        _scheduler_state[source] = {
-            "name": entry.get("name", source),
-            "status": "cancelled",
-            "last_sync": time.time(),
-        }
+        set_state(status="cancelled", last_sync=time.time())
         log.info(f"Sync cancelled for {source}")
 
     except Exception as e:
-        _scheduler_state[source] = {
-            "name": entry.get("name", source),
-            "status": "error",
-            "last_sync": time.time(),
-            "error": str(e),
-        }
+        set_state(status="error", last_sync=time.time(), error=str(e))
         record_sync(
             source=source,
             status="error",
@@ -412,7 +375,8 @@ async def _run_entry_locked(entry: dict, dry_run: bool = False) -> dict | None:
             )
         log.error(f"Sync failed for {source}: {e}")
 
-        await _send_notification(entry, {
+        for member in entries:
+            await _send_notification(member, {
                 "source": source,
                 "kb_id": kb_id,
                 "status": "error",
@@ -492,6 +456,8 @@ def start_daemon(
     from oikb.logging import configure_logging
     configure_logging(log_format=log_format)
 
+    from oikb.kb_sync import group_entries_by_kb
+    group_entries_by_kb(entries)  # Validate shared destinations before starting tasks.
     _entries = entries
     _history = SyncHistory()
     set_build_info(__version__)

src/oikb/kb_sync.py

@@ -0,0 +1,108 @@
+"""Sync all configured sources for a KB with one complete manifest."""
+
+from __future__ import annotations
+
+from contextlib import ExitStack
+from dataclasses import replace
+from typing import Callable
+
+from oikb.client import OikbClient
+from oikb.connectors import BaseConnector, ManifestEntry
+from oikb.sync import SyncCancelled, SyncResult, build_manifest_filter, parse_size, run_sync
+
+
+def group_entries_by_kb(entries: list[dict]) -> list[list[dict]]:
+    groups: dict[str, list[dict]] = {}
+    for entry in entries:
+        if not entry.get("source") or not entry.get("kb-id"):
+            raise ValueError("Each source requires source and kb-id")
+        group = groups.setdefault(entry["kb-id"], [])
+        if group and any(entry.get(key) != group[0].get(key) for key in ("url", "token")):
+            raise ValueError(f"Sources for KB {entry['kb-id']} must use the same url and token")
+        group.append(entry)
+    return list(groups.values())
+
+
+class _CombinedConnector(BaseConnector):
+    def __init__(
+        self,
+        manifest: list[ManifestEntry],
+        routes: dict[tuple[str, str], tuple[BaseConnector, str]],
+    ):
+        self._manifest = manifest
+        self._routes = routes
+
+    def build_manifest(self) -> list[ManifestEntry]:
+        return self._manifest
+
+    def read_file(self, path: str, filename: str) -> bytes:
+        connector, original_path = self._routes[(path, filename)]
+        return connector.read_file(original_path, filename)
+
+
+def run_entries_sync(
+    client: OikbClient,
+    entries: list[dict],
+    *,
+    resolve_connector: Callable[..., BaseConnector],
+    dry_run: bool = False,
+    verbose: bool = False,
+    quiet: bool = False,
+    max_file_size: str | None = None,
+    concurrency: int = 1,
+    cancel_requested: Callable[[], bool] | None = None,
+) -> SyncResult:
+    """Scan and filter every source before allowing a KB-wide diff or deletion.
+
+    Filters apply to source paths. target-path adds a destination directory;
+    multi-source Confluence entries default to their space key as a prefix.
+    ExitStack owns connectors, including when resolution or scanning fails.
+    """
+    groups = group_entries_by_kb(entries)
+    if len(groups) != 1:
+        raise ValueError("Expected sources for exactly one KB")
+    manifest: list[ManifestEntry] = []
+    routes: dict[tuple[str, str], tuple[BaseConnector, str]] = {}
+    with ExitStack() as stack:
+        for entry in entries:
+            if cancel_requested and cancel_requested():
+                raise SyncCancelled("sync cancelled")
+            prefix = entry.get("target-path", "")
+            if "target-path" not in entry and len(entries) > 1 and entry["source"].startswith("confluence:"):
+                from oikb.connectors.confluence import parse_confluence_source
+                prefix = parse_confluence_source(entry["source"])["space_key"]
+            if not isinstance(prefix, str) or (prefix and any(
+                part in {"", ".", ".."} for part in prefix.split("/")
+            )) or "\\" in prefix:
+                raise ValueError("target-path must be a relative directory path")
+            filters = entry.get("filter", {})
+            manifest_filter = build_manifest_filter(
+                include=filters.get("include"),
+                exclude=filters.get("exclude"),
+                max_size=parse_size(filters.get("max-size") or max_file_size),
+            )
+            connector = stack.enter_context(resolve_connector(
+                entry["source"], branch=entry.get("branch"), path=entry.get("path"),
+                auth=entry.get("auth", {}),
+            ))
+            source_manifest = connector.build_manifest()
+            if manifest_filter:
+                source_manifest = manifest_filter(source_manifest)
+            for item in source_manifest:
+                path = "/".join(p for p in (prefix, item.path) if p)
+                key = (path, item.filename)
+                if key in routes:
+                    raise ValueError(f"Duplicate manifest path: {path}/{item.filename}. Set distinct target-path values.")
+                routes[key] = (connector, item.path)
+                manifest.append(replace(item, path=path))
+
+        return run_sync(
+            client=client,
+            connector=_CombinedConnector(manifest, routes),
+            kb_id=entries[0]["kb-id"],
+            dry_run=dry_run,
+            verbose=verbose,
+            quiet=quiet,
+            concurrency=max(entry.get("concurrency", concurrency) for entry in entries),
+            cancel_requested=cancel_requested,
+        )

src/oikb/sync.py

@@ -350,6 +350,12 @@ def _run_sync_inner(
             check_stop()
             try:
                 content = connector.read_file(path, filename)
+                if not content:
+                    if progress is not None:
+                        progress.update(task_id, advance=1, description=f"[yellow]⚠ {display}[/yellow]")
+                    else:
+                        click.echo(click.style(f"  ⚠ {display}: empty content, skipping", fg="yellow"), err=True)
+                    return ("warning", f"{display}: empty content, skipping")
                 check_stop()
                 directory_id = directory_map.get(path) if path else None
                 client.upload_file(
@@ -375,7 +381,14 @@ def _run_sync_inner(
                     check_stop()
                     last_err = e
                     continue
-                last_err = e
+                detail = e.response.text.strip()
+                try:
+                    payload = e.response.json()
+                    if isinstance(payload, dict) and payload.get("detail"):
+                        detail = str(payload["detail"])
+                except ValueError:
+                    pass
+                last_err = RuntimeError(f"{e} — {detail}") if detail else e
                 break
             except SyncCancelled:
                 raise

tests/test_confluence.py

@@ -0,0 +1,298 @@
+from __future__ import annotations
+
+import httpx
+import pytest
+import respx
+
+from oikb.connectors.confluence import ConfluenceConnector, parse_confluence_source
+from oikb.sync import build_manifest_filter
+
+
+def test_parse_hierarchical_source() -> None:
+    assert parse_confluence_source("confluence:ABC?structure=hierarchical") == {
+        "base_url": None,
+        "space_key": "ABC",
+        "structure": "hierarchical",
+    }
+
+
+def test_parse_url_source_with_structure() -> None:
+    assert parse_confluence_source(
+        "https://company.atlassian.net/ABC?structure=hierarchical"
+    ) == {
+        "base_url": "https://company.atlassian.net",
+        "space_key": "ABC",
+        "structure": "hierarchical",
+    }
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        "confluence:ABC?structure=invalid",
+        "confluence:ABC?unexpected=value",
+    ],
+)
+def test_parse_rejects_invalid_structure(source: str) -> None:
+    with pytest.raises(ValueError):
+        parse_confluence_source(source)
+
+
+@respx.mock
+def test_resolves_space_key_to_id() -> None:
+    lookup = respx.get("https://test.atlassian.net/wiki/api/v2/spaces").mock(
+        return_value=httpx.Response(
+            200, json={"results": [{"id": "123456789", "key": "ABC"}]}
+        )
+    )
+    pages = respx.get(
+        "https://test.atlassian.net/wiki/api/v2/spaces/123456789/pages"
+    ).mock(return_value=httpx.Response(200, json={"results": []}))
+
+    connector = ConfluenceConnector(
+        space_key="ABC", base_url="https://test.atlassian.net", token="token"
+    )
+    try:
+        assert connector.build_manifest() == []
+    finally:
+        connector.close()
+
+    assert lookup.calls[0].request.url.params["keys"] == "ABC"
+    assert pages.called
+
+
+@respx.mock
+def test_rejects_mismatched_space_lookup() -> None:
+    respx.get("https://test.atlassian.net/wiki/api/v2/spaces").mock(
+        return_value=httpx.Response(
+            200, json={"results": [{"id": "123456789", "key": "OTHER"}]}
+        )
+    )
+
+    with pytest.raises(ValueError, match="Confluence space 'ABC' not found"):
+        ConfluenceConnector(
+            space_key="ABC",
+            base_url="https://test.atlassian.net",
+            token="token",
+        )
+
+
+def _pages() -> list[dict[str, object]]:
+    return [
+        {"id": "1", "title": "FAQ", "version": {"number": 1}},
+        {
+            "id": "2",
+            "title": "Benefits",
+            "parentId": "1",
+            "version": {"number": 2},
+        },
+        {"id": "3", "title": "Internal Notes", "version": {"number": 1}},
+    ]
+
+
+@pytest.mark.parametrize(
+    ("structure", "expected"),
+    [
+        ("flat", ["Benefits.txt", "FAQ.txt", "Internal Notes.txt"]),
+        ("hierarchical", ["FAQ.txt", "FAQ/Benefits.txt", "Internal Notes.txt"]),
+    ],
+)
+@respx.mock
+def test_manifest_in_both_modes(structure: str, expected: list[str]) -> None:
+    respx.get(
+        "https://test.atlassian.net/wiki/api/v2/spaces/123456789/pages"
+    ).mock(return_value=httpx.Response(200, json={"results": _pages()}))
+
+    connector = ConfluenceConnector(
+        space_key="123456789",
+        base_url="https://test.atlassian.net",
+        token="token",
+        structure=structure,
+    )
+    try:
+        manifest = connector.build_manifest()
+    finally:
+        connector.close()
+
+    assert [entry.display_path for entry in manifest] == expected
+
+
+@respx.mock
+def test_hierarchical_paths_work_with_filter_and_read_file() -> None:
+    respx.get(
+        "https://test.atlassian.net/wiki/api/v2/spaces/123456789/pages"
+    ).mock(return_value=httpx.Response(200, json={"results": _pages()}))
+    content = respx.get("https://test.atlassian.net/wiki/api/v2/pages/2").mock(
+        return_value=httpx.Response(
+            200, json={"body": {"storage": {"value": "<p>Benefits</p>"}}}
+        )
+    )
+
+    connector = ConfluenceConnector(
+        space_key="123456789",
+        base_url="https://test.atlassian.net",
+        token="token",
+        structure="hierarchical",
+    )
+    try:
+        manifest = connector.build_manifest()
+        selected = build_manifest_filter(include=["FAQ*"])(manifest)
+        assert [entry.display_path for entry in selected] == [
+            "FAQ.txt",
+            "FAQ/Benefits.txt",
+        ]
+        assert connector.read_file("FAQ", "Benefits.txt") == b"Benefits"
+    finally:
+        connector.close()
+
+    assert content.called
+
+
+@respx.mock
+def test_hierarchical_mode_disambiguates_duplicate_paths() -> None:
+    pages = [
+        {"id": "1", "title": "FAQ", "version": {"number": 1}},
+        {"id": "2", "title": "FAQ", "version": {"number": 1}},
+    ]
+    respx.get(
+        "https://test.atlassian.net/wiki/api/v2/spaces/123456789/pages"
+    ).mock(return_value=httpx.Response(200, json={"results": pages}))
+    connector = ConfluenceConnector(
+        space_key="123456789",
+        base_url="https://test.atlassian.net",
+        token="token",
+        structure="hierarchical",
+    )
+
+    try:
+        assert [e.display_path for e in connector.build_manifest()] == ["FAQ_1.txt", "FAQ_2.txt"]
+    finally:
+        connector.close()
+
+
+@respx.mock
+def test_manifest_can_be_built_repeatedly() -> None:
+    respx.get(
+        "https://test.atlassian.net/wiki/api/v2/spaces/123456789/pages"
+    ).mock(return_value=httpx.Response(200, json={"results": _pages()}))
+    connector = ConfluenceConnector(
+        space_key="123456789",
+        base_url="https://test.atlassian.net",
+        token="token",
+        structure="hierarchical",
+    )
+
+    try:
+        first = connector.build_manifest()
+        second = connector.build_manifest()
+    finally:
+        connector.close()
+
+    assert first == second
+
+
+@pytest.mark.parametrize("structure", ["flat", "hierarchical"])
+@respx.mock
+def test_v1_pagination_context_path_auth_and_content(structure, monkeypatch):
+    monkeypatch.setenv("CONFLUENCE_API_VERSION", "v1")
+    monkeypatch.setenv("CONFLUENCE_USER", "")
+    endpoint = "https://wiki.example/confluence/rest/api/content"
+    route = respx.get(endpoint).mock(side_effect=[
+        httpx.Response(200, json={
+            "results": [{"id": "1", "title": "Parent", "version": {"number": 1}}],
+            # The server caps the requested 250 to 1. The next link is authoritative.
+            "limit": 1, "_links": {"next": "/rest/api/content?start=1&limit=1"},
+        }),
+        httpx.Response(200, json={"results": [{
+            "id": "2", "title": "Child", "version": {"number": 3},
+            "ancestors": [{"id": "1", "title": "Parent"}],
+        }]}),
+    ])
+    content = respx.get(endpoint + "/2").respond(200, json={
+        "body": {"storage": {"value": '<ac:plain-text-body><![CDATA[if x < 2:\n    run()]]></ac:plain-text-body>'}},
+    })
+    with ConfluenceConnector("ENG", base_url="https://wiki.example/confluence", token="pat", structure=structure) as connector:
+        manifest = connector.build_manifest()
+        child = next(e for e in manifest if e.filename == "Child.txt")
+        assert child.path == ("Parent" if structure == "hierarchical" else "")
+        assert connector.read_file(child.path, child.filename) == b"if x < 2:\n    run()"
+    assert [call.request.url.params["start"] for call in route.calls] == ["0", "1"]
+    assert route.calls[0].request.url.params["spaceKey"] == "ENG"
+    assert route.calls[0].request.url.params["expand"] == "ancestors,version"
+    assert content.calls[0].request.headers["Authorization"] == "Bearer pat"
+    assert content.calls[0].request.url.params["expand"] == "body.storage"
+
+
+@respx.mock
+def test_v2_cursor_is_decoded_and_wiki_path_is_not_duplicated():
+    route = respx.get("https://test.atlassian.net/wiki/api/v2/spaces/123/pages").mock(side_effect=[
+        httpx.Response(200, json={"results": [{"id": "1", "title": "One"}], "_links": {"next": "?cursor=a%2Bb%3D"}}),
+        httpx.Response(200, json={"results": [{"id": "2", "title": "Two"}]}),
+    ])
+    with ConfluenceConnector("123", base_url="https://test.atlassian.net/wiki", user="email", token="secret") as connector:
+        assert len(connector.build_manifest()) == 2
+    assert route.calls[1].request.url.params["cursor"] == "a+b="
+    assert route.calls[0].request.headers["Authorization"] == "Basic ZW1haWw6c2VjcmV0"
+
+
+@respx.mock
+def test_colliding_names_are_stable_and_route_to_distinct_pages():
+    pages = [
+        {"id": "1", "title": "A/B"}, {"id": "2", "title": "A:B"},
+        {"id": "3", "title": "A_B_1"},
+    ]
+    respx.get("https://wiki.example/wiki/api/v2/spaces/123/pages").mock(side_effect=[
+        httpx.Response(200, json={"results": pages}),
+        httpx.Response(200, json={"results": list(reversed(pages))}),
+    ])
+    for page in pages:
+        respx.get(f"https://wiki.example/wiki/api/v2/pages/{page['id']}").respond(
+            200, json={"body": {"storage": {"value": page["title"]}}},
+        )
+    with ConfluenceConnector("123", base_url="https://wiki.example", token="pat") as connector:
+        manifest = connector.build_manifest()
+        assert len({e.display_path for e in manifest}) == 3
+        assert {connector.read_file(e.path, e.filename) for e in manifest} == {b"A/B", b"A:B", b"A_B_1"}
+        assert connector.build_manifest() == manifest
+
+
+@pytest.mark.parametrize("next_link", ["?limit=1", "?start=0", "?start=-1"])
+@respx.mock
+def test_invalid_pagination_fails_instead_of_returning_partial_manifest(next_link):
+    respx.get("https://wiki.example/rest/api/content").respond(200, json={
+        "results": [{"id": "1", "title": "One"}], "_links": {"next": next_link},
+    })
+    with ConfluenceConnector("ENG", base_url="https://wiki.example", token="pat", api_version="v1") as connector:
+        with pytest.raises(ValueError, match="pagination"):
+            connector.build_manifest()
+
+
+@respx.mock
+def test_empty_pages_remain_skipped():
+    from oikb.connectors import SourceFileUnavailable
+    respx.get("https://wiki.example/rest/api/content").respond(200, json={"results": [{"id": "1", "title": "Index"}]})
+    respx.get("https://wiki.example/rest/api/content/1").respond(200, json={
+        "body": {"storage": {"value": '<ac:structured-macro ac:name="children"/>'}},
+    })
+    with ConfluenceConnector("ENG", base_url="https://wiki.example", token="pat", api_version="v1") as connector:
+        connector.build_manifest()
+        with pytest.raises(SourceFileUnavailable):
+            connector.read_file("", "Index.txt")
+
+
+def test_source_context_path_and_invalid_version():
+    assert parse_confluence_source("confluence:https://wiki.example/confluence/ENG")["base_url"] == "https://wiki.example/confluence"
+    with pytest.raises(ValueError, match="API_VERSION"):
+        ConfluenceConnector("ENG", api_version="v3")
+
+
+@respx.mock
+def test_cli_passes_source_auth_and_can_override_basic_env(monkeypatch):
+    from oikb.cli import _resolve_connector
+    monkeypatch.setenv("CONFLUENCE_USER", "cloud-user")
+    route = respx.get("https://wiki.example/rest/api/content").respond(200, json={"results": []})
+    with _resolve_connector("confluence:ENG", auth={
+        "base_url": "https://wiki.example", "user": "", "token": "pat", "api_version": "v1",
+    }) as connector:
+        connector.build_manifest()
+    assert route.calls[0].request.headers["Authorization"] == "Bearer pat"

tests/test_gdrive.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from typing import Any
+
+from oikb.connectors.gdrive import GDriveConnector
+
+
+class _Request:
+    def __init__(self, payload: dict[str, Any]) -> None:
+        self.payload = payload
+
+    def execute(self) -> dict[str, Any]:
+        return self.payload
+
+
+class _FilesResource:
+    def __init__(self, listings: dict[str, list[dict[str, str]]]) -> None:
+        self.listings = listings
+        self.queried_folders: list[str] = []
+
+    def list(self, **kwargs: Any) -> _Request:
+        parent_id = kwargs["q"].split("'", 2)[1]
+        self.queried_folders.append(parent_id)
+        return _Request({"files": self.listings[parent_id]})
+
+
+class _DriveService:
+    def __init__(self, listings: dict[str, list[dict[str, str]]]) -> None:
+        self.files_resource = _FilesResource(listings)
+
+    def files(self) -> _FilesResource:
+        return self.files_resource
+
+
+def test_build_manifest_recurses_into_subfolders() -> None:
+    service = _DriveService(
+        {
+            "root-folder": [
+                {
+                    "id": "root-file",
+                    "name": "root.txt",
+                    "mimeType": "text/plain",
+                    "md5Checksum": "root-checksum",
+                    "size": "4",
+                },
+                {
+                    "id": "nested-folder",
+                    "name": "nested",
+                    "mimeType": "application/vnd.google-apps.folder",
+                },
+            ],
+            "nested-folder": [
+                {
+                    "id": "nested-file",
+                    "name": "child.txt",
+                    "mimeType": "text/plain",
+                    "md5Checksum": "child-checksum",
+                    "size": "5",
+                }
+            ],
+        }
+    )
+    connector = GDriveConnector.__new__(GDriveConnector)
+    connector.folder_id = "root-folder"
+    connector._service = service
+
+    manifest = connector.build_manifest()
+
+    assert [entry.display_path for entry in manifest] == [
+        "nested/child.txt",
+        "root.txt",
+    ]
+    assert service.files_resource.queried_folders == [
+        "root-folder",
+        "nested-folder",
+    ]

tests/test_kb_sync.py

@@ -0,0 +1,147 @@
+from unittest.mock import AsyncMock, Mock
+
+import httpx
+import pytest
+from click.testing import CliRunner
+
+from oikb.connectors import BaseConnector, ManifestEntry
+from oikb.kb_sync import group_entries_by_kb, run_entries_sync
+from oikb.sync import SyncCancelled, SyncResult
+
+
+class Source(BaseConnector):
+    def __init__(self, content, fail=False):
+        self.content = content
+        self.fail = fail
+        self.closed = False
+
+    def build_manifest(self):
+        if self.fail:
+            raise RuntimeError("scan failed")
+        return [ManifestEntry(name, "", str(len(data)), len(data)) for name, data in self.content.items()]
+
+    def read_file(self, path, filename):
+        assert path == ""  # Destination prefixes must not leak into source reads.
+        return self.content[filename]
+
+    def close(self):
+        self.closed = True
+
+
+def test_combined_manifest_filters_routing_and_auth():
+    sources = [Source({"readme.txt": b"first", "skip.bin": b"ignored"}), Source({"readme.txt": b"second"})]
+    resolver = Mock(side_effect=sources)
+    client = Mock()
+    client.sync_diff.side_effect = lambda kb, manifest: {"added": manifest}
+    entries = [
+        {"source": "confluence:ENG", "kb-id": "kb", "filter": {"include": ["*.txt"]}, "auth": {"token": "one"}},
+        {"source": "confluence:HR", "kb-id": "kb", "auth": {"token": "two"}},
+    ]
+    result = run_entries_sync(client, entries, resolve_connector=resolver, quiet=True)
+    assert result.added == 2
+    assert [e["path"] for e in client.sync_diff.call_args.args[1]] == ["ENG", "HR"]
+    assert [call.kwargs["file_content"] for call in client.upload_file.call_args_list] == [b"first", b"second"]
+    assert [call.kwargs["auth"] for call in resolver.call_args_list] == [{"token": "one"}, {"token": "two"}]
+    client.sync_diff.assert_called_once()
+    client.sync_cleanup.assert_not_called()
+    assert all(source.closed for source in sources)
+
+
+@pytest.mark.parametrize("failure", ["scan", "resolve", "duplicate", "filter"])
+def test_failure_prevents_any_kb_mutation_and_closes_sources(failure):
+    first = Source({"a.txt": b"a"})
+    second = Source({"a.txt" if failure == "duplicate" else "b.txt": b"b"}, fail=failure == "scan")
+    resolver = Mock(side_effect=[first, RuntimeError("resolve failed") if failure == "resolve" else second])
+    entries = [{"source": "one", "kb-id": "kb"}, {"source": "two", "kb-id": "kb"}]
+    if failure == "filter":
+        entries[1]["filter"] = {"max-size": "bad"}
+    client = Mock()
+    with pytest.raises((ValueError, RuntimeError)):
+        run_entries_sync(client, entries, resolve_connector=resolver, quiet=True)
+    assert not client.mock_calls
+    assert first.closed
+    if failure not in {"resolve", "filter"}:
+        assert second.closed
+
+
+def test_single_source_keeps_paths_and_target_path_is_optional():
+    for extra, expected in [({}, ""), ({"target-path": "docs/api"}, "docs/api")]:
+        client = Mock()
+        client.sync_diff.return_value = {}
+        source = Source({"a.txt": b"a"})
+        run_entries_sync(client, [{"source": "confluence:ENG", "kb-id": "kb", **extra}], resolve_connector=lambda *a, **kw: source, quiet=True)
+        assert client.sync_diff.call_args.args[1][0]["path"] == expected
+        assert source.closed
+
+
+@pytest.mark.parametrize("prefix", ["/absolute", "../parent", "a/../b", "a\\b", "a//b"])
+def test_invalid_target_path_fails_before_scan(prefix):
+    resolver = Mock()
+    with pytest.raises(ValueError, match="target-path"):
+        run_entries_sync(Mock(), [{"source": "one", "kb-id": "kb", "target-path": prefix}], resolve_connector=resolver)
+    resolver.assert_not_called()
+
+
+def test_group_validation_and_cancellation():
+    with pytest.raises(ValueError, match="source and kb-id"):
+        group_entries_by_kb([{"source": "one"}])
+    with pytest.raises(ValueError, match="same url and token"):
+        group_entries_by_kb([{"source": "one", "kb-id": "kb", "url": "a"}, {"source": "two", "kb-id": "kb", "url": "b"}])
+    client, resolver = Mock(), Mock()
+    with pytest.raises(SyncCancelled):
+        run_entries_sync(client, [{"source": "one", "kb-id": "kb"}], resolve_connector=resolver, cancel_requested=lambda: True)
+    assert not client.mock_calls and not resolver.mock_calls
+
+
+def test_cli_name_selects_entire_kb_group_and_closes_client(monkeypatch):
+    import oikb.cli as cli
+    import oikb.kb_sync as kb_sync
+    entries = [{"name": "one", "source": "one", "kb-id": "kb"}, {"source": "two", "kb-id": "kb"}, {"source": "other", "kb-id": "other"}]
+    monkeypatch.setattr(cli, "_load_oikb_yaml", lambda: entries)
+    client = Mock()
+    monkeypatch.setattr(cli, "_make_client", lambda *a: client)
+    sync = Mock(return_value=SyncResult())
+    monkeypatch.setattr(kb_sync, "run_entries_sync", sync)
+    result = CliRunner().invoke(cli.cli, ["sync", "--name", "one", "--dry-run"])
+    assert result.exit_code == 0, result.output
+    assert sync.call_args.args[1] == entries[:2]
+    assert sync.call_args.kwargs["dry_run"] is True
+    client.close.assert_called_once()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("trigger", ["webhook", "alias", "kb"])
+async def test_daemon_triggers_include_sibling_sources(monkeypatch, trigger):
+    import oikb.cli as cli
+    import oikb.daemon as daemon
+    import oikb.kb_sync as kb_sync
+    entries = [{"name": "one", "source": "one", "kb-id": "kb"}, {"source": "two", "kb-id": "kb"}]
+    monkeypatch.setattr(daemon, "_entries", entries)
+    monkeypatch.setattr(daemon, "_sync_locks", {})
+    monkeypatch.setattr(daemon, "_scheduler_state", {})
+    monkeypatch.setattr(daemon, "_shutdown_event", None)
+    monkeypatch.setattr(daemon, "_history", None)
+    monkeypatch.setattr(daemon, "_send_notification", AsyncMock())
+    monkeypatch.setattr(daemon, "record_sync", Mock())
+    client = Mock()
+    monkeypatch.setattr(cli, "_make_client", lambda **kw: client)
+    sync = Mock(return_value=SyncResult(added=2))
+    monkeypatch.setattr(kb_sync, "run_entries_sync", sync)
+    if trigger == "webhook":
+        await daemon._run_entry(entries[0])  # The same callback registered with webhooks.
+        assert all(daemon._scheduler_state[s]["files_added"] == 2 for s in ("one", "two"))
+    else:
+        result = await daemon.trigger_sync("one" if trigger == "alias" else "kb", dry_run=True)
+        assert result["result"]["added"] == 2
+        assert daemon._scheduler_state == {}
+    assert sync.call_args.args[1] == entries
+    client.close.assert_called_once()
+
+
+def test_upload_error_includes_server_detail():
+    client = Mock()
+    client.sync_diff.side_effect = lambda kb, manifest: {"added": manifest}
+    response = httpx.Response(400, json={"detail": "extraction failed"}, request=httpx.Request("POST", "https://webui.example/files"))
+    client.upload_file.side_effect = httpx.HTTPStatusError("Bad Request", request=response.request, response=response)
+    result = run_entries_sync(client, [{"source": "one", "kb-id": "kb"}], resolve_connector=lambda *a, **kw: Source({"a.txt": b"a"}), quiet=True)
+    assert "extraction failed" in result.errors[0]