# Creating Plugins for RAG2F

Guide to building plugins that extend RAG2F functionality.

This document is written for code agents and humans implementing plugins.
Focus on stable contracts, registration points, and what belongs in a plugin vs. in the core.

## Start Here for Code Agents

If you need the shortest path to a correct plugin, read
[plugin_agent_playbook.txt](./plugin_agent_playbook.txt) first.

Use this file for:
- larger examples
- lifecycle and hook reference
- packaging examples
- publishing notes

## Runtime Facts That Matter During Implementation

These points are easy to miss if you only read examples:

- `plugin.id` is the plugin folder name.
- Morpheus scans Python files recursively inside the plugin folder.
- `activated()` runs during plugin discovery.
- discovery order is entry points first, filesystem second.
- duplicate registry keys are rejected by ATeam, OptimusPrime, and XFiles.
- `XFiles.execute_register()` and `XFiles.execute_get()` return Result objects.
- a repository must satisfy `BaseRepository`.
- an agent adapter must satisfy `AgentAdapter`.
- if `pyproject.toml` exists, plugin activation installs from it before considering `requirements.txt`.

## Plugin Architecture

RAG2F Core contains these modules: Johnny5, IndianaJones, OptimusPrime, XFiles, ATeam. All modules connect to Morpheus (the hook system). Plugins extend the system by registering hooks with Morpheus and by registering concrete runtime objects during plugin activation.

**Plugin types:**
- **Embedder plugins** — register embedders during plugin activation (OptimusPrime registry)
- **Repository plugins** — register repositories during plugin activation (XFiles registry)
- **Agent plugins** — register one or more agents during plugin activation (ATeam registry)
- **Custom plugins** — extend any processing pipeline via hooks

## Minimal Plugin Structure

```
my_plugin/
├── __init__.py          # Required (can be empty)
├── plugin.json          # Optional plugin metadata (recommended)
├── pyproject.toml       # Optional packaging metadata (recommended for distributable plugins)
├── plugin_context.py    # Recommended: store plugin_id in ContextVar (template pattern)
└── hooks.py             # Hook implementations
```

### plugin.json (recommended)

`plugin_id` is the plugin folder name. `plugin.json` is optional metadata.

```json
{
    "name": "My Plugin",
    "version": "1.0.0",
    "description": "What this plugin does",
    "keywords": ["rag2f", "plugin"],
    "author": {"name": "Your Name", "email": "you@example.com"}
}
```

### hooks.py

```python
from rag2f.core.morpheus.decorators import hook

@hook("handle_text_foreground", priority=10)
def my_handler(done, track_id, text, *, rag2f):
    """Process incoming text."""
    if done:
        return done  # Already handled by higher-priority hook
    
    # Your processing logic
    process_text(track_id, text)
    
    return True  # Mark as handled
```

---

## Hook Types

### 1. Plugin Lifecycle Overrides (activated/deactivated)

Registration of embedders, repositories, and agents is done in plugin lifecycle overrides,
not via hooks.

```python
from rag2f.core.morpheus.decorators.plugin_decorator import plugin

from .plugin_context import set_plugin_id

@plugin
def activated(plugin, rag2f_instance):
    plugin_id = plugin.id
    set_plugin_id(plugin_id)
    config = rag2f_instance.spock.get_plugin_config(plugin_id) or {}

    # Example: register embedder
    embedder = MyEmbedder(api_key=config.get("api_key"))
    # Choose a stable registry key (often plugin_id). Ensure rag2f.embedder_default matches.
    rag2f_instance.optimus_prime.register(plugin_id, embedder)

    # Example: register an agent adapter
    writer = MyAgentAdapter(raw_agent=MyProviderAgent(...))
    rag2f_instance.a_team.register(
        f"{plugin_id}.writer",
        writer,
        plugin_id=plugin_id,
        is_default=True,
    )

@plugin
def deactivated(plugin, rag2f_instance):
    """Optional cleanup on deactivation."""
    # Example: unregister embedder/repo if you want
    # rag2f_instance.optimus_prime.unregister(plugin.id)
    return
```

### Plugin context (`plugin_context.py`) and `plugin_id`

The plugin template is expected to include a `plugin_context.py` module that stores the
current `plugin_id` in a `ContextVar`.

Why it matters:
- In `@plugin` lifecycle overrides, you already have `plugin.id`.
- In `@hook` functions you can compute the id via `rag2f.morpheus.self_plugin_id()`, but a
    cached `ContextVar` is often simpler and avoids repeating that logic.

Typical pattern:
1. In `activated()`: call `set_plugin_id(plugin.id)`.
2. In hooks: call `get_plugin_id(rag2f)` (falls back to `self_plugin_id()` if needed).

This makes config access consistent:

```python
plugin_id = get_plugin_id(rag2f)
config = rag2f.spock.get_plugin_config(plugin_id) or {}
```

### 2. Processing Hooks (Piped Value)

First argument is piped through all hooks in priority order.

```python
@hook("handle_text_foreground", priority=5)
def process_text(done: bool, track_id: str, text: str, *, rag2f):
    """Process text if not already handled."""
    if done:
        return done
    
    # Store in vector DB
    embedder = rag2f.optimus_prime.get_default()
    vector = embedder.getEmbedding(text)
    
    repo_result = rag2f.xfiles.execute_get("vectors")
    if repo_result.repository:
        repo_result.repository.insert(track_id, vector, {"text": text})
    
    return True
```

### 3. Transform Hooks (Modify and Return)

Transform data flowing through the pipeline.

```python
@hook("indiana_jones_retrieve", priority=5)
def enrich_results(result, query, k, return_mode, for_synthesize, *, rag2f):
    """Add extra metadata to retrieved items."""
    for item in result.items:
        item.extra["enriched_at"] = datetime.now().isoformat()
    return result
```

---

## Available Hooks

| Hook Name | Module | Purpose | Signature |
|-----------|--------|---------|-----------|
| `get_id_input_text` | Johnny5 | Generate input ID | `(track_id, text, *, rag2f)` |
| `check_duplicated_input_text` | Johnny5 | Check duplicate | `(duplicated, track_id, text, *, rag2f)` |
| `handle_text_foreground` | Johnny5 | Process input | `(done, track_id, text, *, rag2f)` |
| `indiana_jones_retrieve` | IndianaJones | Retrieval | `(result, query, k, return_mode, for_synthesize, *, rag2f)` |
| `indiana_jones_synthesize` | IndianaJones | Synthesize | `(result, retrieve_result, return_mode, kwargs, *, rag2f)` |
| `agent_collect_prompt_fragments` | ATeam | Collect prompt fragments | `(fragments, context, *, rag2f)` |
| `agent_finalize_prompt` | ATeam | Final prompt rewrite | `(resolved_prompt, context, *, rag2f)` |

### ATeam plugin pattern

Use ATeam when a plugin needs to expose LLM agents or prompt-driven agents.

Recommended key pattern:
- `plugin_id.default`
- `plugin_id.writer`
- `plugin_id.reviewer`
- `plugin_id.router`

Recommended flow:
1. create provider-specific raw agents inside `activated()`
2. wrap them in lightweight adapters implementing `AgentAdapter`
3. register them through `rag2f_instance.a_team.register(...)`
4. optionally add `agent_collect_prompt_fragments` and `agent_finalize_prompt` hooks
5. clean up registrations in `deactivated()`

Minimal example:

```python
from rag2f.core.morpheus.decorators.plugin_decorator import plugin


class MyRawAgent:
    def __init__(self, model: str):
        self.model = model


class MyAgentAdapter:
    def __init__(self, raw_agent):
        self._raw_agent = raw_agent

    @property
    def raw(self):
        return self._raw_agent

    @property
    def metadata(self):
        return {"provider": "my_sdk", "model": self._raw_agent.model}

    def run(self, request, *, rag2f):
        return self._raw_agent.generate(request.prompt.text)

    async def run_async(self, request, *, rag2f):
        return await self._raw_agent.generate_async(request.prompt.text)


@plugin
def activated(plugin, rag2f_instance):
    writer = MyAgentAdapter(MyRawAgent("writer-model"))
    rag2f_instance.a_team.register(
        f"{plugin.id}.writer",
        writer,
        plugin_id=plugin.id,
        is_default=True,
    )
```

If prompt logic belongs to the plugin, implement hooks:

```python
from rag2f.core.dto import PromptFragment
from rag2f.core.morpheus.decorators import hook


@hook("agent_collect_prompt_fragments", priority=20)
def add_prompt_parts(fragments, context, rag2f):
    if context.purpose == "draft":
        fragments.append(
            PromptFragment(
                text="Write a concise draft for engineers.",
                plugin_id=context.plugin_id,
            )
        )
    return fragments
```

## Johnny5 and FluxCapacitor

If a plugin starts a FluxCapacitor workflow from Johnny5, keep `track_id` as the public identifier.

- generate `track_id` in `get_id_input_text`
- process or schedule work in `handle_text_foreground`
- if external callers must ask for async status later, keep the plugin-specific mapping from `track_id` to any internal Flux task ids

This keeps the Johnny5 API stable and avoids leaking internal workflow ids into the public contract.

---

## Complete Plugin Example: Vector Store

```

rag2f_qdrant_plugin/
├── __init__.py
├── plugin.json
├── plugin_overrides.py
├── hooks.py
└── requirements.txt
```

### plugin.json

```json
{
    "name": "Qdrant Vector Store",
    "version": "1.0.0",
    "description": "Qdrant-backed vector repository for rag2f"
}
```

### requirements.txt

```
qdrant-client>=1.7.0
```

### plugin_overrides.py (registration during activation)

```python
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

from rag2f.core.morpheus.decorators.plugin_decorator import plugin

from .hooks import QdrantRepository


@plugin
def activated(plugin, rag2f_instance):
    """Register Qdrant repository when the plugin is activated."""
    plugin_id = plugin.id
    config = rag2f_instance.spock.get_plugin_config(plugin_id) or {}
    if not config:
        return

    client = QdrantClient(
        host=config.get("host", "localhost"),
        port=config.get("port", 6333),
    )

    collection = config.get("collection", "documents")
    vector_size = config.get("vector_size", 1536)

    # Ensure collection exists
    try:
        client.get_collection(collection)
    except Exception:
        client.create_collection(
            collection_name=collection,
            vectors_config=VectorParams(size=vector_size, distance=Distance.COSINE),
        )

    repo = QdrantRepository(client, collection, vector_size)
    rag2f_instance.xfiles.execute_register(
        "qdrant",
        repo,
        meta={"type": "vector", "backend": "qdrant", "plugin": plugin_id},
    )
```

### hooks.py (pipelines)

```python
"""Qdrant vector store plugin for RAG2F."""

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

from rag2f.core.morpheus.decorators import hook
from rag2f.core.xfiles.capabilities import Capabilities, NativeCapability, VectorSearchCapability
from rag2f.core.xfiles.repository import BaseRepository, RepositoryNativeMixin


class QdrantRepository(RepositoryNativeMixin, BaseRepository):
    """Qdrant-backed vector repository."""
    
    def __init__(self, client: QdrantClient, collection: str, vector_size: int):
        self._client = client
        self._collection = collection
        self._vector_size = vector_size
    
    def capabilities(self) -> Capabilities:
        return Capabilities(
            vector_search=VectorSearchCapability(
                supported=True,
                dimensions=self._vector_size,
                distance_metrics=("cosine",),
            ),
            native=NativeCapability(supported=True, kinds=("primary",)),
        )
    
    @property
    def name(self) -> str:
        return "qdrant"

    def get(self, id: str, select: list[str] | None = None) -> dict:
        raise NotImplementedError()

    def update(self, id: str, patch: dict) -> None:
        raise NotImplementedError()

    def delete(self, id: str) -> None:
        raise NotImplementedError()

    def _get_native_handle(self, kind: str) -> object:
        return self._client
    
    def vector_search(self, vector: list[float], k: int = 10) -> list[dict]:
        results = self._client.search(
            collection_name=self._collection,
            query_vector=vector,
            limit=k
        )
        return [{"id": r.id, "score": r.score, **r.payload} for r in results]
    
    def insert(self, id: str, item: dict) -> None:
        vector = item["vector"]
        payload = {k: v for k, v in item.items() if k != "vector"}
        self._client.upsert(
            collection_name=self._collection,
            points=[PointStruct(id=id, vector=vector, payload=payload)]
        )


@hook("handle_text_foreground", priority=5)
def store_in_qdrant(done, track_id, text, *, rag2f):
    """Store text embeddings in Qdrant."""
    if done:
        return done
    
    # Get Qdrant repo
    result = rag2f.xfiles.execute_get("qdrant")
    if not result.repository:
        return False
    
    # Get embedder and embed text
    try:
        embedder = rag2f.optimus_prime.get_default()
        vector = embedder.getEmbedding(text)
    except LookupError:
        return False  # No embedder available
    
    # Store
    result.repository.insert(track_id, {"text": text, "vector": vector})
    return True


@hook("indiana_jones_retrieve", priority=10)
def retrieve_from_qdrant(result, query, k, return_mode, for_synthesize, *, rag2f):
    """Retrieve from Qdrant."""
    from rag2f.core.dto.indiana_jones_dto import RetrieveResult, RetrievedItem
    
    repo_result = rag2f.xfiles.execute_get("qdrant")
    if not repo_result.repository:
        return result
    
    try:
        embedder = rag2f.optimus_prime.get_default()
        query_vector = embedder.getEmbedding(query)
    except LookupError:
        return result
    
    results = repo_result.repository.vector_search(query_vector, k)
    
    items = [
        RetrievedItem(
            id=r["id"],
            text=r.get("text", ""),
            score=r.get("score"),
            metadata={"source": "qdrant"}
        )
        for r in results
    ]
    
    return RetrieveResult.success(query=query, items=items)
```

---

## Configuration for Your Plugin

### config.json

```json
{
  "plugins": {
        "rag2f_qdrant_plugin": {
      "host": "localhost",
      "port": 6333,
      "collection": "documents",
      "vector_size": 1536
    }
  }
}
```

### Accessing Config in Hooks

```python
from rag2f.core.morpheus.decorators import hook

@hook("handle_text_foreground", priority=5)
def my_hook(done, track_id, text, *, rag2f):
    plugin_id = rag2f.morpheus.self_plugin_id()
    config = rag2f.spock.get_plugin_config(plugin_id) or {}
    api_key = config.get("api_key")
    timeout = (config.get("http") or {}).get("timeout")
    return done
```

---

## Hook Priority Guidelines

| Priority | Use Case |
|----------|----------|
| 100+ | Override everything (testing) |
| 50-99 | Primary implementation |
| 10-49 | Standard plugins |
| 1-9 | Fallback/default behavior |

Higher priority executes **first**.

---

## Optional: Testing Your Plugin

See [testing.txt](./testing.txt) for complete testing guide.

```python
import pytest
import pytest_asyncio
from rag2f.core.rag2f import RAG2F

@pytest_asyncio.fixture
async def rag2f_with_plugin():
    return await RAG2F.create(
        plugins_folder="./plugins/",
        config={
            "plugins": {
                "my_plugin": {"setting": "value"}
            }
        }
    )

@pytest.mark.asyncio
async def test_plugin_hooks(rag2f_with_plugin):
    assert rag2f_with_plugin.morpheus.plugin_exists("my_plugin")
    
    result = rag2f_with_plugin.johnny5.execute_handle_text_foreground("test")
    assert result.is_ok()
```

The `plugins_folder` must point to the parent folder that contains plugin directories.
Do not pass the plugin directory itself.

---

## Publishing Your Plugin

### Via PyPI (Recommended)

1. Create `pyproject.toml` with entry point:

```toml
[project]
name = "rag2f-my-plugin"
version = "1.0.0"
dependencies = ["rag2f>=0.1.0"]

[project.entry-points."rag2f.plugins"]
my_plugin = "rag2f_my_plugin:get_plugin_path"
```

2. Add path function to `__init__.py`:

```python
from pathlib import Path

def get_plugin_path() -> str:
    return str(Path(__file__).parent)
```

3. Publish:

```bash
pip install build twine
python -m build
twine upload dist/*
```

### Local Development

Just place in `plugins/` folder — RAG2F discovers it automatically.
