# OpenContracts LLM Framework

Last Updated: 2026-01-09

OpenContract's API for creating document and corpus agents.

## Philosophy

- **Extensible Framework Design**: Built on PydanticAI with a pluggable architecture for future framework adapters
- **Rich Responses**: Every interaction returns structured data with sources, metadata, and conversation tracking
- **Conversation Management**: Persistent conversations with automatic message storage and retrieval
- **Tool Integration**: Extensible tool system for document analysis and data retrieval
- **Type Safety**: Full type hints and structured responses throughout
- **Optimized Extraction**: Leverages pydantic_ai's native capabilities for reliable structured data extraction

## Quick Start

```python
from opencontractserver.llms import agents
# Assuming you have a document_id and corpus_id or objects
# from opencontractserver.corpuses.models import Corpus
# from opencontractserver.documents.models import Document
# document_obj = Document.objects.get(id=123)
# corpus_obj = Corpus.objects.get(id=1)

# Create a document agent
# Note: The `corpus` parameter is optional - use None for documents not in a corpus
agent = await agents.for_document(document=123, corpus=1) # Replace 1 with your actual corpus_id or object, or use None

# Chat with rich responses
response = await agent.chat("What are the key terms in this contract?")
print(f"Response: {response.content}")
print(f"Sources: {len(response.sources)} found")
print(f"Message ID: {response.llm_message_id}")

# Stream responses
async for chunk in agent.stream("Summarize the main obligations"):
    print(chunk.content, end="")

# NEW: Structured data extraction (one-shot, no conversation persistence)
from pydantic import BaseModel, Field

class ContractDates(BaseModel):
    effective_date: str = Field(description="Contract effective date")
    expiration_date: str = Field(description="Contract expiration date")

dates = await agent.structured_response(
    "Extract the contract dates",
    ContractDates
)
if dates:
    print(f"Effective: {dates.effective_date}")
    print(f"Expires: {dates.expiration_date}")

# NEW: Document agents now work without corpus!
# Perfect for analyzing standalone documents or one-off extractions
standalone_agent = await agents.for_document(document=456, corpus=None)
simple_result = await standalone_agent.structured_response(
    "What type of document is this?", 
    str
)
print(f"Document type: {simple_result}")
```

## Core Concepts

### High-Level APIs

The `opencontractserver.llms` module provides several high-level API entry points:

- **`agents`**: (`AgentAPI`) For creating and interacting with document and corpus agents. This is the most common entry point. Also provides convenience methods for structured data extraction.
- **`embeddings`**: (`EmbeddingAPI`) For generating text embeddings.
- **`vector_stores`**: (`VectorStoreAPI`) For creating and interacting with vector stores for similarity search.
- **`tools`**: (`ToolAPI`) For creating and managing `CoreTool` instances.

```python
from opencontractserver.llms import agents, embeddings, vector_stores, tools
from opencontractserver.llms.tools.tool_factory import CoreTool

# Example: Creating a tool using the ToolAPI
def my_custom_function(text: str) -> str:
    """A simple custom tool."""
    return f"Processed: {text}"

custom_tool = tools.from_function(
    func=my_custom_function,
    name="MyCustomTool",
    description="A demonstration tool."
)
# This custom_tool can then be passed to an agent.

# Example: Using AgentAPI convenience methods for structured extraction
result = await agents.get_structured_response_from_document(
    document=123,
    corpus=1,  # Or None for standalone documents
    prompt="Extract key contract terms",
    target_type=ContractTerms,
    framework=AgentFramework.PYDANTIC_AI,
    user_id=456  # Optional
)

# Or extract from an entire corpus
insights = await agents.get_structured_response_from_corpus(
    corpus=1,
    prompt="Analyze patterns across all contracts",
    target_type=CorpusInsights,
    framework=AgentFramework.PYDANTIC_AI
)
```

### Agents

Agents are the primary interface for interacting with documents and corpora. They provide:

- **Document Agents**: Work with individual documents (corpus optional; pass None for standalone documents).
- **Corpus Agents**: Work with collections of documents.
- **Framework Flexibility**: Built on PydanticAI with extensible architecture for custom framework adapters.
- **Conversation Persistence**: Automatic conversation management and message storage.
- **Structured Data Extraction**: One-shot typed data extraction without conversation persistence.
- **Nested Streaming**: Real-time visibility into child agent execution through stream observers.

#### Creating Agents

```python
from opencontractserver.llms import agents
from opencontractserver.llms.types import AgentFramework
# from opencontractserver.corpuses.models import Corpus # For corpus_obj
# from opencontractserver.documents.models import Document # For document_obj
# corpus_obj = Corpus.objects.get(id=1) # Example corpus
# document_obj = Document.objects.get(id=123) # Example document

# Document agent with default framework (PydanticAI)
# The `corpus` parameter is optional - can be None for standalone documents
agent = await agents.for_document(document=123, corpus=1) # Use actual document/corpus IDs or objects

# Document agent for standalone document (not in any corpus)
standalone_agent = await agents.for_document(document=123, corpus=None)
# Corpus-dependent tools are automatically filtered out - agent still works!

# Corpus agent with specific framework
agent = await agents.for_corpus(
    corpus=456, # Use actual corpus ID or object
    framework=AgentFramework.PYDANTIC_AI
)

# With custom configuration
# The `corpus` parameter is optional for document agents (use None for standalone).
agent = await agents.for_document(
    document=123, # Use actual document ID or object
    corpus=1,     # Use actual corpus ID or object
    user_id=789,
    system_prompt="You are a legal contract analyzer...",  # Note: Completely overrides default prompt
    model="gpt-4",
    temperature=0.1,
    tools=["load_md_summary", "get_notes_for_document_corpus"]
)

# Advanced: With stream observer for nested agent visibility
async def my_stream_observer(event):
    """Receives events from nested agent calls."""
    print(f"[Nested] {event.type}: {getattr(event, 'content', getattr(event, 'thought', ''))}")

agent = await agents.for_corpus(
    corpus=456,
    framework=AgentFramework.PYDANTIC_AI,
    stream_observer=my_stream_observer  # Will receive events from child document agents
)

# Advanced: Using existing conversation or preloaded messages
# from opencontractserver.conversations.models import Conversation, ChatMessage
# existing_conversation = Conversation.objects.aget(id=your_conversation_id)
# preloaded_messages = await ChatMessage.objects.filter(conversation_id=your_conversation_id).order_by('''created_at''')
# agent = await agents.for_document(
#     document=123,
#     corpus=1,
#     user_id=789,
#     conversation=existing_conversation, # Optionally pass an existing Conversation object
#     # loaded_messages=list(preloaded_messages), # Optionally pass preloaded messages
#     # override_conversation=True # Set to True to use only loaded_messages and ignore others from conversation object
# )

```

#### Agent Responses

All agent interactions return rich, structured responses with complete metadata:

```python
# UnifiedChatResponse structure
response = await agent.chat("What are the payment terms?")

response.content              # The LLM's response text
response.sources              # List of SourceNode objects with citations
response.user_message_id      # ID of stored user message (if persistence enabled)
response.llm_message_id       # ID of stored LLM response (if persistence enabled)  
response.metadata             # Additional response metadata (framework-specific)

# New, event-based streaming (>= v0.9)
# -----------------------------------
#
# The streaming API now yields a *typed* event union instead of a single
# response shape.  Each event has a ``type`` discriminator so it's trivial to
# branch logic without ``isinstance`` checks.
#
#     ThoughtEvent  – short messages about the agent's reasoning (e.g. tool call
#                     decisions, framework-specific "thinking" lines).
#     ContentEvent  – textual delta that forms part of the **final** answer.
#     SourceEvent   – a batch of SourceNode objects discovered mid-stream.
#     ApprovalNeededEvent – tool requires human approval before execution.
#     ApprovalResultEvent – approval decision has been recorded.
#     ResumeEvent   – execution is resuming after approval.
#     FinalEvent    – emitted once; contains the full answer, sources, usage…
#
# All events carry the legacy fields (``user_message_id``, ``llm_message_id``,
# ``content``/``is_complete``) so existing websocket code keeps working.
#
# Example:
# ```python
# async for ev in agent.stream("Analyze the liability clauses"):
#     if ev.type == "thought":
#         print(f"🤔 {ev.thought}")
#     elif ev.type == "content":
#         print(ev.content, end="")
#     elif ev.type == "sources":
#         print(f"\nFound {len(ev.sources)} sources so far…")
#     elif ev.type == "approval_needed":
#         print(f"⚠️ Tool '{ev.pending_tool_call['name']}' needs approval")
#     elif ev.type == "final":
#         print("\nDone! Total tokens:", ev.metadata.get("usage", {}).get("total_tokens"))
# ```
#
# Legacy (pre-v0.9) – UnifiedStreamResponse
# ----------------------------------------
#
# Legacy code using ``UnifiedStreamResponse`` can be updated to use the new event-based
# streaming. Your code can support both by simply checking ``hasattr(chunk, "type")``
# and falling back to the old attributes when the discriminator is absent.

# SourceNode structure (individual source)
for source in response.sources:
    source.annotation_id      # Database ID of the source annotation
    source.content           # Raw text content of the annotation
    source.similarity_score  # Relevance score (0.0 to 1.0)
    source.metadata         # Dict with document_id, corpus_id, page, annotation_label, etc.
    
    # Convenience method for serialization
    source_dict = source.to_dict()  # Returns flattened dict for storage/transmission

# Note: conversation_id is available via agent.get_conversation_id()
```

#### Structured Data Extraction

The framework provides a powerful **structured response API** for one-shot data extraction without conversation persistence. This is perfect for:
- Form filling and data extraction pipelines
- API endpoints that need structured output
- Batch processing where conversation context isn't needed
- Integration with external systems expecting specific data schemas

```python
from pydantic import BaseModel, Field
from typing import List, Optional

# Define your target schema
class ContractParty(BaseModel):
    name: str = Field(description="Full legal name")
    role: str = Field(description="Role (e.g., 'Buyer', 'Seller')")
    address: Optional[str] = Field(None, description="Address if mentioned")

class ContractAnalysis(BaseModel):
    title: str = Field(description="Contract title or type")
    parties: List[ContractParty] = Field(description="All parties")
    total_value: Optional[float] = Field(None, description="Total value")
    governing_law: Optional[str] = Field(None, description="Governing law")

# Extract structured data - no conversation persistence
result = await agent.structured_response(
    "Analyze this contract and extract key information",
    ContractAnalysis
)

if result:
    print(f"Contract: {result.title}")
    print(f"Value: ${result.total_value:,.2f}" if result.total_value else "No value specified")
    for party in result.parties:
        print(f"- {party.name} ({party.role})")
```

**Key Features:**

1. **Type Safety**: Supports `str`, `int`, `float`, `bool`, `List[T]`, and Pydantic models
2. **No Persistence**: Messages are not stored in the database (ephemeral)
3. **Error Handling**: Returns `None` on failure instead of raising exceptions
4. **Parameter Overrides**: Supports per-call customization:
   ```python
   result = await agent.structured_response(
       prompt="Extract payment terms",
       target_type=PaymentTerms,
       system_prompt="You are a financial analyst. Be precise.",
       model="gpt-4",
       temperature=0.1,
       max_tokens=1000
   )
   ```

5. **Default Verification Behavior**: The default system prompt includes verification steps to:
   - Ensure accurate extraction from the actual document content
   - Prevent placeholder values (e.g., "N/A", "Not Available") unless they actually appear in the document
   - Return `None` for missing data rather than inventing values
   - This behavior can be overridden with a custom `system_prompt`

6. **Extra Context Support**: Pass additional guidance via `extra_context`:
   ```python
   result = await agent.structured_response(
       prompt="Extract warranty terms",
       target_type=WarrantyTerms,
       extra_context="""
       This is a software license agreement.
       Warranties are typically in Section 7 or Exhibit C.
       Look for both express warranties and warranty disclaimers.
       """
   )
   ```

**Framework Support:**
- ✅ **PydanticAI**: Fully implemented with automatic output strategy selection
  - Uses pydantic_ai's native capabilities to choose the best extraction method
  - Automatically selects between tool calling, JSON mode, or prompted extraction based on the model
  - Simplified system prompts for better reliability and efficiency
- Note: Custom framework adapters can implement `_structured_response_raw()` to support structured extraction

**Best Practices:**

```python
# Always check for None (indicates extraction failure)
result = await agent.structured_response("Extract dates", DateInfo)
if result is None:
    # Handle extraction failure
    logger.warning("Failed to extract dates")
    return

# Use specific prompts for better results
result = await agent.structured_response(
    "Extract the effective date and termination date from Section 3.1",
    ContractDates
)

# For simple types, be explicit about format
page_count = await agent.structured_response(
    "How many pages does this document have? Return only the number.",
    int
)
```

#### Source Structure

All sources returned by agents follow a standardized format that includes annotation metadata, similarity scores, and coordinate information via the `json` field:

```python
# Example source object structure (PDF document)
pdf_source = {
    "annotation_id": 123,
    "rawText": "This is the annotation content",
    "similarity_score": 0.85,
    "document_id": 456,
    "corpus_id": 789,
    "page": 2,
    "annotation_label": "Contract Clause",
    "json": {  # Full MultipageAnnotationJson for PDFs
        "0": {  # Page 0
            "bounds": {"top": 100.5, "bottom": 120.3, "left": 50.2, "right": 250.8},
            "tokensJsons": [
                {"pageIndex": 0, "tokenIndex": 10},
                {"pageIndex": 0, "tokenIndex": 11}
            ],
            "rawText": "annotation content"
        }
    }
}

# Example source object structure (text document)
text_source = {
    "annotation_id": -1,  # Negative ID for synthetic sources
    "rawText": "exact text match",
    "similarity_score": 1.0,
    "document_id": 456,
    "page": 1,
    "char_start": 1234,
    "char_end": 1250,
    "json": {"start": 1234, "end": 1250}  # Simple format for text files
}

# Sources are consistent across all contexts:
response = await agent.chat("What are the payment terms?")
for source in response.sources:
    print(f"Source: {source.annotation_id} (score: {source.similarity_score})")
    print(f"Content: {source.content}")
    print(f"Metadata: {source.metadata}")

    # Access the source as a dictionary (for serialization/storage)
    source_dict = source.to_dict()  # Automatically generates `json` field
```

**Key Source Features:**

- **Automatic `json` Field Generation**: `SourceNode.to_dict()` intelligently constructs the `json` field based on available metadata:
  - **PDF Sources**: Uses `annotation_json` from metadata → full `MultipageAnnotationJson` with token positions and bounding boxes
  - **Text Sources**: Uses `char_start`/`char_end` from metadata → simple `{start, end}` format
  - **Legacy Sources**: Passes through existing `json` field if present

- **Document Type Agnostic**: Tools store raw metadata (`annotation_json`, `char_start`, `char_end`); transformation happens automatically

- **Frontend Compatibility**: The `json` field format matches frontend TypeScript interfaces (`WebSocketSources`)

This format is used consistently in:
- Database storage (ChatMessage.data['sources'])
- WebSocket streaming (ASYNC_FINISH messages)
- API responses (UnifiedChatResponse.sources)
- Vector store search results
- Exact text search results

### Conversation Management

The framework provides sophisticated conversation management through the `CoreConversationManager`:

#### Persistent Conversations

```python
# Create agent with persistent conversation
# The `corpus` parameter is optional for document agents (use None for standalone).
agent = await agents.for_document(
    document=123, # Use actual document ID or object
    corpus=1,     # Use actual corpus ID or object
    user_id=456,  # Required for persistence
    conversation_id=789  # Optional: resume existing conversation
)

# Messages are automatically stored
response1 = await agent.chat("What is this document about?")
response2 = await agent.chat("Can you elaborate on section 2?")  # Context maintained

# Access conversation info
conversation_id = agent.get_conversation_id()
conversation_info = agent.get_conversation_info()
print(f"Conversation has {conversation_info['message_count']} messages")
```

#### Anonymous Conversations

```python
# Anonymous sessions - context maintained in memory only
# The `corpus` parameter is optional; standalone works for anonymous too.
agent = await agents.for_document(document=123, corpus=None)  # No user_id
response1 = await agent.chat("What is this document about?")
response2 = await agent.chat("Can you elaborate on section 2?")  # Context maintained in memory

# Anonymous conversations are session-only and not persisted
conversation_id = agent.get_conversation_id()  # Returns None for anonymous
conversation_info = agent.get_conversation_info()  # Returns basic info with no persistence

# Important: Anonymous conversations cannot be restored later
```

#### Message Storage Control

```python
# Control message storage per interaction
response = await agent.chat(
    "Sensitive query that shouldn't be stored",
    store_messages=False  # Skip database storage
)

# Manual message storage
user_msg_id = await agent.store_user_message("Custom user message")
llm_msg_id = await agent.store_llm_message("Custom LLM response")
```

### Context Management

Conversations can outgrow an LLM's context window, both during a single
high-tool-traffic turn and across many turns of a long-lived thread. The
framework ships two complementary subsystems for this:

#### Context Guardrails (single-turn protection)

`opencontractserver/llms/context_guardrails.py` provides three layers of
runtime protection against context overflow:

1. **Token estimation** — heuristics that approximate token counts for prompts,
   tool outputs, and message history without paying the cost of a real
   tokeniser.
2. **Conversation compaction** — old messages are summarised in place when the
   estimated context budget would be exceeded, preserving recent turns and key
   instructions verbatim while collapsing older context into a summary.
3. **Tool-output truncation** — long tool results are clipped (with a marker)
   before being injected back into the conversation.

The guardrails fire automatically inside the WebSocket consumer's per-turn
loop. When the budget is exhausted to the point that no further safe
compaction is possible, the consumer emits an `ASYNC_ERROR` with `error =
WS_ERROR_CONTEXT_EXHAUSTED` (see `opencontractserver.constants.context_guardrails`).

Tunable constants live in `opencontractserver/constants/context_guardrails.py`.

#### Memory Curation (cross-conversation learning)

`opencontractserver/tasks/memory_tasks.py` is a Celery-based pipeline that
turns idle conversations into long-term corpus memory:

1. A scheduled task scans for conversations whose latest message is older than
   the idle threshold and that have not yet been curated.
2. Each idle conversation is summarised by a curation LLM.
3. The summary is merged into the corpus's memory document.
4. The conversation is marked `curated=True` so it is not re-processed.

Memory is then read by agents via the `read_corpus_memory` tool
(`opencontractserver/llms/tools/core_tools/memory.py`), so a corpus's agents
gradually accumulate knowledge from prior interactions without inflating
per-turn context.

Operators should monitor curation cost and tune the idle threshold to balance
freshness against LLM spend.

### Tools

The framework provides a unified tool system that works across all supported frameworks. Core tools often have synchronous and asynchronous versions (e.g., `load_document_md_summary` and `aload_document_md_summary`).

#### Built-in Tools

The framework ships a substantial library of built-in tools under
`opencontractserver/llms/tools/core_tools/`. Highlights:

| Tool family | File | Purpose |
|---|---|---|
| Annotations | `annotations.py` | Read / create annotations and labels |
| Documents | `documents.py` | Document metadata, summaries, file access |
| Notes | `notes.py` | Read / write per-document notes |
| Document summaries | `document_summaries.py`, `md_summaries.py` | Generate and read markdown summaries |
| Page images | `page_images.py` | Vision tools — fetch page images for multimodal models |
| Search | `search.py`, `document_indexing.py`, `text_extracts.py` | Vector search, keyword search, sourced text-span retrieval |
| Links | `links.py` | Create / read `OC_URL` link annotations |
| CAML / corpus articles | `caml_article.py` | Read / write the corpus's markdown "README" article |
| Memory | `memory.py` | Read the curated corpus memory document |
| Descriptions | `descriptions.py` | Read / update corpus and document descriptions |
| **PII** | `pii.py` | Scan documents for PII via the Privacy Filter microservice and persist results as annotations |
| Privacy Filter client | `_privacy_filter_client.py` | HTTP client for the external Privacy Filter service |

In addition, `opencontractserver/llms/tools/web_search_tools.py` exposes a
real-time web search tool that agents can call mid-conversation. Set
`SEARCH_PROVIDER_API_KEY` (or the relevant env var for your provider) to enable
it. Be aware of the privacy and cost implications before enabling it on
documents that contain sensitive data.

```python
from opencontractserver.llms.tools import create_document_tools # Convenience function
from opencontractserver.llms.tools.core_tools import (
    load_document_md_summary, # Sync version
    aload_document_md_summary, # Async version
    get_notes_for_document_corpus,
    aget_notes_for_document_corpus, # Async version
    get_md_summary_token_length
)

# Use built-in tools by name (async versions preferred when available)
# The `corpus` parameter is optional for document agents (use None for standalone).
agent = await agents.for_document(
    document=123, # Use actual document ID or object
    corpus=1,     # Use actual corpus ID or object
    tools=["load_md_summary", "get_notes_for_document_corpus", "get_md_summary_token_length"]
)

# Or use CoreTool objects directly (e.g., from the convenience function)
# create_document_tools() provides a list of pre-configured CoreTool instances.
document_tools = create_document_tools()
agent = await agents.for_document(document=123, corpus=1, tools=document_tools) # Use actual document/corpus IDs
```

#### Custom Tools

```python
from opencontractserver.llms.tools.tool_factory import CoreTool

def analyze_contract_risk(contract_text: str) -> str:
    """Analyze contract risk factors."""
    # Your custom analysis logic
    return "Risk analysis results..."

# Create CoreTool from function
risk_tool = CoreTool.from_function(
    analyze_contract_risk,
    description="Analyze contract risk factors"
)

# The `corpus` parameter is optional for document agents (use None for standalone).
agent = await agents.for_document(
    document=123, # Use actual document ID or object
    corpus=None,
    tools=[risk_tool]
)
```

#### Corpus-Dependent Tools

Some tools require a corpus context to function properly (e.g., tools that create notes or search across corpus documents). The framework provides automatic tool filtering for documents not in a corpus:

```python
from opencontractserver.llms.tools.tool_factory import CoreTool

def create_corpus_note(title: str, content: str, corpus_id: int) -> str:
    """Create a note within the corpus context."""
    # This tool needs a corpus to function
    return f"Created note '{title}' in corpus {corpus_id}"

# Mark tool as requiring corpus
corpus_tool = CoreTool.from_function(
    create_corpus_note,
    description="Create a note in the corpus",
    requires_corpus=True  # ← Corpus requirement flag
)

# For documents IN a corpus - tool is available
agent = await agents.for_document(
    document=123,
    corpus=456,  # Document is in this corpus
    tools=[corpus_tool]  # Tool will be included
)

# For documents NOT in a corpus - tool is automatically filtered out
agent = await agents.for_document(
    document=123,
    corpus=None,  # No corpus context
    tools=[corpus_tool]  # Tool will be skipped with info log
)
# Agent creation succeeds, corpus-dependent tools are gracefully omitted
```

**Key Features:**

- **Graceful Degradation**: Corpus-dependent tools are automatically filtered when `corpus=None`
- **Informative Logging**: Framework logs which tools are skipped and why
- **No Errors**: Agent creation never fails due to missing corpus - tools just adapt
- **Backward Compatibility**: Existing tools without `requires_corpus=True` work everywhere

**Built-in Tools with Corpus Requirements:**

Most built-in document tools work without corpus context, but some corpus-specific operations may require it:

```python
# These tools work with or without corpus
safe_tools = [
    "load_md_summary",           # Document summary access
    "get_md_summary_token_length", # Token counting
    "load_document_txt_extract"  # Raw text access
]

# These tools may require corpus context (marked with requires_corpus=True)
corpus_tools = [
    "get_notes_for_document_corpus",  # Notes tied to corpus context
    "add_document_note",              # Creating corpus-scoped notes
    # Framework handles filtering automatically
]

# Use both safely - framework filters as needed
agent = await agents.for_document(
    document=123,
    corpus=None,  # No corpus
    tools=safe_tools + corpus_tools  # Mixed tools - filtering handled automatically
)
```

#### Extract & Analyzer Tools

Agents can also discover and dispatch **Extracts** (Fieldset-driven structured-data
extractions) and **Analyses** (Gremlin or task-based analyzers) — the same actions a
human triggers from the UI or a static `CorpusAction`. These tools live in
[`tools/core_tools/extracts_and_analyzers.py`](../../../opencontractserver/llms/tools/core_tools/extracts_and_analyzers.py)
and slot into the standard registry alongside notes, summaries, and annotations.

| Tool | Category | Approval | Requires write | Purpose |
|------|----------|----------|----------------|---------|
| `list_fieldsets` | extracts | – | – | List Fieldsets visible to the user, with column metadata, so the agent can pick one |
| `start_extract` | extracts | ✓ | ✓ | Create a new `Extract` from an existing `fieldset_id` and queue `run_extract` |
| `list_recent_extracts` | extracts | – | – | Most recent Extracts on this corpus, with status |
| `list_analyzers` | analyzers | – | – | List non-disabled Analyzers visible to the user |
| `start_analysis` | analyzers | ✓ | ✓ | Run an existing Analyzer; mirrors `process_analyzer` |
| `list_recent_analyses` | analyzers | – | – | Most recent Analyses on this corpus, with status |

**Design rules:**

- The agent **picks from configured Fieldsets/Analyzers** — it cannot invent
  new column schemas or analyzers. This keeps the run surface aligned with
  what the corpus admin has explicitly defined.
- All six tools require corpus context (`requires_corpus=True`); they are
  filtered out for standalone documents.
- `start_extract` and `start_analysis` are gated by the standard approval
  flow (`requires_approval=True`) and require WRITE permission on the
  corpus (`requires_write_permission=True`). In an agent-based
  `CorpusAction`, list them under `pre_authorized_tools` to skip the
  approval prompt.
- Document scope follows the agent context:
    - **Corpus agent**: omitting `document_ids` runs on the full corpus.
    - **Document agent**: omitting `document_ids` runs on the current
      document only. Any `document_ids` the LLM passes are intersected
      with `corpus.get_documents()` so the agent cannot reach documents
      outside its scope.
- Both run tools dispatch the same Celery pipeline as their GraphQL
  counterparts (`StartExtract`, `StartDocumentAnalysisMutation`) — same
  `Extract` / `Analysis` records, same permissions, same
  `corpus_action_id` lineage tracking.

**Example: discover and run an extract**

```python
from opencontractserver.llms import agents

agent = await agents.for_corpus(
    corpus=corpus_id,
    user_id=user_id,
    tools=[
        "list_fieldsets",
        "start_extract",
        "list_recent_extracts",
        "list_analyzers",
        "start_analysis",
        "list_recent_analyses",
    ],
)

# The agent will call list_fieldsets to pick a Fieldset, then call
# start_extract with the chosen fieldset_id. The approval modal surfaces
# the tool name + arguments to the user before execution.
async for event in agent.stream("Run the standard MSA extract on this corpus"):
    ...
```

**Opt-in for CorpusActions**

These tools are **not** added to `DEFAULT_DOCUMENT_ACTION_TOOLS`. To let
an agent-based `CorpusAction` dispatch extracts/analyses automatically on
new documents, list them on the `CorpusAction`:

```python
CorpusAction.objects.create(
    corpus=corpus,
    creator=user,
    trigger=CorpusActionTrigger.ADD_DOCUMENT,
    agent_config=my_agent_config,           # or task_instructions=...
    task_instructions="If this is an MSA, run the MSA extract.",
    pre_authorized_tools=["start_extract"], # skips approval for this tool
)
```

#### Framework-Specific Tools

The framework automatically converts tools to the appropriate format. `CoreTool` objects are converted to framework-specific wrappers via `UnifiedToolFactory.create_tool()` in [`tools/tool_factory.py`](../../../opencontractserver/llms/tools/tool_factory.py). For PydanticAI, this produces `PydanticAIToolWrapper` instances (see [`tools/pydantic_ai_tools.py`](../../../opencontractserver/llms/tools/pydantic_ai_tools.py)).

```python
# Tools work seamlessly with the framework
agent = await agents.for_document(
    document=123,
    corpus=None,
    framework=AgentFramework.PYDANTIC_AI,
    tools=["load_md_summary"]
)
```

#### Tool Approval & Human-in-the-Loop

Some tools might be *dangerous* (e.g. deleting data) or simply require legal review before execution. OpenContracts supports a **durable approval gate** that pauses the agent right before such a tool would run, persists all state, and lets a human approve or reject the call at a later time—even after a server restart.

##### Approval Flow Overview

When a tool requiring approval is called, the framework:

1. **Pauses execution** before running the tool
2. **Emits an `ApprovalNeededEvent`** with the pending tool call details
3. **Persists state** in the database with `state=AWAITING_APPROVAL`
4. **Waits for human decision** via `resume_with_approval()`

Upon approval/rejection:

1. **Emits `ApprovalResultEvent`** with the decision
2. If approved:
   - **Emits `ResumeEvent`** to signal continuation
   - **Executes the tool** with the original arguments
   - **Continues normal agent execution** (can call more tools)
3. If rejected:
   - **Emits final event** with rejection message
   - **Ends the conversation turn**

##### Flagging Tools for Approval

```python
from opencontractserver.llms import tools

async def delete_user_account(user_id: int) -> str:
    """Permanently delete a user (⚠ irreversible)."""
    # Dangerous operation implementation
    return f"Account {user_id} deleted"

# Mark tool as requiring approval
danger_tool = tools.from_function(
    delete_user_account,
    name="delete_user_account", 
    description="Delete a user – requires admin approval.",
    requires_approval=True,  # ← approval flag
)

agent = await agents.for_document(
    document=123, corpus=1,
    tools=[danger_tool]
)
```

##### Handling Approval Events

When the LLM attempts to call a flagged tool, the agent pauses and emits an `ApprovalNeededEvent`:

```python
from opencontractserver.llms.agents.core_agents import (
    ApprovalNeededEvent,
    ApprovalResultEvent,
    ResumeEvent
)

async for event in agent.stream("Delete user account 42"):
    match event.type:
        case "approval_needed":
            # Agent has paused, waiting for approval
            tool_call = event.pending_tool_call
            print(f"Tool '{tool_call['name']}' needs approval")
            print(f"Arguments: {tool_call['arguments']}")
            
            # Get human decision (via UI, CLI, etc.)
            approved = await get_user_decision()
            
            # Resume execution
            async for resume_event in agent.resume_with_approval(
                llm_message_id=event.llm_message_id,
                approved=approved
            ):
                # Handle approval result and continuation events
                match resume_event.type:
                    case "approval_result":
                        print(f"Decision: {resume_event.decision}")
                    case "resume":
                        print(f"Execution resuming...")
                    case "thought" | "content" | "sources" | "final":
                        # Normal event processing continues
                        pass
                 
        case "thought" | "content" | "sources" | "final":
            # Handle other events normally
            pass
```

##### New Event Types

The approval flow introduces two new event types:

| Event Type | Purpose | Key Fields | When Emitted |
|------------|---------|------------|--------------|
| `ApprovalResultEvent` | Confirms decision was recorded | `decision` ("approved"/"rejected"), `pending_tool_call` | Immediately after `resume_with_approval()` |
| `ResumeEvent` | Signals execution restart | Standard event fields | After approval, before tool execution |
| `ErrorEvent` | Error occurred during execution | `error`, `metadata`, `is_complete` (always True) | Unrecoverable errors (e.g., rate limits, API failures) |

##### Approval Event Structure

`ApprovalNeededEvent` contains:

| Field | Type | Description |
|-------|------|-------------|
| `type` | `"approval_needed"` | Event discriminator |
| `pending_tool_call` | `dict` | `{name, arguments, tool_call_id}` |
| `user_message_id` | `int` | Database message ID |
| `llm_message_id` | `int` | Database message ID (use for resume) |
| `metadata` | `dict` | Additional state information |

##### Resumption API

```python
# Resume execution after approval
# Returns an async generator of events
async for event in agent.resume_with_approval(
    llm_message_id=paused_message_id,
    approved=True
):
    # Process events (approval_result, resume, thought, content, etc.)
    if event.type == "final":
        print(f"Final answer: {event.accumulated_content}")

# Reject the tool execution
async for event in agent.resume_with_approval(
    llm_message_id=paused_message_id,
    approved=False
):
    # Will receive approval_result + final event with rejection message
    pass
```

**Multi-Tool Execution**: After approval, the agent continues its normal execution flow. It can:
- Process the tool result
- Call additional tools (including other approval-gated tools)
- Generate a final response incorporating all tool results

##### Implementation Details

The full implementation of approval gating lives in two files:
- **Veto gate**: [`tools/pydantic_ai_tools.py`](../../../opencontractserver/llms/tools/pydantic_ai_tools.py) — `PydanticAIToolWrapper._maybe_raise()` raises `ToolConfirmationRequired` before tool execution
- **Pause/resume**: [`agents/pydantic_ai_agents.py`](../../../opencontractserver/llms/agents/pydantic_ai_agents.py) — `PydanticAICoreAgent.resume_with_approval()` handles state validation and continuation

###### State Validation

The `resume_with_approval()` method performs strict state validation:

1. Loads the paused `ChatMessage` from database by `llm_message_id`
2. Checks `data['state']` is exactly `AWAITING_APPROVAL`
3. Handles duplicate requests gracefully (returns empty generator if already completed/cancelled)
4. Raises `ValueError` for invalid states

###### Tool Execution Paths (Approved)

When approved, the framework attempts to execute the tool using two fallback paths:

1. **Agent Config Tools**: Searches `self.config.tools` for a matching tool name and executes it
2. **PydanticAI Registry** (fallback): Looks up the tool via `_get_function_tools(self.pydantic_ai_agent)` and extracts the underlying callable

Both paths use an empty context with `skip_approval_gate=True` to prevent re-triggering the gate, and preserve the original `tool_call_id`.

After execution, a continuation prompt containing the tool result is fed back into `_stream_core()` with a fresh `ChatMessage` for the resumed run, linked to the original user message.

###### Rejection Path

When `approved=False`:
1. Emits `ApprovalResultEvent` with `decision="rejected"`
2. Marks the paused message as `CANCELLED` in the database
3. Emits a `FinalEvent` with a rejection message and stops

###### Event Sequence

**Approval Flow (approved=True):**
```
1. ApprovalResultEvent (decision="approved")
2. ResumeEvent (signals execution restart)
3. [Stream events from continuation...] (ThoughtEvent, ContentEvent, etc.)
4. FinalEvent (with approval_decision in metadata)
```

**Rejection Flow (approved=False):**
```
1. ApprovalResultEvent (decision="rejected")
2. FinalEvent (rejection message, conversation ends)
```

**Key Notes:**
- Approval state persists across server restarts (stored in database)
- Only PydanticAI agents support approval gating currently
- Custom framework adapters should implement approval handling in their `_stream_raw()` method
- All events include `user_message_id` and `llm_message_id` for tracking

##### Nested Approval Gates (Corpus → Document Sub-Agents)

When a corpus agent delegates a question to a document sub-agent via
`ask_document`, the sub-agent may itself encounter tools that require
approval. These nested approval events propagate up to the corpus agent
and ultimately to the user through the following mechanism:

**Propagation Flow:**

```
User ──▶ CorpusAgent ──ask_document──▶ DocumentSubAgent
                                            │
                        ┌───────────────────┘
                        ▼
           Sub-agent tool requires approval
                        │
                        ▼
           ApprovalNeededEvent emitted in sub-agent stream
                        │
                        ▼
           ask_document_tool detects event, raises
           ToolConfirmationRequired with nested metadata
                        │
                        ▼
           Corpus agent pauses, surfaces to user via WebSocket
                        │
                        ▼
           User approves/rejects via frontend modal
                        │
                        ▼
           resume_with_approval() re-invokes ask_document
           with config._approval_bypass_allowed = True
                        │
                        ▼
           Sub-agent created with skip_approval_gate=True
           (tool executes without re-prompting)
```

**Metadata Preservation:**

The `ToolConfirmationRequired` exception carries sub-agent tool details
in `_`-prefixed keys for frontend display:

```python
raise ToolConfirmationRequired(
    tool_name="ask_document",
    tool_args={
        "document_id": document_id,
        "question": question,
        # Metadata for UI display (stripped before execution):
        "_sub_tool_name": "update_document_summary",
        "_sub_tool_arguments": {"new_content": "..."},
    },
)
```

The frontend (CorpusChat) unwraps these to show the actual sub-tool name
and arguments in the approval modal, rather than the generic
`ask_document` wrapper.

**Security: Approval Bypass Protection:**

The `skip_approval` mechanism uses `config._approval_bypass_allowed`
(a runtime flag on `AgentConfig`) rather than a function parameter, to
prevent LLM prompt injection from bypassing approval gates:

- Normal execution: `_approval_bypass_allowed` is `False` (default)
- Post-approval resume: `resume_with_approval()` sets it to `True` in a
  `try/finally` block, ensuring it is always reset
- The flag is read inside `ask_document_tool`'s closure and passed to
  `_agents_api.for_document(skip_approval_gate=bypass)`

**Defensive Handling:**

Malformed `approval_needed` events (missing `name`, non-dict
`pending_tool_call`, empty strings) are logged and skipped rather than
crashing the tool. The sub-agent stream continues normally.

**Key Files:**
- `agents/pydantic_ai_agents.py`: `ask_document_tool()` closure,
  `resume_with_approval()` bypass flag
- `frontend/src/components/corpuses/CorpusChat.tsx`: Sub-tool unwrapping
  in `ASYNC_APPROVAL_NEEDED` handler
- `tests/test_nested_approval_gates.py`: Comprehensive test suite
- The method always returns an async generator (even for rejection) for consistent API

### Nested Agent Streaming

The framework now supports **real-time visibility into nested agent execution** through the stream observer pattern. This is particularly powerful when corpus agents delegate work to document agents.

#### The Stream Observer Pattern

When a parent agent (e.g., corpus agent) calls a child agent (e.g., document agent via `ask_document` tool), the child's stream events can be forwarded to a configured observer:

```python
from opencontractserver.llms.agents.core_agents import AgentConfig
from opencontractserver.llms.agents.pydantic_ai_agents import PydanticAICorpusAgent

# Define your observer
async def websocket_forwarder(event):
    """Forward nested events to WebSocket clients."""
    await websocket.send_json({
        "type": event.type,
        "content": getattr(event, "content", ""),
        "thought": getattr(event, "thought", ""),
        "sources": [s.to_dict() for s in getattr(event, "sources", [])]
    })

# NOTE: stream_observer is only available via AgentConfig + low-level factory,
# not the high-level agents.for_corpus() API.
config = AgentConfig(
    user_id=user_id,
    stream_observer=websocket_forwarder,
)
corpus_agent = await PydanticAICorpusAgent.create(corpus=corpus_id, config=config)

# When streaming, nested events bubble up automatically
async for event in corpus_agent.stream("Analyze payment terms across all contracts"):
    # Parent agent events
    if event.type == "thought" and "[ask_document]" in event.thought:
        # These are relayed child agent thoughts
        print(f"Child agent: {event.thought}")
    else:
        # Direct parent agent events
        print(f"Parent: {event.type} - {event.content}")
```

#### How It Works

1. **Configuration**: Set `stream_observer` in `AgentConfig` or pass it when creating agents
2. **Automatic Forwarding**: Framework adapters call the observer for every emitted event
3. **Child Agent Integration**: Tools like `ask_document` forward their stream to the observer
4. **WebSocket Ready**: Perfect for real-time UI updates showing nested reasoning

#### Example: Corpus Agent with Live Document Analysis

```python
# In your WebSocket handler
async def handle_corpus_query(websocket, corpus_id, query):
    # Create observer that forwards to WebSocket
    async def forward_to_client(event):
        await websocket.send_json({
            "event": event.type,
            "data": {
                "content": getattr(event, "content", ""),
                "thought": getattr(event, "thought", ""),
                "sources": [s.to_dict() for s in getattr(event, "sources", [])],
                "metadata": getattr(event, "metadata", {})
            }
        })
    
    # Create corpus agent with observer (requires low-level factory)
    config = AgentConfig(stream_observer=forward_to_client)
    agent = await PydanticAICorpusAgent.create(corpus=corpus_id, config=config)
    
    # Stream response - client sees EVERYTHING including nested calls
    async for event in agent.stream(query):
        # Parent events also go to client
        await forward_to_client(event)
```

#### Benefits

- **Complete Visibility**: See exactly what child agents are doing in real-time
- **Better UX**: Users see progress even during long-running nested operations
- **Debugging**: Full execution trace across agent boundaries
- **No Blocking**: Parent agent continues streaming while child executes

#### Implementation Details

The stream observer is implemented at the framework adapter level:

- **PydanticAI**: `ask_document_tool` (inline in [`PydanticAICorpusAgent.create()`](../../../opencontractserver/llms/agents/pydantic_ai_agents.py)) explicitly forwards child events
- **CoreAgentBase**: `_emit_observer_event()` helper (in [`agents/core_agents.py`](../../../opencontractserver/llms/agents/core_agents.py)) ensures safe forwarding — observer exceptions are caught and logged, never breaking the stream

This pattern ensures that even deeply nested agent calls remain visible and debuggable, providing transparency into multi-agent workflows.

### Streaming

All agents support streaming responses for real-time interaction. The framework now provides **event-based streaming** for rich, granular interaction visibility.

#### Event-Based Streaming (Recommended)

**PydanticAI agents** emit granular events that expose the agent's reasoning process:

```python
# Rich event streaming with PydanticAI
agent = await agents.for_document(
    document=123, corpus=1, 
    framework=AgentFramework.PYDANTIC_AI
)

async for event in agent.stream("What are the key contract terms?"):
    match event.type:
        case "thought":
            print(f"🤔 Agent thinking: {event.thought}")
            # event.metadata may contain tool info for tool-related thoughts
            
        case "content":
            print(event.content, end="", flush=True)
            # event.metadata contains tool details if content is from tool calls
            
        case "sources":
            print(f"\n📚 Found {len(event.sources)} relevant sources")
            for source in event.sources:
                print(f"  - {source.annotation_id}: {source.content[:50]}...")
                
        case "error":
            print(f"\n❌ Error: {event.error}")
            print(f"Error type: {event.metadata.get('error_type', 'Unknown')}")
            # Handle error gracefully - stream ends after error event
            break
                
        case "final":
            print(f"\n✅ Complete! Usage: {event.metadata.get('usage', {})}")
            print(f"Total sources: {len(event.sources)}")

# All events include message IDs for tracking
print(f"Conversation: {event.user_message_id} → {event.llm_message_id}")
```

**Example PydanticAI Event Sequence:**
```
🤔 Agent thinking: Received user prompt; beginning reasoning cycle…
🤔 Agent thinking: Sending request to language model…
🤔 Agent thinking: Processing model response – may invoke tools…
🤔 Agent thinking: Calling tool `similarity_search` with args {'query': 'key contract terms', 'k': 10}
📚 Found 5 relevant sources
🤔 Agent thinking: Tool `similarity_search` returned a result.
🤔 Agent thinking: Run finished; aggregating final results…
Based on the contract analysis, the key terms include...
✅ Complete! Usage: {'requests': 2, 'total_tokens': 1247}
```

#### Legacy Streaming (Backward Compatibility)

Legacy code using the traditional streaming approach can be updated to use event-based streaming:

```python
# Traditional streaming - still supported
async for chunk in agent.stream("Analyze liability clauses"):
    print(chunk.content, end="")
    
    # Access metadata during streaming
    if chunk.is_complete:
        print(f"\nSources: {len(chunk.sources)}")
        print(f"Message ID: {chunk.llm_message_id}")

# Detect streaming type at runtime
async for event in agent.stream("Your query"):
    if hasattr(event, 'type'):  # New event-based streaming
        handle_event_based_streaming(event)
    else:  # Legacy UnifiedStreamResponse
        handle_legacy_streaming(event)
```

#### Advanced Streaming Patterns

```python
# Stream with custom message storage control
async for event in agent.stream("Sensitive analysis", store_messages=False):
    # Process events without persisting to database
    if event.type == "content":
        secure_output_handler(event.content)

# Real-time UI updates with event metadata
async for event in agent.stream("Complex analysis"):
    if event.type == "thought":
        ui.show_thinking_indicator(event.thought)
        if "tool_name" in event.metadata:
            ui.show_tool_usage(event.metadata["tool_name"])
    elif event.type == "content":
        ui.append_content(event.content)
    elif event.type == "sources":
        ui.update_source_panel(event.sources)
```

### Embeddings

The framework provides sync embeddings APIs via `opencontractserver.llms.embeddings`:

```python
from opencontractserver.llms import embeddings

# Sync version (currently the only exposed API)
embedder_path, vector = embeddings.generate("Contract analysis text")
print(f"Using embedder: {embedder_path}")
print(f"Vector dimension: {len(vector)}")
print(f"Vector type: {type(vector)}")  # numpy.ndarray

# The embeddings integrate with the vector stores for document search

# Note: For async embedding generation, use the underlying utility directly:
from opencontractserver.utils.embeddings import generate_embeddings_from_text
embedder_path, vector = generate_embeddings_from_text("Contract analysis text")
```

### Vector Stores

Vector stores provide both sync and async search methods, accessible via `opencontractserver.llms.vector_stores`.

```python
from opencontractserver.llms import vector_stores
from opencontractserver.llms.vector_stores.core_vector_stores import VectorSearchQuery

# Create vector store (framework-specific store will be chosen based on config or default)
store = vector_stores.create(
    framework="pydantic_ai", # Or custom-defined framework, or omit for default
    user_id=123, # Optional, for user-specific data if applicable
    corpus_id=456 # Or document_id for document-specific vector store context
)

# Search annotations
query = VectorSearchQuery(
    query_text="payment obligations",
    similarity_top_k=10
)

# Async search (recommended)
results = await store.async_search(query)

# Sync search (for compatibility)
results = store.search(query)

for result in results:
    print(f"Score: {result.similarity_score}")
    print(f"Text: {result.annotation.raw_text[:100]}...")
```

## Architecture

The framework follows a layered architecture that separates concerns and enables framework flexibility:

```
┌─────────────────────────────────────────┐
│           API Layer                     │  ← api.py (agents, embeddings, vector_stores, tools)
├─────────────────────────────────────────┤
│        Framework Adapter Layer          │  ← agents/pydantic_ai_agents.py
│ (Implements CoreAgent for specific SDK) │     agents/pydantic_ai_agents.py
├─────────────────────────────────────────┤
│         Core Agent Protocol             │  ← agents/core_agents.py (Defines .chat, .stream)
│         & Unified Tool System           │  ← tools/ (CoreTool, UnifiedToolFactory)
├─────────────────────────────────────────┤
│         Core Business Logic             │  ← Framework-agnostic utils, config
│         & Conversation Management       │     conversations/ (CoreConversationManager)
├─────────────────────────────────────────┤
│      Django Models & Vector Stores      │  ← Your documents + annotation data & persistence
└─────────────────────────────────────────┘
```

### How It Works

1. **Beautiful API (`api.py`)**:
   - `agents.for_document(document=123, corpus=1)` provides the elegant entry point.
   - `agents.get_structured_response_from_document()` and `agents.get_structured_response_from_corpus()` provide convenience methods for one-shot structured extraction.
   - Handles parameter validation, type conversion, and defaults.
   - Routes to the appropriate factory based on framework choice.
   - Similar entry points exist for `embeddings`, `vector_stores`, and `tools`.

2. **Unified Factory (`agents/agent_factory.py`, `vector_stores/vector_store_factory.py`, etc.)**:
   - E.g., `UnifiedAgentFactory.create_document_agent()` orchestrates agent creation.
   - Converts string framework names to enums, resolves tools, creates contexts.
   - Delegates to framework-specific implementations.

3. **Framework Adapters** (e.g., `agents/pydantic_ai_agents.py`):
   - E.g., `PydanticAIDocumentAgent.create()` builds the actual LLM integration.
   - Creates vector stores, configures embeddings, sets up the underlying PydanticAI agent.
   - Returns a framework-specific agent that implements the `CoreAgent` protocol.

4. **CoreAgent Protocol (`agents/core_agents.py`)**:
   - The returned agent object (e.g., an instance of `PydanticAIDocumentAgent`) inherits from `CoreAgentBase`, which provides universal `chat()`, `stream()`, and `structured_response()` wrappers that handle all database persistence, approval gating, and message lifecycle management.
   - Framework adapters only implement low-level `_chat_raw()`, `_stream_raw()`, and `_structured_response_raw()` methods that return pure content without any database side-effects.
   - When you call `await agent.chat("Your query")`, the `CoreAgentBase` wrapper automatically handles user message storage, LLM placeholder creation, calling the adapter's `_chat_raw()` method, and completing the stored message with results.
   - The `structured_response()` method provides ephemeral, typed data extraction without any database persistence—perfect for one-shot extractions.
   - This architecture ensures that adapters cannot "forget" to persist conversations or handle approval flows—all database operations are centralized and automatic.
   - **Approval Flow**: When a tool requiring approval is called, the framework automatically pauses execution, emits `ApprovalNeededEvent`, and waits for `resume_with_approval()` to be called.
   - **Resume Capability**: The `resume_with_approval()` method allows continuation of paused executions, emitting `ApprovalResultEvent` and `ResumeEvent` before resuming normal agent flow.
   - PydanticAI agents provide granular event-based streaming that exposes the agent's execution graph in real-time.
   - The `_emit_observer_event()` helper enables stream observers to receive events from nested agent calls, providing complete visibility across agent boundaries.

5. **Conversation Management**:
   - `CoreConversationManager` handles message persistence and retrieval.
   - Automatically stores user and LLM messages with proper relationships.
   - Supports both persistent (database) and anonymous (memory-only) conversations.

6. **Tool System**:
   - `CoreTool` provides framework-agnostic tool definitions.
   - Framework-specific factories convert tools to appropriate formats.
   - Built-in tools (e.g., via `create_document_tools()`) for document analysis, note retrieval, and content access. Async versions of core tools are often available.

7. **Timeline Streaming System** (`TimelineStreamMixin`):
   - A mixin class that framework adapters can inherit to get automatic timeline construction
   - Wraps the adapter's `_stream_core()` method and intercepts all emitted events
   - Uses `TimelineBuilder` to incrementally build a reasoning timeline from stream events
   - Automatically injects the complete timeline into `FinalEvent.metadata['timeline']`
   - Persists the timeline to the database via `_finalise_llm_message()` helper
   - Ensures every framework produces consistent timeline structure without duplicate code
   - Timeline includes: thoughts, tool calls, tool results, source discoveries, and status markers
   - Used by PydanticAI adapters (`PydanticAICoreAgent` inherits from `TimelineStreamMixin`)

### Framework Support

#### LlamaIndex Integration

> **⚠️ Warning: LlamaIndex Support Removed**
> 
> LlamaIndex integration has been removed from the OpenContracts LLM framework. All agents now use PydanticAI as the underlying framework. If you need LlamaIndex support, you can implement a custom adapter by following the `CoreAgent` protocol in `agents/core_agents.py`.


#### PydanticAI Integration

```python
# PydanticAI agents use:
# - Modern async patterns with proper type safety
# - Execution graph streaming via agent.iter() for granular visibility
# - Rich event-based streaming (ThoughtEvent, ContentEvent, SourceEvent, FinalEvent)
# - Structured tool definitions with Pydantic models
# - Real-time tool call observation with arguments and results
# - Automatic output strategy selection (tool calling, JSON mode, or prompting)
# - Optimized structured extraction using pydantic_ai's native capabilities

from opencontractserver.llms.agents.pydantic_ai_agents import PydanticAIDocumentAgent
from opencontractserver.llms.vector_stores.pydantic_ai_vector_stores import PydanticAIAnnotationVectorStore

# Framework-specific features
# agent = await PydanticAIDocumentAgent.create(document_obj, corpus_obj, config, conversation_manager, tools)
# vector_search_tool = PydanticAIAnnotationVectorStore(...).create_vector_search_tool()

# PydanticAI streaming produces rich UnifiedStreamEvent objects
async for event in pydantic_agent.stream("Analyze contract"):
    event.type                 # "thought", "content", "sources", or "final"
    event.metadata             # Rich metadata (tool names, args, usage, etc.)
    
    # Event-specific fields:
    if event.type == "thought":
        event.thought          # Agent's reasoning step
    elif event.type == "content":
        event.content          # Text delta for final answer
    elif event.type == "sources":
        event.sources          # List of SourceNode objects
    elif event.type == "final":
        event.accumulated_content  # Complete final answer
        event.sources              # All sources found
        event.metadata['usage']    # Token usage statistics

# Structured extraction leverages pydantic_ai's automatic strategy selection
# The framework automatically chooses the best method based on the model:
# - Tool calling for models like GPT-4
# - Native JSON mode where available
# - Prompted extraction as fallback
result = await agent.structured_response(
    "Extract contract dates",
    ContractDates  # pydantic_ai handles the output strategy
)
```

#### PydanticAI Pitfalls

**`instructions` vs `system_prompt` — Use `instructions` for Agent System Prompts**

When creating a `PydanticAIAgent`, pydantic-ai offers two parameters for providing system-level context:

| Parameter | Behavior |
|-----------|----------|
| `instructions` | Attached to every `ModelRequest` via the `.instructions` field. **Always sent to the model**, even when `message_history` is provided. |
| `system_prompt` | Added as `SystemPromptPart` entries via `_sys_parts()`. **Only included when `message_history` is `None`** (i.e., the first turn of a brand-new conversation with no prior messages). |

In OpenContracts, the `chat()` method persists a HUMAN message *before* calling the underlying pydantic-ai `run()`. This means `_get_message_history()` returns a non-empty list by the time pydantic-ai processes the request. Pydantic-ai then takes the `if message_history:` branch in `UserPromptNode._prepare_messages()` and **skips `_sys_parts()` entirely**, silently dropping any `system_prompt` content.

**Result**: The model receives the user message and tool definitions but **no system prompt at all**, leading to confused responses like "Could you please clarify which specific instructions…"

**Rule**: Always use `instructions=` (not `system_prompt=`) when constructing `PydanticAIAgent` instances:

```python
# CORRECT — instructions are always sent
agent = PydanticAIAgent(
    model="gpt-4.1",
    instructions=my_system_prompt,   # ← Always attached to requests
    tools=my_tools,
)

# WRONG — system_prompt is silently dropped when message_history exists
agent = PydanticAIAgent(
    model="gpt-4.1",
    system_prompt=my_system_prompt,  # ← Lost if any message history is present
    tools=my_tools,
)
```

This applies to all three agent creation sites in `pydantic_ai_agents.py`: `PydanticAIDocumentAgent.create()`, `PydanticAICorpusAgent.create()`, and the structured extraction agent in `structured_response()`.

> **Version note**: Verified against pydantic-ai 0.2.20. Future versions may unify these parameters.

**All production tools MUST be async (`a`-prefixed in `core_tools.py`)**

`PydanticAIToolWrapper` accepts both sync and async functions, but the wrapper does **NOT** push sync callables onto a thread pool. A sync function that touches the Django ORM (e.g., `Document.objects.get(...)`) will raise `SynchronousOnlyOperation` the moment it runs inside an async agent loop.

Use the `a`-prefixed async versions defined in `opencontractserver/llms/tools/core_tools/`. The sync versions exist for testing and lightweight helpers only.

**Operational errors become tool error strings; security errors propagate (issue #820)**

The wrapper has a deliberate fault-tolerance contract:

- **Operational exceptions** (network errors, parsing errors, missing data, etc.) are caught and returned to the LLM as an error string. This lets the model recover or apologise gracefully instead of crashing the conversation.
- **Security exceptions** — specifically `PermissionError` and `ToolConfirmationRequired` — propagate to the consumer. The agent stops; the user sees an approval prompt or an auth error. Tools should raise these explicitly when access is denied or confirmation is needed.

#### Framework Selection

Choose your framework based on your needs:

| Framework | Best For | Streaming Type | Structured Response | Visibility |
|-----------|----------|----------------|---------------------|------------|
| **PydanticAI** | Production use, all features | Event-based (thought/content/sources/final) | ✅ Optimized with automatic strategy selection | Full execution graph visibility |
| **Custom Adapters** | Specialized requirements | Implement in `_stream_raw()` | Implement in `_structured_response_raw()` | Depends on implementation |

```python
# Specify framework explicitly (PydanticAI is the default and recommended framework)
agent = await agents.for_document(
    document=123, corpus=1,
    framework=AgentFramework.PYDANTIC_AI
)

# Or set globally via Django settings
# LLMS_DEFAULT_AGENT_FRAMEWORK = "pydantic_ai"
```

## Agent Permission Model

The LLM framework implements a comprehensive permission system that ensures agents always execute with the calling user's permissions, never escalating privileges. This security model applies at multiple layers: consumer validation, agent factory filtering, and runtime tool execution.

### Permission Architecture Overview

The permission system operates on three key principles:

1. **User Context Inheritance**: Agents inherit and operate within the permissions of the user who invokes them
2. **Defense in Depth**: Multiple validation layers ensure permission checks cannot be bypassed
3. **Graceful Degradation**: Tools requiring unavailable permissions are filtered out rather than causing errors

### Tool Permission Flags

Every tool in the system can declare its permission requirements through three boolean flags:

```python
from opencontractserver.llms.tools.tool_factory import CoreTool

# Example tool with permission flags
dangerous_tool = CoreTool.from_function(
    my_function,
    name="delete_data",
    description="Delete data from the system",
    requires_corpus=True,          # Tool needs corpus context
    requires_approval=True,         # Tool needs user confirmation before execution
    requires_write_permission=True  # Tool performs write operations (NEW)
)
```

#### Permission Flag Definitions

| Flag | Type | Default | Description |
|------|------|---------|-------------|
| `requires_corpus` | `bool` | `False` | Tool requires a corpus context to function. Automatically filtered when `corpus=None`. |
| `requires_approval` | `bool` | `False` | Tool execution pauses for human approval before running. See "Tool Approval & Human-in-the-Loop" section. |
| `requires_write_permission` | `bool` | `False` | Tool performs write operations (create, update, delete). Filtered if user lacks WRITE permission on the corpus. |

### Permission Filtering Flow

The framework enforces permissions at three distinct layers, creating a defense-in-depth security model:

#### Layer 1: Consumer Validation (Entry Point)

**Location**: `opencontractserver/consumers/unified_agent_consumer.py`

The WebSocket consumer (or API endpoint) performs initial READ permission validation:

```python
# Consumer checks READ access before creating agent
if not corpus.user_can(user, "READ"):
    raise PermissionDenied("User lacks READ access to corpus")

# Similarly for documents
if not document.user_can(user, "READ"):
    raise PermissionDenied("User lacks READ access to document")
```

**Purpose**: Ensure the user can even access the document/corpus before creating an agent.

#### Layer 2: Agent Factory Filtering (Agent Creation)

**Location**: [`opencontractserver/llms/agents/agent_factory.py`](../../../opencontractserver/llms/agents/agent_factory.py)

When creating an agent, the `_user_has_write_permission()` helper (line ~29) checks `PermissionTypes.CRUD` on the resource. The filtering logic is inline in `create_document_agent()` and `create_corpus_agent()` — tools with `requires_write_permission=True` are skipped when the user lacks CRUD permission, and tools with `requires_corpus=True` are skipped when `corpus=None`. Tools with `requires_approval=True` are also stripped in public (unauthenticated) contexts.

**Purpose**: Prevent the agent from even seeing tools it shouldn't be allowed to use, reducing attack surface.

#### Layer 3: Runtime Validation (Tool Execution)

**Location**: [`opencontractserver/llms/tools/pydantic_ai_tools.py`](../../../opencontractserver/llms/tools/pydantic_ai_tools.py)

Even if a tool somehow makes it to execution, the `PydanticAIToolWrapper` (line ~223) wraps every tool call with two module-level pre-execution checks:

1. **`_check_user_permissions(ctx)`** (line ~20): Validates the user in `RunContext[PydanticAIDependencies]` has READ permission on the bound document/corpus. This is intentionally **not cached** — each tool call triggers fresh DB queries to detect mid-session permission revocations.
2. **`_validate_resource_id_params(ctx, **kwargs)`** (line ~132): Ensures `document_id`/`corpus_id` arguments match the agent's context, preventing prompt-injection attacks that attempt cross-resource access.

These checks run inside the generated `async_wrapper`/`sync_wrapper` functions, not as methods on the wrapper class.

**Purpose**: Final safety check that catches any bypass attempts or edge cases, ensuring operations never execute without proper permissions.

### Permission Enforcement by Tool Type

Different tools have different permission requirements based on their operations. The actual permission flags for each tool are set during agent initialization in [`agents/pydantic_ai_agents.py`](../../../opencontractserver/llms/agents/pydantic_ai_agents.py) (see `PydanticAIDocumentAgent.create()` ~line 1496 and `PydanticAICorpusAgent.create()` ~line 2111).

#### Read-Only Tools (No Special Permissions)

Tools like `load_document_summary`, `get_summary_token_length`, and `similarity_search` have no special permission flags. These are available to any user with READ permission on the resource.

#### Write Tools (Require Approval + Corpus)

Tools that modify data — such as `add_document_note`, `update_document_summary`, and `duplicate_annotations` — are marked with `requires_approval=True` and `requires_corpus=True`. These are automatically filtered out when no corpus is present, and pause for human approval before execution.

> **Note**: The `requires_write_permission` flag exists on `CoreTool` but is not currently set on any built-in tools. It's available for custom tools that need write-permission gating at the factory level.

### Security Guarantees

The permission model provides several critical security guarantees:

#### 1. No Privilege Escalation

Even if an agent's creator has higher permissions, the agent executes with the caller's permissions:

```python
# Scenario: Admin creates a shared agent, regular user calls it
# Admin has WRITE permission, regular user has READ only

# User invokes agent
agent = await agents.for_document(
    document=doc_id,
    corpus=corpus_id,
    user_id=regular_user_id  # ← Regular user's ID, not admin's
)

# Agent filters tools based on regular_user_id's permissions
# Write tools are automatically excluded, even though admin could use them
```

#### 2. Consistent Permission Checks

All validation layers use the `user_can()` permission API (`Model.objects.user_can()` / `obj.user_can()`). Layer 2 (factory) checks `PermissionTypes.CRUD` for write-gating, while Layer 3 (runtime) checks `PermissionTypes.READ` on every tool call.

#### 3. Fail-Safe Defaults

Tools default to the most restrictive permission model:

- `requires_write_permission=False` by default (must explicitly opt-in)
- Tools without permission flags are treated as read-only
- Missing user context blocks all write operations

### Permission Model in Action

Here's a complete example showing how permissions flow through the system:

```python
from opencontractserver.llms import agents

# User with READ-only permission attempts to use agent
user_id = 123  # User with READ permission, no WRITE
corpus_id = 456

# Create agent for document
agent = await agents.for_document(
    document=789,
    corpus=corpus_id,
    user_id=user_id
)

# Agent creation succeeds, but write tools are filtered:
# ✅ Available: load_md_summary, get_notes_for_document_corpus, similarity_search
# ❌ Filtered: add_document_note, update_document_summary, duplicate_annotations

# User can query the agent
response = await agent.chat("What are the key contract terms?")
# ✅ Works - uses read-only tools (similarity_search, load_md_summary)

# User asks agent to create a note
response = await agent.chat("Create a note summarizing the payment terms")
# ✅ Request succeeds, but agent can't call add_document_note (not available)
# Agent responds: "I cannot create notes as that tool is not available"

# If write tool somehow executed (impossible due to multi-layer checks):
# ❌ Layer 3 validation raises PermissionDenied exception
```

### Key Implementation Files

Understanding the permission model requires familiarity with these key files:

| File | Purpose | Key Functions/Classes |
|------|---------|----------------------|
| [`tools/tool_factory.py`](../../../opencontractserver/llms/tools/tool_factory.py) | Core tool definition with permission flags | `CoreTool` dataclass with `requires_approval`, `requires_corpus`, `requires_write_permission` flags |
| [`tools/tool_registry.py`](../../../opencontractserver/llms/tools/tool_registry.py) | Metadata registry of available tools | `AVAILABLE_TOOLS` tuple of `ToolDefinition` entries (metadata-only, not used for tool resolution) |
| [`agents/agent_factory.py`](../../../opencontractserver/llms/agents/agent_factory.py) | Agent creation with tool filtering | `_user_has_write_permission()`, inline filtering in `create_document_agent()`/`create_corpus_agent()` |
| [`agents/pydantic_ai_agents.py`](../../../opencontractserver/llms/agents/pydantic_ai_agents.py) | Tool assembly with flag assignment | `PydanticAIDocumentAgent.create()`, `PydanticAICorpusAgent.create()` |
| [`tools/pydantic_ai_tools.py`](../../../opencontractserver/llms/tools/pydantic_ai_tools.py) | Runtime permission validation | `_check_user_permissions()` (module-level), `_validate_resource_id_params()`, `PydanticAIToolWrapper` |
| [`utils/permissioning.py`](../../../opencontractserver/utils/permissioning.py) | Core permission checking utilities | `_default_user_can()` (the body behind `Manager.user_can` / `obj.user_can`) |
| `consumers/unified_agent_consumer.py` | WebSocket consumer with entry validation | Initial READ permission checks |

### Best Practices for Tool Development

When creating custom tools, follow these guidelines:

#### 1. Always Declare Write Requirements

```python
# ❌ Bad - doesn't declare write requirement
custom_tool = CoreTool.from_function(
    delete_annotations,
    name="delete_annotations",
    description="Delete annotations"
    # Missing requires_write_permission=True
)

# ✅ Good - explicitly declares write requirement
custom_tool = CoreTool.from_function(
    delete_annotations,
    name="delete_annotations",
    description="Delete annotations",
    requires_write_permission=True  # ← Explicit declaration
)
```

#### 2. Combine Permission Flags Appropriately

Some tools need multiple permission checks:

```python
# Tool that writes data AND needs user confirmation
risky_tool = CoreTool.from_function(
    bulk_delete_notes,
    name="bulk_delete_notes",
    description="Delete multiple notes at once",
    requires_corpus=True,          # Needs corpus context
    requires_approval=True,         # Dangerous operation
    requires_write_permission=True  # Modifies data
)
```

#### 3. Document Permission Requirements

```python
def create_annotation(document_id: int, label: str, text: str) -> str:
    """Create a new annotation on the document.

    **Permission Requirements**:
    - User must have WRITE permission on the corpus
    - Tool requires corpus context (requires_corpus=True)
    - Does not require approval (low-risk operation)

    Args:
        document_id: ID of the document to annotate
        label: Annotation label
        text: Text content to annotate

    Returns:
        Success message with annotation ID
    """
    # Implementation
    pass
```

### Testing Permission Enforcement

When testing agents with permission-gated tools:

```python
import pytest
from opencontractserver.llms import agents

@pytest.mark.asyncio
async def test_write_permission_filtering(user_factory, corpus_factory, document_factory):
    # Create user with READ-only permission
    user = await user_factory.create()
    corpus = await corpus_factory.create()
    document = await document_factory.create(corpus=corpus)

    # Grant READ permission only
    await grant_permission(user, corpus, "READ")

    # Create agent with write tools
    agent = await agents.for_document(
        document=document.id,
        corpus=corpus.id,
        user_id=user.id,
        tools=["add_document_note", "update_document_summary"]  # Write tools
    )

    # Verify write tools were filtered
    available_tool_names = [t.name for t in agent.config.tools]
    assert "add_document_note" not in available_tool_names
    assert "update_document_summary" not in available_tool_names

    # Verify read-only tools still work
    response = await agent.chat("What is this document about?")
    assert response.content  # Should succeed with read-only tools
```

### Migration Notes for Existing Tools

If you have existing custom tools that perform write operations:

1. **Add the `requires_write_permission` flag**:
   ```python
   # Update your tool definition
   my_tool = CoreTool.from_function(
       my_function,
       name="my_tool",
       description="Tool description",
       requires_write_permission=True  # ← Add this line
   )
   ```

2. **Test with read-only users**: Verify that users without WRITE permission can still use your agent (without the write tool)

3. **Document the permission requirement**: Update your tool's docstring to note the WRITE permission requirement

---

## Advanced Usage

### Custom Configuration

```python
from opencontractserver.llms.agents.core_agents import AgentConfig

# Create custom configuration (for low-level PydanticAIDocumentAgent.create())
config = AgentConfig(
    model_name="gpt-4-turbo",  # Note: field is model_name, not model
    temperature=0.2,
    max_tokens=2000,
    system_prompt="You are an expert legal analyst...",  # Note: Completely replaces any default prompt
    embedder_path="sentence-transformers/all-MiniLM-L6-v2",
    tools=["load_md_summary", "get_notes_for_document_corpus"], # Ensure tools are appropriate for context
    verbose=True,
    stream_observer=my_observer_function  # Optional: receive nested agent events
)

# Important: Custom system_prompt behavior
# - For chat/stream: Adds context about document analysis
# - For structured_response: Default includes verification steps to prevent hallucination
# - Any custom system_prompt completely replaces these defaults

# The `corpus` parameter is optional for document agents (use None for standalone).
# Note: agents.for_document() does not accept a `config` object directly.
# Pass individual parameters instead:
agent = await agents.for_document(
    document=123,
    corpus=None,
    model="gpt-4-turbo",
    temperature=0.2,
    max_tokens=2000,
    system_prompt="You are an expert legal analyst...",
    tools=["load_md_summary", "get_notes_for_document_corpus"],
    verbose=True,
)
```

### Conversation Patterns

#### Multi-turn Analysis

```python
# Persistent conversation for complex analysis
# The `corpus` parameter is optional for document agents (use None for standalone).
agent = await agents.for_document(
    document=123, # Use actual document ID or object
    corpus=1,     # Use actual corpus ID or object
    user_id=456,
    system_prompt="You are analyzing a legal contract. Build context across multiple questions."
)

# Build context over multiple interactions
overview = await agent.chat("Provide an overview of this contract")
risks = await agent.chat("What are the main risks for the buyer?")
recommendations = await agent.chat("Based on our discussion, what are your recommendations?")

# Access full conversation
info = agent.get_conversation_info()
print(f"Analyzed contract in {info['message_count']} messages")
```

#### Anonymous Sessions

```python
# Anonymous sessions - context maintained in memory only
# The `corpus` parameter is required for document agents.
agent = await agents.for_document(document=123, corpus=1)  # No storage, use actual document/corpus IDs
response1 = await agent.chat("What is this document about?")
response2 = await agent.chat("What are the key risks mentioned?")
response3 = await agent.chat("How do these risks compare?")
# Context flows between messages within the same session but is lost when session ends
```

#### Conversation Restoration

```python
# Resume a previous conversation
# The `corpus` parameter is optional for document agents (use None for standalone).
agent = await agents.for_document(
    document=123, # Use actual document ID or object
    corpus=1,     # Use actual corpus ID or object
    user_id=456,
    conversation_id=789  # Resume existing conversation
)

# Continue where you left off
response = await agent.chat("Following up on our previous discussion...")
```

### Advanced Tool Usage

#### Custom Tool Development

```python
from opencontractserver.llms.tools.tool_factory import CoreTool, ToolMetadata
from typing import List, Dict, Any

async def analyze_contract_clauses(document_id: int, clause_types: List[str]) -> Dict[str, Any]:
    """Analyze specific types of clauses in a contract.
    
    Args:
        document_id: The document to analyze
        clause_types: Types of clauses to look for (e.g., ['''payment''', '''termination'''])
    
    Returns:
        Dictionary with clause analysis results
    """
    # Your custom analysis logic here
    # Ensure this tool has access to the document_id context if needed,
    # or adapt it to receive necessary data directly.
    print(f"Analyzing document {document_id} for clauses: {clause_types}")
    return {
        "found_clauses": clause_types,
        "analysis": "Detailed analysis results...",
        "recommendations": ["Recommendation 1", "Recommendation 2"]
    }

# Create tool with rich metadata
clause_tool = CoreTool(
    function=analyze_contract_clauses, # This is an async function
    metadata=ToolMetadata(
        name="analyze_contract_clauses",
        description="Analyze specific types of clauses in a contract",
        parameter_descriptions={
            "document_id": "The ID of the document to analyze", # Agent context usually provides this
            "clause_types": "List of clause types to search for"
        }
    )
)

# Use in agent
# The `corpus` parameter is required for document agents.
agent = await agents.for_document(
    document=123, # Use actual document ID or object
    corpus=1,     # Use actual corpus ID or object
    tools=[clause_tool]
)
```

#### Tool Composition

```python
from opencontractserver.llms.tools import create_document_tools # Assuming this is already imported
# from opencontractserver.llms.tools.tool_factory import CoreTool # For custom_tools if defined elsewhere

# Assume clause_tool, risk_tool, compliance_tool are defined CoreTool instances
# For example:
# risk_tool = CoreTool.from_function(...)
# compliance_tool = CoreTool.from_function(...)

# Combine built-in and custom tools
standard_tools = create_document_tools()
# custom_tools = [clause_tool, risk_tool, compliance_tool] # Ensure these are defined

# The `corpus` parameter is required for document agents.
# agent = await agents.for_document(
#     document=123, # Use actual document ID or object
#     corpus=1,     # Use actual corpus ID or object
#     tools=standard_tools + custom_tools
# )
```

#### Tool Precedence and Overrides

When you pass tools to an agent, **caller-provided tools take precedence** over built-in defaults if there's a name conflict. This allows you to customize tool behavior without modifying the framework.

**When conflicts occur:**
- You pass a tool with `__name__` matching a default tool (e.g., `"update_document_description"`)
- The framework detects the duplicate and uses YOUR tool's configuration

**What gets overridden:**
- Tool description (affects LLM's understanding of the tool)
- `requires_approval` flag (enables/disables human-in-the-loop)
- `parameter_descriptions` (affects LLM parameter usage)
- The actual function implementation

**Example: Disabling approval for a tool**
```python
from opencontractserver.llms.tools.tool_factory import CoreTool
from opencontractserver.llms.tools.core_tools import update_document_description

# Create a version of update_document_description that doesn't require approval
no_approval_tool = CoreTool.from_function(
    update_document_description,
    name="update_document_description",  # Same name as default
    description="Update the document description (auto-approved)",
    requires_approval=False,  # Override the default's requires_approval=True
)

agent = await agents.for_document(
    document=123,
    corpus=1,
    tools=[no_approval_tool],  # Your tool replaces the default
)
# Now update_document_description won't pause for approval
```

**Precedence rules:**
1. Per-call `tools` parameter → highest priority
2. `AgentConfig.tools` → used if no per-call tools
3. Built-in defaults → lowest priority (replaced by above)

**Logging:** When a caller tool overrides a default, an INFO-level log is emitted:
```
Caller tool 'update_document_description' overrides default - using caller's configuration
```

**Security Considerations:**
- Only pass **trusted tools** via the `tools` parameter
- Overriding tools can **bypass `requires_approval` safeguards** that provide human-in-the-loop protection
- If tools originate from user-controlled configurations (e.g., stored in database), **validate them against an approved registry** before passing to agents
- The `deduplicate_tools()` utility in `opencontractserver/utils/tools.py` documents these security implications

### Vector Store Integration

#### Advanced Search

The `CoreAnnotationVectorStore` (which underlies framework-specific stores) allows for rich filtering in `VectorSearchQuery`.

```python
from opencontractserver.llms.vector_stores.core_vector_stores import (
    CoreAnnotationVectorStore, # Typically not instantiated directly by user, but via vector_stores.create()
    VectorSearchQuery
)
# from opencontractserver.llms import vector_stores # For vector_stores.create()

# Example: Creating a store instance (usually done via vector_stores.create())
# store = vector_stores.create(
#     user_id=123,
#     corpus_id=456,
#     embedder_path="sentence-transformers/all-MiniLM-L6-v2" # Handled by config
# )
# For demonstration, let's assume '''store''' is an instance of a CoreAnnotationVectorStore compatible store.

# Complex search with filters
# Available filters include Django ORM lookups on Annotation fields,
# and related fields like '''document__title''', '''annotation_label__name'''.
# Also supports: '''label_id''', '''annotation_type''', '''custom_metadata_filters''' (for JSONField queries),
# and '''text_content_filters'''.
query = VectorSearchQuery(
    query_text="payment obligations and penalties",
    similarity_top_k=20,
    filters={
        "annotation_label__name": "payment_clause", # Filter by label name
        "document__title__icontains": "service_agreement", # Filter by document title
        # "custom_metadata_filters": {"client_id": "XYZ"}, # Example for JSONField
        # "annotation_type": "TYPE_A" # Example for annotation type
    }
)

# results = await store.async_search(query) # Assuming store is available

# Process results
# for result in results:
#     annotation = result.annotation
#     print(f"Document: {annotation.document.title}")
#     print(f"Score: {result.similarity_score:.3f}")
#     print(f"Text: {annotation.raw_text[:200]}...")
#     print("---")
```

#### Framework-Specific Vector Stores

```python
# Custom vector stores can be implemented by extending CoreAnnotationVectorStore
# See opencontractserver/llms/vector_stores/core_vector_stores.py for the interface

# PydanticAI vector store
from opencontractserver.llms.vector_stores.pydantic_ai_vector_stores import PydanticAIAnnotationVectorStore

# pydantic_store = PydanticAIAnnotationVectorStore(
#     user_id=123, # Optional
#     corpus_id=456 # Or document_id
# )

# Both provide the same core functionality (search, async_search)
# with framework-specific optimizations and integration patterns.
# Typically created via opencontractserver.llms.vector_stores.create().
```

### Configuration Reference

The framework has two levels of configuration:

1. **`AgentConfig`** (in [`agents/core_agents.py`](../../../opencontractserver/llms/agents/core_agents.py)): Low-level dataclass used internally by agent implementations.
2. **`AgentAPI.for_document()` / `AgentAPI.for_corpus()`** (in [`api.py`](../../../opencontractserver/llms/api.py)): High-level API with individual keyword arguments. This is the recommended entry point.

> **Important**: `agents.for_document()` does **not** accept an `AgentConfig` object. It takes individual parameters (`model`, `temperature`, etc.) and constructs the config internally. Use `AgentConfig` directly only when working with the lower-level `PydanticAIDocumentAgent.create()` factory.

#### AgentConfig Fields

| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `user_id` | `Optional[int]` | `None` | User ID for conversation persistence; `None` for anonymous sessions |
| `model_name` | `str` | `"gpt-4o-mini"` | LLM model identifier (e.g., "gpt-4", "gpt-4-turbo") |
| `api_key` | `Optional[str]` | `None` | API key for LLM provider (defaults to Django settings) |
| `embedder_path` | `Optional[str]` | `None` | Embedding model path; auto-detected from corpus/document if not specified |
| `similarity_top_k` | `int` | `10` | Number of similar results to retrieve in vector searches |
| `streaming` | `bool` | `True` | Enable/disable streaming responses |
| `verbose` | `bool` | `True` | Enable verbose logging for debugging |
| `system_prompt` | `Optional[str]` | `None` | Custom system prompt; replaces default if provided |
| `temperature` | `float` | `0.7` | LLM temperature (0.0-2.0); lower = more deterministic |
| `max_tokens` | `Optional[int]` | `None` | Maximum tokens in response; `None` uses model default |
| `stream_update_freq` | `int` | `50` | Token interval for database updates during streaming |
| `stream_observer` | `Optional[Callable]` | `None` | Async callback for receiving nested agent events |
| `conversation` | `Optional[Conversation]` | `None` | Existing conversation object to resume |
| `conversation_id` | `Optional[int]` | `None` | ID of conversation to resume |
| `loaded_messages` | `Optional[list[ChatMessage]]` | `None` | Pre-loaded conversation messages |
| `store_user_messages` | `bool` | `True` | Whether to persist user messages to database |
| `store_llm_messages` | `bool` | `True` | Whether to persist LLM responses to database |
| `tools` | `list[Any]` | `[]` | List of tools available to the agent |

> **Default difference**: `AgentConfig.verbose` defaults to `True`, but `AgentAPI.for_document()` passes `verbose=False` by default. The API-level default takes precedence when using the high-level API.

#### Django Settings

The default framework for agents is configured via Django settings:

| Setting | Default | Description |
|---------|---------|-------------|
| `LLMS_DOCUMENT_AGENT_FRAMEWORK` | `AgentFramework.PYDANTIC_AI` | Framework for document agents |
| `LLMS_CORPUS_AGENT_FRAMEWORK` | `AgentFramework.PYDANTIC_AI` | Framework for corpus agents |
| `LLMS_VECTOR_STORE_FRAMEWORK` | `AgentFramework.PYDANTIC_AI` | Framework for vector stores. Independent of the agent framework — set this if vector store framework selection should differ from `LLMS_DOCUMENT_AGENT_FRAMEWORK`. |

#### Usage Examples

```python
from opencontractserver.llms import agents

# High-level API (recommended) — pass config as individual kwargs
agent = await agents.for_document(
    document=123,
    corpus=1,
    user_id=456,
    model="gpt-4-turbo",
    temperature=0.2,
    max_tokens=2000,
    system_prompt="You are an expert legal analyst...",
    verbose=True,
)

# Per-method overrides (don't affect the agent's base config)
response = await agent.structured_response(
    "Extract dates",
    DateInfo,
    model="gpt-4",  # Override just for this call
    temperature=0.0
)
```

#### Advanced Configuration

```python
# Anonymous agent (no persistence) — just omit user_id
agent = await agents.for_document(document=123, corpus=1)
# conversation_id will be None — nothing persisted

# Stream observer for nested visibility
# NOTE: stream_observer is only available on AgentConfig, not the high-level API.
# Use the lower-level PydanticAIDocumentAgent.create() when you need it:
from opencontractserver.llms.agents.core_agents import AgentConfig
from opencontractserver.llms.agents.pydantic_ai_agents import PydanticAIDocumentAgent

async def event_forwarder(event):
    await websocket.send_json({"type": event.type, "data": event.content})

config = AgentConfig(
    stream_observer=event_forwarder,  # Receives all nested events
    stream_update_freq=25  # More frequent updates
)

# Resume existing conversation (via high-level API)
agent = await agents.for_document(
    document=123,
    corpus=1,
    user_id=123,
    conversation_id=789,  # Resume this conversation
    temperature=0.5
)
```

## Error Handling

The framework provides structured error handling with specific exception types:

```python
from opencontractserver.llms import agents
from opencontractserver.documents.models import Document
from opencontractserver.corpuses.models import Corpus

try:
    agent = await agents.for_document(document=999999, corpus=999)
except Document.DoesNotExist:
    print("Document not found")
except Corpus.DoesNotExist:
    print("Corpus not found")
except Exception as e:
    print(f"Unexpected error: {e}")

# Graceful handling of LLM errors (example with a hypothetical agent)
# try:
#     response = await agent.chat("Complex query that might fail")
# except Exception as e:
#     # Framework handles LLM errors gracefully
#     print(f"LLM error: {e}")
#     # Conversation state is preserved
```

### Common Error Patterns

```python
# from opencontractserver.documents.models import Document # For Document.DoesNotExist
# from opencontractserver.llms.agents.core_agents import FinalEvent, UnifiedStreamResponse # For streaming errors
# import logging # For logger
# logger = logging.getLogger(__name__)

# Handle missing documents/corpuses
# async def get_agent_for_doc(document_id, corpus_id):
#     try:
#         agent = await agents.for_document(document=document_id, corpus=corpus_id)
#         return agent
#     except Document.DoesNotExist:
#         return {"error": "Document not found"}
#     except Corpus.DoesNotExist:
#         return {"error": "Corpus not found"}


# Handle conversation errors
# async def process_chat(agent, user_message):
#     try:
#         response = await agent.chat(user_message)
#         return response
#     except Exception as e:
#         # Log error but preserve conversation
#         logger.error(f"Chat error: {e}")
#         return {"error": "Failed to process message", "conversation_id": agent.get_conversation_id()}

# Handle streaming errors
# async def stream_message_handler(agent, message):
#     try:
#         async for event in agent.stream(message):
#             yield event
#     except Exception as e:
#         # Send error event based on streaming type
#         if hasattr(agent, '_uses_event_streaming'):  # Event-based streaming
#             # Note: Errors are now handled internally by the framework and
#             # emitted as ErrorEvent objects. This try/except is only needed
#             # for truly unexpected errors outside the agent's control.
#             yield ErrorEvent(
#                 error=str(e),
#                 content=f"Error: {e}",
#                 metadata={"error": str(e), "error_type": type(e).__name__}
#             )
#         else:  # Legacy streaming
#             yield UnifiedStreamResponse(
#                 content=f"Error: {e}",
#                 is_complete=True,
#                 metadata={"error": str(e)}
#             )
```

## Additional Modules

### Moderation Tools

The framework includes a set of moderation tools for threaded discussion management in [`tools/moderation_tools.py`](../../../opencontractserver/llms/tools/moderation_tools.py). These are used by moderation agents to manage conversations:

**Read-Only Tools** (no approval required):
- `get_thread_context` / `aget_thread_context` — Get thread metadata (title, lock/pin status, message count)
- `get_thread_messages` / `aget_thread_messages` — List messages in a thread (paginated)
- `get_message_content` / `aget_message_content` — Get a single message's full content

**Write Tools** (require approval):
- `lock_thread` / `alock_thread` — Lock a thread to prevent new messages
- `unlock_thread` / `aunlock_thread` — Unlock a locked thread
- `pin_thread` / `apin_thread` — Pin a thread for visibility
- `unpin_thread` / `aunpin_thread` — Unpin a pinned thread
- `delete_message` / `adelete_message` — Soft-delete a message with a reason

**Posting Tools**:
- `add_thread_message` / `aadd_thread_message` — Post a new message to a thread

Moderation tools can be included by name via the `tools=` parameter (e.g., `tools=["lock_thread", "get_thread_context"]`). The `_resolve_tools()` function in [`api.py`](../../../opencontractserver/llms/api.py) maps these names to their async versions automatically.

### One-shot completions (`agenerate_text`)

For lightweight, non-agentic "infra" LLM calls that don't need tool calling or
conversation persistence (e.g. conversation-title generation), use
[`completions.py`](../../../opencontractserver/llms/completions.py)'s
`agenerate_text`:

```python
from opencontractserver.llms.completions import agenerate_text

title = await agenerate_text(
    "Create a brief title for: What indemnities does this contract carry?",
    instructions="You create concise (max 5 word) titles.",
    corpus_preferred=corpus.preferred_llm,  # honour the corpus default
)
```

Unlike a hardcoded client, `agenerate_text` walks the **same model-resolution
chain as the agent factory** — per-call `model=` → per-corpus
`corpus_preferred` (`Corpus.preferred_llm`) → install-wide
`PipelineSettings.default_llm` → Django settings — and builds the model through
the credential-aware [`model_factory`](../../../opencontractserver/llms/model_factory.py),
so it is provider-agnostic (OpenAI, Anthropic, Google, Ollama, …). Incidental
LLM calls must use this helper rather than a hardcoded provider/model so they
respect the configured Singleton LLM instead of silently bypassing it.

### Conversation Vector Stores

In addition to the annotation-based vector stores, the framework includes conversation-specific vector stores for semantic search over chat messages:

- [`vector_stores/core_conversation_vector_stores.py`](../../../opencontractserver/llms/vector_stores/core_conversation_vector_stores.py) — Framework-agnostic conversation search
- [`vector_stores/pydantic_ai_conversation_vector_stores.py`](../../../opencontractserver/llms/vector_stores/pydantic_ai_conversation_vector_stores.py) — PydanticAI-specific wrapper

These enable agents to search their own conversation history for context, supporting long multi-turn interactions.

---

## Performance Considerations

The framework is designed for production use with several performance optimizations:

### Database Optimization

- **Async ORM**: All database operations use Django's async ORM capabilities.
- **Prefetch Related**: Vector stores prefetch related objects to avoid N+1 queries.
- **Connection Pooling**: Efficient database connection management (handled by Django).
- **Bulk Operations**: Message storage uses bulk operations where possible.

```python
# Example of optimized queryset in CoreAnnotationVectorStore
# from opencontractserver.annotations.models import Annotation
# queryset = Annotation.objects.select_related(
#     '''document''', '''annotation_label'''
# ).prefetch_related(
#     '''document__doc_type''' # Example of prefetching deeper relation
# ).filter(...)
```

### Caching Strategy

- **Embedding Caching**: Vector embeddings can be cached to avoid recomputation (implementation specific, may depend on embedder).
- **Model Caching**: LLM models are cached and reused across requests (handled by the underlying framework).
- **Vector Store Caching**: Search results can be cached for repeated queries (application-level or via custom store decorators).

### Memory Management

- **Streaming Responses**: Large responses are streamed to avoid memory issues.
- **Lazy Loading**: Django models use lazy loading for related objects unless explicitly prefetched.
- **Context Windows**: Conversation context is managed within model limits by the agent implementations.

### Source Management

- **Consistent Serialization**: Sources use a unified format across database storage and WebSocket transmission to eliminate conversion overhead.
- **Metadata Flattening**: Source metadata is flattened to top-level fields for efficient access and reduced nesting.
- **Similarity Scores**: All sources include similarity scores for relevance ranking and filtering.
- **Lazy Source Loading**: Sources are only populated when complete responses are available (final streaming chunk).

### Concurrency

- **Async Throughout**: All core operations are async-compatible.
- **Connection Limits**: Proper database connection pooling prevents resource exhaustion.
- **Rate Limiting**: Consider implementing rate limiting at the application or API gateway level for external LLM APIs.

```python
# Example of concurrent agent usage
import asyncio

# async def analyze_documents_concurrently(document_corpus_pairs): # List of (doc_id, corpus_id) tuples
#     agents_list = []
#     for doc_id, corpus_id in document_corpus_pairs:
#         agent = await agents.for_document(document=doc_id, corpus=corpus_id)
#         agents_list.append(agent)
    
#     tasks = [
#         agent.chat("Summarize key points")
#         for agent in agents_list
#     ]
    
#     results = await asyncio.gather(*tasks)
#     return results
```

## Testing

The framework includes comprehensive test coverage:

```python
# Example test patterns
import pytest
from opencontractserver.llms import agents
from opencontractserver.llms.agents.core_agents import UnifiedChatResponse
# from opencontractserver.documents.models import Document # For test setup
# from opencontractserver.corpuses.models import Corpus # For test setup

@pytest.mark.asyncio
async def test_document_agent_chat(db, document_factory, corpus_factory): # Assuming db and factories for setup
    # test_corpus = await corpus_factory.create()
    # test_document = await document_factory.create(corpus=test_corpus)
    # agent = await agents.for_document(document=test_document.id, corpus=test_corpus.id)
    # response = await agent.chat("Test message")
    
    # assert isinstance(response, UnifiedChatResponse)
    # assert response.content
    # assert response.user_message_id
    # assert response.llm_message_id
    pass # Placeholder for actual test structure

@pytest.mark.asyncio
async def test_conversation_persistence(db, document_factory, corpus_factory, user_factory): # Assuming factories
    # test_user = await user_factory.create()
    # test_corpus = await corpus_factory.create()
    # test_document = await document_factory.create(corpus=test_corpus)
    # agent = await agents.for_document(
    #     document=test_document.id,
    #     corpus=test_corpus.id,
    #     user_id=test_user.id
    # )
    
    # response1 = await agent.chat("First message")
    # response2 = await agent.chat("Second message")
    
    # # Verify conversation continuity via agent method
    # assert agent.get_conversation_id() is not None
    
    # # Verify message storage
    # info = agent.get_conversation_info()
    # assert info is not None
    # assert info.get('''message_count''', 0) >= 4  # 2 user + 2 LLM messages
    pass # Placeholder for actual test structure

@pytest.mark.asyncio
async def test_structured_response(db, document_factory, corpus_factory): # Assuming factories
    # from pydantic import BaseModel, Field
    # 
    # class ContractDates(BaseModel):
    #     effective_date: str = Field(description="Effective date")
    #     expiration_date: str = Field(description="Expiration date")
    # 
    # test_corpus = await corpus_factory.create()
    # test_document = await document_factory.create(corpus=test_corpus)
    # agent = await agents.for_document(
    #     document=test_document.id,
    #     corpus=test_corpus.id
    # )
    # 
    # # Test structured extraction
    # result = await agent.structured_response(
    #     "Extract the contract dates",
    #     ContractDates
    # )
    # 
    # # Verify result
    # assert result is None or isinstance(result, ContractDates)
    # if result:
    #     assert result.effective_date  # Required field
    # 
    # # Verify no conversation persistence
    # assert agent.get_conversation_id() is None  # Anonymous agent
    pass # Placeholder for actual test structure
```

## Contributing

The framework is designed for extensibility. Here's how to contribute:

### Adding Core Functionality

1. **Core Logic**: Add to `core_*.py` modules (e.g., `agents/core_agents.py`) or to the relevant submodule of the `tools/core_tools/` package.
2. **Framework Adapters**: Create new adapter in `agents/` (see "Adding a New Framework" below).
3. **Tools**: Add to the relevant submodule of the `tools/core_tools/` package (e.g. `tools/core_tools/notes.py`, `tools/core_tools/annotations.py`) for general tools, or within framework adapters for framework-specific tool handling. Re-export new public symbols from `tools/core_tools/__init__.py`. Ensure async versions are provided where appropriate.
4. **API**: Extend `api.py` for new high-level functionality if needed (e.g., new API classes like `AgentAPI`, `ToolAPI`).

### Adding a New Framework

To add support for a new LLM framework (e.g., LangChain, Haystack):

1. **Add Framework Enum** (in [`types.py`](../../../opencontractserver/llms/types.py)):
   ```python
   class AgentFramework(Enum):
       PYDANTIC_AI = "pydantic_ai"
       LANGCHAIN = "langchain"  # New framework
   ```

2. **Implement Agent Adapters**:
   - Create `agents/langchain_agents.py`
   - Define classes inheriting from `CoreAgentBase` (from [`agents/core_agents.py`](../../../opencontractserver/llms/agents/core_agents.py)).
   
   ```python
   # agents/langchain_agents.py
   from typing import AsyncGenerator, Type, TypeVar, Optional # For Python < 3.9, else from collections.abc import AsyncGenerator
   from opencontractserver.llms.agents.core_agents import (
       CoreAgentBase, SourceNode, AgentConfig, 
       UnifiedStreamEvent, ThoughtEvent, ContentEvent, FinalEvent
    )
   
   T = TypeVar("T")
    # from opencontractserver.documents.models import Document
    # from opencontractserver.corpuses.models import Corpus

    class LangChainDocumentAgent(CoreAgentBase):
        # def __init__(self, config: AgentConfig, conversation_manager: CoreConversationManager, underlying_agent: Any):
        #     super().__init__(config, conversation_manager)
        #     self.underlying_agent = underlying_agent
        pass # Simplified for brevity
        
        @classmethod
        async def create(
            cls, 
            # document: Document, 
            # corpus: Corpus, 
            config: AgentConfig, 
            conversation_manager: CoreConversationManager,
            tools: list = None
        ): # -> "LangChainDocumentAgent":
            # Initialize your LangChain agent here (e.g., langchain_agent = ...)
            # return cls(config, conversation_manager, langchain_agent)
            pass
        
        async def _chat_raw(self, message: str, **kwargs) -> tuple[str, list[SourceNode], dict]:
            # Implement raw chat using your framework (no DB operations)
            # Return tuple of (content, sources, metadata)
            # CoreAgentBase will handle all message storage automatically
            pass
        
        async def _stream_raw(self, message: str, **kwargs) -> AsyncGenerator[UnifiedStreamEvent, None]:
            # Implement raw streaming using your framework (no DB operations)
            # Yield UnifiedStreamEvent objects (ThoughtEvent, ContentEvent, etc.)
            # CoreAgentBase wrapper will handle message storage and incremental updates automatically
            # Call self._emit_observer_event(event) to forward events to any configured observer
            pass
        
        async def _structured_response_raw(
            self, 
            prompt: str, 
            target_type: Type[T],
            *,
            system_prompt: Optional[str] = None,
            model: Optional[str] = None,
            tools: Optional[list] = None,
            temperature: Optional[float] = None,
            max_tokens: Optional[int] = None,
            **kwargs
        ) -> Optional[T]:
            # Implement structured data extraction using your framework
            # Return instance of target_type or None on failure
            # No DB operations - this is ephemeral extraction
            pass
    ```

3. **Integrate into `UnifiedAgentFactory`** (in [`agents/agent_factory.py`](../../../opencontractserver/llms/agents/agent_factory.py)):
   ```python
   # elif framework == AgentFramework.LANGCHAIN:
   #     from opencontractserver.llms.agents.langchain_agents import LangChainDocumentAgent # Or CorpusAgent
   #     if for_document:
   #         return await LangChainDocumentAgent.create(
   #             document=document_obj, # Ensure document_obj and corpus_obj are passed
   #             corpus=corpus_obj,
   #             config=config,
   #             conversation_manager=conversation_manager,
   #             tools=framework_tools
   #         )
   #     else: # for_corpus
   #         # return await LangChainCorpusAgent.create(...)
   #         pass
   pass # Simplified
   ```

4. **Add Tool Support**:
   - Create `tools/langchain_tools.py` if needed.
   - Implement tool conversion from `CoreTool` to your framework's tool format.
   - Update [`tools/tool_factory.py`](../../../opencontractserver/llms/tools/tool_factory.py) (`UnifiedToolFactory`) to handle the new framework.

5. **Add Vector Store Support**:
   - Create `vector_stores/langchain_vector_stores.py`.
   - Implement adapter around `CoreAnnotationVectorStore` (see [`vector_stores/core_vector_stores.py`](../../../opencontractserver/llms/vector_stores/core_vector_stores.py)).
   - Update [`vector_stores/vector_store_factory.py`](../../../opencontractserver/llms/vector_stores/vector_store_factory.py).

6. **Testing**:
   - Create comprehensive tests following the patterns in existing test files (e.g., `test_pydantic_ai_agents.py`).
   - Test the public `chat()`, `stream()`, and `structured_response()` methods (which are provided by `CoreAgentBase`), conversation management, tool usage, and error handling.
   - Note that `_chat_raw()`, `_stream_raw()`, and `_structured_response_raw()` methods are internal implementation details and typically don't require separate testing—the public API tests exercise them indirectly.

By following these steps, you can extend the OpenContracts LLM framework to support new LLM technologies while maintaining the consistent, rich API with conversation management, source tracking, and structured responses.

### Code Style Guidelines

- **Type Hints**: All functions must have complete type hints.
- **Docstrings**: Use Google-style docstrings for all public methods.
- **Async/Await**: Use async patterns consistently throughout. Core functionalities should be async-first.
- **Error Handling**: Provide meaningful error messages and proper exception handling.
- **Testing**: Include comprehensive tests for all new functionality.

### Documentation Standards

- **API Documentation**: Document all public interfaces with examples.
- **Architecture Decisions**: Document significant design choices.
- **Migration Guides**: Provide migration paths for breaking changes.
- **Performance Notes**: Document performance characteristics and limitations.

### Event-Based Streaming Architecture

The framework's event-based streaming (available in PydanticAI) provides unprecedented visibility into agent execution:

```
User Query → PydanticAI Agent → Execution Graph Stream
                    ↓
┌─────────────────────────────────────────────────────┐
│ ThoughtEvent: "Received user prompt..."            │
├─────────────────────────────────────────────────────┤
│ ThoughtEvent: "Sending request to language model…" │
├─────────────────────────────────────────────────────┤
│ ContentEvent: "Based on the"                       │
│ ContentEvent: " contract analysis"                 │
│ ContentEvent: "..."                                │
├─────────────────────────────────────────────────────┤
│ ThoughtEvent: "Calling tool similarity_search(...)"│
│ SourceEvent: [SourceNode, SourceNode, ...]         │
│ ThoughtEvent: "Tool returned result"               │
├─────────────────────────────────────────────────────┤
│ FinalEvent: Complete answer + all sources + usage  │
└─────────────────────────────────────────────────────┘
                    ↓
           WebSocket Consumer
                    ↓
              Frontend UI
```

#### Event Types Detail

| Event Type | Purpose | Fields | When Emitted |
|------------|---------|--------|--------------|
| `ThoughtEvent` | Agent reasoning steps | `thought`, `metadata` | Execution graph transitions, tool decisions |
| `ContentEvent` | Answer content deltas | `content`, `accumulated_content`, `metadata` | Model text generation |
| `SourceEvent` | Source discovery | `sources`, `metadata` | Vector search results |
| `ApprovalNeededEvent` | Tool approval required | `pending_tool_call`, `metadata` | Flagged tool execution paused |
| `ApprovalResultEvent` | Approval decision recorded | `decision`, `pending_tool_call`, `metadata` | After resume_with_approval() called |
| `ResumeEvent` | Execution restarting | Standard event fields | After approval, before tool runs |
| `FinalEvent` | Complete results | `accumulated_content`, `sources`, `metadata` | End of execution |
| `ErrorEvent` | Error occurred during execution | `error`, `metadata` | Unrecoverable errors (e.g., rate limits, API failures) |

#### Implementation Benefits

- **Real-time Debugging**: See exactly where agents get stuck or make wrong decisions
- **Rich UI/UX**: Build sophisticated interfaces showing agent "thinking"
- **Performance Monitoring**: Track tool usage, token consumption, and execution time
- **Audit Trails**: Complete visibility into agent decision-making process

```python
# Example: Building a debug UI
async for event in agent.stream("Complex legal analysis"):
    timestamp = time.time()
    
    if event.type == "thought":
        debug_panel.add_thought(timestamp, event.thought, event.metadata)
    elif event.type == "content":
        answer_panel.append_text(event.content)
    elif event.type == "sources":
        source_panel.update_sources(event.sources)
    elif event.type == "approval_needed":
        # Human-in-the-loop: pause execution, request approval
        approval_panel.show_approval_request(
            tool_name=event.pending_tool_call["name"],
            tool_args=event.pending_tool_call["arguments"],
            message_id=event.llm_message_id
        )
        # UI triggers approval flow, which calls resume_with_approval()
    elif event.type == "final":
        debug_panel.add_summary(timestamp, event.metadata)
        performance_monitor.log_usage(event.metadata.get("usage", {}))
```

---

## Recent Improvements (v0.10+)

### Critical API Fixes
- **Fixed pydantic_ai Integration**: Corrected API usage to match pydantic_ai's actual interface
  - Changed `result_type` → `output_type` in Agent initialization
  - Changed `run_result.data` → `run_result.output` for accessing results
  - Framework now properly uses pydantic_ai's documented API

### Source Handling Enhancements
- **Automatic `json` Field Generation**: `SourceNode.to_dict()` now intelligently constructs coordinate information
  - **PDF Sources**: Automatically includes full `MultipageAnnotationJson` with token positions and bounding boxes from PlasmaPDF
  - **Text Sources**: Automatically generates simple `{start, end}` format from character positions
  - **Tool Simplification**: Tools only need to store raw metadata (`annotation_json`, `char_start`, `char_end`); transformation happens automatically
  - **Frontend Compatibility**: Ensures all sources have the `json` field required by frontend TypeScript interfaces
  - **Exact Text Search**: Fixed issue where exact text search sources weren't displaying in frontend
  - **Centralized Logic**: Single transformation point prevents duplication and ensures consistency across all source types

### Performance Optimizations
- **Simplified Structured Extraction**: Removed 200+ lines of complex prompt engineering
  - Leverages pydantic_ai's automatic output strategy selection
  - Framework automatically chooses between tool calling, JSON mode, or prompted extraction
  - Reduced token usage by eliminating repetitive instruction blocks
  - More reliable extraction with model-specific optimizations

### Code Quality Improvements
- **Cleaner Architecture**: Removed unnecessary complexity from structured extraction
- **Better Maintainability**: Aligned with pydantic_ai best practices
- **Future-proof Design**: Will automatically benefit from pydantic_ai improvements

### Migration Notes
If you were using the framework before these fixes, no changes are needed to your code. The improvements are internal optimizations that maintain the same public API while providing better reliability and performance. Sources now automatically include properly formatted coordinate information for both PDF and text documents.

---

This framework represents the evolution of OpenContracts' LLM capabilities, providing a foundation for sophisticated document analysis while maintaining simplicity and elegance in its API design.

---

### Standalone Document Agents (corpus=None)

Document agents now support a corpus-less mode by passing `corpus=None`.

- **When to use**: Analyze a single document outside any corpus context, or when a corpus has not (yet) been created.
- **Tool availability**: Tools marked with `requires_corpus=True` are automatically filtered out. Core tools like `load_document_md_summary`, `get_md_summary_token_length`, `load_document_txt_extract`, and vector search remain available.
- **Embedder selection**: If the document already has structural annotation embeddings, the framework prefers that embedder. Otherwise, it falls back to `settings.DEFAULT_EMBEDDER`.
- **Vector search**: Uses a document-level vector store (no `corpus_id` required) as long as an `embedder_path` is known.
- **Conversations**:
  - Authenticated users get normal conversation persistence and can resume later (via `conversation_id`).
  - Anonymous users create ephemeral sessions (no database writes, `get_conversation_id()` returns `None`).

Example:

```python
from opencontractserver.llms import agents

# Standalone document chat (no corpus) – corpus-dependent tools are filtered automatically
agent = await agents.for_document(document=123, corpus=None)

async for ev in agent.stream("Summarize the main points"):
    if ev.type == "content":
        print(ev.content, end="")
    elif ev.type == "final":
        print("\nDone. Sources:", len(ev.sources))
```

Notes:
- `get_document_summary` automatically falls back to loading the document’s markdown summary when no corpus is present.
- You can still supply custom tools; anything tagged `requires_corpus=True` is omitted gracefully.

---

### Default Tool Reference

Tools are **not** pre-defined as static `CoreTool` instances. They are assembled dynamically during agent creation in [`PydanticAIDocumentAgent.create()`](../../../opencontractserver/llms/agents/pydantic_ai_agents.py) and [`PydanticAICorpusAgent.create()`](../../../opencontractserver/llms/agents/pydantic_ai_agents.py). The table below reflects the tool names as registered by those factory methods.

Additionally, a smaller subset of **static** `CoreTool` instances is available via [`create_document_tools()`](../../../opencontractserver/llms/tools/tool_factory.py) in `tool_factory.py`. These are lower-level tools used when building agents outside the standard factory path.

#### Document Agent Tools

These tools are available on agents created via `agents.for_document()`:

| Tool Name (as registered) | Short Description | Requires Approval | Requires Corpus | Source |
|---------------------------|------------------|-------------------|-----------------|--------|
| `similarity_search` | Semantic vector search for relevant passages | No | No | Bound method on `PydanticAIAnnotationVectorStore` |
| `load_document_summary` | Load markdown summary (optionally truncated) | No | No | Wraps `aload_document_md_summary` |
| `get_summary_token_length` | Approximate token length of markdown summary | No | No | Wraps `aget_md_summary_token_length` |
| `get_document_text_length` | Total character length of plain-text extract | No | No | Inline closure in `create()` |
| `load_document_text` | Load full or partial plain-text extract | No | No | Wraps `aload_document_txt_extract` |
| `search_exact_text` | Find exact text matches with page/position info | No | No | Wraps `asearch_exact_text_as_sources` |
| `get_document_description` | Get document's description field | No | No | Wraps `aget_document_description` |
| `update_document_description` | Update document's description field | **Yes** | No | Wraps `aupdate_document_description` |
| `get_document_notes` | Retrieve notes for document in corpus | No | **Yes** | Wraps `aget_notes_for_document_corpus` |
| `search_document_notes` | Search notes by keyword | No | **Yes** | Wraps `asearch_document_notes` |
| `add_document_note` | Create a new note for the document | **Yes** | **Yes** | Wraps `aadd_document_note` |
| `update_document_note` | Update an existing note (new revision) | **Yes** | No | Wraps `aupdate_document_note` |
| `duplicate_annotations` | Duplicate annotations with a new label | **Yes** | **Yes** | Wraps `aduplicate_annotations_with_label` |
| `add_exact_string_annotations` | Add annotations for exact string matches | **Yes** | **Yes** | Wraps `aadd_annotations_from_exact_strings` |
| `get_document_summary` | Get latest summary content (falls back to md summary without corpus) | No | **Yes** | Wraps `aget_document_summary` |
| `get_document_summary_versions` | Version history of document summary | No | **Yes** | Wraps `aget_document_summary_versions` |
| `get_document_summary_diff` | Unified diff between two summary versions | No | **Yes** | Wraps `aget_document_summary_diff` |
| `update_document_summary` | Create / update document summary | **Yes** | **Yes** | Wraps `aupdate_document_summary` |

> Tools marked **Requires Corpus** are only included when `corpus` is not `None`. They are automatically filtered out for standalone document agents.

#### Corpus Agent Tools

Corpus agents (via `agents.for_corpus()`) get these **additional** tools on top of the vector search tool:

| Tool Name (as registered) | Short Description | Requires Approval | Source |
|---------------------------|------------------|-------------------|--------|
| `get_corpus_description` | Retrieve corpus markdown description | No | Wraps `aget_corpus_description` |
| `update_corpus_description` | Update corpus description | No | Wraps `aupdate_corpus_description` |
| `list_documents` | List documents in the current corpus | No | Inline closure in `create()` |
| `ask_document` | Ask a nested question to a per-document agent | No | Inline closure in `create()` |

#### Image Tools (from `create_document_tools()`)

The static [`create_document_tools()`](../../../opencontractserver/llms/tools/tool_factory.py) function also includes these image tools. They are available when using `create_document_tools()` directly but are **not** currently added by the standard `PydanticAIDocumentAgent.create()` factory:

| Tool Name | Short Description | Source |
|-----------|------------------|--------|
| `aget_page_image` | Get a visual image of a specific page from a PDF | `core_tools/page_images.py` |
| `alist_document_images` | List all images in a document (metadata only) | `image_tools.py` |
| `aget_document_image` | Get image data (base64) for a specific image | `image_tools.py` |
| `aget_annotation_images` | Get all images referenced by an annotation | `image_tools.py` |

#### String Tool Aliases

When passing tools by name to `agents.for_document()`, the [`_resolve_tools()`](../../../opencontractserver/llms/api.py) function in `api.py` supports these string aliases:

| String Alias | Resolves To |
|-------------|-------------|
| `"load_md_summary"`, `"summarize"` | `load_document_md_summary` |
| `"md_summary_length"` | `get_md_summary_token_length` |
| `"get_notes"`, `"notes"` | `get_notes_for_document_corpus` |
| `"note_length"` | `get_note_content_token_length` |
| `"partial_note"` | `get_partial_note_content` |
| `"load_document_text"` | `load_document_txt_extract` |
| `"get_document_summary"` | `get_document_summary` (sync) |
| `"update_document_summary"` | `update_document_summary` (sync) |
| `"add_document_note"` | `add_document_note` (sync) |
| `"search_document_notes"` | `search_document_notes` (sync) |
| `"get_page_image"` | `get_page_image` (sync) |

> **Note**: String-resolved tools use the **sync** versions from the `core_tools/` package. The standard agent factory uses **async** versions wrapped as inline closures. For production use, prefer passing `CoreTool` instances or letting the factory assemble defaults.

**Legend**
- **Requires Approval**: Tool execution pauses until a human approves the call.
- **Requires Corpus**: Tool is automatically filtered when `corpus=None`.

> **Note**: You can create your own tools with `CoreTool.from_function(...)` and set `requires_approval=True` or `requires_corpus=True` as needed. The framework will enforce approval gates and automatic corpus filtering for you.

---

## Runtime LLM configuration

OpenContracts lets you pick the LLM that powers an agent at three layers:

1. **Per-call override** — `agents.for_corpus(corpus=X, model="anthropic:claude-opus-4-6")`.
2. **Per-agent override** — `AgentConfiguration.preferred_llm`. Wins whenever the @-mention task or delegation sub-agent path runs this agent.
3. **Per-corpus default** — `Corpus.preferred_llm`. Wins when no explicit/per-agent override is set.
4. **Settings default** — `settings.DEFAULT_LLM` (preferred) → legacy `settings.OPENAI_MODEL` (back-compat) → hard fallback `"gpt-4o"`.

The resolver lives in [`opencontractserver/llms/llm_registry.py`](../../../opencontractserver/llms/llm_registry.py):

```python
from opencontractserver.llms.llm_registry import resolve_model_spec

# Returns "anthropic:claude-opus-4-6"
resolve_model_spec(
    explicit=None,
    agent_preferred=None,
    corpus_preferred="anthropic:claude-opus-4-6",
)
```

The [agent factory](../../../opencontractserver/llms/agents/agent_factory.py) calls the resolver automatically. Callers pick the slot that matches semantics:

- `agents.for_*(model="anthropic:claude-opus-4-6")` — per-call override. Wins over every persisted default.
- `agents.for_*(agent_preferred_llm=agent_config.preferred_llm)` — per-agent override. Wins over the corpus default but yields to a per-call `model=`. The @-mention task ([`agent_tasks.py`](../../../opencontractserver/tasks/agent_tasks.py)) and the delegation sub-agent path ([`delegation_tools.py`](../../../opencontractserver/llms/tools/delegation_tools.py)) use this slot.

### Model spec format

Specs follow [`pydantic-ai`](https://ai.pydantic.dev)'s `"{provider_key}:{model_name}"` convention:

| Spec                                    | Provider           | Model                |
| --------------------------------------- | ------------------ | -------------------- |
| `"openai:gpt-4o"`                       | OpenAI             | gpt-4o               |
| `"anthropic:claude-opus-4-6"`           | Anthropic          | claude-opus-4-6      |
| `"google-gla:gemini-2.0-flash"`         | Google (AI Studio) | gemini-2.0-flash     |
| `"ollama:llama3.3"`                     | Ollama (local)     | llama3.3             |

Bare strings (no colon — e.g. `"gpt-4o"`) are treated as `openai` models so legacy `OPENAI_MODEL` values keep working.

### Provider registry

LLM providers are registered as pluggable [pipeline components](../../pipelines/pipeline_overview.md) under [`opencontractserver/pipeline/llm_providers/`](../../../opencontractserver/pipeline/llm_providers/). Each subclass of [`BaseLLMProvider`](../../../opencontractserver/pipeline/base/llm_provider.py) declares:

- `provider_key` — pydantic-ai prefix used to build the spec and route credential lookups.
- `supported_models` — bare model names suggested to the UI (e.g. dropdown for `Corpus.preferred_llm`). Not strictly enforced at runtime so newly-released models can be used without a code change.
- `requires_api_key` — whether the provider needs a credential (`False` for `ollama`).
- a nested `Settings` dataclass — the standard pipeline-component config schema, carrying an `api_key` (`SECRET`) and, where applicable, a `base_url` (`OPTIONAL`). Built with the `llm_api_key_field` / `llm_base_url_field` helpers so the credentials are stored in the `PipelineSettings` singleton (the key encrypted) and editable live in the System Settings UI.

Add a new provider by dropping a file in `pipeline/llm_providers/` — the `PipelineComponentRegistry` walks the package on first access and registers every concrete subclass. Discover the registered providers programmatically via `get_all_llm_providers_cached()` or through the GraphQL `pipelineComponents { llmProviders { providerKey, supportedModels, requiresApiKey, settingsSchema { name settingType hasValue } } }` query.

### API keys & endpoints

LLM credentials are configurable **live** — exactly like a parser's or embedder's — without editing environment variables or redeploying. A superuser sets a provider's `api_key` (and optional `base_url`) in **System Settings → Pipeline Components**; the key is stored encrypted in the `PipelineSettings` singleton and the endpoint as a plaintext setting.

Resolution is **DB-wins / env-fallback**, applied by [`opencontractserver/llms/model_factory.py`](../../../opencontractserver/llms/model_factory.py):

- When a provider has a DB-configured `api_key`/`base_url`, `build_agent_model()` returns a concrete pydantic-ai model whose `Provider` carries those credentials — overriding the environment. A custom `base_url` lets you point OpenAI/Ollama at an OpenAI-compatible gateway or self-hosted server.
- When nothing is configured (the default), it returns the bare `"{provider}:{model}"` spec string and pydantic-ai resolves the credential from the provider-native environment variable (`OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `GEMINI_API_KEY`, …) exactly as before.

Any failure to build a credentialed model degrades to the env-fallback string, so a misconfiguration can never take the chat path down. The factory is invoked at every `make_pydantic_ai_agent` call site (document, corpus, and structured-output agents, plus the memory-curation tasks); it performs ORM access, so async call sites use the `abuild_agent_model()` wrapper.

Because the factory runs on every agent build, the resolved per-provider credentials are memoized in-process keyed on `(class_path, PipelineSettings.modified)` — the same cache key the reranker/embedder instance caches use. This skips the Fernet/PBKDF2 secret decryption on repeat builds while keeping rotation live: a superuser key change calls `PipelineSettings.save()`, which bumps `modified` and clears the singleton cache, so the next build misses the memo and re-decrypts (no redeploy, no staleness beyond the existing 5-minute `PipelineSettings` cache TTL). An out-of-band write that bypasses `save()` (e.g. `QuerySet.update`) should call `invalidate_credential_cache()`.

### Validation

`Corpus.save()` and `AgentConfiguration.save()` both run the resolver's `validate_model_spec()` — a malformed string or a provider with no registered `BaseLLMProvider` subclass raises `ValidationError({"preferred_llm": ...})`. The validator does not gate against `supported_models` so users aren't blocked from passing newly-released model names. Specs are normalised to canonical `"{provider}:{model}"` form on the way into the database.
