# Zotero KB V1 Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Build a first working Zotero KB service that can create projects, import currently selected Zotero items through a minimal bridge, read local Zotero data and attachments, generate Markdown cards plus JSON indexes, and provide two project-scoped writing endpoints.

**Architecture:** Use a Python FastAPI service with a file-backed workspace, a SQLite-powered Zotero reader, a card builder that normalizes source bundles before LLM generation, and project-scoped views layered on top of a canonical global library. Add a minimal Zotero bridge that exports selected item keys, and ship local SKILL files that read only one target project's content.

**Tech Stack:** Python 3.10, FastAPI, Uvicorn, pytest, sqlite3, pathlib, subprocess (`pdftotext`), standard-library JSON/HTML handling

---

### Task 1: Bootstrap the service and workspace model

**Files:**
- Create: `pyproject.toml`
- Create: `src/zotero_kb/__init__.py`
- Create: `src/zotero_kb/config.py`
- Create: `src/zotero_kb/workspace.py`
- Create: `tests/test_workspace.py`

- [ ] **Step 1: Write the failing workspace tests**

```python
from pathlib import Path

from zotero_kb.config import AppConfig
from zotero_kb.workspace import Workspace


def test_workspace_initialization_creates_required_directories(tmp_path: Path) -> None:
    config = AppConfig(workspace_dir=tmp_path / "workspace", zotero_data_dir=tmp_path / "zotero")
    workspace = Workspace(config)

    workspace.ensure_layout()

    assert (config.workspace_dir / "library" / "collections").is_dir()
    assert (config.workspace_dir / "library" / "index").is_dir()
    assert (config.workspace_dir / "library" / "cache" / "source-bundles").is_dir()
    assert (config.workspace_dir / "projects").is_dir()


def test_create_project_writes_project_files(tmp_path: Path) -> None:
    config = AppConfig(workspace_dir=tmp_path / "workspace", zotero_data_dir=tmp_path / "zotero")
    workspace = Workspace(config)
    workspace.ensure_layout()

    project = workspace.create_project(
        project_id="thesis-ch2",
        name="Thesis Chapter 2",
        llm_provider="openai",
        llm_model="gpt-5-mini",
    )

    assert project.project_id == "thesis-ch2"
    assert (config.workspace_dir / "projects" / "thesis-ch2" / "project.json").is_file()
    assert (config.workspace_dir / "projects" / "thesis-ch2" / "selected-items.json").is_file()
    assert (config.workspace_dir / "projects" / "thesis-ch2" / "project-index.json").is_file()
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/test_workspace.py -q`
Expected: FAIL with `ModuleNotFoundError` for `zotero_kb`

- [ ] **Step 3: Write minimal implementation**

```toml
[project]
name = "zotero-kb"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
  "fastapi>=0.115,<1",
  "uvicorn>=0.30,<1",
  "pydantic>=2.8,<3",
]

[project.optional-dependencies]
dev = [
  "pytest>=8.3,<9",
  "httpx>=0.27,<0.28",
]

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[tool.pytest.ini_options]
pythonpath = ["src"]
```

```python
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class AppConfig:
    workspace_dir: Path
    zotero_data_dir: Path
    bridge_file: Path | None = None
```

```python
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

from .config import AppConfig


@dataclass(frozen=True)
class ProjectRecord:
    project_id: str
    name: str
    project_dir: Path


class Workspace:
    def __init__(self, config: AppConfig) -> None:
        self.config = config

    def ensure_layout(self) -> None:
        for path in (
            self.config.workspace_dir / "library" / "collections",
            self.config.workspace_dir / "library" / "index",
            self.config.workspace_dir / "library" / "cache" / "source-bundles",
            self.config.workspace_dir / "projects",
        ):
            path.mkdir(parents=True, exist_ok=True)

    def create_project(self, project_id: str, name: str, llm_provider: str, llm_model: str) -> ProjectRecord:
        self.ensure_layout()
        project_dir = self.config.workspace_dir / "projects" / project_id
        project_dir.mkdir(parents=True, exist_ok=True)
        payload = {
            "id": project_id,
            "name": name,
            "zotero_data_dir": str(self.config.zotero_data_dir),
            "selection_mode": "zotero-bridge",
            "llm": {"provider": llm_provider, "model": llm_model, "base_url": None},
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
        (project_dir / "project.json").write_text(json.dumps(payload, indent=2), encoding="utf-8")
        (project_dir / "selected-items.json").write_text("[]\n", encoding="utf-8")
        (project_dir / "project-index.json").write_text("{\"items\": []}\n", encoding="utf-8")
        return ProjectRecord(project_id=project_id, name=name, project_dir=project_dir)
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/test_workspace.py -q`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add pyproject.toml src/zotero_kb/__init__.py src/zotero_kb/config.py src/zotero_kb/workspace.py tests/test_workspace.py
git commit -m "feat: bootstrap zotero kb workspace"
```

### Task 2: Implement the Zotero reader and bridge contract

**Files:**
- Create: `src/zotero_kb/zotero_reader.py`
- Create: `src/zotero_kb/bridge.py`
- Create: `tests/fixtures/build_zotero_fixture.py`
- Create: `tests/test_zotero_reader.py`

- [ ] **Step 1: Write the failing reader tests**

```python
from pathlib import Path

from zotero_kb.zotero_reader import ZoteroReader


def test_read_selected_items_from_fixture(tmp_path: Path) -> None:
    fixture_dir = tmp_path / "zotero"
    fixture_dir.mkdir()
    build_fixture_zotero_dir(fixture_dir)
    reader = ZoteroReader(fixture_dir)

    items = reader.read_items(["PAPER0001"])

    assert len(items) == 1
    item = items[0]
    assert item.item_key == "PAPER0001"
    assert item.title == "Card Pipelines for Research Writing"
    assert item.tags == ["llm", "writing"]
    assert item.collection_paths == [["Theory", "Drafting"]]
    assert item.attachment_texts[0].startswith("This paper studies")


def test_read_selected_keys_from_bridge_snapshot(tmp_path: Path) -> None:
    bridge_file = tmp_path / "selected-items.json"
    bridge_file.write_text("{\"selected_keys\": [\"PAPER0001\", \"PAPER0002\"]}", encoding="utf-8")

    assert read_selected_keys(bridge_file) == ["PAPER0001", "PAPER0002"]
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/test_zotero_reader.py -q`
Expected: FAIL because `ZoteroReader` and `read_selected_keys` do not exist

- [ ] **Step 3: Write minimal implementation**

```python
import json
from pathlib import Path


def read_selected_keys(bridge_file: Path) -> list[str]:
    payload = json.loads(bridge_file.read_text(encoding="utf-8"))
    return [str(item) for item in payload.get("selected_keys", [])]
```

```python
@dataclass(frozen=True)
class ZoteroItemRecord:
    item_key: str
    title: str
    creators: list[str]
    year: str | None
    tags: list[str]
    collection_paths: list[list[str]]
    notes: list[str]
    attachment_texts: list[str]
```

```python
class ZoteroReader:
    def __init__(self, zotero_data_dir: Path) -> None:
        self.zotero_data_dir = zotero_data_dir

    def read_items(self, item_keys: list[str]) -> list[ZoteroItemRecord]:
        # query zotero.sqlite for items, creators, tags, notes, collection paths
        # resolve attachments through itemAttachments.path and extract text
        ...
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/test_zotero_reader.py -q`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add src/zotero_kb/zotero_reader.py src/zotero_kb/bridge.py tests/fixtures/build_zotero_fixture.py tests/test_zotero_reader.py
git commit -m "feat: add zotero reader and bridge snapshot support"
```

### Task 3: Build cards and canonical indexes

**Files:**
- Create: `src/zotero_kb/cards.py`
- Create: `src/zotero_kb/llm.py`
- Create: `tests/test_cards.py`

- [ ] **Step 1: Write the failing card-builder tests**

```python
from pathlib import Path

from zotero_kb.cards import CardBuilder
from zotero_kb.zotero_reader import ZoteroItemRecord


def test_build_card_writes_markdown_and_indexes(tmp_path: Path) -> None:
    item = ZoteroItemRecord(
        item_key="PAPER0001",
        title="Card Pipelines for Research Writing",
        creators=["Alice Smith", "Bob Li"],
        year="2024",
        tags=["llm", "writing"],
        collection_paths=[["Theory", "Drafting"]],
        notes=["Merged notes matter."],
        attachment_texts=["This paper studies card pipelines for research writing."],
    )
    builder = CardBuilder(workspace_dir=tmp_path, llm_client=FakeLlmClient())

    result = builder.build_or_update(item)

    assert result.card_path == tmp_path / "library" / "collections" / "Theory" / "Drafting" / "Card Pipelines for Research Writing [PAPER0001].md"
    assert result.card_path.read_text(encoding="utf-8").startswith("---")
    cards_index = json.loads((tmp_path / "library" / "index" / "cards.json").read_text(encoding="utf-8"))
    assert cards_index["PAPER0001"]["title"] == "Card Pipelines for Research Writing"
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/test_cards.py -q`
Expected: FAIL because `CardBuilder` does not exist

- [ ] **Step 3: Write minimal implementation**

```python
class LlmClient(Protocol):
    def generate_card(self, source_bundle: dict[str, object]) -> dict[str, object]:
        ...
```

```python
class CardBuilder:
    def __init__(self, workspace_dir: Path, llm_client: LlmClient) -> None:
        ...

    def build_or_update(self, item: ZoteroItemRecord) -> CardBuildResult:
        # write source bundle
        # compute source_hash
        # ask llm_client for structured sections
        # render markdown card
        # update items.json, cards.json, collections.json
        ...
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/test_cards.py -q`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add src/zotero_kb/cards.py src/zotero_kb/llm.py tests/test_cards.py
git commit -m "feat: build markdown cards and canonical indexes"
```

### Task 4: Add project views and writing services

**Files:**
- Create: `src/zotero_kb/projects.py`
- Create: `src/zotero_kb/writing.py`
- Create: `tests/test_projects.py`
- Create: `tests/test_writing.py`

- [ ] **Step 1: Write the failing project and writing tests**

```python
def test_add_item_to_project_updates_selected_items_and_project_index(tmp_path: Path) -> None:
    ...
    assert payload["selected_items"] == ["PAPER0001"]
    assert payload["cards"][0]["item_key"] == "PAPER0001"


def test_recommend_citations_only_reads_project_items(tmp_path: Path) -> None:
    result = service.recommend_citations(project_id="thesis-ch2", prompt="support scoped retrieval")
    assert [item["item_key"] for item in result["results"]] == ["PAPER0001"]


def test_generate_plan_returns_structured_sections(tmp_path: Path) -> None:
    plan = service.generate_plan(project_id="thesis-ch2", prompt="argue that project scoping improves drafting")
    assert "sections" in plan
    assert plan["sections"][0]["citations"][0]["item_key"] == "PAPER0001"
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/test_projects.py tests/test_writing.py -q`
Expected: FAIL because project and writing services do not exist

- [ ] **Step 3: Write minimal implementation**

```python
class ProjectService:
    def add_items(self, project_id: str, item_keys: list[str]) -> dict[str, object]:
        ...

    def remove_item(self, project_id: str, item_key: str) -> dict[str, object]:
        ...
```

```python
class WritingService:
    def recommend_citations(self, project_id: str, prompt: str) -> dict[str, object]:
        ...

    def generate_plan(self, project_id: str, prompt: str, stance: str | None = None) -> dict[str, object]:
        ...
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/test_projects.py tests/test_writing.py -q`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add src/zotero_kb/projects.py src/zotero_kb/writing.py tests/test_projects.py tests/test_writing.py
git commit -m "feat: add project views and writing services"
```

### Task 5: Expose API and Web console

**Files:**
- Create: `src/zotero_kb/api.py`
- Create: `src/zotero_kb/main.py`
- Create: `src/zotero_kb/templates/index.html`
- Create: `tests/test_api.py`

- [ ] **Step 1: Write the failing API tests**

```python
from fastapi.testclient import TestClient

from zotero_kb.api import create_app


def test_create_project_endpoint(tmp_path: Path) -> None:
    client = TestClient(create_app(make_test_config(tmp_path)))
    response = client.post("/api/projects", json={"project_id": "thesis-ch2", "name": "Thesis Chapter 2"})
    assert response.status_code == 201
    assert response.json()["id"] == "thesis-ch2"


def test_import_selected_items_endpoint(tmp_path: Path) -> None:
    client = TestClient(create_app(make_test_config(tmp_path)))
    response = client.post("/api/projects/thesis-ch2/imports/selected-items")
    assert response.status_code == 200
    assert response.json()["imported_item_keys"] == ["PAPER0001"]
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/test_api.py -q`
Expected: FAIL because `create_app` does not exist

- [ ] **Step 3: Write minimal implementation**

```python
def create_app(config: AppConfig) -> FastAPI:
    app = FastAPI()

    @app.get("/")
    def index() -> HTMLResponse:
        ...

    @app.post("/api/projects", status_code=201)
    def create_project(payload: CreateProjectRequest) -> dict[str, object]:
        ...

    @app.post("/api/projects/{project_id}/imports/selected-items")
    def import_selected_items(project_id: str) -> dict[str, object]:
        ...

    return app
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/test_api.py -q`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add src/zotero_kb/api.py src/zotero_kb/main.py src/zotero_kb/templates/index.html tests/test_api.py
git commit -m "feat: expose zotero kb api and web console"
```

### Task 6: Ship skill files and Zotero bridge scaffold

**Files:**
- Create: `skills/zotero-citation-recommender/SKILL.md`
- Create: `skills/zotero-citation-planner/SKILL.md`
- Create: `zotero-bridge/src/bootstrap.js`
- Create: `zotero-bridge/src/manifest.json`
- Create: `README.md`
- Create: `tests/test_skill_assets.py`

- [ ] **Step 1: Write the failing asset tests**

```python
def test_skill_files_exist() -> None:
    assert Path("skills/zotero-citation-recommender/SKILL.md").is_file()
    assert Path("skills/zotero-citation-planner/SKILL.md").is_file()


def test_bridge_manifest_exists() -> None:
    assert Path("zotero-bridge/src/manifest.json").is_file()
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest tests/test_skill_assets.py -q`
Expected: FAIL because skill and bridge files do not exist

- [ ] **Step 3: Write minimal implementation**

```markdown
# zotero-citation-recommender

Read `projects/<project-id>/project-index.json`, then open only the referenced card files from `library/collections/`. Recommend citations from those files only.
```

```javascript
// Export selected item keys from Zotero into a bridge snapshot file.
async function exportSelectedItems() {
  const selectedItems = Zotero.getMainWindow().ZoteroPane.getSelectedItems();
  ...
}
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest tests/test_skill_assets.py -q`
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add skills/zotero-citation-recommender/SKILL.md skills/zotero-citation-planner/SKILL.md zotero-bridge/src/bootstrap.js zotero-bridge/src/manifest.json README.md tests/test_skill_assets.py
git commit -m "feat: add skills and zotero bridge scaffold"
```