Documenter MCP tools and general support + improved workflow

agents/blueprints/backend/be-documenter.md

@@ -90,29 +90,31 @@ If none: `roboco_agent_idle()`
 - Understand what was built and why
 
 ### 6. WRITE
-**File Paths** - Write documentation to `/app/docs/`:
+Use `roboco_docs_write()` - system handles paths and deduplication automatically:
-- `/app/docs/backend/` - Backend documentation
-- `/app/docs/backend/api/` - API documentation
-- `/app/docs/backend/changelog.md` - Changelog
 
-**API Documentation** (if new/changed endpoints)
+```python
-- Endpoint URL, method
+roboco_docs_write({
-- Request/response schemas
+    "task_id": "your-task-uuid",
-- Example requests/responses
+    "filename": "user-api.md",
-- Error cases
+    "doc_type": "api",  # api, qa, guide, readme, changelog, architecture, design
-
+    "title": "User API Endpoints",
-**README Updates** (if new features)
+    "content": "# User API\n\n## GET /api/users\n..."
-- Feature description
+})
-- Usage examples
-- Configuration options
-
-**Changelog Entry**
-```markdown
-## [version] - YYYY-MM-DD
-### Added/Changed/Fixed
-- {Description}
 ```
 
+**SMART DEDUPLICATION**: System searches RAG for similar existing docs.
+- If similar doc exists → updates it (no duplicates)
+- If no match → creates new doc
+- You don't need to remember paths or check if doc exists
+
+**Doc Types:**
+- `api` - API documentation
+- `qa` - QA test plans
+- `guide` - User guides
+- `readme` - README updates
+- `changelog` - Changelog entries
+- `architecture` - Architecture docs
+
 Update progress: `roboco_task_progress(task_id, "Completed API docs...", 50)`
 
 ### 7. SUBMIT TO PM
@@ -232,6 +234,9 @@ tools:
   - roboco_channel_list, roboco_channel_history
   - roboco_message_send, roboco_message_get, roboco_ask_question
   - roboco_session_history_for_task  # Get discussion history for your task
+  # Documentation (auto-deduplication via RAG)
+  - roboco_docs_write  # Write/update docs (handles dedup automatically)
+  - roboco_docs_read, roboco_docs_list, roboco_docs_delete
 ```
 
 ## Permissions

agents/blueprints/frontend/fe-documenter.md

@@ -90,29 +90,30 @@ If none: `roboco_agent_idle()`
 - Understand usage patterns
 
 ### 6. WRITE
-**File Paths** - Write documentation to `/app/docs/`:
+Use `roboco_docs_write()` - system handles paths and deduplication automatically:
-- `/app/docs/frontend/` - Frontend documentation
-- `/app/docs/frontend/components/` - Component documentation
-- `/app/docs/frontend/changelog.md` - Changelog
 
-**Component Documentation**
+```python
-- Props interface
+roboco_docs_write({
-- Usage examples
+    "task_id": "your-task-uuid",
-- States and variants
+    "filename": "button-component.md",
-- Accessibility notes
+    "doc_type": "api",  # api, qa, guide, readme, changelog, architecture, design
-
+    "title": "Button Component",
-**README Updates** (if new features)
+    "content": "# Button Component\n\n## Props\n..."
-- Feature description
+})
-- Installation/setup
-- Usage examples
-
-**Changelog Entry**
-```markdown
-## [version] - YYYY-MM-DD
-### Added/Changed/Fixed
-- {Description}
 ```
 
+**SMART DEDUPLICATION**: System searches RAG for similar existing docs.
+- If similar doc exists → updates it (no duplicates)
+- If no match → creates new doc
+- You don't need to remember paths or check if doc exists
+
+**Doc Types:**
+- `api` - Component/API documentation
+- `guide` - Usage guides
+- `readme` - README updates
+- `changelog` - Changelog entries
+- `design` - Design documentation
+
 ### 7. SUBMIT TO PM
 `roboco_task_docs_complete(task_id, doc_notes?)` - Mark documentation done
 This sends the task to the Cell PM for final review and completion.
@@ -230,6 +231,9 @@ tools:
   - roboco_channel_list, roboco_channel_history
   - roboco_message_send, roboco_message_get, roboco_ask_question
   - roboco_session_history_for_task  # Get discussion history for your task
+  # Documentation (auto-deduplication via RAG)
+  - roboco_docs_write  # Write/update docs (handles dedup automatically)
+  - roboco_docs_read, roboco_docs_list, roboco_docs_delete
 ```
 
 ## Permissions

agents/blueprints/ux_ui/ux-documenter.md

@@ -90,29 +90,29 @@ If none: `roboco_agent_idle()`
 - Understand usage guidelines
 
 ### 6. WRITE
-**File Paths** - Write documentation to `/app/docs/`:
+Use `roboco_docs_write()` - system handles paths and deduplication automatically:
-- `/app/docs/ux_ui/` - UX/UI documentation
-- `/app/docs/ux_ui/design-system/` - Design system documentation
-- `/app/docs/ux_ui/changelog.md` - Changelog
 
-**Component Guidelines**
+```python
-- When to use this component
+roboco_docs_write({
-- Variants and states
+    "task_id": "your-task-uuid",
-- Dos and don'ts
+    "filename": "button-guidelines.md",
-- Accessibility notes
+    "doc_type": "design",  # api, qa, guide, readme, changelog, architecture, design
-
+    "title": "Button Component Guidelines",
-**Design System Updates**
+    "content": "# Button Guidelines\n\n## When to Use\n..."
-- Token additions/changes
+})
-- Pattern documentation
-- Usage examples
-
-**Changelog Entry**
-```markdown
-## [version] - YYYY-MM-DD
-### Added/Changed/Fixed
-- {Description}
 ```
 
+**SMART DEDUPLICATION**: System searches RAG for similar existing docs.
+- If similar doc exists → updates it (no duplicates)
+- If no match → creates new doc
+- You don't need to remember paths or check if doc exists
+
+**Doc Types:**
+- `design` - Design system documentation
+- `guide` - Usage guidelines
+- `readme` - README updates
+- `changelog` - Changelog entries
+
 ### 7. SUBMIT TO PM
 `roboco_task_docs_complete(task_id, doc_notes?)` - Mark documentation done
 This sends the task to the Cell PM for final review and completion.
@@ -230,6 +230,9 @@ tools:
   - roboco_channel_list, roboco_channel_history
   - roboco_message_send, roboco_message_get, roboco_ask_question
   - roboco_session_history_for_task  # Get discussion history for your task
+  # Documentation (auto-deduplication via RAG)
+  - roboco_docs_write  # Write/update docs (handles dedup automatically)
+  - roboco_docs_read, roboco_docs_list, roboco_docs_delete
 ```
 
 ## Permissions

agents/prompts/roles/documenter.md

@@ -11,7 +11,7 @@ For communication structure: `roboco_kb_search("communication hierarchy")`
 ## Workflow
 
 ```
-SCAN → CLAIM → START → CHECKOUT → GATHER → WRITE → COMMIT → REFLECT → INDEX → SUBMIT
+SCAN → CLAIM → START → CHECKOUT → GATHER → WRITE → COMMIT → REFLECT → VERIFY → SUBMIT
 ```
 
 **You work in PARALLEL with the developer during `awaiting_documentation`.**
@@ -42,9 +42,24 @@ Use `roboco_task_start()` then `roboco_message_send()` to announce.
 4. Review the actual code changes via git
 
 ### 6. WRITE
-Create documentation: API docs, usage examples, architecture notes, README updates.
+Create documentation using `roboco_docs_write()`:
-- Update progress: `roboco_task_progress()`
+
-- Write to correct paths (see "Your Write Access" below)
+```
+roboco_docs_write({
+  task_id: "current-task-uuid",
+  filename: "api-endpoints.md",
+  doc_type: "api",           # api, qa, guide, readme, changelog, architecture, design
+  title: "User API Endpoints",
+  content: "# User API\n\n..."
+})
+```
+
+**SMART DEDUPLICATION**: The system automatically searches for similar existing docs.
+- If similar doc exists → updates it instead of creating duplicate
+- If no similar doc → creates new doc
+- You don't need to remember paths or check if doc exists
+
+Update progress: `roboco_task_progress()`
 
 ### 7. COMMIT (Git Tasks)
 **For tasks with `requires_git=True`:**
@@ -56,8 +71,9 @@ Create documentation: API docs, usage examples, architecture notes, README updat
 ### 8. REFLECT
 Use `roboco_journal_reflect()` before submitting. REQUIRED.
 
-### 9. INDEX
+### 9. VERIFY
-Use `roboco_kb_index_docs()` to make docs searchable. REQUIRED.
+Docs are auto-indexed in RAG when written via `roboco_docs_write()`.
+Use `roboco_docs_list(task_id)` to verify your docs are tracked.
 
 ### 10. SUBMIT
 Use `roboco_task_docs_complete()`. This sets `docs_complete=True`.
@@ -96,6 +112,12 @@ Use `roboco_task_docs_complete()`. This sets `docs_complete=True`.
 - `roboco_kb_search`, `roboco_rag_query`, `roboco_kb_stats`
 - `roboco_kb_index_docs` (index documentation for search)
 
+**Documentation:**
+- `roboco_docs_write(task_id, filename, doc_type, title, content)` - Write/update docs
+- `roboco_docs_read(path)` - Read existing doc
+- `roboco_docs_list(task_id)` - List docs for task
+- `roboco_docs_delete(path)` - Delete doc (rarely needed)
+
 ## NOT Your Tools
 
 - `roboco_task_create`, `roboco_task_assign`, `roboco_task_activate` → PM only
@@ -105,17 +127,6 @@ Use `roboco_task_docs_complete()`. This sets `docs_complete=True`.
 - `roboco_task_qa_pass`, `roboco_task_qa_fail` → QA only
 - `roboco_notify_send` → PM only
 
-## Your Write Access
-
-| Directory | When to Use |
-|-----------|-------------|
-| `/docs/{your-team}/` | Team documentation (APIs, services) |
-| `/docs/features/{your-team}/` | Feature docs for your team's work |
-| `/docs/bugs/{your-team}/` | Bug documentation, root cause analysis |
-| `/docs/features/shared/` | Cross-team feature documentation |
-
-**You CANNOT write to:** `/docs/internal/`, `/docs/standards/`, `/docs/workflows/`, `/docs/self/`, other team directories.
-
 ## Rules
 
 1. **Only claim awaiting_documentation or pending** - Can't claim dev tasks
@@ -123,10 +134,9 @@ Use `roboco_task_docs_complete()`. This sets `docs_complete=True`.
 3. **Message when starting** - Announce to cell
 4. **Read dev's journey** - `roboco_journal_read_team()` required
 5. **Reflect before submit** - `roboco_journal_reflect()` required
-6. **Index your docs** - `roboco_kb_index_docs()` for future search
+6. **Use roboco_docs_write** - System handles paths and deduplication
 7. **Quality docs** - Future developers depend on this
 8. **Cannot complete** - Only PM completes after review
-9. **Write to correct paths** - Use team-scoped directories only
 
 ## CRITICAL: Self-Documentation Prevention
 
@@ -141,5 +151,4 @@ If you try to claim a task where you were the original developer:
 Before critical actions, verify with RAG:
 - **Communication structure**: `roboco_kb_search("communication hierarchy")`
 - **Full workflow example**: `roboco_kb_search("documenter workflow")`
-- **Documentation structure**: `roboco_kb_search("documentation directories")`
 - **Tool parameters**: `roboco_kb_search("mcp tools")`

docs/rag/roles/documenter.md

@@ -19,7 +19,7 @@
 - Claim tasks in `awaiting_documentation` status
 - Claim `pending` tasks (direct documentation tasks from PM)
 - Complete documentation (`docs_complete`)
-- Index documentation: `roboco_kb_index_docs()`
+- Write documentation: `roboco_docs_write()` (auto-indexes in RAG)
 - Search and query knowledge base
 
 ## What You CANNOT Do
@@ -47,9 +47,9 @@ awaiting_documentation → claim → start → write → docs_complete
 |------|---------|
 | `roboco_task_claim` | Take ownership |
 | `roboco_task_start` | Begin documentation |
+| `roboco_docs_write` | Write/update docs (auto-dedup via RAG) |
 | `roboco_task_docs_complete` | Submit for PM review |
 | `roboco_journal_read_team` | Read developer's journey |
-| `roboco_kb_index_docs` | Index new documentation |
 
 ## Gather Context First
 
@@ -66,14 +66,26 @@ roboco_kb_search("similar documentation")
 roboco_channel_history("backend-cell")
 ```
 
-## Documentation Deliverables
+## Writing Documentation
 
-Depending on task, create:
+Use `roboco_docs_write()` - handles paths and deduplication automatically:
-- API documentation
+
-- Usage examples with code snippets
+```python
-- Architecture notes
+roboco_docs_write({
-- README updates
+    "task_id": "your-task-uuid",
-- Changelog entries
+    "filename": "feature-api.md",
+    "doc_type": "api",  # api, qa, guide, readme, changelog, architecture, design
+    "title": "Feature API Documentation",
+    "content": "# Feature API\n\n..."
+})
+```
+
+**SMART DEDUPLICATION**: RAG searches for similar existing docs.
+- If similar doc exists → updates it (no duplicates)
+- If no match → creates new doc
+- Auto-indexed for search
+
+**Doc Types**: `api`, `qa`, `guide`, `readme`, `changelog`, `architecture`, `design`
 
 ## Completing Documentation
 
@@ -105,9 +117,9 @@ If documenter == original_developer, the claim is FORBIDDEN.
 
 ## Before Completing
 
-1. Journal your work: `roboco_journal_entry({type: "documentation"})`
+1. Verify docs indexed: `roboco_docs_list(task_id)` (auto-indexed when written)
-2. Write reflection: `roboco_journal_reflect()`
+2. Journal your work: `roboco_journal_entry({type: "documentation"})`
-3. Index new docs: `roboco_kb_index_docs(["docs/new-feature.md"])`
+3. Write reflection: `roboco_journal_reflect()`
 
 ## Escalation
 

docs/rag/tools/api-endpoints.md

@@ -85,6 +85,15 @@ Base URL: `http://{host}:{port}/api/v1`
 | GET | `/journals/me/stats` | My stats |
 | GET | `/journals/{agent}/entries` | Read team journal |
 
+## Documentation
+
+| Method | Endpoint | Description |
+|--------|----------|-------------|
+| POST | `/docs/write` | Write/update doc (RAG dedup) |
+| GET | `/docs/read` | Read documentation |
+| GET | `/docs/list` | List docs (by task or team) |
+| DELETE | `/docs/delete` | Delete documentation |
+
 ## Knowledge Base
 
 | Method | Endpoint | Description |
@@ -94,7 +103,7 @@ Base URL: `http://{host}:{port}/api/v1`
 | POST | `/optimal/mentor/ask` | Ask mentor |
 | GET | `/optimal/stats` | KB stats |
 | POST | `/optimal/index/code` | Index code |
-| POST | `/optimal/index/docs` | Index docs |
+| POST | `/optimal/index/docs` | Bulk index docs |
 
 ## Health
 

docs/rag/tools/kb-tools.md

@@ -44,7 +44,29 @@ roboco_ask_mentor(
 )
 ```
 
-## Indexing
+## Documentation Writing (Documenter, Cell PM)
+
+```python
+# Write/update documentation (auto-dedup via RAG)
+roboco_docs_write({
+    "task_id": "task-uuid",
+    "filename": "api-endpoints.md",
+    "doc_type": "api",  # api, qa, guide, readme, changelog, architecture, design
+    "title": "API Endpoints",
+    "content": "# API Endpoints\n\n..."
+})
+
+# List docs for a task
+roboco_docs_list(task_id="task-uuid")
+
+# Read a doc
+roboco_docs_read(path="backend/api/endpoints.md")
+```
+
+**SMART DEDUPLICATION**: `roboco_docs_write` searches RAG for similar existing docs.
+If high-similarity match found, updates instead of creating duplicate.
+
+## Bulk Indexing
 
 ```python
 # Index code (PM, Developer)
@@ -53,7 +75,8 @@ roboco_kb_index_code(
     project="roboco"
 )
 
-# Index docs (PM, Documenter)
+# Index docs (PM, Documenter) - for bulk/explicit indexing
+# Note: roboco_docs_write() auto-indexes when writing
 roboco_kb_index_docs(
     sources=["docs/**/*.md"],
     project="roboco"

docs/rag/troubleshooting/common-issues.md

@@ -87,3 +87,36 @@ Fix all issues before submitting.
 - Read your journal for this task
 - Get proactive context: `roboco_get_proactive_context(task_id)`
 - Read channel history for discussions
+
+## Documentation Path Confusion
+
+**Problem**: Unsure where to write documentation
+
+**Solution**: Use `roboco_docs_write()` - system handles paths automatically
+
+```python
+roboco_docs_write({
+    "task_id": "your-task-uuid",
+    "filename": "feature.md",
+    "doc_type": "api",  # api, qa, guide, readme, changelog, architecture, design
+    "title": "Feature Documentation",
+    "content": "..."
+})
+```
+
+- Team folder: Determined from your agent ID
+- Subfolder: Determined by doc_type
+- No path decisions needed
+
+## Documentation Already Exists
+
+**Problem**: Want to update existing doc but created duplicate
+
+**Cause**: Content was too different from existing doc (RAG similarity < 0.75)
+
+**Solution**:
+- Ensure content covers the same topic
+- Or delete duplicate: `roboco_docs_delete(path)`
+- Check existing: `roboco_docs_list(task_id)` or `roboco_kb_search("topic")`
+
+**Note**: `roboco_docs_write()` auto-deduplicates via RAG by **content similarity**. If content is semantically similar (~75%+), it updates instead of creating new.

docs/rag/troubleshooting/kb-errors.md

@@ -57,6 +57,39 @@
 - Check file permissions
 - Verify Ollama is running
 
+## Documentation Write Failed
+
+**Problem**: `roboco_docs_write()` fails
+
+**Causes**:
+1. Invalid doc_type (must be: api, qa, guide, readme, changelog, architecture, design)
+2. Missing required fields (task_id, filename, title, content)
+3. Agent not authorized (only documenter and cell_pm roles)
+4. Task not found
+
+**Solutions**:
+- Verify doc_type is valid
+- Ensure all required fields provided
+- Check your role has write permission
+- Verify task_id exists
+
+## Duplicate Documentation Created
+
+**Problem**: Created duplicate docs instead of updating existing
+
+**Causes**:
+1. Content too different from existing doc (RAG similarity < 0.75)
+2. Doc in different team folder
+3. RAG search failed (but write still succeeded)
+
+**Solutions**:
+- Ensure content covers same topic as existing doc
+- Check existing docs first: `roboco_docs_list(task_id)`
+- Search KB: `roboco_kb_search("topic keywords")`
+- Delete duplicate if needed: `roboco_docs_delete(path)`
+
+**Note**: `roboco_docs_write()` uses RAG to auto-deduplicate by **content similarity** (not just title). If content is semantically similar (>75% similarity), it updates instead of creating new.
+
 ## Cannot Clear Index
 
 **Problem**: "Not authorized to clear index"

roboco/agents/documenter.py

@@ -10,7 +10,6 @@ from datetime import UTC, datetime
 from pathlib import Path
 from uuid import UUID
 
-import aiofiles
 import structlog
 
 from roboco.agents.base import Agent, AgentConfig
@@ -450,21 +449,29 @@ good,complete,clear,helpful,None
         """
         PUBLISH phase: Documentation goes live.
 
-        - Write files to disk
+        - Write docs via API (auto-dedup, auto-index)
         - Link docs to task
         - Update task status
         """
         self.log.info("PUBLISH phase", task_id=str(ctx.task_id))
 
-        # Write documentation files
+        # Write documentation via docs API (handles dedup, indexing, task linking)
         for doc_spec in ctx.documents_needed:
             if doc_spec.content:
                 try:
-                    path = Path(doc_spec.path)
+                    # Use docs API - handles path, dedup, indexing automatically
-                    path.parent.mkdir(parents=True, exist_ok=True)
+                    await self._api_call(
-                    async with aiofiles.open(path, "w") as f:
+                        "POST",
-                        await f.write(doc_spec.content)
+                        "/docs/write",
-                    self.log.info("Published", path=doc_spec.path)
+                        json={
+                            "task_id": str(ctx.task_id),
+                            "filename": Path(doc_spec.path).name,
+                            "doc_type": doc_spec.doc_type.value,
+                            "title": doc_spec.title,
+                            "content": doc_spec.content,
+                        },
+                    )
+                    self.log.info("Published via docs API", path=doc_spec.path)
                 except Exception as e:
                     self.log.error(
                         "Failed to publish", path=doc_spec.path, error=str(e)

roboco/api/app.py

@@ -17,6 +17,7 @@ from roboco.api.routes.a2a import wellknown_router as a2a_wellknown_router
 from roboco.api.routes.agents import router as agents_router
 from roboco.api.routes.channels import router as channels_router
 from roboco.api.routes.dashboard import router as dashboard_router
+from roboco.api.routes.docs import router as docs_router
 from roboco.api.routes.git import router as git_router
 from roboco.api.routes.groups import router as groups_router
 from roboco.api.routes.health import router as health_router
@@ -278,6 +279,13 @@ def create_app() -> FastAPI:
         tags=["Work Sessions"],
     )
 
+    # Documentation
+    app.include_router(
+        docs_router,
+        prefix=f"{api_prefix}/docs",
+        tags=["Documentation"],
+    )
+
     # ==========================================================================
     # WebSocket
     # ==========================================================================

roboco/api/routes/__init__.py

@@ -8,6 +8,7 @@ from roboco.api.routes import (
     agents,
     channels,
     dashboard,
+    docs,
     git,
     health,
     journals,
@@ -28,6 +29,7 @@ __all__ = [
     "agents",
     "channels",
     "dashboard",
+    "docs",
     "git",
     "health",
     "journals",

roboco/api/routes/docs.py

@@ -0,0 +1,234 @@
+"""
+Documentation API Routes
+
+File management endpoints for agent documentation.
+Agents use these to write/read docs without path confusion.
+"""
+
+from uuid import UUID
+
+from fastapi import APIRouter, HTTPException, Query, status
+
+from roboco.agents_config import get_agent_team
+from roboco.api.deps import CurrentAgentContext, DbSession
+from roboco.api.schemas.docs import (
+    DocRefResponse,
+    ListDocsResponse,
+    ReadDocResponse,
+    WriteDocRequest,
+    WriteDocResponse,
+)
+from roboco.services.base import NotFoundError, UnauthorizedError, ValidationError
+from roboco.services.docs import get_docs_service
+
+router = APIRouter()
+
+# Module-level Query defaults
+_list_task_id_query: UUID | None = Query(None, description="Filter by task ID")
+_read_path_query: str = Query(
+    ..., description="Normalized path (e.g., 'backend/api/endpoints.md')"
+)
+
+
+# =============================================================================
+# WRITE ENDPOINT
+# =============================================================================
+
+
+@router.post("/write", response_model=WriteDocResponse)
+async def write_doc(
+    data: WriteDocRequest,
+    db: DbSession,
+    agent: CurrentAgentContext,
+) -> WriteDocResponse:
+    """
+    Write a documentation file.
+
+    Team folder is determined automatically from agent ID.
+    Subfolder is determined by doc_type.
+
+    The document will be:
+    - Written to /app/docs/{team}/{type_folder}/{filename}
+    - Tracked in task.documents for database traceability
+    - Indexed in RAG for searchability
+    """
+    service = get_docs_service(db)
+
+    try:
+        rel_path, doc_ref, is_update = await service.write_doc(
+            agent_id=str(agent.agent_id),
+            req=data,
+        )
+        await db.commit()
+
+        return WriteDocResponse(
+            status="updated" if is_update else "created",
+            path=rel_path,
+            doc_ref=DocRefResponse(
+                path=doc_ref.path,
+                title=doc_ref.title,
+                doc_type=doc_ref.doc_type,
+                version=doc_ref.version,
+                created_by=doc_ref.created_by,
+                created_at=doc_ref.created_at,
+                updated_by=doc_ref.updated_by,
+                updated_at=doc_ref.updated_at,
+            ),
+        )
+    except ValidationError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=e.message,
+        ) from e
+    except UnauthorizedError as e:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=e.message,
+        ) from e
+    except NotFoundError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=e.message,
+        ) from e
+
+
+# =============================================================================
+# READ ENDPOINT
+# =============================================================================
+
+
+@router.get("/read", response_model=ReadDocResponse)
+async def read_doc(
+    db: DbSession,
+    agent: CurrentAgentContext,
+    path: str = _read_path_query,
+) -> ReadDocResponse:
+    """
+    Read a documentation file by path.
+
+    Path should be normalized (e.g., "backend/api/endpoints.md").
+    """
+    service = get_docs_service(db)
+
+    try:
+        content, size_bytes = await service.read_doc(
+            agent_id=str(agent.agent_id),
+            path=path,
+        )
+
+        return ReadDocResponse(
+            path=path,
+            content=content,
+            size_bytes=size_bytes,
+        )
+    except ValidationError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=e.message,
+        ) from e
+    except UnauthorizedError as e:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=e.message,
+        ) from e
+    except NotFoundError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=e.message,
+        ) from e
+
+
+# =============================================================================
+# LIST ENDPOINT
+# =============================================================================
+
+
+@router.get("/list", response_model=ListDocsResponse)
+async def list_docs(
+    db: DbSession,
+    agent: CurrentAgentContext,
+    task_id: UUID | None = _list_task_id_query,
+) -> ListDocsResponse:
+    """
+    List documentation files.
+
+    If task_id is provided, lists docs for that task.
+    Otherwise, lists all docs for the agent's team.
+    """
+    service = get_docs_service(db)
+
+    try:
+        docs = await service.list_docs(
+            agent_id=str(agent.agent_id),
+            task_id=task_id,
+        )
+
+        # Get team for response
+        team = get_agent_team(str(agent.agent_id)) or "unknown"
+
+        return ListDocsResponse(
+            documents=[
+                DocRefResponse(
+                    path=d.path,
+                    title=d.title,
+                    doc_type=d.doc_type,
+                    version=d.version,
+                    created_by=d.created_by,
+                    created_at=d.created_at,
+                    updated_by=d.updated_by,
+                    updated_at=d.updated_at,
+                )
+                for d in docs
+            ],
+            team=team,
+            count=len(docs),
+        )
+    except UnauthorizedError as e:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=e.message,
+        ) from e
+    except NotFoundError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=e.message,
+        ) from e
+
+
+# =============================================================================
+# DELETE ENDPOINT
+# =============================================================================
+
+
+@router.delete("/delete", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_doc(
+    db: DbSession,
+    agent: CurrentAgentContext,
+    path: str = _read_path_query,
+) -> None:
+    """
+    Delete a documentation file.
+    """
+    service = get_docs_service(db)
+
+    try:
+        await service.delete_doc(
+            agent_id=str(agent.agent_id),
+            path=path,
+        )
+        await db.commit()
+    except ValidationError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=e.message,
+        ) from e
+    except UnauthorizedError as e:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=e.message,
+        ) from e
+    except NotFoundError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=e.message,
+        ) from e

roboco/api/schemas/docs.py

@@ -0,0 +1,119 @@
+"""
+Documentation API Schemas
+
+Request/response models for documentation file management endpoints.
+"""
+
+from enum import Enum
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+# =============================================================================
+# ENUMS
+# =============================================================================
+
+
+class DocType(str, Enum):
+    """Documentation file types determining folder placement."""
+
+    API = "api"  # /docs/{team}/api/
+    QA = "qa"  # /docs/{team}/qa/
+    GUIDE = "guide"  # /docs/{team}/guides/
+    README = "readme"  # /docs/{team}/
+    CHANGELOG = "changelog"  # /docs/{team}/
+    ARCHITECTURE = "architecture"  # /docs/{team}/architecture/
+    DESIGN = "design"  # /docs/{team}/design/ (UX/UI)
+
+
+# =============================================================================
+# REQUEST MODELS
+# =============================================================================
+
+
+class WriteDocRequest(BaseModel):
+    """Request to write a documentation file."""
+
+    task_id: UUID = Field(..., description="Task this documentation belongs to")
+    filename: str = Field(
+        ...,
+        min_length=1,
+        max_length=255,
+        description="Filename (e.g., 'endpoints.md') - no path",
+    )
+    doc_type: DocType = Field(..., description="Type determines subfolder placement")
+    title: str = Field(
+        ..., min_length=1, max_length=500, description="Human-readable title"
+    )
+    content: str = Field(..., min_length=1, description="Full markdown content")
+
+
+class ReadDocRequest(BaseModel):
+    """Request to read a documentation file."""
+
+    path: str = Field(
+        ..., description="Normalized path (e.g., 'backend/api/endpoints.md')"
+    )
+
+
+class UpdateDocRequest(BaseModel):
+    """Request to update a documentation file."""
+
+    path: str = Field(
+        ..., description="Normalized path (e.g., 'backend/api/endpoints.md')"
+    )
+    title: str | None = Field(default=None, max_length=500, description="New title")
+    content: str | None = Field(default=None, description="New content")
+
+
+class DeleteDocRequest(BaseModel):
+    """Request to delete a documentation file."""
+
+    path: str = Field(
+        ..., description="Normalized path (e.g., 'backend/api/endpoints.md')"
+    )
+
+
+# =============================================================================
+# RESPONSE MODELS
+# =============================================================================
+
+
+class DocRefResponse(BaseModel):
+    """Response model for a document reference."""
+
+    path: str
+    title: str
+    doc_type: str
+    version: str | None = None
+    created_by: str | None = None
+    created_at: str | None = None
+    updated_by: str | None = None
+    updated_at: str | None = None
+
+    class Config:
+        from_attributes = True
+
+
+class WriteDocResponse(BaseModel):
+    """Response after writing a documentation file."""
+
+    status: str = Field(..., description="'created' or 'updated'")
+    path: str = Field(..., description="Full path where doc was written")
+    doc_ref: DocRefResponse = Field(..., description="The created document reference")
+
+
+class ReadDocResponse(BaseModel):
+    """Response containing documentation file content."""
+
+    path: str
+    content: str
+    size_bytes: int
+
+
+class ListDocsResponse(BaseModel):
+    """Response listing documentation files."""
+
+    documents: list[DocRefResponse]
+    team: str
+    count: int

roboco/mcp/docs_server.py

@@ -0,0 +1,238 @@
+"""
+Documentation MCP Server
+
+Exposes documentation file management tools to Claude Code agents.
+Bypasses Claude Code's file permission system by going through the API.
+
+Tools:
+- roboco_docs_write: Write/update documentation (RAG-based deduplication)
+- roboco_docs_read: Read a documentation file
+- roboco_docs_list: List documentation files
+- roboco_docs_delete: Delete a documentation file
+"""
+
+from typing import Any
+
+from mcp.server.fastmcp import FastMCP
+
+from roboco.mcp.schemas import WriteDocInput
+from roboco.mcp.utils import ApiClient, format_error_response
+
+# =============================================================================
+# HANDLER FUNCTIONS
+# =============================================================================
+
+
+async def _handle_write(
+    data: WriteDocInput,
+    client: ApiClient,
+) -> dict[str, Any]:
+    """Handle documentation write via API."""
+    payload = {
+        "task_id": data.task_id,
+        "filename": data.filename,
+        "doc_type": data.doc_type,
+        "title": data.title,
+        "content": data.content,
+    }
+
+    result, error = await client.post_or_error(
+        "/docs/write",
+        json=payload,
+        error_code="WRITE_FAILED",
+        error_message="Failed to write documentation",
+    )
+    if error or result is None:
+        return error or format_error_response("WRITE_FAILED", "No result")
+
+    status = result.get("status", "created")
+    path = result.get("path")
+    is_update = status == "updated"
+
+    if is_update:
+        guidance = (
+            f"Updated existing documentation at /app/docs/{path}. "
+            "RAG found similar doc and updated it instead of creating duplicate."
+        )
+    else:
+        guidance = (
+            f"Created new documentation at /app/docs/{path}. "
+            "The file has been indexed in RAG and linked to the task."
+        )
+
+    return {
+        "status": status,
+        "path": path,
+        "doc_ref": result.get("doc_ref"),
+        "is_update": is_update,
+        "guidance": guidance,
+    }
+
+
+async def _handle_read(
+    path: str,
+    client: ApiClient,
+) -> dict[str, Any]:
+    """Handle documentation read via API."""
+    result, error = await client.get_or_error(
+        "/docs/read",
+        params={"path": path},
+        error_code="READ_FAILED",
+        error_message="Failed to read documentation",
+    )
+    if error or result is None:
+        return error or format_error_response("READ_FAILED", "No result")
+
+    return {
+        "path": result.get("path"),
+        "content": result.get("content"),
+        "size_bytes": result.get("size_bytes"),
+    }
+
+
+async def _handle_list(
+    task_id: str | None,
+    client: ApiClient,
+) -> dict[str, Any]:
+    """Handle documentation listing via API."""
+    params = {}
+    if task_id:
+        params["task_id"] = task_id
+
+    result, error = await client.get_or_error(
+        "/docs/list",
+        params=params if params else None,
+        error_code="LIST_FAILED",
+        error_message="Failed to list documentation",
+    )
+    if error or result is None:
+        return error or format_error_response("LIST_FAILED", "No result")
+
+    return {
+        "documents": result.get("documents", []),
+        "team": result.get("team"),
+        "count": result.get("count", 0),
+        "guidance": (
+            f"Found {result.get('count', 0)} documentation files. "
+            "Use roboco_docs_read to view contents."
+        ),
+    }
+
+
+async def _handle_delete(
+    path: str,
+    client: ApiClient,
+) -> dict[str, Any]:
+    """Handle documentation deletion via API."""
+    resp = await client.delete(f"/docs/delete?path={path}")
+
+    if not resp.ok:
+        return format_error_response(
+            "DELETE_FAILED",
+            f"Failed to delete documentation: {resp.text}",
+        )
+
+    return {
+        "status": "deleted",
+        "path": path,
+        "guidance": f"Documentation at {path} has been deleted.",
+    }
+
+
+# =============================================================================
+# MCP SERVER FACTORY
+# =============================================================================
+
+
+def create_docs_mcp_server(agent_id: str) -> FastMCP:
+    """
+    Create a Documentation MCP server for a specific agent.
+
+    Args:
+        agent_id: The agent identifier (e.g., "be-doc")
+
+    Returns:
+        Configured FastMCP server
+    """
+    mcp = FastMCP(f"roboco-docs-{agent_id}", json_response=True)
+    client = ApiClient(agent_id)
+
+    @mcp.tool()
+    async def roboco_docs_write(data: WriteDocInput) -> dict[str, Any]:
+        """
+        Write or update documentation for your current task.
+
+        SMART DEDUPLICATION: Before creating a new doc, RAG searches for
+        existing documentation with similar CONTENT (not just title).
+        If a high-similarity match is found, the existing doc is updated
+        instead of creating a duplicate.
+
+        Team folder is determined automatically from your agent ID.
+        Subfolder is determined by doc_type.
+
+        The document will be:
+        - Checked against existing docs via content similarity search
+        - Updated if semantically similar doc exists, or created if new
+        - Tracked in task.documents for database traceability
+        - Indexed in RAG for searchability
+
+        Args:
+            data: WriteDocInput with task_id, filename, doc_type, title, content
+
+        Returns:
+            status: "created" or "updated"
+            is_update: True if existing doc was updated
+        """
+        return await _handle_write(data, client)
+
+    @mcp.tool()
+    async def roboco_docs_read(path: str) -> dict[str, Any]:
+        """
+        Read a documentation file by path.
+
+        Args:
+            path: Normalized path (e.g., "backend/api/endpoints.md")
+        """
+        return await _handle_read(path, client)
+
+    @mcp.tool()
+    async def roboco_docs_list(task_id: str | None = None) -> dict[str, Any]:
+        """
+        List documentation files.
+
+        If task_id is provided, lists docs for that task.
+        Otherwise, lists all docs for your team.
+
+        Args:
+            task_id: Optional task UUID to filter by
+        """
+        return await _handle_list(task_id, client)
+
+    @mcp.tool()
+    async def roboco_docs_delete(path: str) -> dict[str, Any]:
+        """
+        Delete a documentation file.
+
+        Args:
+            path: Normalized path (e.g., "backend/api/endpoints.md")
+        """
+        return await _handle_delete(path, client)
+
+    return mcp
+
+
+# =============================================================================
+# STANDALONE RUNNER
+# =============================================================================
+
+if __name__ == "__main__":
+    import sys
+
+    MIN_ARGS = 2
+    if len(sys.argv) < MIN_ARGS:
+        print("Usage: python docs_server.py <agent_id>")
+        sys.exit(1)
+
+    agent_id_arg = sys.argv[1]
+    server = create_docs_mcp_server(agent_id_arg)
+    server.run()

roboco/mcp/schemas/__init__.py

@@ -307,3 +307,42 @@ class GroupCreateInput(BaseModel):
         le=4,
         description="Access level: 0=CEO, 1=Board, 2=Main PM, 3=Cell PM, 4=Members",
     )
+
+
+# =============================================================================
+# DOCUMENTATION SCHEMAS
+# =============================================================================
+
+
+class WriteDocInput(BaseModel):
+    """Input for writing a documentation file."""
+
+    task_id: str = Field(..., description="Task UUID this documentation belongs to")
+    filename: str = Field(
+        ...,
+        min_length=1,
+        max_length=255,
+        description="Filename (e.g., 'endpoints.md') - no path separators",
+    )
+    doc_type: str = Field(
+        ...,
+        description="Type: api, qa, guide, readme, changelog, architecture, design",
+    )
+    title: str = Field(
+        ...,
+        min_length=1,
+        max_length=500,
+        description="Human-readable title",
+    )
+    content: str = Field(..., min_length=1, description="Full markdown content")
+
+
+class UpdateDocInput(BaseModel):
+    """Input for updating a documentation file."""
+
+    path: str = Field(
+        ...,
+        description="Normalized path to update (e.g., 'backend/api/endpoints.md')",
+    )
+    title: str | None = Field(default=None, description="New title (optional)")
+    content: str | None = Field(default=None, description="New content (optional)")

roboco/models/task.py

@@ -47,6 +47,16 @@ class DocRef(RobocoBase):
         ..., description="Type of document (api, readme, architecture, etc.)"
     )
     version: str | None = Field(default=None, description="Document version")
+    created_by: str | None = Field(default=None, description="Agent slug who created")
+    created_at: str | None = Field(
+        default=None, description="ISO timestamp of creation"
+    )
+    updated_by: str | None = Field(
+        default=None, description="Agent slug who last updated"
+    )
+    updated_at: str | None = Field(
+        default=None, description="ISO timestamp of last update"
+    )
 
 
 class FileRef(RobocoBase):

roboco/runtime/orchestrator.py

@@ -285,6 +285,8 @@ class AgentOrchestrator:
             "mcp__roboco-a2a__*",
             # Test tools - run tests, lint, format
             "mcp__roboco-test__*",
+            # Documentation file management
+            "mcp__roboco-docs__*",
             # File operations for documenters and developers
             # Note: // prefix = absolute path (container paths like /app/docs)
             "Write(//app/docs/**)",
@@ -592,6 +594,7 @@ class AgentOrchestrator:
         - roboco-git: Git operations (role-based at handler level)
         - roboco-a2a: Agent-to-Agent protocol
         - roboco-test: Test/lint/format tools
+        - roboco-docs: Documentation file management
 
         Git context is passed to MCP servers so git tools can use defaults.
         """
@@ -717,6 +720,23 @@ class AgentOrchestrator:
             "env": mcp_env,
         }
 
+        # Docs server - documentation file management
+        # Only for documenter and cell_pm roles (they write docs)
+        # Other roles can read via API but don't need the MCP tools
+        agent_role = get_agent_role(agent_id)
+        if agent_role in ("documenter", "cell_pm"):
+            mcp_servers["roboco-docs"] = {
+                "command": "uv",
+                "args": [
+                    "run",
+                    "python",
+                    "-m",
+                    "roboco.mcp.docs_server",
+                    agent_id,
+                ],
+                "env": mcp_env,
+            }
+
         config: dict[str, Any] = {"mcpServers": mcp_servers}
 
         # Write to shared config directory (mounted in both orchestrator and agents)

roboco/services/docs.py

@@ -0,0 +1,609 @@
+"""
+Documentation Service
+
+Handles documentation file management for agents:
+- Write documentation files with automatic team/path resolution
+- Track documents in task.documents for database traceability
+- Auto-index in RAG for searchability
+"""
+
+import asyncio
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import TYPE_CHECKING, ClassVar
+from uuid import UUID
+
+from sqlalchemy import select
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+from roboco.agents_config import get_agent_role, get_agent_team
+from roboco.api.schemas.docs import WriteDocRequest
+from roboco.db.tables import TaskTable
+from roboco.models.task import DocRef
+from roboco.services.base import (
+    BaseService,
+    NotFoundError,
+    UnauthorizedError,
+    ValidationError,
+)
+
+# =============================================================================
+# CONSTANTS
+# =============================================================================
+
+# Base path for documentation in container
+DOCS_BASE_PATH = Path("/app/docs")
+
+# Team to folder mapping
+TEAM_PATHS: dict[str, str] = {
+    "backend": "backend",
+    "frontend": "frontend",
+    "ux_ui": "ux_ui",
+}
+
+# Doc type to subfolder mapping
+TYPE_SUBFOLDERS: dict[str, str] = {
+    "api": "api",
+    "qa": "qa",
+    "guide": "guides",
+    "readme": "",  # Root of team folder
+    "changelog": "",
+    "architecture": "architecture",
+    "design": "design",
+}
+
+# Roles that can write documentation
+WRITE_ROLES: frozenset[str] = frozenset({"documenter", "cell_pm"})
+
+# Roles that can read documentation
+READ_ROLES: frozenset[str] = frozenset(
+    {
+        "documenter",
+        "cell_pm",
+        "main_pm",
+        "developer",
+        "qa",
+        "auditor",
+        "ceo",
+    }
+)
+
+# Path parsing constants
+_MIN_PATH_PARTS_FOR_SUBFOLDER = 2
+_SUBFOLDER_INDEX = 1
+
+# RAG similarity threshold for doc deduplication
+# Above this score, we update existing doc instead of creating new
+_SIMILARITY_THRESHOLD = 0.75
+
+# Max chars of content to use for similarity search
+_CONTENT_SUMMARY_LENGTH = 500
+
+
+# =============================================================================
+# SERVICE
+# =============================================================================
+
+
+class DocsService(BaseService):
+    """Service for documentation file management."""
+
+    service_name: ClassVar[str] = "docs"
+
+    async def write_doc(
+        self,
+        agent_id: str,
+        req: WriteDocRequest,
+    ) -> tuple[str, DocRef, bool]:
+        """
+        Write or update a documentation file with RAG-based deduplication.
+
+        Before creating a new doc, searches RAG for similar existing docs.
+        If a high-similarity match is found, updates that doc instead.
+
+        Args:
+            agent_id: Agent slug or UUID writing the doc
+            req: Write request with task_id, filename, doc_type, title, content
+
+        Returns:
+            Tuple of (relative_path, DocRef, is_update)
+            - is_update: True if existing doc was updated, False if new created
+
+        Raises:
+            ValidationError: If team unknown or invalid doc_type
+            UnauthorizedError: If agent cannot write docs
+            NotFoundError: If task not found
+        """
+        # 1. Validate agent and get team
+        team = get_agent_team(agent_id)
+        if not team:
+            raise ValidationError(
+                f"Unknown agent team for {agent_id}",
+                field="agent_id",
+            )
+
+        role = get_agent_role(agent_id)
+        if role not in WRITE_ROLES:
+            raise UnauthorizedError(
+                action="write_doc",
+                reason=(
+                    f"Role '{role}' cannot write documentation. "
+                    "Only documenters and cell PMs allowed."
+                ),
+            )
+
+        doc_type = req.doc_type.value
+
+        # 2. Validate doc_type
+        if doc_type not in TYPE_SUBFOLDERS:
+            valid_types = list(TYPE_SUBFOLDERS.keys())
+            raise ValidationError(
+                f"Unknown doc_type: {doc_type}. Valid types: {valid_types}",
+                field="doc_type",
+            )
+
+        # 3. Validate filename (no path traversal)
+        if "/" in req.filename or "\\" in req.filename or ".." in req.filename:
+            raise ValidationError(
+                "Filename cannot contain path separators or '..'",
+                field="filename",
+            )
+
+        # 4. Search RAG for similar existing documentation (by content, not just title)
+        existing_path = await self._find_similar_doc(req.title, req.content, team)
+
+        if existing_path:
+            # UPDATE existing doc instead of creating new
+            return await self._update_existing_doc(
+                agent_id=agent_id,
+                existing_path=existing_path,
+                req=req,
+                doc_type=doc_type,
+            )
+
+        # 5. No similar doc found - create new
+        return await self._create_new_doc(
+            agent_id=agent_id,
+            team=team,
+            req=req,
+            doc_type=doc_type,
+        )
+
+    async def _find_similar_doc(
+        self,
+        title: str,
+        content: str,
+        team: str,
+    ) -> str | None:
+        """
+        Search RAG for existing doc with similar content.
+
+        Uses content (not just title) to find semantically similar documents.
+        Returns the path of the similar doc if found above threshold,
+        None otherwise.
+        """
+        try:
+            from roboco.models.optimal import IndexType, QueryContext
+            from roboco.services.optimal import get_optimal_service
+
+            optimal = await get_optimal_service()
+
+            # Build search query from title + content summary
+            content_summary = (
+                content[:_CONTENT_SUMMARY_LENGTH]
+                if len(content) > _CONTENT_SUMMARY_LENGTH
+                else content
+            )
+            search_query = f"{title}\n\n{content_summary}"
+
+            # Search documentation index by content similarity
+            context = QueryContext(index_types=[IndexType.DOCUMENTATION])
+            results = await optimal.search(
+                query=search_query,
+                context=context,
+                top_k=5,
+            )
+
+            if not results:
+                return None
+
+            # Check for high-similarity match in same team
+            for result in results:
+                if result.score >= _SIMILARITY_THRESHOLD:
+                    # Check if it's in the same team's docs
+                    source = result.source or ""
+                    if source.startswith(f"{team}/") or f"/{team}/" in source:
+                        self.log.info(
+                            "Found similar existing doc by content",
+                            title=title,
+                            existing_path=source,
+                            score=result.score,
+                        )
+                        return source
+
+            return None
+
+        except Exception as e:
+            # RAG search failure shouldn't block doc creation
+            self.log.warning(
+                "RAG search for similar docs failed",
+                title=title,
+                error=str(e),
+            )
+            return None
+
+    async def _create_new_doc(
+        self,
+        agent_id: str,
+        team: str,
+        req: WriteDocRequest,
+        doc_type: str,
+    ) -> tuple[str, DocRef, bool]:
+        """Create a new documentation file."""
+        # Build relative path: {team}/{type_subfolder}/{filename}
+        team_path = TEAM_PATHS.get(team, team)
+        subfolder = TYPE_SUBFOLDERS[doc_type]
+        if subfolder:
+            rel_path = f"{team_path}/{subfolder}/{req.filename}"
+        else:
+            rel_path = f"{team_path}/{req.filename}"
+
+        full_path = DOCS_BASE_PATH / rel_path
+
+        self.log.info(
+            "Creating new documentation",
+            agent_id=agent_id,
+            task_id=str(req.task_id),
+            path=rel_path,
+            doc_type=doc_type,
+        )
+
+        # Write file
+        await self._write_file(full_path, req.content)
+
+        # Create DocRef
+        doc_ref = DocRef(
+            path=rel_path,
+            title=req.title,
+            doc_type=doc_type,
+            created_by=agent_id,
+            created_at=datetime.now(UTC).isoformat(),
+        )
+
+        # Add to task.documents
+        await self._add_doc_to_task(req.task_id, doc_ref)
+
+        # Index in RAG
+        await self._index_doc_in_rag(full_path)
+
+        return rel_path, doc_ref, False  # is_update=False
+
+    async def _update_existing_doc(
+        self,
+        agent_id: str,
+        existing_path: str,
+        req: WriteDocRequest,
+        doc_type: str,
+    ) -> tuple[str, DocRef, bool]:
+        """Update an existing documentation file."""
+        full_path = DOCS_BASE_PATH / existing_path
+
+        self.log.info(
+            "Updating existing documentation (RAG dedup)",
+            agent_id=agent_id,
+            task_id=str(req.task_id),
+            existing_path=existing_path,
+            new_title=req.title,
+        )
+
+        # Write updated content to existing path
+        await self._write_file(full_path, req.content)
+
+        # Get existing DocRef to preserve created_by/created_at
+        existing_ref = await self._get_existing_doc_ref(req.task_id, existing_path)
+
+        now = datetime.now(UTC).isoformat()
+
+        # Create updated DocRef - preserve original creation info, set update info
+        doc_ref = DocRef(
+            path=existing_path,
+            title=req.title,
+            doc_type=doc_type,
+            created_by=existing_ref.created_by if existing_ref else agent_id,
+            created_at=existing_ref.created_at if existing_ref else now,
+            updated_by=agent_id,
+            updated_at=now,
+        )
+
+        # Link to new task (doc evolves across tasks)
+        await self._add_doc_to_task(req.task_id, doc_ref)
+
+        # Re-index in RAG
+        await self._index_doc_in_rag(full_path)
+
+        return existing_path, doc_ref, True  # is_update=True
+
+    async def read_doc(
+        self,
+        agent_id: str,
+        path: str,
+    ) -> tuple[str, int]:
+        """
+        Read a documentation file.
+
+        Args:
+            agent_id: Agent slug or UUID reading the doc
+            path: Normalized path (e.g., "backend/api/endpoints.md")
+
+        Returns:
+            Tuple of (content, size_bytes)
+
+        Raises:
+            UnauthorizedError: If agent cannot read docs
+            NotFoundError: If file not found
+        """
+        # 1. Check read permission
+        role = get_agent_role(agent_id)
+        if role not in READ_ROLES:
+            raise UnauthorizedError(
+                action="read_doc",
+                reason=f"Role '{role}' cannot read documentation.",
+            )
+
+        # 2. Validate path (no traversal)
+        if ".." in path:
+            raise ValidationError(
+                "Path cannot contain '..'",
+                field="path",
+            )
+
+        # 3. Build full path
+        full_path = DOCS_BASE_PATH / path
+
+        # 4. Read file
+        content = await self._read_file(full_path)
+        size_bytes = len(content.encode("utf-8"))
+
+        return content, size_bytes
+
+    async def list_docs(
+        self,
+        agent_id: str,
+        task_id: UUID | None = None,
+    ) -> list[DocRef]:
+        """
+        List documentation files.
+
+        Args:
+            agent_id: Agent slug or UUID
+            task_id: Optional task to filter by
+
+        Returns:
+            List of DocRef objects
+        """
+        # Check read permission
+        role = get_agent_role(agent_id)
+        if role not in READ_ROLES:
+            raise UnauthorizedError(
+                action="list_docs",
+                reason=f"Role '{role}' cannot list documentation.",
+            )
+
+        if task_id:
+            # Get docs for specific task
+            result = await self.session.execute(
+                select(TaskTable).where(TaskTable.id == task_id)
+            )
+            task = result.scalar_one_or_none()
+            if not task:
+                raise NotFoundError("Task", str(task_id))
+            return [DocRef(**d) for d in (task.documents or [])]
+        else:
+            # Get agent's team and list files from filesystem
+            team = get_agent_team(agent_id)
+            if not team:
+                return []
+            return await self._list_docs_for_team(team)
+
+    async def delete_doc(
+        self,
+        agent_id: str,
+        path: str,
+    ) -> bool:
+        """
+        Delete a documentation file.
+
+        Args:
+            agent_id: Agent slug or UUID
+            path: Normalized path (e.g., "backend/api/endpoints.md")
+
+        Returns:
+            True if deleted
+
+        Raises:
+            UnauthorizedError: If agent cannot write docs
+            NotFoundError: If file not found
+        """
+        # 1. Check write permission
+        role = get_agent_role(agent_id)
+        if role not in WRITE_ROLES:
+            raise UnauthorizedError(
+                action="delete_doc",
+                reason=(
+                    f"Role '{role}' cannot delete documentation. "
+                    "Only documenters and cell PMs allowed."
+                ),
+            )
+
+        # 2. Validate path
+        if ".." in path:
+            raise ValidationError("Path cannot contain '..'", field="path")
+
+        # 3. Build full path and verify file exists
+        full_path = DOCS_BASE_PATH / path
+        if not full_path.exists():
+            raise NotFoundError("Documentation file", path)
+
+        # 4. Delete file
+        await self._delete_file(full_path)
+
+        self.log.info("Documentation deleted", agent_id=agent_id, path=path)
+        return True
+
+    def _infer_doc_type(self, path: str) -> str:
+        """Infer doc_type from path."""
+        parts = path.split("/")
+        if len(parts) >= _MIN_PATH_PARTS_FOR_SUBFOLDER:
+            has_subfolder = len(parts) > _MIN_PATH_PARTS_FOR_SUBFOLDER
+            subfolder = parts[_SUBFOLDER_INDEX] if has_subfolder else ""
+            return next(
+                (k for k, v in TYPE_SUBFOLDERS.items() if v == subfolder),
+                "readme",
+            )
+        return "readme"
+
+    # =========================================================================
+    # INTERNAL METHODS
+    # =========================================================================
+
+    async def _write_file(self, path: Path, content: str) -> None:
+        """Write file using thread pool."""
+
+        def _do_write() -> None:
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(content, encoding="utf-8")
+
+        await asyncio.to_thread(_do_write)
+        self.log.debug("File written", path=str(path), size=len(content))
+
+    async def _read_file(self, path: Path) -> str:
+        """Read file using thread pool."""
+
+        def _do_read() -> str:
+            if not path.exists():
+                raise NotFoundError("Documentation file", str(path))
+            return path.read_text(encoding="utf-8")
+
+        return await asyncio.to_thread(_do_read)
+
+    async def _delete_file(self, path: Path) -> None:
+        """Delete file using thread pool."""
+
+        def _do_delete() -> None:
+            if path.exists():
+                path.unlink()
+
+        await asyncio.to_thread(_do_delete)
+        self.log.debug("File deleted", path=str(path))
+
+    async def _get_existing_doc_ref(
+        self, task_id: UUID, path: str
+    ) -> DocRef | None:
+        """Get existing DocRef from task.documents by path."""
+        result = await self.session.execute(
+            select(TaskTable).where(TaskTable.id == task_id)
+        )
+        task = result.scalar_one_or_none()
+        if not task or not task.documents:
+            return None
+
+        for doc in task.documents:
+            if doc.get("path") == path:
+                return DocRef(**doc)
+        return None
+
+    async def _add_doc_to_task(self, task_id: UUID, doc_ref: DocRef) -> None:
+        """Add DocRef to task.documents."""
+        result = await self.session.execute(
+            select(TaskTable).where(TaskTable.id == task_id)
+        )
+        task = result.scalar_one_or_none()
+        if not task:
+            raise NotFoundError("Task", str(task_id))
+
+        # Append to documents list
+        docs = list(task.documents or [])
+
+        # Check if doc with same path exists (update) or new (create)
+        existing_idx = next(
+            (i for i, d in enumerate(docs) if d.get("path") == doc_ref.path),
+            None,
+        )
+
+        if existing_idx is not None:
+            docs[existing_idx] = doc_ref.model_dump()
+            status = "updated"
+        else:
+            docs.append(doc_ref.model_dump())
+            status = "created"
+
+        task.documents = docs
+        self.log.info(
+            f"DocRef {status} in task",
+            task_id=str(task_id),
+            path=doc_ref.path,
+        )
+
+    async def _index_doc_in_rag(self, path: Path) -> None:
+        """Index document in RAG. Failures are logged but don't break flow."""
+        try:
+            # Import here to avoid circular dependency
+            from roboco.services.optimal import get_optimal_service
+
+            optimal = await get_optimal_service()
+            await optimal.index_documentation(sources=[str(path)])
+            self.log.debug("Document indexed in RAG", path=str(path))
+        except Exception as e:
+            # Log but don't fail - RAG indexing is nice-to-have
+            self.log.warning(
+                "Failed to index document in RAG",
+                path=str(path),
+                error=str(e),
+            )
+
+    async def _list_docs_for_team(self, team: str) -> list[DocRef]:
+        """List documentation files for a team from filesystem."""
+        team_path = TEAM_PATHS.get(team, team)
+        base = DOCS_BASE_PATH / team_path
+
+        def _scan_docs() -> list[DocRef]:
+            docs: list[DocRef] = []
+            if not base.exists():
+                return docs
+
+            for md_file in base.rglob("*.md"):
+                rel_path = str(md_file.relative_to(DOCS_BASE_PATH))
+                # Infer doc_type from path
+                parts = rel_path.split("/")
+                if len(parts) >= _MIN_PATH_PARTS_FOR_SUBFOLDER:
+                    has_subfolder = len(parts) > _MIN_PATH_PARTS_FOR_SUBFOLDER
+                    subfolder = parts[_SUBFOLDER_INDEX] if has_subfolder else ""
+                    doc_type = next(
+                        (k for k, v in TYPE_SUBFOLDERS.items() if v == subfolder),
+                        "readme",
+                    )
+                else:
+                    doc_type = "readme"
+
+                docs.append(
+                    DocRef(
+                        path=rel_path,
+                        title=md_file.stem.replace("_", " ").replace("-", " ").title(),
+                        doc_type=doc_type,
+                    )
+                )
+            return docs
+
+        return await asyncio.to_thread(_scan_docs)
+
+
+# =============================================================================
+# FACTORY
+# =============================================================================
+
+
+def get_docs_service(session: "AsyncSession") -> DocsService:
+    """Factory function for DocsService."""
+    return DocsService(session)