Merge branch 'main' into codex/fix/prefix-based-cache-optim

# Conflicts:
#	src/schemas/api.py
This commit is contained in:
adavyas 2026-03-28 19:59:43 -07:00
commit 131905d14a
36 changed files with 4008 additions and 443 deletions

View File

@ -32,7 +32,7 @@ LOG_LEVEL=INFO
# =============================================================================
# Connection URI for PostgreSQL database with pgvector support
# Must use postgresql+psycopg prefix for SQLAlchemy compatibility
DB_CONNECTION_URI=postgresql+psycopg://testuser:testpwd@localhost:5432/honcho
DB_CONNECTION_URI=postgresql+psycopg://postgres:postgres@localhost:5432/postgres
# Optional database settings
# DB_SCHEMA=public

View File

@ -31,7 +31,7 @@ For more information on closing issues using keywords, please check https://docs
## **Changelog**
<!-- 📛📛📛📛
Log of changes introduced in this release in the style fo https://keepachangelog.com/en/1.1.0/
Log of changes introduced in this release in the style of https://keepachangelog.com/en/1.1.0/
📛📛📛📛 -->
### **Added**

View File

@ -5,6 +5,19 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](http://keepachangelog.com/)
and this project adheres to [Semantic Versioning](http://semver.org/).
## [Unreleased]
### Fixed
- Memory leak in deriver: `_observation_locks` dict grew unboundedly with every unique (workspace, observer, observed) combination; replaced with `WeakValueDictionary` so locks are automatically evicted when no longer in use (DEV-1412)
- SQL injection vector in `dependencies.py`: parameterized `SET application_name` queries using `set_config()` instead of f-string interpolation (DEV-1400)
- NUL byte (`\x00`) crashes: all user-facing text inputs (message content, metadata, peer cards, queries) are now sanitized at the Pydantic schema level before reaching PostgreSQL (DEV-1400)
### Added
- JSONB metadata validation: max 100 top-level keys and max nesting depth of 5 on all metadata input fields (DEV-1400)
- Filter recursion depth limit: `_build_filter_conditions()` now enforces a max depth of 5 to prevent stack overflow from deeply nested filter dicts (DEV-1400)
## [3.0.3] - 2026-02-25
### Added
@ -454,7 +467,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
### Changed
- `/list` endpoints to not require a request body
- `metamessage_type` to `label` with backwards compatability
- `metamessage_type` to `label` with backwards compatibility
- Database Provisioning to rely on alembic
- Database Session Manager to explicitly rollback transactions before closing
the connection
@ -628,7 +641,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
- Authentication Middleware now implemented using built-in FastAPI Security
module
- Get by name routes for users and collections now include "name" in slug
- Python SDK moved to separate [respository](https://github.com/plastic-labs/honcho-python)
- Python SDK moved to separate [repository](https://github.com/plastic-labs/honcho-python)
### Fixed
@ -699,7 +712,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
### Changed
- session_data is now metadata
- session_data is a JSON field used python `dict` for compatability
- session_data is a JSON field used python `dict` for compatibility
## [0.0.2] — 2024-02-01

View File

@ -32,9 +32,11 @@ RUN --mount=type=cache,target=/root/.cache/uv \
# Place executables in the environment at the front of the path
ENV PATH="/app/.venv/bin:$PATH"
ENV HOME=/app
ENV UV_CACHE_DIR=/tmp/uv-cache
# Create non-root user and set ownership
RUN addgroup --system app && adduser --system --group app && chown -R app:app /app
RUN addgroup --system app && adduser --system --group app && mkdir -p /tmp/uv-cache && chown -R app:app /app /tmp/uv-cache
COPY --chown=app:app src/ /app/src/
COPY --chown=app:app migrations/ /app/migrations/

View File

@ -4,29 +4,43 @@ services:
build:
context: .
dockerfile: Dockerfile
entrypoint: ["sh", "docker/entrypoint.sh"]
depends_on:
database:
condition: service_healthy
redis:
condition: service_healthy
ports:
- 8000:8000
volumes:
- .:/app
- venv:/app/.venv
environment:
- DB_CONNECTION_URI=postgresql+psycopg://postgres:postgres@database:5432/postgres
- CACHE_URL=redis://redis:6379/0?suppress=true
env_file:
- .env
- path: .env
required: false
deriver:
build:
context: .
dockerfile: Dockerfile
entrypoint: ["uv", "run", "python", "-m", "src.deriver"]
entrypoint: ["/app/.venv/bin/python", "-m", "src.deriver"]
depends_on:
database:
condition: service_healthy
redis:
condition: service_healthy
volumes:
- .:/app
- venv:/app/.venv
environment:
- DB_CONNECTION_URI=postgresql+psycopg://postgres:postgres@database:5432/postgres
- CACHE_URL=redis://redis:6379/0?suppress=true
- METRICS_ENABLED=true
env_file:
- .env
- path: .env
required: false
database:
image: pgvector/pgvector:pg15
restart: always
@ -34,16 +48,16 @@ services:
- 5432:5432
command: ["postgres", "-c", "max_connections=800"]
environment:
- POSTGRES_DB=honcho
- POSTGRES_USER=testuser
- POSTGRES_PASSWORD=testpwd
- POSTGRES_DB=postgres
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=postgres
- POSTGRES_HOST_AUTH_METHOD=trust
- PGDATA=/var/lib/postgresql/data/pgdata
volumes:
- ./database/init.sql:/docker-entrypoint-initdb.d/init.sql
- pgdata:/var/lib/postgresql/data/
healthcheck:
test: ["CMD-SHELL", "pg_isready -U testuser -d honcho"]
test: ["CMD-SHELL", "pg_isready -U postgres -d postgres"]
interval: 5s
timeout: 5s
retries: 5
@ -59,6 +73,16 @@ services:
interval: 5s
timeout: 5s
retries: 5
prometheus:
image: prom/prometheus:v3.2.1
ports:
- 9090:9090
volumes:
- ./docker/prometheus.yml:/etc/prometheus/prometheus.yml:ro
- prometheus-data:/prometheus
depends_on:
api:
condition: service_started
grafana:
image: grafana/grafana:11.4.0
ports:
@ -70,6 +94,11 @@ services:
- GF_AUTH_ANONYMOUS_ORG_ROLE=Viewer
volumes:
- ./grafana-data:/var/lib/grafana
- ./docker/grafana-datasource.yml:/etc/grafana/provisioning/datasources/datasource.yml:ro
depends_on:
prometheus:
condition: service_started
volumes:
pgdata:
venv:
prometheus-data:

8
docker/entrypoint.sh Executable file
View File

@ -0,0 +1,8 @@
#!/bin/sh
set -e
echo "Running database migrations..."
/app/.venv/bin/python scripts/provision_db.py
echo "Starting API server..."
exec /app/.venv/bin/fastapi run --host 0.0.0.0 src/main.py

View File

@ -0,0 +1,9 @@
apiVersion: 1
datasources:
- name: Prometheus
type: prometheus
access: proxy
url: http://prometheus:9090
isDefault: true
editable: false

10
docker/prometheus.yml Normal file
View File

@ -0,0 +1,10 @@
global:
scrape_interval: 15s
scrape_configs:
- job_name: honcho-api
static_configs:
- targets: ["api:8000"]
- job_name: honcho-deriver
static_configs:
- targets: ["deriver:9090"]

View File

@ -27,6 +27,19 @@ Welcome to the Honcho changelog! This section documents all notable changes to t
### Honcho API and SDK Changelogs
<Tabs>
<Tab title="Honcho API">
<Update label="v3.0.4 (Unreleased)">
### Fixed
- Memory leak in deriver: `_observation_locks` dict grew unboundedly with every unique (workspace, observer, observed) combination; replaced with `WeakValueDictionary` so locks are automatically evicted when no longer in use (DEV-1412)
- SQL injection vector in `dependencies.py`: parameterized `SET application_name` queries using `set_config()` instead of f-string interpolation (DEV-1400)
- NUL byte (`\x00`) crashes: all user-facing text inputs (message content, metadata, peer cards, queries) are now sanitized at the Pydantic schema level before reaching PostgreSQL (DEV-1400)
### Added
- JSONB metadata validation: max 100 top-level keys and max nesting depth of 5 on all metadata input fields (DEV-1400)
- Filter recursion depth limit: `_build_filter_conditions()` now enforces a max depth of 5 to prevent stack overflow from deeply nested filter dicts (DEV-1400)
</Update>
<Update label="v3.0.3 (Current)">
### Added
@ -477,7 +490,7 @@ Welcome to the Honcho changelog! This section documents all notable changes to t
### Changed
- `/list` endpoints to not require a request body
- `metamessage_type` to `label` with backwards compatability
- `metamessage_type` to `label` with backwards compatibility
- Database Provisioning to rely on alembic
- Database Session Manager to explicitly rollback transactions before closing
the connection

View File

@ -97,22 +97,24 @@
"v3/guides/integrations/langgraph",
"v3/guides/integrations/mcp",
"v3/guides/integrations/n8n",
"v3/guides/integrations/openclaw"
"v3/guides/integrations/openclaw",
"v3/guides/integrations/hermes"
]
},
{
"group": "Tutorials",
"pages": [
"v3/guides/discord",
"v3/guides/granola",
"v3/guides/telegram",
"v3/guides/integrations/reachy-mini"
"v3/guides/integrations/reachy-mini",
"v3/guides/gmail"
]
},
{
"group": "Community Integrations",
"pages": [
"v3/guides/community/agent0",
"v3/guides/community/hermes"
"v3/guides/community/agent0"
]
},
{
@ -302,13 +304,12 @@
"pages": [
"v2/integrations/crewai",
"v2/integrations/langgraph",
"v2/integrations/mcp",
"v2/integrations/n8n"
"v2/integrations/mcp"
]
},
{
"group": "Application Interfaces",
"pages": ["v2/guides/discord", "v2/guides/telegram"]
"pages": ["v2/guides/discord", "v2/guides/n8n", "v2/guides/telegram"]
}
]
},

File diff suppressed because one or more lines are too long

Before

Width:  |  Height:  |  Size: 38 KiB

After

Width:  |  Height:  |  Size: 38 KiB

File diff suppressed because one or more lines are too long

Before

Width:  |  Height:  |  Size: 38 KiB

After

Width:  |  Height:  |  Size: 38 KiB

View File

@ -25,7 +25,7 @@ Use Honcho to build with Honcho! The [plugin](/v3/guides/integrations/claudecode
/plugin install honcho-dev@honcho # Skills to teach claude how to integrate Honcho
```
The markeplace also includes all the agent skills below, so you can use `/honcho-dev:integrate` directly after installing.
The marketplace also includes all the agent skills below, so you can use `/honcho-dev:integrate` directly after installing.
See the [full Claude Code integration guide](/v3/guides/integrations/claudecode) for setup details.

View File

@ -1,33 +0,0 @@
---
title: "Hermes Agent"
icon: 'bolt'
description: "Add AI-native memory to Hermes Agent"
sidebarTitle: 'Hermes Agent'
---
[Hermes Agent](https://github.com/NousResearch/hermes-agent) is an open-source AI agent from Nous Research with advanced tool-calling capabilities, terminal access, a skills system, and multi-platform deployment (Telegram, Discord, Slack, WhatsApp). The Honcho integration gives Hermes persistent cross-session memory and user modeling.
## Getting Started
Honcho support is built into Hermes Agent. See the [Hermes Agent README](https://github.com/NousResearch/hermes-agent) for full installation and configuration instructions.
The integration is opt-in and requires:
1. A Honcho API key from [app.honcho.dev](https://app.honcho.dev)
2. The `honcho-ai` package (`pip install hermes-agent[honcho]`)
3. Enabling Honcho in your Hermes config
## How It Works
The integration runs alongside Hermes's existing `USER.md` memory system. Honcho adds cross-session reasoning — prefetching user context into each turn, syncing exchanges for ongoing modeling, and exposing a dialectic tool (`query_user_context`) for the agent to query its understanding mid-conversation.
## Next Steps
<CardGroup cols={2}>
<Card title="Hermes Agent" icon="github" href="https://github.com/NousResearch/hermes-agent">
Source code, installation, and full documentation.
</Card>
<Card title="Honcho Architecture" icon="sitemap" href="../../documentation/core-concepts/architecture">
Learn about peers, sessions, and dialectic reasoning.
</Card>
</CardGroup>

631
docs/v3/guides/gmail.mdx Normal file
View File

@ -0,0 +1,631 @@
---
title: "Gmail"
icon: 'envelope'
description: "Load Gmail threads into Honcho to give your AI agents memory of email conversations."
sidebarTitle: 'Gmail'
---
In this tutorial, we'll walk through how to ingest your Gmail emails into Honcho. By the end, each email thread will be a Honcho session and each participant will be a peer — giving your agents memory of who said what across your email history.
This guide includes a ready-to-run Python script that handles everything: Gmail OAuth, thread fetching, participant extraction, and Honcho ingestion. You can run it as-is or use the full tutorial below to understand each piece as you go.
<Note>
The full script is available on [GitHub](https://github.com/plastic-labs/honcho/tree/main/examples/gmail). This is a developer-focused tutorial — it requires creating a Google Cloud project and OAuth credentials.
</Note>
## TL;DR
If you just want to get your emails into Honcho, here's everything you need.
### 1. Set Up Google Cloud Credentials
Follow Google's official [Gmail API Python Quickstart](https://developers.google.com/gmail/api/quickstart/python) to:
1. Create a Google Cloud project and enable the Gmail API
2. Configure the OAuth consent screen
3. Create OAuth credentials (select **Desktop app** as the application type)
4. Download the credentials JSON into the same directory as the script
The script auto-detects Google's default `client_secret_*.json` filename, so no renaming needed. The script only needs the `gmail.readonly` scope.
### 2. Install Dependencies
<CodeGroup>
```bash uv
uv pip install google-api-python-client google-auth-oauthlib honcho-ai
```
```bash pip
pip install google-api-python-client google-auth-oauthlib honcho-ai
```
</CodeGroup>
### 3. Preview with a Dry Run
<CodeGroup>
```bash uv
uv run honcho_gmail.py --dry-run --max-threads 5
```
```bash python
python honcho_gmail.py --dry-run --max-threads 5
```
</CodeGroup>
On first run, a browser window opens for OAuth consent. After authorizing, a `token.json` file is created — future runs skip this step.
### 4. Load into Honcho
<CodeGroup>
```bash uv
export HONCHO_API_KEY=your_api_key
uv run honcho_gmail.py --workspace gmail-inbox --max-threads 20
```
```bash python
export HONCHO_API_KEY=your_api_key
python honcho_gmail.py --workspace gmail-inbox --max-threads 20
```
</CodeGroup>
You can filter threads with Gmail search syntax:
<CodeGroup>
```bash uv
uv run honcho_gmail.py --query "from:alice@example.com"
uv run honcho_gmail.py --label INBOX
uv run honcho_gmail.py --query "after:2024/01/01 has:attachment" --max-threads 50
```
```bash python
python honcho_gmail.py --query "from:alice@example.com"
python honcho_gmail.py --label INBOX
python honcho_gmail.py --query "after:2024/01/01 has:attachment" --max-threads 50
```
</CodeGroup>
That's it — your emails are now queryable in Honcho. Read on if you want to understand how the script works and the design decisions behind it.
---
## Full Tutorial
### How Gmail Maps to Honcho
The core idea is straightforward: each Gmail thread becomes a Honcho session, and each email participant becomes a peer. Here's the full mapping:
| Gmail Concept | Honcho Concept | Details |
|---------------|----------------|---------|
| Your Gmail account | Workspace (`gmail`) | One workspace for all email data |
| Email participant | Peer | Email address as ID for deduplication |
| Email thread | Session (`gmail-thread-{id}`) | One session per thread, all participants attached |
| Individual email | Message | Attributed to the sender with original timestamp |
### Email as Peer ID
The script normalizes email addresses into URL-safe peer IDs — `alice@example.com` becomes `alice-example-com`. This means the same person is automatically deduplicated across threads. If Alice emails you in 10 different threads, all of those conversations accumulate under a single peer.
```python
def peer_id_from_email(email: str) -> str:
"""Convert email to a valid Honcho peer ID."""
return email.replace("@", "-").replace(".", "-")
```
This also means peers are consistent across data sources. If you import Granola meetings and Gmail threads for the same person, they merge under the same peer ID.
### Extracting Participants
Every email has a sender, recipients, and optionally CC/BCC addresses. The script extracts all of these to build a complete picture of who's involved in each thread:
```python
for m in msgs:
register_peer(m["from"])
for addr in parse_address_list(m["to"]):
register_peer(addr)
for addr in parse_address_list(m["cc"]):
register_peer(addr)
for addr in parse_address_list(m["bcc"]):
register_peer(addr)
```
Display names are extracted when available (e.g., `Alice Smith <alice@example.com>` → name: "Alice Smith"). When only an email is present, the script generates a name from the local part.
### Message Attribution and Timestamps
Each email becomes a message attributed to its sender via `peer.message()`. The original email timestamp is preserved using `created_at`, so Honcho sees the conversation in chronological order — not the order you imported it.
```python
honcho_msgs.append(peer.message(
content,
metadata={
"gmail_id": m["id"],
"subject": m["subject"],
"from": m["from"],
"to": m["to"],
"labels": m["labels"],
},
created_at=m["timestamp"],
))
```
### Multi-Peer Sessions
Each thread's session is linked to all participants using `session.add_peers()`. This means when you query Honcho about a peer, it has context not just from their messages but from the full conversations they participated in.
```python
session = honcho.session(session_id, metadata={
"gmail_thread_id": tid,
"subject": subject,
"source": "gmail",
"message_count": len(msgs),
})
session.add_peers(thread_peers)
```
### Stripping Quoted Replies
Email threads are full of quoted replies — each message repeats everything above it. The script strips these out so only the new content is stored per message, avoiding duplication in Honcho's memory:
```python
def strip_quoted_replies(text: str) -> str:
"""Strip quoted reply text, keeping only the new content."""
lines = text.split("\n")
clean_lines = []
for line in lines:
stripped = line.strip()
if re.match(r"^On .+wrote:\s*$", stripped):
break
if stripped.startswith(">"):
break
# ... other reply markers
clean_lines.append(line)
return "\n".join(clean_lines).rstrip()
```
### Querying After Import
Once your emails are in Honcho, you can query any peer:
```python
import os
from honcho import Honcho
honcho = Honcho(workspace_id="gmail-inbox", api_key=os.environ["HONCHO_API_KEY"])
alice = honcho.peer("alice-example-com")
print(alice.chat("What has Alice been discussing with me?"))
print(alice.chat("What action items has Alice mentioned?"))
```
---
## CLI Reference
```
usage: honcho_gmail.py [-h] [--workspace WORKSPACE] [--query QUERY]
[--label LABEL] [--max-threads N] [--dry-run]
[--credentials PATH] [--token PATH]
options:
--workspace, -w Honcho workspace ID (default: gmail)
--query, -q Gmail search query (e.g., 'from:alice@example.com')
--label, -l Gmail label to filter by (e.g., INBOX)
--max-threads, -n Max threads to fetch (default: 10)
--dry-run Preview without writing to Honcho
--credentials, -c Path to OAuth credentials JSON (auto-detects client_secret*.json)
--token, -t Path to store access token (default: token.json)
```
## Troubleshooting
### "No client_secret*.json file found"
Download OAuth credentials from Google Cloud Console and place the `client_secret_*.json` file in the same directory as the script.
### "Access blocked: This app's request is invalid"
Your OAuth consent screen may not be configured correctly. Ensure you've added the `gmail.readonly` scope.
### "Token has been expired or revoked"
Delete `token.json` and run the script again to re-authenticate.
### Rate Limits
The script includes a small delay when creating peers to avoid hitting Honcho's rate limits. For large imports (100+ threads), consider running in batches.
### Unique Messages
Use an AI assistant in your inbox? Want to parse out its messages differently? Feel free to modify and improve the structure of this script to fit your bespoke email setup. This script was written for agents and as such is easy to update with your coding assistant.
## Full Script
<Accordion title="honcho_gmail.py">
```python
#!/usr/bin/env python3
"""Load Gmail messages into Honcho.
Uses the Gmail API directly (with OAuth) to fetch emails and the Honcho Python SDK to store them.
Each Gmail thread becomes a Honcho session, each sender becomes a peer.
Prerequisites:
1. Create a Google Cloud project and enable the Gmail API
2. Create OAuth 2.0 credentials (Desktop app type)
3. Download the credentials JSON (client_secret_*.json) into this directory
4. Install dependencies:
pip install google-api-python-client google-auth-oauthlib honcho-ai
On first run, a browser window will open for OAuth consent. After authorizing,
a 'token.json' file will be created to store your credentials for future runs.
"""
import argparse
import base64
import glob
import os
import re
import time
from datetime import datetime, timezone
from email.header import decode_header, make_header
from email.utils import getaddresses, parseaddr
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
SCOPES = ["https://www.googleapis.com/auth/gmail.readonly"]
PEER_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$")
def find_credentials() -> str:
"""Find a Google OAuth credentials file in the current directory."""
matches = glob.glob("client_secret*.json")
if matches:
return matches[0]
raise FileNotFoundError(
"No client_secret*.json file found.\n"
"Download OAuth credentials from Google Cloud Console:\n"
"1. Go to console.cloud.google.com\n"
"2. Create/select a project and enable Gmail API\n"
"3. Create OAuth 2.0 credentials (Desktop app)\n"
"4. Download the JSON into this directory"
)
def get_gmail_service(credentials_file: str | None = None, token_file: str = "token.json"):
"""Authenticate and return a Gmail API service instance."""
creds = None
if os.path.exists(token_file):
creds = Credentials.from_authorized_user_file(token_file, SCOPES)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
print("Refreshing expired credentials...")
creds.refresh(Request())
else:
if credentials_file is None:
credentials_file = find_credentials()
print(f"Using credentials: {credentials_file}")
print("Opening browser for OAuth consent...")
flow = InstalledAppFlow.from_client_secrets_file(credentials_file, SCOPES)
creds = flow.run_local_server(port=0)
with open(token_file, "w") as token:
token.write(creds.to_json())
print(f"Credentials saved to {token_file}")
return build("gmail", "v1", credentials=creds)
def list_threads(service, query: str = None, label_ids: list = None, max_results: int = 10) -> list[dict]:
"""List Gmail threads with pagination support."""
all_threads = []
page_token = None
while len(all_threads) < max_results:
try:
params = {
"userId": "me",
"maxResults": min(100, max_results - len(all_threads)),
}
if query:
params["q"] = query
if label_ids:
params["labelIds"] = label_ids
if page_token:
params["pageToken"] = page_token
response = service.users().threads().list(**params).execute()
threads = response.get("threads", [])
all_threads.extend(threads)
page_token = response.get("nextPageToken")
if not page_token:
break
except HttpError as e:
print(f"Error listing threads: {e}")
break
return all_threads[:max_results]
def get_thread(service, thread_id: str) -> dict:
"""Fetch a complete Gmail thread with all messages."""
try:
return service.users().threads().get(
userId="me",
id=thread_id,
format="full"
).execute()
except HttpError as e:
print(f"Error fetching thread {thread_id}: {e}")
return {}
def _decode_header_str(header: str) -> str:
"""Decode an RFC 2047 encoded header string to plain Unicode."""
return str(make_header(decode_header(header)))
def extract_email(from_header: str) -> str:
"""Extract bare email from an RFC 5322 header value."""
_, addr = parseaddr(_decode_header_str(from_header))
return addr.lower().strip()
def extract_name(from_header: str) -> str:
"""Extract display name from an RFC 5322 header value."""
name, _ = parseaddr(_decode_header_str(from_header))
return name.strip() or from_header.strip()
def decode_body(payload: dict) -> str:
"""Recursively extract plain text from a Gmail message payload."""
if payload.get("mimeType") == "text/plain":
data = payload.get("body", {}).get("data", "")
if data:
return base64.urlsafe_b64decode(data).decode("utf-8", errors="replace")
parts = payload.get("parts", [])
for part in parts:
text = decode_body(part)
if text:
return text
return ""
def strip_quoted_replies(text: str) -> str:
"""Strip quoted reply text from an email body, keeping only the new content."""
lines = text.split("\n")
clean_lines = []
for line in lines:
stripped = line.strip()
if re.match(r"^On .+wrote:\s*$", stripped):
break
if stripped.startswith("---------- Forwarded message"):
break
if stripped.startswith(">"):
break
if re.match(r"^[-_]{10,}$", stripped):
break
clean_lines.append(line)
return "\n".join(clean_lines).rstrip()
def parse_address_list(header: str) -> list[str]:
"""Parse a comma-separated email header into individual addresses."""
if not header.strip():
return []
decoded = _decode_header_str(header)
return [
f"{name} <{addr}>" if name else addr
for name, addr in getaddresses([decoded])
if addr
]
def peer_id_from_email(email: str) -> str:
"""Convert email to a valid Honcho peer ID."""
peer_id = re.sub(r"[^A-Za-z0-9_-]+", "-", email).strip("-").lower()
peer_id = re.sub(r"-{2,}", "-", peer_id)
if not peer_id:
peer_id = "unknown-peer"
if not PEER_ID_PATTERN.fullmatch(peer_id):
raise ValueError(f"Generated peer ID is invalid: {peer_id!r}")
return peer_id
def fetch_thread_messages(service, thread_id: str) -> list[dict]:
"""Fetch all messages in a Gmail thread with full content."""
data = get_thread(service, thread_id)
messages = []
for msg in data.get("messages", []):
headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])}
body = strip_quoted_replies(decode_body(msg.get("payload", {})))
ts = int(msg.get("internalDate", "0")) / 1000
messages.append({
"id": msg["id"],
"thread_id": msg["threadId"],
"from": headers.get("From", ""),
"to": headers.get("To", ""),
"cc": headers.get("Cc", ""),
"bcc": headers.get("Bcc", ""),
"subject": headers.get("Subject", ""),
"date": headers.get("Date", ""),
"timestamp": datetime.fromtimestamp(ts, tz=timezone.utc),
"body": body.strip(),
"labels": msg.get("labelIds", []),
"snippet": msg.get("snippet", ""),
})
return messages
def main():
parser = argparse.ArgumentParser(description="Load Gmail messages into Honcho")
parser.add_argument("--workspace", "-w", default="gmail", help="Honcho workspace ID (default: gmail)")
parser.add_argument("--query", "-q", default=None, help="Gmail search query (e.g. 'from:alice@example.com')")
parser.add_argument("--label", "-l", default=None, help="Gmail label to filter by (e.g. INBOX)")
parser.add_argument("--max-threads", "-n", type=int, default=10, help="Max threads to fetch (default: 10)")
parser.add_argument("--dry-run", action="store_true", help="Print what would be loaded without writing to Honcho")
parser.add_argument("--credentials", "-c", default=None, help="Path to OAuth credentials JSON (auto-detects client_secret*.json)")
parser.add_argument("--token", "-t", default="token.json", help="Path to store/load access token")
args = parser.parse_args()
# Authenticate
print("Authenticating with Gmail API...")
service = get_gmail_service(args.credentials, args.token)
print(" Authenticated successfully!")
label_ids = [args.label] if args.label else None
# List threads
print(f"\nFetching up to {args.max_threads} threads from Gmail...")
threads = list_threads(service, query=args.query, label_ids=label_ids, max_results=args.max_threads)
print(f" Found {len(threads)} threads")
if not threads:
print("No threads found. Try adjusting --query or --label.")
return
# Fetch full messages for each thread
all_thread_messages = {}
seen_peers = {}
def register_peer(addr: str):
email = extract_email(addr)
if email and email not in seen_peers:
name = extract_name(addr)
if name.lower().strip() == email or "@" in name:
name = email.split("@")[0].replace(".", " ").title()
seen_peers[email] = {
"name": name,
"peer_id": peer_id_from_email(email),
"email": email,
}
for i, t in enumerate(threads):
tid = t["id"]
print(f" Fetching thread {i+1}/{len(threads)}: {tid}")
msgs = fetch_thread_messages(service, tid)
all_thread_messages[tid] = msgs
for m in msgs:
register_peer(m["from"])
for addr in parse_address_list(m["to"]):
register_peer(addr)
for addr in parse_address_list(m["cc"]):
register_peer(addr)
for addr in parse_address_list(m["bcc"]):
register_peer(addr)
# Summary
total_msgs = sum(len(v) for v in all_thread_messages.values())
print("\nSummary:")
print(f" Threads: {len(all_thread_messages)}")
print(f" Messages: {total_msgs}")
print(f" Unique participants: {len(seen_peers)}")
for email, info in seen_peers.items():
print(f" {info['peer_id']} ({info['name']} <{email}>)")
if args.dry_run:
print("\n[DRY RUN] Would create the above in Honcho. Showing first message per thread:")
for tid, msgs in all_thread_messages.items():
m = msgs[0]
body_preview = m["body"][:120].replace("\n", " ") if m["body"] else m["snippet"][:120]
print(f" Thread {tid}: {m['subject']}")
print(f" {m['from']} @ {m['date']}")
print(f" {body_preview}...")
return
# Load into Honcho
from honcho import Honcho
print(f"\nLoading into Honcho workspace '{args.workspace}'...")
honcho = Honcho(workspace_id=args.workspace)
# Create peers
peers = {}
for i, (email, info) in enumerate(seen_peers.items()):
if i > 0 and i % 4 == 0:
time.sleep(1)
peers[email] = honcho.peer(info["peer_id"], metadata={
"email": email,
"name": info["name"],
"source": "gmail",
})
print(f" Peer: {info['peer_id']}")
# Create sessions and messages per thread
for tid, msgs in all_thread_messages.items():
subject = msgs[0]["subject"] if msgs else "No subject"
session_id = f"gmail-thread-{tid}"
thread_peer_emails = set()
for m in msgs:
thread_peer_emails.add(extract_email(m["from"]))
for addr in parse_address_list(m["to"]):
thread_peer_emails.add(extract_email(addr))
for addr in parse_address_list(m["cc"]):
thread_peer_emails.add(extract_email(addr))
for addr in parse_address_list(m["bcc"]):
thread_peer_emails.add(extract_email(addr))
thread_peers = [peers[e] for e in thread_peer_emails if e in peers]
session = honcho.session(session_id, metadata={
"gmail_thread_id": tid,
"subject": subject,
"source": "gmail",
"message_count": len(msgs),
})
session.add_peers(thread_peers)
honcho_msgs = []
for m in msgs:
email = extract_email(m["from"])
peer = peers.get(email)
if not peer:
continue
content = m["body"] if m["body"] else m["snippet"]
if not content:
continue
honcho_msgs.append(peer.message(
content,
metadata={
"gmail_id": m["id"],
"subject": m["subject"],
"from": m["from"],
"to": m["to"],
"labels": m["labels"],
},
created_at=m["timestamp"],
))
if honcho_msgs:
session.add_messages(honcho_msgs)
print(f" Session {session_id}: {len(honcho_msgs)} messages — {subject[:60]}")
print(f"\nDone! Loaded {total_msgs} messages into workspace '{args.workspace}'.")
if __name__ == "__main__":
main()
```
</Accordion>
## Next Steps
<CardGroup cols={2}>
<Card title="Design Patterns" icon="cubes" href="/v3/documentation/core-concepts/design-patterns">
See how the Granola integration maps to common Honcho patterns.
</Card>
<Card title="GitHub Repository" icon="github" href="https://github.com/plastic-labs/honcho/tree/main/examples/gmail">
Source code and example script.
</Card>
</CardGroup>

953
docs/v3/guides/granola.mdx Normal file
View File

@ -0,0 +1,953 @@
---
title: "Granola"
icon: 'microphone'
description: "Import meeting notes and transcripts from Granola into Honcho"
sidebarTitle: 'Granola'
---
In this tutorial, we'll walk through how to import your [Granola](https://granola.ai) meeting data into Honcho. By the end, your meeting participants, transcripts, and summaries will be mapped onto Honcho's peer and session model — giving your agents queryable memory of the people you meet with.
This guide includes a ready-to-run Python script that handles everything: Granola OAuth, meeting fetching, participant detection, and interactive import. You can run it as-is or use the full tutorial below to understand each design decision.
<Note>
The full script is available on [GitHub](https://github.com/plastic-labs/honcho/tree/main/examples/granola).
</Note>
## TL;DR
If you just want to get your meetings into Honcho, here's everything you need.
### 1. Install Dependencies
<CodeGroup>
```bash uv
uv pip install honcho-ai httpx
```
```bash pip
pip install honcho-ai httpx
```
</CodeGroup>
### 2. Set Your API Key
```bash
export HONCHO_API_KEY="your-key-from-app.honcho.dev"
```
### 3. Run the Script
<CodeGroup>
```bash uv
uv run python honcho_granola.py
```
```bash python
python honcho_granola.py
```
</CodeGroup>
The script will:
1. Open your browser for Granola OAuth authentication
2. Fetch all meetings and their content
3. Walk you through each meeting interactively — confirm peers, choose import mode, skip meetings you don't want
4. Print a summary of what was transferred
That's it — your meetings are now queryable in Honcho. Read on if you want to understand how the script works and the design decisions behind it.
---
## Full Tutorial
### How Granola Maps to Honcho
The core idea is straightforward: each Granola meeting becomes a Honcho session, and each participant becomes a peer. Here's the full mapping:
| Granola Concept | Honcho Concept | Details |
|-----------------|----------------|---------|
| Your Granola account | Workspace (`granola`) | One workspace for all meetings |
| Meeting participant | Peer | Email as ID for deduplication across meetings |
| Individual meeting | Session (`meeting-{id}`) | One session per meeting |
| Transcript turns | Messages with attribution | Two-person calls get full speaker attribution |
| Meeting summary | Message from note creator | Multi-person calls store the summary |
### Email as Peer ID
The script uses email addresses as the basis for peer IDs, normalized to a URL-safe format (e.g., `alice@example.com` becomes `alice-example-com`). This ensures consistent identification across meetings — if you meet someone in 5 different calls, all conversations accumulate under the same peer.
```python
# These all resolve to the same peer:
honcho.peer("alice-example-com") # From Meeting A
honcho.peer("alice-example-com") # From Meeting B
```
This also means peers are consistent across data sources. If you import both Granola meetings and Gmail threads for the same person, they merge under the same peer ID.
### Auto-Detecting "Me"
Granola marks the note creator in its participant list with `(note creator)`. The script uses this to identify you automatically — no configuration needed.
```
Participants: You (note creator) from Your Company <you@example.com>,
Alice from Acme Corp <alice@example.com>
```
### Two-Person Calls: Full Attribution
When exactly one other participant is present *and* the transcript contains `Them:` turns, the script stores the transcript with speaker-attributed messages. Consecutive same-speaker turns are merged before storing, cleaning up the fragmentation that's common in raw transcripts.
```python
session.add_messages([
me.message("What's your timeline for the launch?"),
them.message("We're targeting Q2, but it depends on the API integration."),
])
```
### Multi-Person Calls: Summary Mode
Granola's transcript uses `Them:` for all non-creator speakers with no disambiguation — in a 4-person call, everyone else is just `Them:`. Rather than guess incorrectly, the script stores Granola's summary as your record of the meeting, with participants in metadata.
```python
session.add_messages([
me.message(
f"Meeting: Product Planning\n"
f"Date: Mar 5, 2026 2:00 PM\n"
f"Participants: Alice from Acme Corp, Bob from Widgets Inc\n\n"
f"{meeting_summary}",
metadata={
"participants": "Alice from Acme Corp, Bob from Widgets Inc",
"mode": "summary",
"granola_meeting_id": meeting_id,
}
)
])
```
The summary is attributed to you because it's *your* record of what happened. Granola captured your notes from a meeting where those people were present.
### Interactive Confirmation
For each meeting, you choose the import mode: two-person (full attribution), summary, or skip. For multi-person calls that are actually 1:1s (extra participants listed but didn't speak), you can override the detection and select the actual speaker.
### Noisy Transcripts Preserved
Granola's raw transcripts are often fragmented (`Me: Yeah. Them: Yeah. Me: And.`). The script merges consecutive same-speaker turns but otherwise preserves the raw content. Honcho's reasoning extracts signal from noisy data.
### Querying After Import
Once your meetings are in Honcho, you can query any peer:
```python
import os
from honcho import Honcho
honcho = Honcho(workspace_id="granola", api_key=os.environ["HONCHO_API_KEY"])
# Peer IDs are normalized from emails: alice@example.com -> alice-example-com
alice = honcho.peer("alice-example-com")
print(alice.chat("What is Alice working on?"))
print(alice.chat("What concerns has Alice raised?"))
me = honcho.peer("you-example-com")
print(me.chat("What topics do I discuss most frequently?"))
```
### Combining with Other Sources
Because meetings live in a standard Honcho workspace, you can enrich peer representations with data from other channels:
```python
# Same workspace, same peer — data accumulates
alice = honcho.peer("alice-example-com")
me = honcho.peer("you-example-com")
discord_session = honcho.session("discord-general-2024-03")
discord_session.add_messages([
alice.message("Just shipped the new API version!"),
me.message("Congrats! How's the migration guide coming?"),
])
# Queries now draw from both meeting transcripts AND Discord history
alice.chat("What has Alice shipped recently?")
```
---
## Troubleshooting
| Issue | Fix |
|-------|-----|
| Granola OAuth fails | Ensure you have a paid Granola plan (MCP requires Pro+). Clear cached token and retry. |
| Missing transcripts | Free tier has no transcript access. The script falls back to summary content. |
| 500 errors from Honcho | Check for null bytes or control characters in transcript content. The script sanitizes these automatically. |
| Rate limiting with many meetings | The script processes sequentially with delays. Honcho ingestion is async — don't poll for immediate results. |
## Full Script
<Accordion title="honcho_granola.py">
```python
#!/usr/bin/env python3
"""Load Granola meeting notes into Honcho.
Uses the Granola MCP server (with OAuth) to fetch meetings and the Honcho Python SDK
to store them. Each meeting becomes a Honcho session. Two-person meetings get full
speaker attribution; multi-person meetings are stored as summaries.
Prerequisites:
pip install honcho-ai httpx
Environment Variables:
HONCHO_API_KEY - Your Honcho API key (get from app.honcho.dev/api-keys)
Usage:
python honcho_granola.py
"""
import asyncio
import base64
import hashlib
import json
import os
import re
import secrets
import sys
import threading
import traceback
import webbrowser
from dataclasses import dataclass, field
from datetime import datetime, timezone
from http.server import HTTPServer, BaseHTTPRequestHandler
from typing import Any
from urllib.parse import parse_qs, urlencode, urlparse
import httpx
@dataclass
class Participant:
name: str
email: str | None = None
org: str | None = None
@dataclass
class ParsedParticipants:
note_creator: Participant | None = None
others: list[Participant] = field(default_factory=list)
@dataclass
class TranscriptTurn:
speaker: str
text: str
# Granola MCP + OAuth endpoints
GRANOLA_MCP_URL = "https://mcp.granola.ai/mcp"
AUTH_BASE = "https://mcp-auth.granola.ai"
OAUTH_REDIRECT_PORT = 8765
OAUTH_REDIRECT_URI = f"http://localhost:{OAUTH_REDIRECT_PORT}/callback"
# Honcho message size limit (25000 max, leave headroom)
MAX_MESSAGE_LEN = 24000
# ---------------------------------------------------------------------------
# OAuth callback handler (must be a class for BaseHTTPRequestHandler)
# ---------------------------------------------------------------------------
class _OAuthCallback(BaseHTTPRequestHandler):
auth_result: dict[str, str | None] = {"code": None, "error": None}
def do_GET(self):
params = parse_qs(urlparse(self.path).query)
if "code" in params:
_OAuthCallback.auth_result["code"] = params["code"][0]
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(b"<h1>Authenticated! You can close this window.</h1>")
elif "error" in params:
_OAuthCallback.auth_result["error"] = params.get("error_description", params["error"])[0]
self.send_response(400)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(f"<h1>Error: {_OAuthCallback.auth_result['error']}</h1>".encode())
else:
self.send_response(404)
self.end_headers()
def log_message(self, fmt, *args):
pass
# ---------------------------------------------------------------------------
# Granola OAuth + MCP
# ---------------------------------------------------------------------------
async def authenticate(http_client: httpx.AsyncClient) -> str:
"""Perform OAuth (DCR + PKCE) with Granola. Returns access token."""
_OAuthCallback.auth_result = {"code": None, "error": None}
print("\nAuthenticating with Granola...")
# Register client (DCR)
resp = await http_client.post(
f"{AUTH_BASE}/oauth2/register",
json={
"client_name": "Granola to Honcho Transfer",
"redirect_uris": [OAUTH_REDIRECT_URI],
"grant_types": ["authorization_code"],
"response_types": ["code"],
"token_endpoint_auth_method": "none",
},
)
if resp.status_code not in (200, 201):
raise RuntimeError(f"Client registration failed: {resp.status_code}")
client_id = resp.json().get("client_id")
# PKCE
verifier = secrets.token_urlsafe(32)
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
# Browser auth
auth_url = f"{AUTH_BASE}/oauth2/authorize?" + urlencode({
"client_id": client_id,
"redirect_uri": OAUTH_REDIRECT_URI,
"response_type": "code",
"state": "granola-honcho-transfer",
"code_challenge": challenge,
"code_challenge_method": "S256",
})
server = HTTPServer(("localhost", OAUTH_REDIRECT_PORT), _OAuthCallback)
thread = threading.Thread(target=server.handle_request)
thread.start()
print(" Opening browser for authentication...")
webbrowser.open(auth_url)
thread.join(timeout=120)
server.server_close()
auth_result = _OAuthCallback.auth_result
if auth_result["error"]:
raise RuntimeError(f"Authentication failed: {auth_result['error']}")
if not auth_result["code"]:
raise RuntimeError("Authentication timed out")
# Exchange code for token
resp = await http_client.post(
f"{AUTH_BASE}/oauth2/token",
data={
"grant_type": "authorization_code",
"code": auth_result["code"],
"redirect_uri": OAUTH_REDIRECT_URI,
"client_id": client_id,
"code_verifier": verifier,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
if resp.status_code != 200:
raise RuntimeError(f"Token exchange failed: {resp.status_code}")
print(" Authenticated successfully!")
return resp.json()["access_token"]
async def call_mcp_tool(
http_client: httpx.AsyncClient,
access_token: str,
tool_name: str,
arguments: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Call a Granola MCP tool, handling both JSON and SSE responses."""
resp = await http_client.post(
GRANOLA_MCP_URL,
json={
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {"name": tool_name, "arguments": arguments or {}},
},
headers={
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
},
)
if resp.status_code != 200:
raise RuntimeError(f"MCP call failed: {resp.status_code} - {resp.text}")
# SSE response
if "text/event-stream" in resp.headers.get("content-type", ""):
result = None
for line in resp.text.split("\n"):
if line.strip().startswith("data: "):
try:
parsed = json.loads(line.strip()[6:])
if "result" in parsed:
result = parsed
elif "error" in parsed:
raise RuntimeError(f"MCP error: {parsed['error']}")
except json.JSONDecodeError:
continue
if result:
final = result.get("result", {})
return final if isinstance(final, dict) else {"result": final}
raise RuntimeError("No result in SSE response")
# JSON response
result = resp.json()
if "error" in result:
raise RuntimeError(f"MCP error: {result['error']}")
return result.get("result", {})
def extract_mcp_text(result: dict[str, Any]) -> str:
"""Extract text from the first content block of an MCP result.
Raises ValueError if the response structure is unexpected.
"""
content = result.get("content", [])
if not isinstance(content, list) or not content:
raise ValueError(f"MCP response missing content array: {list(result.keys())}")
first = content[0]
if not isinstance(first, dict) or "text" not in first:
raise ValueError(f"MCP content block missing 'text' field: {first}")
return str(first["text"])
# ---------------------------------------------------------------------------
# Granola data fetching
# ---------------------------------------------------------------------------
async def list_meetings(
http_client: httpx.AsyncClient, access_token: str, limit: int = 100,
) -> list[dict[str, Any]]:
"""List meetings from Granola MCP. Parses Granola's XML-like response format."""
result = await call_mcp_tool(http_client, access_token, "list_meetings", {"limit": limit})
text = extract_mcp_text(result)
meetings: list[dict[str, Any]] = []
for match in re.finditer(r'<meeting\s+id="([^"]+)"\s+title="([^"]+)"\s+date="([^"]+)"', text):
mid, title, date = match.groups()
block_end = text.find("</meeting>", match.end())
block = text[match.end():block_end] if block_end != -1 else ""
p_match = re.search(r"<known_participants>\s*(.*?)\s*</known_participants>", block, re.DOTALL)
meetings.append({
"id": mid,
"title": title,
"date": date,
"participants": p_match.group(1).strip() if p_match else "",
})
return meetings
async def get_meeting_details(
http_client: httpx.AsyncClient, access_token: str, meeting_id: str,
) -> dict[str, Any]:
"""Get full meeting details including notes."""
result = await call_mcp_tool(http_client, access_token, "get_meetings", {"meeting_ids": [meeting_id]})
text = extract_mcp_text(result)
return {"id": meeting_id, "raw_content": text}
async def get_meeting_transcript(
http_client: httpx.AsyncClient, access_token: str, meeting_id: str,
max_retries: int = 3,
) -> str | None:
"""Get transcript for a meeting (paid tiers only).
Retries on rate limit responses with exponential backoff.
"""
for attempt in range(max_retries):
try:
result = await call_mcp_tool(http_client, access_token, "get_meeting_transcript", {"meeting_id": meeting_id})
text = extract_mcp_text(result)
except Exception as e:
print(f" Transcript unavailable: {e}")
return None
if not text or "no transcript" in text.lower():
return None
# Granola returns rate limit errors as content text, not HTTP errors
if "rate limit" in text.lower():
wait = 2 ** attempt * 3 # 3s, 6s, 12s
print(f" ⚠ Granola rate limit hit (attempt {attempt + 1}/{max_retries}), waiting {wait}s...")
await asyncio.sleep(wait)
continue
return text
print(f" ⚠ Transcript skipped after {max_retries} rate limit retries")
return None
async def fetch_all_meetings(
http_client: httpx.AsyncClient, access_token: str,
) -> list[dict[str, Any]]:
"""Fetch meeting list and enrich each with transcript and details."""
print("\nFetching meetings from Granola...")
meetings = await list_meetings(http_client, access_token, limit=500)
if not meetings:
print("No meetings found.")
return []
print(f" Found {len(meetings)} meetings. Fetching content...\n")
for i, m in enumerate(meetings, 1):
mid = m.get("id")
if not mid:
continue
transcript = await get_meeting_transcript(http_client, access_token, mid)
if transcript:
m["transcript"] = transcript
try:
m.update(await get_meeting_details(http_client, access_token, mid))
except Exception as exc:
print(f" Failed to fetch details for {mid}: {exc}")
has_t = "transcript" in m
has_s = bool(extract_summary(m))
label = "transcript+summary" if has_t and has_s else "transcript only" if has_t else "summary only" if has_s else "basic only"
print(f" [{i}/{len(meetings)}] {label}: {m.get('title', 'Untitled')[:45]}")
await asyncio.sleep(1.5) # rate limit
return meetings
# ---------------------------------------------------------------------------
# Parsing helpers
# ---------------------------------------------------------------------------
def parse_participants(participants_str: str) -> ParsedParticipants:
"""Parse Granola's participant string into structured participants.
Warns on unparsable entries instead of silently dropping them.
"""
result = ParsedParticipants()
if not participants_str:
return result
# Split on commas, but not inside angle brackets
entries, current, depth = [], [], 0
for ch in participants_str:
if ch == "<":
depth += 1
elif ch == ">":
depth = max(depth - 1, 0)
elif ch == "," and depth == 0:
entries.append("".join(current))
current = []
continue
current.append(ch)
if current:
entries.append("".join(current))
for entry in entries:
entry = entry.strip()
if not entry:
continue
is_creator = "(note creator)" in entry
clean = entry.replace("(note creator)", "").strip()
email_match = re.search(r"<([^>]+)>", clean)
email = email_match.group(1) if email_match else None
name = re.sub(r"\s*<[^>]+>", "", clean).strip()
if not name:
print(f" Warning: could not parse participant entry: {entry!r}")
continue
org = None
org_match = re.match(r"(.+?)\s+from\s+(.+)", name)
if org_match:
name, org = org_match.group(1).strip(), org_match.group(2).strip()
person = Participant(name=name, email=email, org=org)
if is_creator:
result.note_creator = person
else:
result.others.append(person)
return result
def parse_transcript_turns(raw: str) -> list[TranscriptTurn]:
"""Split a Granola transcript into speaker turns."""
# Unwrap JSON wrapper if present
try:
parsed = json.loads(raw)
if isinstance(parsed, dict) and "transcript" in parsed:
raw = str(parsed["transcript"])
except (json.JSONDecodeError, TypeError):
pass
parts = re.split(r"(?:^|\s{2,})(Me|Them):\s*", raw)
turns: list[TranscriptTurn] = []
i = 1
while i < len(parts) - 1:
text = parts[i + 1].strip()
if text:
turns.append(TranscriptTurn(speaker=parts[i], text=text))
i += 2
return turns
def extract_summary(meeting: dict[str, Any]) -> str:
"""Extract best available summary text from meeting data."""
candidates = []
for key in ("summary", "notes", "note", "meeting_notes", "description"):
val = meeting.get(key)
if isinstance(val, str) and val.strip():
candidates.append(val.strip())
raw = meeting.get("raw_content")
if isinstance(raw, str) and raw.strip():
candidates.append(raw.strip())
for c in candidates:
for tag in ("summary", "notes"):
m = re.search(rf"<{tag}>\s*(.*?)\s*</{tag}>", c, re.DOTALL)
if m:
return m.group(1).strip()
return candidates[0] if candidates else ""
def peer_id_from(value: str) -> str:
"""Normalize a name or email into a Honcho-safe peer ID."""
norm = re.sub(r"[^a-z0-9_-]+", "-", value.strip().lower())
norm = re.sub(r"-{2,}", "-", norm).strip("-_")
return (norm or "peer")[:100]
def sanitize(text: str) -> str:
"""Remove null bytes and control characters."""
return re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
def parse_date(date_str: str) -> datetime:
"""Parse Granola's date format into a timezone-aware datetime.
Raises ValueError if the date string doesn't match any known format.
"""
for fmt in ["%b %d, %Y %I:%M %p", "%b %d, %Y %I:%M:%S %p", "%B %d, %Y %I:%M %p"]:
try:
return datetime.strptime(date_str, fmt).replace(tzinfo=timezone.utc)
except ValueError:
continue
raise ValueError(f"Unrecognized date format: {date_str!r}")
# ---------------------------------------------------------------------------
# Honcho import helpers
# ---------------------------------------------------------------------------
def build_messages(
peer: Any,
content: str,
metadata: dict[str, object] | None,
created_at: datetime,
) -> list[Any]:
"""Build chunked messages for a single peer, attaching metadata to the first chunk."""
messages = []
content = sanitize(content)
for start in range(0, len(content), MAX_MESSAGE_LEN):
chunk = content[start:start + MAX_MESSAGE_LEN]
msg_meta = metadata if start == 0 else None
messages.append(peer.message(chunk, metadata=msg_meta, created_at=created_at))
return messages
def send_messages(session: Any, messages: list[Any]) -> None:
"""Send messages to a session in batches of 100."""
for batch_start in range(0, len(messages), 100):
session.add_messages(messages[batch_start:batch_start + 100])
def import_two_person(
honcho: Any,
session: Any,
me_peer_id: str,
them_peer_id: str,
turns: list[TranscriptTurn],
metadata: dict[str, object],
created_at: datetime,
) -> None:
"""Import a two-person meeting with speaker attribution."""
me_peer = honcho.peer(me_peer_id)
them_peer = honcho.peer(them_peer_id)
# Merge consecutive same-speaker turns
merged: list[TranscriptTurn] = []
for t in turns:
if merged and merged[-1].speaker == t.speaker:
merged[-1].text += " " + t.text
else:
merged.append(TranscriptTurn(speaker=t.speaker, text=t.text))
messages: list[Any] = []
for i, t in enumerate(merged):
peer = me_peer if t.speaker == "Me" else them_peer
msg_meta = metadata if i == 0 else None
messages.extend(build_messages(peer, t.text, msg_meta, created_at))
send_messages(session, messages)
print(f" -> Imported as 2-person ({me_peer_id} + {them_peer_id})")
def import_summary(
honcho: Any,
session: Any,
me_peer_id: str,
meeting: dict[str, Any],
metadata: dict[str, object],
created_at: datetime,
) -> None:
"""Import a meeting as a summary message."""
me_peer = honcho.peer(me_peer_id)
summary = extract_summary(meeting)
if not summary:
raw_t = meeting.get("transcript", "")
try:
parsed = json.loads(raw_t)
summary = str(parsed.get("transcript", "")) if isinstance(parsed, dict) else raw_t
except (json.JSONDecodeError, TypeError):
summary = raw_t
summary = summary or "No content available"
title = meeting.get("title", "Untitled")
date = meeting.get("date", "")
header = f"Meeting: {title}\nDate: {date}\nParticipants: {meeting.get('participants', '')}\n\n"
messages = build_messages(me_peer, header + summary, metadata, created_at)
send_messages(session, messages)
print(" -> Imported as summary")
def resolve_them_participant(others: list[Participant]) -> Participant | None:
"""Ask user to pick which participant is 'Them' from a multi-person meeting."""
for j, p in enumerate(others, 1):
email_str = f" <{p.email}>" if p.email else ""
print(f" {j}. {p.name}{email_str}")
idx_str = input(f" Who is 'Them'? [1-{len(others)}]: ").strip()
try:
return others[int(idx_str) - 1]
except (ValueError, IndexError):
print(" Invalid selection.")
return None
def review_meeting(
index: int,
total: int,
meeting: dict[str, Any],
participants: ParsedParticipants,
turns: list[TranscriptTurn],
) -> tuple[str, Participant | None]:
"""Display meeting info and get user's import choice.
Returns (mode, them_participant) where mode is one of:
- "two_person": import with speaker attribution using them_participant
- "summary": import as a single summary message
- "skip": skip this meeting
"""
title = meeting.get("title", "Untitled")
date = meeting.get("date", "")
creator = participants.note_creator
others = participants.others
me_turns = sum(1 for t in turns if t.speaker == "Me")
them_turns = len(turns) - me_turns
total_words = sum(len(t.text.split()) for t in turns)
print(f"\n{'─' * 60}")
print(f" [{index}/{total}] {title}")
print(f" Date: {date}")
if creator:
print(f" You: {creator.name} <{creator.email}>")
for j, p in enumerate(others, 1):
email_str = f" <{p.email}>" if p.email else ""
org_str = f" ({p.org})" if p.org else ""
print(f" {j}. {p.name}{email_str}{org_str}")
has_transcript = bool(meeting.get("transcript"))
if turns:
print(f" Transcript: {me_turns} Me, {them_turns} Them, ~{total_words} words")
if them_turns == 0:
print(" ** No 'Them' turns — nobody else spoke **")
if total_words < 30:
print(" ** Very short — might be empty **")
elif has_transcript:
raw = meeting["transcript"]
print(f" Transcript: present ({len(raw)} chars) but could not parse speaker turns")
print(f" Preview: {raw[:200]!r}")
else:
print(f" Content: {'summary available' if extract_summary(meeting) else 'metadata only'}")
# Two-person default: exactly one other participant with transcript
if len(others) == 1 and them_turns > 0:
them_label = others[0].name + (f" <{others[0].email}>" if others[0].email else "")
print(f"\n Detected: 2-person call (you + {them_label})")
choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower()
while choice not in ("", "s", "k"):
choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower()
if choice == "k":
return ("skip", None)
if choice == "s":
return ("summary", None)
return ("two_person", others[0])
# Multi-person with transcript
if len(others) > 1 and them_turns > 0:
print(f"\n {len(others)} participants")
choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower()
while choice not in ("", "2", "k"):
choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower()
if choice == "k":
return ("skip", None)
if choice == "2":
them = resolve_them_participant(others)
if them is None:
return ("summary", None)
return ("two_person", them)
return ("summary", None)
# No transcript or no other speakers
choice = input(" [Enter] summary / [k] skip: ").strip().lower()
while choice not in ("", "k"):
choice = input(" [Enter] summary / [k] skip: ").strip().lower()
if choice == "k":
return ("skip", None)
return ("summary", None)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
async def main():
print("=" * 60)
print(" Granola -> Honcho Meeting Notes Transfer")
print("=" * 60)
if not os.environ.get("HONCHO_API_KEY"):
print("\nError: HONCHO_API_KEY not set.")
print(" Get your key at: https://app.honcho.dev/api-keys")
sys.exit(1)
async with httpx.AsyncClient(timeout=60.0) as http_client:
try:
access_token = await authenticate(http_client)
meetings = await fetch_all_meetings(http_client, access_token)
if not meetings:
sys.exit(0)
from honcho import Honcho
honcho = Honcho(workspace_id="granola")
seen_peers: set[str] = set()
results = {"imported": 0, "skipped": 0, "failed": 0}
print("\n" + "=" * 60)
print(" Review each meeting")
print("=" * 60)
for i, m in enumerate(meetings, 1):
mid = m.get("id")
if not mid:
continue
participants = parse_participants(m.get("participants", ""))
turns = parse_transcript_turns(m["transcript"]) if m.get("transcript") else []
mode, them = review_meeting(i, len(meetings), m, participants, turns)
if mode == "skip":
print(" -> Skipped")
results["skipped"] += 1
continue
# Resolve creator peer
creator = participants.note_creator
me_source = (creator.email or creator.name) if creator else None
if not me_source:
print(" -> Skipped (no creator identifier)")
results["skipped"] += 1
continue
me_peer_id = peer_id_from(me_source)
if me_peer_id not in seen_peers:
print(f" New peer: {me_source} ({me_peer_id})")
seen_peers.add(me_peer_id)
try:
created_at = parse_date(m.get("date", ""))
session = honcho.session(f"meeting-{mid}")
metadata: dict[str, object] = {
"title": m.get("title", "Untitled"),
"date": m.get("date", ""),
"granola_meeting_id": mid,
"mode": mode,
}
if mode == "two_person" and them is not None:
them_source = them.email or them.name
them_peer_id = peer_id_from(them_source)
if them_peer_id not in seen_peers:
print(f" New peer: {them_source} ({them_peer_id})")
seen_peers.add(them_peer_id)
import_two_person(honcho, session, me_peer_id, them_peer_id, turns, metadata, created_at)
else:
import_summary(honcho, session, me_peer_id, m, metadata, created_at)
results["imported"] += 1
except ValueError as e:
print(f" -> FAILED: {e}")
results["failed"] += 1
except Exception as e:
print(f" -> FAILED: {e}")
traceback.print_exc()
results["failed"] += 1
# Done
print("\n" + "=" * 60)
print(" Transfer Complete!")
print("=" * 60)
print(f"\n Imported: {results['imported']}")
print(f" Skipped: {results['skipped']}")
print(f" Failed: {results['failed']}")
print(" Workspace: granola")
print(f" Peers: {sorted(seen_peers)}")
except KeyboardInterrupt:
print("\n\nAborted.")
sys.exit(0)
except Exception as e:
print(f"\nTransfer failed: {e}")
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
```
</Accordion>
## Next Steps
<CardGroup cols={2}>
<Card title="Design Patterns" icon="cubes" href="/v3/documentation/core-concepts/design-patterns">
See how the Granola integration maps to common Honcho patterns.
</Card>
<Card title="GitHub Repository" icon="github" href="https://github.com/plastic-labs/honcho/tree/main/examples/granola">
Source code and example script.
</Card>
</CardGroup>

View File

@ -0,0 +1,187 @@
---
title: "Hermes Agent + Honcho"
sidebarTitle: "Hermes Agent"
description: "How Hermes Agent uses Honcho for persistent cross-session memory and user modeling"
icon: "message-bot"
---
[Hermes Agent](https://github.com/NousResearch/hermes-agent) is an open-source AI agent from [Nous Research](https://nousresearch.com) with tool-calling, terminal access, a skills system, and multi-platform deployment (Telegram, Discord, Slack, WhatsApp). Honcho gives Hermes persistent cross-session memory and user modeling.
For setup, configuration, and CLI commands, see the [Hermes Agent Honcho docs](https://hermes-agent.nousresearch.com/docs/user-guide/features/honcho).
## What Honcho provides
Honcho acts as a long-term memory and user-model layer alongside Hermes' built-in memory files (`MEMORY.md` and `USER.md`).
It gives Hermes three capabilities:
1. **Prompt-time context injection** -- durable context about a user loaded into the prompt before generating a response.
2. **Cross-session continuity** -- recall of stable preferences, project history, and working context across conversations.
3. **Durable writeback** -- stable facts learned during a conversation stored back for future turns.
These sit alongside Hermes' local session history. Session history remembers the current conversation. Honcho remembers what should still matter later.
## Dual-peer architecture
Both the user and the AI agent have peer representations in Honcho:
- **User peer**: observed from user messages. Learns preferences, goals, communication style.
- **AI peer**: observed from assistant messages. Builds the agent's knowledge representation.
Both representations are injected into the system prompt, giving Hermes awareness of both who it's talking to and what it knows.
## Available tools
Hermes exposes four Honcho tools to the agent:
| Tool | What it does |
|---|---|
| `honcho_profile` | Fast peer card retrieval (no LLM). Returns curated key facts about the user. |
| `honcho_search` | Semantic search over memory. Returns raw excerpts ranked by relevance. |
| `honcho_context` | Dialectic Q&A powered by Honcho's LLM. Synthesizes answers from conversation history. |
| `honcho_conclude` | Writes durable facts to Honcho when the user states preferences, corrections, or important context. |
## Two memory layers
When Honcho is enabled, Hermes operates with two layer memory by default (`hybrid`):
**Local session history** -- the immediate transcript for the current chat, thread, or CLI session. Use it for recent turns, short-lived task context, and follow-up questions.
**Honcho memory** -- the semantic, cross-session layer. Use it for user preferences, durable project facts, cross-session continuity, and synthesized peer context.
## Running Honcho locally with Hermes
If you want to point Hermes at a local Honcho instance instead of the hosted API:
### Docker (quickest)
```bash
git clone https://github.com/plastic-labs/honcho.git
cd honcho
cp .env.template .env
cp docker-compose.yml.example docker-compose.yml
```
Edit `.env`:
```bash
OPENAI_API_KEY=your-openai-api-key
ANTHROPIC_API_KEY=your-anthropic-api-key
DB_CONNECTION_URI=postgresql+psycopg://postgres:postgres@database:5432/honcho
AUTH_USE_AUTH=false
```
```bash
docker compose up -d
curl http://localhost:8000/health
```
### Manual
```bash
git clone https://github.com/plastic-labs/honcho.git
cd honcho
uv sync
cp .env.template .env
```
Edit `.env` with a local or cloud Postgres connection string and API keys, then:
```bash
uv run alembic upgrade head
uv run fastapi dev src/main.py
```
Then update `~/.honcho/config.json` to point at your local instance:
```json
{
"apiKey": "not-needed-with-auth-disabled",
"baseUrl": "http://localhost:8000",
"hosts": {
"hermes": {
"workspace": "hermes",
"peerName": "your-name",
"aiPeer": "hermes",
"memoryMode": "hybrid",
"enabled": true
}
}
}
```
The `baseUrl` field overrides the default hosted API. With `AUTH_USE_AUTH=false` on the server, the `apiKey` value is ignored but the field must still be present.
See the full [self-hosting guide](/v3/contributing/self-hosting) for database options, cloud setup, and troubleshooting.
## Verifying the integration
Steps to test the integration via CLI and agentically by speaking to Hermes agent in natural language.
### 1. Check configuration
```bash
hermes honcho status
```
### 2. Test cross-session recall
In one conversation:
```text
Remember that my test phrase is velvet circuit.
```
In a fresh conversation (different thread, new CLI session):
```text
What is my test phrase?
```
If Hermes recalls "velvet circuit" after short-term context is gone, Honcho is working.
### 3. Test writeback
Tell Hermes a preference:
```text
Remember that I prefer terse answers.
```
Wait briefly if writes are asynchronous. Open a fresh conversation:
```text
How should you respond to me?
```
If Hermes answers with the stored preference, writeback is functioning.
## Session strategy
| Scope | When to use |
|----------------------|---------------------------------------------------------|
| Per-Session | A honcho session starts fresh each time a new Hermes session is created. Hermes remembers the user across sessions. |
| Per Directory | One honcho session per project directory. Context is scoped to each directory. Coding/project memory scoped to each repository/workspace. |
| Global (per user) | Continuity across all chats, threads, and projects. One honcho session globally for the user and Hermes agent. |
## Next steps
<CardGroup cols={2}>
<Card title="Hermes Agent Honcho Docs" icon="book" href="https://hermes-agent.nousresearch.com/docs/user-guide/features/honcho">
Setup, configuration, CLI commands, and all config options.
</Card>
<Card title="Hermes Agent Source" icon="github" href="https://github.com/NousResearch/hermes-agent">
Source code, installation, and full documentation.
</Card>
<Card title="Honcho Architecture" icon="sitemap" href="/v3/documentation/core-concepts/architecture">
Peers, sessions, and how reasoning works.
</Card>
<Card title="Self-Hosting Guide" icon="server" href="/v3/contributing/self-hosting">
Full local environment setup, database options, and troubleshooting.
</Card>
</CardGroup>

View File

@ -149,7 +149,7 @@ uv run python main.py
<Card title="Get Context" icon="database" href="/v3/documentation/features/get-context">
Retrieve formatted conversation history
</Card>
<Card title="Github code" icon="robot" href="https://github.com/plastic-labs/reachy-mini-honcho">
<Card title="GitHub code" icon="robot" href="https://github.com/plastic-labs/reachy-mini-honcho">
Dig into the code
</Card>
</CardGroup>

View File

@ -2,30 +2,34 @@
title: "Guides, Cookbooks, and Integrations"
sidebarTitle: 'Overview'
description: 'Helpful guides and design patterns for building with Honcho'
icon: 'hat-wizard'
icon: 'puzzle-piece'
---
<Note> Before you start a guide, follow [Quickstart](/v3/documentation/introduction/quickstart) to get up and running with Honcho in your language of choice. </Note>
Honcho plugs into whatever you're already building. Add memory to an AI assistant, connect an external data source, wire Honcho into your agent framework, or migrate from another provider.
These guides provide concrete examples and implementation patterns for building with Honcho. Whether you're integrating Honcho into existing platforms, exploring advanced features, or getting up and running quickly, you'll find working code you can adapt to your needs.
Each guide focuses on a specific use case with practical examples. The goal is to get you from idea to working prototype as quickly as possible, then provide the depth you need to scale and customize.
## Getting Started
Quick integration guides to get up and running:
## AI Assistants
Add persistent memory to AI assistants and agents:
<CardGroup cols={2}>
<Card title="MCP Integration" icon="link" href="/v3/guides/integrations/mcp">
Get Honcho running with a single prompt in Claude Code
<Card title="Claude Code" icon="terminal" href="/v3/guides/integrations/claude-code">
Long-term memory that survives context wipes, session restarts, and project switches
</Card>
<Card title="LangGraph" icon="diagram-project" href="/v3/guides/integrations/langgraph">
Add persistent memory and theory of mind to your LangGraph agents
<Card title="MCP Server" icon="star-of-life" href="/v3/guides/integrations/mcp">
Add Honcho memory to Claude Desktop, Cursor, Windsurf, Cline, and any MCP client
</Card>
<Card title="Hermes Agent" icon="bolt" href="/v3/guides/community/hermes">
Cross-session memory for Nous Research's Hermes agent
</Card>
<Card title="OpenClaw" icon="lobster" href="/v3/guides/integrations/openclaw">
Memory across every channel — WhatsApp, Telegram, Discord, Slack, and more
</Card>
<Card title="Agent Zero" icon="triangle" href="/v3/guides/community/agent0">
Persistent memory plugin for the Agent Zero framework
</Card>
</CardGroup>
## Showcase
Real-world examples of what you can build with Honcho:
## Platform Connectors
Connect external platforms to Honcho:
<CardGroup cols={2}>
<Card title="Discord Bot" icon="discord" href="/v3/guides/discord">
@ -34,7 +38,37 @@ Real-world examples of what you can build with Honcho:
<Card title="Telegram Bot" icon="telegram" href="/v3/guides/telegram">
Create a Telegram bot with persistent user understanding
</Card>
<Card title="Gmail" icon="envelope" href="/v3/guides/gmail">
Import email threads into Honcho — peers, sessions, and messages from your inbox
</Card>
<Card title="Granola" icon="calendar" href="/v3/guides/granola">
Ingest meeting transcripts with speaker turns and participant data
</Card>
<Card title="Reachy Mini" icon="robot" href="/v3/guides/integrations/reachy-mini">
Build an embodied voice robot that remembers users across sessions
Build an embodied voice robot with long-term memory
</Card>
</CardGroup>
## Agent Frameworks
Use Honcho as a memory layer in your agent orchestration stack:
<CardGroup cols={2}>
<Card title="LangGraph" icon="diagram-project" href="/v3/guides/integrations/langgraph">
Add persistent memory and theory of mind to your LangGraph agents
</Card>
<Card title="CrewAI" icon="users-gear" href="/v3/guides/integrations/crewai">
Give CrewAI agents memory that persists across sessions
</Card>
<Card title="n8n" icon="share-nodes" href="/v3/guides/integrations/n8n">
Build intelligent automation workflows with persistent memory
</Card>
</CardGroup>
## Migrations
Coming from another memory provider?
<CardGroup cols={2}>
<Card title="Migrate from Mem0" icon="arrow-right-arrow-left" href="/v3/guides/migrations/mem0">
Transfer your data and update your integration code
</Card>
</CardGroup>

View File

@ -0,0 +1,374 @@
#!/usr/bin/env python3
"""Load Gmail messages into Honcho.
Uses the Gmail API directly (with OAuth) to fetch emails and the Honcho Python SDK to store them.
Each Gmail thread becomes a Honcho session, each sender becomes a peer.
Prerequisites:
1. Create a Google Cloud project and enable the Gmail API
2. Create OAuth 2.0 credentials (Desktop app type)
3. Download the credentials JSON (client_secret_*.json) into this directory
4. Install dependencies:
pip install google-api-python-client google-auth-oauthlib honcho-ai
On first run, a browser window will open for OAuth consent. After authorizing,
a 'token.json' file will be created to store your credentials for future runs.
"""
import argparse
import base64
import glob
import os
import re
import time
from datetime import datetime, timezone
from email.header import decode_header, make_header
from email.utils import getaddresses, parseaddr
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
SCOPES = ["https://www.googleapis.com/auth/gmail.readonly"]
PEER_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$")
def find_credentials() -> str:
"""Find a Google OAuth credentials file in the current directory."""
matches = glob.glob("client_secret*.json")
if matches:
return matches[0]
raise FileNotFoundError(
"No client_secret*.json file found.\n"
"Download OAuth credentials from Google Cloud Console:\n"
"1. Go to console.cloud.google.com\n"
"2. Create/select a project and enable Gmail API\n"
"3. Create OAuth 2.0 credentials (Desktop app)\n"
"4. Download the JSON into this directory"
)
def get_gmail_service(credentials_file: str | None = None, token_file: str = "token.json"):
"""Authenticate and return a Gmail API service instance."""
creds = None
if os.path.exists(token_file):
creds = Credentials.from_authorized_user_file(token_file, SCOPES)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
print("Refreshing expired credentials...")
creds.refresh(Request())
else:
if credentials_file is None:
credentials_file = find_credentials()
print(f"Using credentials: {credentials_file}")
print("Opening browser for OAuth consent...")
flow = InstalledAppFlow.from_client_secrets_file(credentials_file, SCOPES)
creds = flow.run_local_server(port=0)
with open(token_file, "w") as token:
token.write(creds.to_json())
print(f"Credentials saved to {token_file}")
return build("gmail", "v1", credentials=creds)
def list_threads(service, query: str = None, label_ids: list = None, max_results: int = 10) -> list[dict]:
"""List Gmail threads with pagination support."""
all_threads = []
page_token = None
while len(all_threads) < max_results:
try:
params = {
"userId": "me",
"maxResults": min(100, max_results - len(all_threads)),
}
if query:
params["q"] = query
if label_ids:
params["labelIds"] = label_ids
if page_token:
params["pageToken"] = page_token
response = service.users().threads().list(**params).execute()
threads = response.get("threads", [])
all_threads.extend(threads)
page_token = response.get("nextPageToken")
if not page_token:
break
except HttpError as e:
print(f"Error listing threads: {e}")
break
return all_threads[:max_results]
def get_thread(service, thread_id: str) -> dict:
"""Fetch a complete Gmail thread with all messages."""
try:
return service.users().threads().get(
userId="me",
id=thread_id,
format="full"
).execute()
except HttpError as e:
print(f"Error fetching thread {thread_id}: {e}")
return {}
def _decode_header_str(header: str) -> str:
"""Decode an RFC 2047 encoded header string to plain Unicode."""
return str(make_header(decode_header(header)))
def extract_email(from_header: str) -> str:
"""Extract bare email from an RFC 5322 header value."""
_, addr = parseaddr(_decode_header_str(from_header))
return addr.lower().strip()
def extract_name(from_header: str) -> str:
"""Extract display name from an RFC 5322 header value."""
name, _ = parseaddr(_decode_header_str(from_header))
return name.strip() or from_header.strip()
def decode_body(payload: dict) -> str:
"""Recursively extract plain text from a Gmail message payload."""
if payload.get("mimeType") == "text/plain":
data = payload.get("body", {}).get("data", "")
if data:
return base64.urlsafe_b64decode(data).decode("utf-8", errors="replace")
parts = payload.get("parts", [])
for part in parts:
text = decode_body(part)
if text:
return text
return ""
def strip_quoted_replies(text: str) -> str:
"""Strip quoted reply text from an email body, keeping only the new content."""
lines = text.split("\n")
clean_lines = []
for line in lines:
stripped = line.strip()
if re.match(r"^On .+wrote:\s*$", stripped):
break
if stripped.startswith("---------- Forwarded message"):
break
if stripped.startswith(">"):
break
if re.match(r"^[-_]{10,}$", stripped):
break
clean_lines.append(line)
return "\n".join(clean_lines).rstrip()
def parse_address_list(header: str) -> list[str]:
"""Parse a comma-separated email header into individual addresses."""
if not header.strip():
return []
decoded = _decode_header_str(header)
return [
f"{name} <{addr}>" if name else addr
for name, addr in getaddresses([decoded])
if addr
]
def peer_id_from_email(email: str) -> str:
"""Convert email to a valid Honcho peer ID."""
peer_id = re.sub(r"[^A-Za-z0-9_-]+", "-", email).strip("-").lower()
peer_id = re.sub(r"-{2,}", "-", peer_id)
if not peer_id:
peer_id = "unknown-peer"
if not PEER_ID_PATTERN.fullmatch(peer_id):
raise ValueError(f"Generated peer ID is invalid: {peer_id!r}")
return peer_id
def fetch_thread_messages(service, thread_id: str) -> list[dict]:
"""Fetch all messages in a Gmail thread with full content."""
data = get_thread(service, thread_id)
messages = []
for msg in data.get("messages", []):
headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])}
body = strip_quoted_replies(decode_body(msg.get("payload", {})))
ts = int(msg.get("internalDate", "0")) / 1000
messages.append({
"id": msg["id"],
"thread_id": msg["threadId"],
"from": headers.get("From", ""),
"to": headers.get("To", ""),
"cc": headers.get("Cc", ""),
"bcc": headers.get("Bcc", ""),
"subject": headers.get("Subject", ""),
"date": headers.get("Date", ""),
"timestamp": datetime.fromtimestamp(ts, tz=timezone.utc),
"body": body.strip(),
"labels": msg.get("labelIds", []),
"snippet": msg.get("snippet", ""),
})
return messages
def main():
parser = argparse.ArgumentParser(description="Load Gmail messages into Honcho")
parser.add_argument("--workspace", "-w", default="gmail", help="Honcho workspace ID (default: gmail)")
parser.add_argument("--query", "-q", default=None, help="Gmail search query (e.g. 'from:alice@example.com')")
parser.add_argument("--label", "-l", default=None, help="Gmail label to filter by (e.g. INBOX)")
parser.add_argument("--max-threads", "-n", type=int, default=10, help="Max threads to fetch (default: 10)")
parser.add_argument("--dry-run", action="store_true", help="Print what would be loaded without writing to Honcho")
parser.add_argument("--credentials", "-c", default=None, help="Path to OAuth credentials JSON (auto-detects client_secret*.json)")
parser.add_argument("--token", "-t", default="token.json", help="Path to store/load access token")
args = parser.parse_args()
# Authenticate
print("Authenticating with Gmail API...")
service = get_gmail_service(args.credentials, args.token)
print(" Authenticated successfully!")
label_ids = [args.label] if args.label else None
# List threads
print(f"\nFetching up to {args.max_threads} threads from Gmail...")
threads = list_threads(service, query=args.query, label_ids=label_ids, max_results=args.max_threads)
print(f" Found {len(threads)} threads")
if not threads:
print("No threads found. Try adjusting --query or --label.")
return
# Fetch full messages for each thread
all_thread_messages = {}
seen_peers = {}
def register_peer(addr: str):
email = extract_email(addr)
if email and email not in seen_peers:
name = extract_name(addr)
if name.lower().strip() == email or "@" in name:
name = email.split("@")[0].replace(".", " ").title()
seen_peers[email] = {
"name": name,
"peer_id": peer_id_from_email(email),
"email": email,
}
for i, t in enumerate(threads):
tid = t["id"]
print(f" Fetching thread {i+1}/{len(threads)}: {tid}")
msgs = fetch_thread_messages(service, tid)
all_thread_messages[tid] = msgs
for m in msgs:
register_peer(m["from"])
for addr in parse_address_list(m["to"]):
register_peer(addr)
for addr in parse_address_list(m["cc"]):
register_peer(addr)
for addr in parse_address_list(m["bcc"]):
register_peer(addr)
# Summary
total_msgs = sum(len(v) for v in all_thread_messages.values())
print("\nSummary:")
print(f" Threads: {len(all_thread_messages)}")
print(f" Messages: {total_msgs}")
print(f" Unique participants: {len(seen_peers)}")
for email, info in seen_peers.items():
print(f" {info['peer_id']} ({info['name']} <{email}>)")
if args.dry_run:
print("\n[DRY RUN] Would create the above in Honcho. Showing first message per thread:")
for tid, msgs in all_thread_messages.items():
m = msgs[0]
body_preview = m["body"][:120].replace("\n", " ") if m["body"] else m["snippet"][:120]
print(f" Thread {tid}: {m['subject']}")
print(f" {m['from']} @ {m['date']}")
print(f" {body_preview}...")
return
# Load into Honcho
from honcho import Honcho
print(f"\nLoading into Honcho workspace '{args.workspace}'...")
honcho = Honcho(workspace_id=args.workspace)
# Create peers
peers = {}
for i, (email, info) in enumerate(seen_peers.items()):
if i > 0 and i % 4 == 0:
time.sleep(1)
peers[email] = honcho.peer(info["peer_id"], metadata={
"email": email,
"name": info["name"],
"source": "gmail",
})
print(f" Peer: {info['peer_id']}")
# Create sessions and messages per thread
for tid, msgs in all_thread_messages.items():
subject = msgs[0]["subject"] if msgs else "No subject"
session_id = f"gmail-thread-{tid}"
thread_peer_emails = set()
for m in msgs:
thread_peer_emails.add(extract_email(m["from"]))
for addr in parse_address_list(m["to"]):
thread_peer_emails.add(extract_email(addr))
for addr in parse_address_list(m["cc"]):
thread_peer_emails.add(extract_email(addr))
for addr in parse_address_list(m["bcc"]):
thread_peer_emails.add(extract_email(addr))
thread_peers = [peers[e] for e in thread_peer_emails if e in peers]
session = honcho.session(session_id, metadata={
"gmail_thread_id": tid,
"subject": subject,
"source": "gmail",
"message_count": len(msgs),
})
session.add_peers(thread_peers)
honcho_msgs = []
for m in msgs:
email = extract_email(m["from"])
peer = peers.get(email)
if not peer:
continue
content = m["body"] if m["body"] else m["snippet"]
if not content:
continue
honcho_msgs.append(peer.message(
content,
metadata={
"gmail_id": m["id"],
"subject": m["subject"],
"from": m["from"],
"to": m["to"],
"labels": m["labels"],
},
created_at=m["timestamp"],
))
if honcho_msgs:
session.add_messages(honcho_msgs)
print(f" Session {session_id}: {len(honcho_msgs)} messages — {subject[:60]}")
print(f"\nDone! Loaded {total_msgs} messages into workspace '{args.workspace}'.")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,751 @@
#!/usr/bin/env python3
"""Load Granola meeting notes into Honcho.
Uses the Granola MCP server (with OAuth) to fetch meetings and the Honcho Python SDK
to store them. Each meeting becomes a Honcho session. Two-person meetings get full
speaker attribution; multi-person meetings are stored as summaries.
Prerequisites:
pip install honcho-ai httpx
Environment Variables:
HONCHO_API_KEY - Your Honcho API key (get from app.honcho.dev/api-keys)
Usage:
python honcho_granola.py
"""
import asyncio
import base64
import hashlib
import json
import os
import re
import secrets
import sys
import threading
import traceback
import webbrowser
from dataclasses import dataclass, field
from datetime import datetime, timezone
from http.server import HTTPServer, BaseHTTPRequestHandler
from typing import Any
from urllib.parse import parse_qs, urlencode, urlparse
import httpx
@dataclass
class Participant:
name: str
email: str | None = None
org: str | None = None
@dataclass
class ParsedParticipants:
note_creator: Participant | None = None
others: list[Participant] = field(default_factory=list)
@dataclass
class TranscriptTurn:
speaker: str
text: str
# Granola MCP + OAuth endpoints
GRANOLA_MCP_URL = "https://mcp.granola.ai/mcp"
AUTH_BASE = "https://mcp-auth.granola.ai"
OAUTH_REDIRECT_PORT = 8765
OAUTH_REDIRECT_URI = f"http://localhost:{OAUTH_REDIRECT_PORT}/callback"
# Honcho message size limit (25000 max, leave headroom)
MAX_MESSAGE_LEN = 24000
# ---------------------------------------------------------------------------
# OAuth callback handler (must be a class for BaseHTTPRequestHandler)
# ---------------------------------------------------------------------------
class _OAuthCallback(BaseHTTPRequestHandler):
auth_result: dict[str, str | None] = {"code": None, "error": None}
def do_GET(self):
params = parse_qs(urlparse(self.path).query)
if "code" in params:
_OAuthCallback.auth_result["code"] = params["code"][0]
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(b"<h1>Authenticated! You can close this window.</h1>")
elif "error" in params:
_OAuthCallback.auth_result["error"] = params.get("error_description", params["error"])[0]
self.send_response(400)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(f"<h1>Error: {_OAuthCallback.auth_result['error']}</h1>".encode())
else:
self.send_response(404)
self.end_headers()
def log_message(self, fmt, *args):
pass
# ---------------------------------------------------------------------------
# Granola OAuth + MCP
# ---------------------------------------------------------------------------
async def authenticate(http_client: httpx.AsyncClient) -> str:
"""Perform OAuth (DCR + PKCE) with Granola. Returns access token."""
_OAuthCallback.auth_result = {"code": None, "error": None}
print("\nAuthenticating with Granola...")
# Register client (DCR)
resp = await http_client.post(
f"{AUTH_BASE}/oauth2/register",
json={
"client_name": "Granola to Honcho Transfer",
"redirect_uris": [OAUTH_REDIRECT_URI],
"grant_types": ["authorization_code"],
"response_types": ["code"],
"token_endpoint_auth_method": "none",
},
)
if resp.status_code not in (200, 201):
raise RuntimeError(f"Client registration failed: {resp.status_code}")
client_id = resp.json().get("client_id")
# PKCE
verifier = secrets.token_urlsafe(32)
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
# Browser auth
auth_url = f"{AUTH_BASE}/oauth2/authorize?" + urlencode({
"client_id": client_id,
"redirect_uri": OAUTH_REDIRECT_URI,
"response_type": "code",
"state": "granola-honcho-transfer",
"code_challenge": challenge,
"code_challenge_method": "S256",
})
server = HTTPServer(("localhost", OAUTH_REDIRECT_PORT), _OAuthCallback)
thread = threading.Thread(target=server.handle_request)
thread.start()
print(" Opening browser for authentication...")
webbrowser.open(auth_url)
thread.join(timeout=120)
server.server_close()
auth_result = _OAuthCallback.auth_result
if auth_result["error"]:
raise RuntimeError(f"Authentication failed: {auth_result['error']}")
if not auth_result["code"]:
raise RuntimeError("Authentication timed out")
# Exchange code for token
resp = await http_client.post(
f"{AUTH_BASE}/oauth2/token",
data={
"grant_type": "authorization_code",
"code": auth_result["code"],
"redirect_uri": OAUTH_REDIRECT_URI,
"client_id": client_id,
"code_verifier": verifier,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
if resp.status_code != 200:
raise RuntimeError(f"Token exchange failed: {resp.status_code}")
print(" Authenticated successfully!")
return resp.json()["access_token"]
async def call_mcp_tool(
http_client: httpx.AsyncClient,
access_token: str,
tool_name: str,
arguments: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Call a Granola MCP tool, handling both JSON and SSE responses."""
resp = await http_client.post(
GRANOLA_MCP_URL,
json={
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {"name": tool_name, "arguments": arguments or {}},
},
headers={
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
},
)
if resp.status_code != 200:
raise RuntimeError(f"MCP call failed: {resp.status_code} - {resp.text}")
# SSE response
if "text/event-stream" in resp.headers.get("content-type", ""):
result = None
for line in resp.text.split("\n"):
if line.strip().startswith("data: "):
try:
parsed = json.loads(line.strip()[6:])
if "result" in parsed:
result = parsed
elif "error" in parsed:
raise RuntimeError(f"MCP error: {parsed['error']}")
except json.JSONDecodeError:
continue
if result:
final = result.get("result", {})
return final if isinstance(final, dict) else {"result": final}
raise RuntimeError("No result in SSE response")
# JSON response
result = resp.json()
if "error" in result:
raise RuntimeError(f"MCP error: {result['error']}")
return result.get("result", {})
def extract_mcp_text(result: dict[str, Any]) -> str:
"""Extract text from the first content block of an MCP result.
Raises ValueError if the response structure is unexpected.
"""
content = result.get("content", [])
if not isinstance(content, list) or not content:
raise ValueError(f"MCP response missing content array: {list(result.keys())}")
first = content[0]
if not isinstance(first, dict) or "text" not in first:
raise ValueError(f"MCP content block missing 'text' field: {first}")
return str(first["text"])
# ---------------------------------------------------------------------------
# Granola data fetching
# ---------------------------------------------------------------------------
async def list_meetings(
http_client: httpx.AsyncClient, access_token: str, limit: int = 100,
) -> list[dict[str, Any]]:
"""List meetings from Granola MCP. Parses Granola's XML-like response format."""
result = await call_mcp_tool(http_client, access_token, "list_meetings", {"limit": limit})
text = extract_mcp_text(result)
meetings: list[dict[str, Any]] = []
for match in re.finditer(r'<meeting\s+id="([^"]+)"\s+title="([^"]+)"\s+date="([^"]+)"', text):
mid, title, date = match.groups()
block_end = text.find("</meeting>", match.end())
block = text[match.end():block_end] if block_end != -1 else ""
p_match = re.search(r"<known_participants>\s*(.*?)\s*</known_participants>", block, re.DOTALL)
meetings.append({
"id": mid,
"title": title,
"date": date,
"participants": p_match.group(1).strip() if p_match else "",
})
return meetings
async def get_meeting_details(
http_client: httpx.AsyncClient, access_token: str, meeting_id: str,
) -> dict[str, Any]:
"""Get full meeting details including notes."""
result = await call_mcp_tool(http_client, access_token, "get_meetings", {"meeting_ids": [meeting_id]})
text = extract_mcp_text(result)
return {"id": meeting_id, "raw_content": text}
async def get_meeting_transcript(
http_client: httpx.AsyncClient, access_token: str, meeting_id: str,
max_retries: int = 3,
) -> str | None:
"""Get transcript for a meeting (paid tiers only).
Retries on rate limit responses with exponential backoff.
"""
for attempt in range(max_retries):
try:
result = await call_mcp_tool(http_client, access_token, "get_meeting_transcript", {"meeting_id": meeting_id})
text = extract_mcp_text(result)
except Exception as e:
print(f" Transcript unavailable: {e}")
return None
if not text or "no transcript" in text.lower():
return None
# Granola returns rate limit errors as content text, not HTTP errors
if "rate limit" in text.lower():
wait = 2 ** attempt * 3 # 3s, 6s, 12s
print(f" ⚠ Granola rate limit hit (attempt {attempt + 1}/{max_retries}), waiting {wait}s...")
await asyncio.sleep(wait)
continue
return text
print(f" ⚠ Transcript skipped after {max_retries} rate limit retries")
return None
async def fetch_all_meetings(
http_client: httpx.AsyncClient, access_token: str,
) -> list[dict[str, Any]]:
"""Fetch meeting list and enrich each with transcript and details."""
print("\nFetching meetings from Granola...")
meetings = await list_meetings(http_client, access_token, limit=500)
if not meetings:
print("No meetings found.")
return []
print(f" Found {len(meetings)} meetings. Fetching content...\n")
for i, m in enumerate(meetings, 1):
mid = m.get("id")
if not mid:
continue
transcript = await get_meeting_transcript(http_client, access_token, mid)
if transcript:
m["transcript"] = transcript
try:
m.update(await get_meeting_details(http_client, access_token, mid))
except Exception as exc:
print(f" Failed to fetch details for {mid}: {exc}")
has_t = "transcript" in m
has_s = bool(extract_summary(m))
label = "transcript+summary" if has_t and has_s else "transcript only" if has_t else "summary only" if has_s else "basic only"
print(f" [{i}/{len(meetings)}] {label}: {m.get('title', 'Untitled')[:45]}")
await asyncio.sleep(1.5) # rate limit
return meetings
# ---------------------------------------------------------------------------
# Parsing helpers
# ---------------------------------------------------------------------------
def parse_participants(participants_str: str) -> ParsedParticipants:
"""Parse Granola's participant string into structured participants.
Warns on unparsable entries instead of silently dropping them.
"""
result = ParsedParticipants()
if not participants_str:
return result
# Split on commas, but not inside angle brackets
entries, current, depth = [], [], 0
for ch in participants_str:
if ch == "<":
depth += 1
elif ch == ">":
depth = max(depth - 1, 0)
elif ch == "," and depth == 0:
entries.append("".join(current))
current = []
continue
current.append(ch)
if current:
entries.append("".join(current))
for entry in entries:
entry = entry.strip()
if not entry:
continue
is_creator = "(note creator)" in entry
clean = entry.replace("(note creator)", "").strip()
email_match = re.search(r"<([^>]+)>", clean)
email = email_match.group(1) if email_match else None
name = re.sub(r"\s*<[^>]+>", "", clean).strip()
if not name:
print(f" Warning: could not parse participant entry: {entry!r}")
continue
org = None
org_match = re.match(r"(.+?)\s+from\s+(.+)", name)
if org_match:
name, org = org_match.group(1).strip(), org_match.group(2).strip()
person = Participant(name=name, email=email, org=org)
if is_creator:
result.note_creator = person
else:
result.others.append(person)
return result
def parse_transcript_turns(raw: str) -> list[TranscriptTurn]:
"""Split a Granola transcript into speaker turns."""
# Unwrap JSON wrapper if present
try:
parsed = json.loads(raw)
if isinstance(parsed, dict) and "transcript" in parsed:
raw = str(parsed["transcript"])
except (json.JSONDecodeError, TypeError):
pass
parts = re.split(r"(?:^|\s{2,})(Me|Them):\s*", raw)
turns: list[TranscriptTurn] = []
i = 1
while i < len(parts) - 1:
text = parts[i + 1].strip()
if text:
turns.append(TranscriptTurn(speaker=parts[i], text=text))
i += 2
return turns
def extract_summary(meeting: dict[str, Any]) -> str:
"""Extract best available summary text from meeting data."""
candidates = []
for key in ("summary", "notes", "note", "meeting_notes", "description"):
val = meeting.get(key)
if isinstance(val, str) and val.strip():
candidates.append(val.strip())
raw = meeting.get("raw_content")
if isinstance(raw, str) and raw.strip():
candidates.append(raw.strip())
for c in candidates:
for tag in ("summary", "notes"):
m = re.search(rf"<{tag}>\s*(.*?)\s*</{tag}>", c, re.DOTALL)
if m:
return m.group(1).strip()
return candidates[0] if candidates else ""
def peer_id_from(value: str) -> str:
"""Normalize a name or email into a Honcho-safe peer ID."""
norm = re.sub(r"[^a-z0-9_-]+", "-", value.strip().lower())
norm = re.sub(r"-{2,}", "-", norm).strip("-_")
return (norm or "peer")[:100]
def sanitize(text: str) -> str:
"""Remove null bytes and control characters."""
return re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
def parse_date(date_str: str) -> datetime:
"""Parse Granola's date format into a timezone-aware datetime.
Raises ValueError if the date string doesn't match any known format.
"""
for fmt in ["%b %d, %Y %I:%M %p", "%b %d, %Y %I:%M:%S %p", "%B %d, %Y %I:%M %p"]:
try:
return datetime.strptime(date_str, fmt).replace(tzinfo=timezone.utc)
except ValueError:
continue
raise ValueError(f"Unrecognized date format: {date_str!r}")
# ---------------------------------------------------------------------------
# Honcho import helpers
# ---------------------------------------------------------------------------
def build_messages(
peer: Any,
content: str,
metadata: dict[str, object] | None,
created_at: datetime,
) -> list[Any]:
"""Build chunked messages for a single peer, attaching metadata to the first chunk."""
messages = []
content = sanitize(content)
for start in range(0, len(content), MAX_MESSAGE_LEN):
chunk = content[start:start + MAX_MESSAGE_LEN]
msg_meta = metadata if start == 0 else None
messages.append(peer.message(chunk, metadata=msg_meta, created_at=created_at))
return messages
def send_messages(session: Any, messages: list[Any]) -> None:
"""Send messages to a session in batches of 100."""
for batch_start in range(0, len(messages), 100):
session.add_messages(messages[batch_start:batch_start + 100])
def import_two_person(
honcho: Any,
session: Any,
me_peer_id: str,
them_peer_id: str,
turns: list[TranscriptTurn],
metadata: dict[str, object],
created_at: datetime,
) -> None:
"""Import a two-person meeting with speaker attribution."""
me_peer = honcho.peer(me_peer_id)
them_peer = honcho.peer(them_peer_id)
# Merge consecutive same-speaker turns
merged: list[TranscriptTurn] = []
for t in turns:
if merged and merged[-1].speaker == t.speaker:
merged[-1].text += " " + t.text
else:
merged.append(TranscriptTurn(speaker=t.speaker, text=t.text))
messages: list[Any] = []
for i, t in enumerate(merged):
peer = me_peer if t.speaker == "Me" else them_peer
msg_meta = metadata if i == 0 else None
messages.extend(build_messages(peer, t.text, msg_meta, created_at))
send_messages(session, messages)
print(f" -> Imported as 2-person ({me_peer_id} + {them_peer_id})")
def import_summary(
honcho: Any,
session: Any,
me_peer_id: str,
meeting: dict[str, Any],
metadata: dict[str, object],
created_at: datetime,
) -> None:
"""Import a meeting as a summary message."""
me_peer = honcho.peer(me_peer_id)
summary = extract_summary(meeting)
if not summary:
raw_t = meeting.get("transcript", "")
try:
parsed = json.loads(raw_t)
summary = str(parsed.get("transcript", "")) if isinstance(parsed, dict) else raw_t
except (json.JSONDecodeError, TypeError):
summary = raw_t
summary = summary or "No content available"
title = meeting.get("title", "Untitled")
date = meeting.get("date", "")
header = f"Meeting: {title}\nDate: {date}\nParticipants: {meeting.get('participants', '')}\n\n"
messages = build_messages(me_peer, header + summary, metadata, created_at)
send_messages(session, messages)
print(" -> Imported as summary")
def resolve_them_participant(others: list[Participant]) -> Participant | None:
"""Ask user to pick which participant is 'Them' from a multi-person meeting."""
for j, p in enumerate(others, 1):
email_str = f" <{p.email}>" if p.email else ""
print(f" {j}. {p.name}{email_str}")
idx_str = input(f" Who is 'Them'? [1-{len(others)}]: ").strip()
try:
return others[int(idx_str) - 1]
except (ValueError, IndexError):
print(" Invalid selection.")
return None
def review_meeting(
index: int,
total: int,
meeting: dict[str, Any],
participants: ParsedParticipants,
turns: list[TranscriptTurn],
) -> tuple[str, Participant | None]:
"""Display meeting info and get user's import choice.
Returns (mode, them_participant) where mode is one of:
- "two_person": import with speaker attribution using them_participant
- "summary": import as a single summary message
- "skip": skip this meeting
"""
title = meeting.get("title", "Untitled")
date = meeting.get("date", "")
creator = participants.note_creator
others = participants.others
me_turns = sum(1 for t in turns if t.speaker == "Me")
them_turns = len(turns) - me_turns
total_words = sum(len(t.text.split()) for t in turns)
print(f"\n{'' * 60}")
print(f" [{index}/{total}] {title}")
print(f" Date: {date}")
if creator:
print(f" You: {creator.name} <{creator.email}>")
for j, p in enumerate(others, 1):
email_str = f" <{p.email}>" if p.email else ""
org_str = f" ({p.org})" if p.org else ""
print(f" {j}. {p.name}{email_str}{org_str}")
has_transcript = bool(meeting.get("transcript"))
if turns:
print(f" Transcript: {me_turns} Me, {them_turns} Them, ~{total_words} words")
if them_turns == 0:
print(" ** No 'Them' turns — nobody else spoke **")
if total_words < 30:
print(" ** Very short — might be empty **")
elif has_transcript:
raw = meeting["transcript"]
print(f" Transcript: present ({len(raw)} chars) but could not parse speaker turns")
print(f" Preview: {raw[:200]!r}")
else:
print(f" Content: {'summary available' if extract_summary(meeting) else 'metadata only'}")
# Two-person default: exactly one other participant with transcript
if len(others) == 1 and them_turns > 0:
them_label = others[0].name + (f" <{others[0].email}>" if others[0].email else "")
print(f"\n Detected: 2-person call (you + {them_label})")
choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower()
while choice not in ("", "s", "k"):
choice = input(" [Enter] 2-person / [s]ummary / [k] skip: ").strip().lower()
if choice == "k":
return ("skip", None)
if choice == "s":
return ("summary", None)
return ("two_person", others[0])
# Multi-person with transcript
if len(others) > 1 and them_turns > 0:
print(f"\n {len(others)} participants")
choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower()
while choice not in ("", "2", "k"):
choice = input(" [Enter] summary / [2] 2-person / [k] skip: ").strip().lower()
if choice == "k":
return ("skip", None)
if choice == "2":
them = resolve_them_participant(others)
if them is None:
return ("summary", None)
return ("two_person", them)
return ("summary", None)
# No transcript or no other speakers
choice = input(" [Enter] summary / [k] skip: ").strip().lower()
while choice not in ("", "k"):
choice = input(" [Enter] summary / [k] skip: ").strip().lower()
if choice == "k":
return ("skip", None)
return ("summary", None)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
async def main():
print("=" * 60)
print(" Granola -> Honcho Meeting Notes Transfer")
print("=" * 60)
if not os.environ.get("HONCHO_API_KEY"):
print("\nError: HONCHO_API_KEY not set.")
print(" Get your key at: https://app.honcho.dev/api-keys")
sys.exit(1)
async with httpx.AsyncClient(timeout=60.0) as http_client:
try:
access_token = await authenticate(http_client)
meetings = await fetch_all_meetings(http_client, access_token)
if not meetings:
sys.exit(0)
from honcho import Honcho
honcho = Honcho(workspace_id="granola_test")
seen_peers: set[str] = set()
results = {"imported": 0, "skipped": 0, "failed": 0}
print("\n" + "=" * 60)
print(" Review each meeting")
print("=" * 60)
for i, m in enumerate(meetings, 1):
mid = m.get("id")
if not mid:
continue
participants = parse_participants(m.get("participants", ""))
turns = parse_transcript_turns(m["transcript"]) if m.get("transcript") else []
mode, them = review_meeting(i, len(meetings), m, participants, turns)
if mode == "skip":
print(" -> Skipped")
results["skipped"] += 1
continue
# Resolve creator peer
creator = participants.note_creator
me_source = (creator.email or creator.name) if creator else None
if not me_source:
print(" -> Skipped (no creator identifier)")
results["skipped"] += 1
continue
me_peer_id = peer_id_from(me_source)
if me_peer_id not in seen_peers:
print(f" New peer: {me_source} ({me_peer_id})")
seen_peers.add(me_peer_id)
try:
created_at = parse_date(m.get("date", ""))
session = honcho.session(f"meeting-{mid}")
metadata: dict[str, object] = {
"title": m.get("title", "Untitled"),
"date": m.get("date", ""),
"granola_meeting_id": mid,
"mode": mode,
}
if mode == "two_person" and them is not None:
them_source = them.email or them.name
them_peer_id = peer_id_from(them_source)
if them_peer_id not in seen_peers:
print(f" New peer: {them_source} ({them_peer_id})")
seen_peers.add(them_peer_id)
import_two_person(honcho, session, me_peer_id, them_peer_id, turns, metadata, created_at)
else:
import_summary(honcho, session, me_peer_id, m, metadata, created_at)
results["imported"] += 1
except ValueError as e:
print(f" -> FAILED: {e}")
results["failed"] += 1
except Exception as e:
print(f" -> FAILED: {e}")
traceback.print_exc()
results["failed"] += 1
# Done
print("\n" + "=" * 60)
print(" Transfer Complete!")
print("=" * 60)
print(f"\n Imported: {results['imported']}")
print(f" Skipped: {results['skipped']}")
print(f" Failed: {results['failed']}")
print(" Workspace: granola")
print(f" Peers: {sorted(seen_peers)}")
except KeyboardInterrupt:
print("\n\nAborted.")
sys.exit(0)
except Exception as e:
print(f"\nTransfer failed: {e}")
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())

View File

@ -17,7 +17,10 @@ async def get_db():
db: AsyncSession = SessionLocal()
try:
if settings.DB.TRACING:
await db.execute(text(f"SET application_name = '{context}'"))
await db.execute(
text("SELECT set_config('application_name', :name, false)"),
{"name": context},
)
yield db
except Exception:
await db.rollback()
@ -45,7 +48,8 @@ async def tracked_db(operation_name: str | None = None):
try:
if settings.DB.TRACING:
await db.execute(
text(f"SET application_name = '{context or f'task:{operation_name}'}'")
text("SELECT set_config('application_name', :name, false)"),
{"name": context or f"task:{operation_name}"},
)
yield db

162
src/schemas/__init__.py Normal file
View File

@ -0,0 +1,162 @@
"""Pydantic schemas for Honcho.
Re-exports all public names from submodules so that existing
``from src.schemas import X`` imports continue to work unchanged.
"""
from src.schemas.api import (
RESOURCE_NAME_PATTERN,
Conclusion,
ConclusionBatchCreate,
ConclusionCreate,
ConclusionGet,
ConclusionQuery,
DialecticOptions,
DialecticResponse,
DialecticStreamChunk,
DialecticStreamDelta,
Message,
MessageBase,
MessageBatchCreate,
MessageCreate,
MessageGet,
MessageSearchOptions,
MessageUpdate,
MessageUploadCreate,
Peer,
PeerBase,
PeerCardResponse,
PeerCardSet,
PeerContext,
PeerCreate,
PeerGet,
PeerRepresentationGet,
PeerUpdate,
QueueStatus,
RepresentationResponse,
ScheduleDreamRequest,
Session,
SessionBase,
SessionContext,
SessionCreate,
SessionGet,
SessionQueueStatus,
SessionSummaries,
SessionUpdate,
Summary,
WebhookEndpoint,
WebhookEndpointBase,
WebhookEndpointCreate,
Workspace,
WorkspaceBase,
WorkspaceCreate,
WorkspaceGet,
WorkspaceUpdate,
)
from src.schemas.configuration import (
DreamConfiguration,
DreamType,
MessageConfiguration,
PeerCardConfiguration,
PeerConfig,
ReasoningConfiguration,
ResolvedConfiguration,
ResolvedDreamConfiguration,
ResolvedPeerCardConfiguration,
ResolvedReasoningConfiguration,
ResolvedSummaryConfiguration,
SessionConfiguration,
SessionPeerConfig,
SummaryConfiguration,
WorkspaceConfiguration,
)
from src.schemas.internal import (
DocumentBase,
DocumentCreate,
DocumentMetadata,
MessageBulkData,
ObservationInput,
QueueCounts,
QueueStatusRow,
ReconcilerType,
SessionCounts,
SessionPeerData,
)
__all__ = [
# configuration
"DreamConfiguration",
"DreamType",
"MessageConfiguration",
"PeerCardConfiguration",
"PeerConfig",
"ReasoningConfiguration",
"ResolvedConfiguration",
"ResolvedDreamConfiguration",
"ResolvedPeerCardConfiguration",
"ResolvedReasoningConfiguration",
"ResolvedSummaryConfiguration",
"SessionConfiguration",
"SessionPeerConfig",
"SummaryConfiguration",
"WorkspaceConfiguration",
# api
"Conclusion",
"ConclusionBatchCreate",
"ConclusionCreate",
"ConclusionGet",
"ConclusionQuery",
"DialecticOptions",
"DialecticResponse",
"DialecticStreamChunk",
"DialecticStreamDelta",
"Message",
"MessageBase",
"MessageBatchCreate",
"MessageCreate",
"MessageGet",
"MessageSearchOptions",
"MessageUpdate",
"MessageUploadCreate",
"Peer",
"PeerBase",
"PeerCardResponse",
"PeerCardSet",
"PeerContext",
"PeerCreate",
"PeerGet",
"PeerRepresentationGet",
"PeerUpdate",
"QueueStatus",
"RESOURCE_NAME_PATTERN",
"RepresentationResponse",
"ScheduleDreamRequest",
"Session",
"SessionBase",
"SessionContext",
"SessionCreate",
"SessionGet",
"SessionQueueStatus",
"SessionSummaries",
"SessionUpdate",
"Summary",
"WebhookEndpoint",
"WebhookEndpointBase",
"WebhookEndpointCreate",
"Workspace",
"WorkspaceBase",
"WorkspaceCreate",
"WorkspaceGet",
"WorkspaceUpdate",
# internal
"DocumentBase",
"DocumentCreate",
"DocumentMetadata",
"MessageBulkData",
"ObservationInput",
"QueueCounts",
"QueueStatusRow",
"ReconcilerType",
"SessionCounts",
"SessionPeerData",
]

View File

@ -1,7 +1,12 @@
"""Pydantic schemas for API request/response validation.
These schemas are consumed by the FastAPI routers and define the public
API contract.
"""
import datetime
import ipaddress
from enum import Enum
from typing import Annotated, Any, Literal, Self, cast
from typing import Annotated, Any, Self, cast
from urllib.parse import urlparse
import tiktoken
@ -18,7 +23,17 @@ from pydantic import (
from pydantic_core import PydanticCustomError
from src.config import ReasoningLevel, settings
from src.utils.types import DocumentLevel
from src.schemas.configuration import (
DreamType,
MessageConfiguration,
SessionConfiguration,
SessionPeerConfig,
WorkspaceConfiguration,
)
# ---------------------------------------------------------------------------
# Metadata validation helpers
# ---------------------------------------------------------------------------
RESOURCE_NAME_PATTERN = r"^[a-zA-Z0-9_-]+$"
@ -38,15 +53,19 @@ def _sanitize_value(v: Any) -> Any:
if isinstance(v, str):
return strip_nul_bytes(v)
if isinstance(v, dict):
data = cast(dict[str, Any], v)
return {_sanitize_value(k): _sanitize_value(val) for k, val in data.items()}
d = cast(dict[str, Any], v)
return {_sanitize_value(k): _sanitize_value(val) for k, val in d.items()}
if isinstance(v, list):
items = cast(list[Any], v)
return [_sanitize_value(item) for item in items]
lst = cast(list[Any], v)
return [_sanitize_value(item) for item in lst]
return v
def _check_metadata_limits(value: Any, *, _current_depth: int = 1) -> None:
def _check_metadata_limits(
value: Any,
*,
_current_depth: int = 1,
) -> None:
"""Validate metadata doesn't exceed key count or nesting depth limits."""
if _current_depth > _METADATA_MAX_DEPTH:
raise ValueError(
@ -76,7 +95,7 @@ def _check_metadata_limits(value: Any, *, _current_depth: int = 1) -> None:
def _validate_metadata(v: Any) -> Any:
"""Validate and sanitize a metadata dict before field parsing."""
"""Validate and sanitize a metadata dict: enforce limits and strip NUL bytes."""
if not isinstance(v, dict):
return v
data = cast(dict[str, Any], v)
@ -86,188 +105,9 @@ def _validate_metadata(v: Any) -> Any:
_SanitizedMetadata = Annotated[dict[str, Any], BeforeValidator(_validate_metadata)]
class DreamType(str, Enum):
"""Types of dreams that can be triggered."""
OMNI = "omni"
class ReconcilerType(str, Enum):
"""Types of reconciler tasks that can be performed."""
SYNC_VECTORS = "sync_vectors"
CLEANUP_QUEUE = "cleanup_queue"
class ReasoningConfiguration(BaseModel):
enabled: bool | None = Field(
default=None,
description="Whether to enable reasoning functionality.",
)
custom_instructions: str | None = Field(
default=None,
description="TODO: currently unused. Custom instructions to use for the reasoning system on this workspace/session/message.",
)
class PeerCardConfiguration(BaseModel):
use: bool | None = Field(
default=None,
description="Whether to use peer card related to this peer during reasoning process.",
)
create: bool | None = Field(
default=None,
description="Whether to generate peer card based on content.",
)
class SummaryConfiguration(BaseModel):
enabled: bool | None = Field(
default=None,
description="Whether to enable summary functionality.",
)
messages_per_short_summary: int | None = Field(
default=None,
ge=10,
description="Number of messages per short summary. Must be positive, greater than or equal to 10, and less than messages_per_long_summary.",
)
messages_per_long_summary: int | None = Field(
default=None,
ge=20,
description="Number of messages per long summary. Must be positive, greater than or equal to 20, and greater than messages_per_short_summary.",
)
@model_validator(mode="after")
def validate_summary_thresholds(self) -> Self:
"""Validate that short summary threshold <= long summary threshold."""
short = self.messages_per_short_summary
long = self.messages_per_long_summary
if short is not None and long is not None and short >= long:
raise ValueError(
"messages_per_short_summary must be less than messages_per_long_summary"
)
return self
class DreamConfiguration(BaseModel):
enabled: bool | None = Field(
default=None,
description="Whether to enable dream functionality. If reasoning is disabled, dreams will also be disabled and this setting will be ignored.",
)
class WorkspaceConfiguration(BaseModel):
"""
The set of options that can be in a workspace DB-level configuration dictionary.
All fields are optional. Session-level configuration overrides workspace-level configuration, which overrides global configuration.
"""
model_config = ConfigDict(extra="allow") # pyright: ignore
reasoning: ReasoningConfiguration | None = Field(
default=None,
description="Configuration for reasoning functionality.",
)
peer_card: PeerCardConfiguration | None = Field(
default=None,
description="Configuration for peer card functionality. If reasoning is disabled, peer cards will also be disabled and these settings will be ignored.",
)
summary: SummaryConfiguration | None = Field(
default=None,
description="Configuration for summary functionality.",
)
dream: DreamConfiguration | None = Field(
default=None,
description="Configuration for dream functionality. If reasoning is disabled, dreams will also be disabled and these settings will be ignored.",
)
class SessionConfiguration(WorkspaceConfiguration):
"""
The set of options that can be in a session DB-level configuration dictionary.
All fields are optional. Session-level configuration overrides workspace-level configuration, which overrides global configuration.
"""
pass
class MessageConfiguration(BaseModel):
"""
The set of options that can be in a message DB-level configuration dictionary.
All fields are optional. Message-level configuration overrides all other configurations.
"""
reasoning: ReasoningConfiguration | None = Field(
default=None,
description="Configuration for reasoning functionality.",
)
class ResolvedReasoningConfiguration(BaseModel):
enabled: bool
class ResolvedPeerCardConfiguration(BaseModel):
use: bool
create: bool
class ResolvedSummaryConfiguration(BaseModel):
enabled: bool
messages_per_short_summary: int
messages_per_long_summary: int
class ResolvedDreamConfiguration(BaseModel):
enabled: bool
class ResolvedConfiguration(BaseModel):
"""
The final resolved configuration for a given message.
Hierarchy: message > session > workspace > global configuration
"""
reasoning: ResolvedReasoningConfiguration
peer_card: ResolvedPeerCardConfiguration
summary: ResolvedSummaryConfiguration
dream: ResolvedDreamConfiguration
@model_validator(mode="before")
@classmethod
def migrate_deriver_to_reasoning(cls, data: Any) -> Any:
"""Handle v3.0.0 migration: 'deriver' was renamed to 'reasoning'."""
if not isinstance(data, dict):
return data
config = cast(dict[str, Any], data)
if "deriver" in config and "reasoning" not in config:
config["reasoning"] = config.pop("deriver")
return config
class PeerConfig(BaseModel):
# TODO: Update description - should say "Whether honcho forms a representation of the peer itself"
observe_me: bool | None = Field(
default=None,
description="Whether Honcho will use reasoning to form a representation of this peer",
)
class SessionPeerConfig(PeerConfig):
# TODO: Update description - should say "Whether this peer forms representations of other peers in the session"
observe_others: bool | None = Field(
default=None,
description="Whether this peer should form a session-level theory-of-mind representation of other peers in the session",
)
# ---------------------------------------------------------------------------
# Workspace schemas
# ---------------------------------------------------------------------------
class WorkspaceBase(BaseModel):
@ -309,6 +149,11 @@ class Workspace(WorkspaceBase):
)
# ---------------------------------------------------------------------------
# Peer schemas
# ---------------------------------------------------------------------------
class PeerBase(BaseModel):
pass
@ -396,6 +241,21 @@ class PeerCardResponse(BaseModel):
class PeerCardSet(BaseModel):
peer_card: list[str] = Field(..., description="The peer card content to set")
@field_validator("peer_card", mode="before")
@classmethod
def sanitize_peer_card(cls, v: Any) -> Any:
if isinstance(v, list):
return [
item.replace("\x00", "") if isinstance(item, str) else item
for item in cast(list[Any], v)
]
return v
# ---------------------------------------------------------------------------
# Message schemas
# ---------------------------------------------------------------------------
class MessageBase(BaseModel):
pass
@ -410,6 +270,11 @@ class MessageCreate(MessageBase):
_encoded_message: list[int] = PrivateAttr(default=[])
@field_validator("content", mode="after")
@classmethod
def sanitize_content(cls, v: str) -> str:
return v.replace("\x00", "")
@property
def encoded_message(self) -> list[int]:
return self._encoded_message
@ -465,6 +330,11 @@ class MessageUploadCreate(BaseModel):
model_config = ConfigDict(populate_by_name=True) # pyright: ignore
# ---------------------------------------------------------------------------
# Session schemas
# ---------------------------------------------------------------------------
class SessionBase(BaseModel):
pass
@ -571,106 +441,9 @@ class SessionSummaries(SessionBase):
)
class DocumentBase(BaseModel):
pass
class DocumentMetadata(BaseModel):
message_ids: list[int] = Field(
description="The ID range(s) of the messages that this document was derived from. Acts as a link to the primary source of the document. Note that as a document gets deduplicated, additional ranges will be added, because the same document could be derived from completely separate message ranges."
)
message_created_at: str = Field(
description="The timestamp of the message that this document was derived from. Note that this is not the same as the created_at timestamp of the document. This timestamp is usually only saved with second-level precision."
)
source_ids: list[str] | None = Field(
default=None,
description="Document IDs of source documents for tree traversal -- required for deductive and inductive documents",
)
premises: list[str] | None = Field(
default=None,
description="Human-readable premise text for display -- only applicable for deductive documents",
)
sources: list[str] | None = Field(
default=None,
description="Human-readable source text for display -- only applicable for inductive documents",
)
pattern_type: str | None = Field(
default=None,
description="Type of pattern identified (preference, behavior, personality, tendency, correlation) -- only applicable for inductive documents",
)
confidence: str | None = Field(
default=None,
description="Confidence level (high, medium, low) -- only applicable for inductive documents",
)
class DocumentCreate(DocumentBase):
content: Annotated[str, Field(min_length=1, max_length=100000)]
session_name: str | None = Field(
default=None,
description="The session from which the document was derived (NULL for global observations)",
)
level: DocumentLevel = Field(
default="explicit",
description="The level of the document (explicit, deductive, inductive, or contradiction)",
)
times_derived: int = Field(
default=1,
ge=1,
description="The number of times that a semantic duplicate document to this one has been derived",
)
metadata: DocumentMetadata = Field()
embedding: list[float] = Field()
# Tree linkage field
source_ids: list[str] | None = Field(
default=None,
description="Document IDs of source/premise documents -- for deductive and inductive documents",
)
class ObservationInput(BaseModel):
"""Validated observation input from LLM tool calls."""
content: str = Field(min_length=1)
level: DocumentLevel = "explicit"
source_ids: list[str] | None = None
premises: list[str] | None = None
sources: list[str] | None = None
pattern_type: (
Literal["preference", "behavior", "personality", "tendency", "correlation"]
| None
) = None
confidence: Literal["high", "medium", "low"] | None = None
@field_validator("content", mode="after")
@classmethod
def sanitize_content(cls, v: str) -> str:
sanitized = cast(str, strip_nul_bytes(v))
if not sanitized:
raise PydanticCustomError(
"string_too_short",
"String should have at least 1 character",
)
return sanitized
@model_validator(mode="after")
def validate_level_fields(self) -> Self:
"""Validate that level-specific fields are present when required."""
if self.level == "deductive" and not self.source_ids:
raise ValueError(
"deductive observations require 'source_ids' field with document IDs of premises"
)
if self.level == "inductive" and not self.source_ids:
raise ValueError(
"inductive observations require 'source_ids' field with document IDs of sources"
)
if self.level == "contradiction" and (
not self.source_ids or len(self.source_ids) < 2
):
raise ValueError(
"contradiction observations require 'source_ids' field with at least 2 IDs of contradicting observations"
)
return self
# ---------------------------------------------------------------------------
# Conclusion schemas
# ---------------------------------------------------------------------------
class ConclusionGet(BaseModel):
@ -773,8 +546,17 @@ class ConclusionBatchCreate(BaseModel):
)
# ---------------------------------------------------------------------------
# Search schemas
# ---------------------------------------------------------------------------
class MessageSearchOptions(BaseModel):
query: str = Field(..., description="Search query")
query: Annotated[
str,
BeforeValidator(strip_nul_bytes),
Field(min_length=1, description="Search query"),
]
filters: dict[str, Any] | None = Field(
default=None, description="Filters to scope the search"
)
@ -788,7 +570,17 @@ class MessageSearchOptions(BaseModel):
@field_validator("query", mode="after")
@classmethod
def sanitize_query(cls, v: str) -> str:
return cast(str, strip_nul_bytes(v))
if not v:
raise PydanticCustomError(
"string_too_short",
"String should have at least 1 character",
)
return v
# ---------------------------------------------------------------------------
# Dialectic schemas
# ---------------------------------------------------------------------------
class DialecticOptions(BaseModel):
@ -841,50 +633,9 @@ class DialecticStreamChunk(BaseModel):
done: bool = False
class SessionCounts(BaseModel):
"""Counts for a specific session in queue processing."""
completed: int
in_progress: int
pending: int
class QueueCounts(BaseModel):
"""Aggregated counts for queue processing status."""
total: int
completed: int
in_progress: int
pending: int
sessions: dict[str, SessionCounts]
class QueueStatusRow(BaseModel):
"""Represents a row from the queue status SQL query result."""
session_id: str | None
total: int
completed: int
in_progress: int
pending: int
session_total: int
session_completed: int
session_in_progress: int
session_pending: int
class SessionPeerData(BaseModel):
"""Data for managing session peer relationships."""
peer_names: dict[str, SessionPeerConfig]
class MessageBulkData(BaseModel):
"""Data for bulk message operations."""
messages: list[MessageCreate]
session_name: str
workspace_name: str
# ---------------------------------------------------------------------------
# Queue status schemas
# ---------------------------------------------------------------------------
class SessionQueueStatus(BaseModel):
@ -926,6 +677,11 @@ class QueueStatus(BaseModel):
)
# ---------------------------------------------------------------------------
# Dream scheduling schemas
# ---------------------------------------------------------------------------
class ScheduleDreamRequest(BaseModel):
observer: str = Field(..., description="Observer peer name")
observed: str | None = Field(
@ -937,7 +693,11 @@ class ScheduleDreamRequest(BaseModel):
)
# Webhook endpoint schemas
# ---------------------------------------------------------------------------
# Webhook schemas
# ---------------------------------------------------------------------------
class WebhookEndpointBase(BaseModel):
pass

View File

@ -0,0 +1,186 @@
"""Configuration schemas for hierarchical settings resolution.
Covers workspace, session, and message-level configuration as well as
the fully-resolved variants used at runtime.
"""
from enum import Enum
from typing import Any, Self, cast
from pydantic import BaseModel, ConfigDict, Field, model_validator
class DreamType(str, Enum):
"""Types of dreams that can be triggered."""
OMNI = "omni"
class ReasoningConfiguration(BaseModel):
enabled: bool | None = Field(
default=None,
description="Whether to enable reasoning functionality.",
)
custom_instructions: str | None = Field(
default=None,
description="TODO: currently unused. Custom instructions to use for the reasoning system on this workspace/session/message.",
)
class PeerCardConfiguration(BaseModel):
use: bool | None = Field(
default=None,
description="Whether to use peer card related to this peer during reasoning process.",
)
create: bool | None = Field(
default=None,
description="Whether to generate peer card based on content.",
)
class SummaryConfiguration(BaseModel):
enabled: bool | None = Field(
default=None,
description="Whether to enable summary functionality.",
)
messages_per_short_summary: int | None = Field(
default=None,
ge=10,
description="Number of messages per short summary. Must be positive, greater than or equal to 10, and less than messages_per_long_summary.",
)
messages_per_long_summary: int | None = Field(
default=None,
ge=20,
description="Number of messages per long summary. Must be positive, greater than or equal to 20, and greater than messages_per_short_summary.",
)
@model_validator(mode="after")
def validate_summary_thresholds(self) -> Self:
"""Validate that short summary threshold <= long summary threshold."""
short = self.messages_per_short_summary
long = self.messages_per_long_summary
if short is not None and long is not None and short >= long:
raise ValueError(
"messages_per_short_summary must be less than messages_per_long_summary"
)
return self
class DreamConfiguration(BaseModel):
enabled: bool | None = Field(
default=None,
description="Whether to enable dream functionality. If reasoning is disabled, dreams will also be disabled and this setting will be ignored.",
)
class WorkspaceConfiguration(BaseModel):
"""
The set of options that can be in a workspace DB-level configuration dictionary.
All fields are optional. Session-level configuration overrides workspace-level configuration, which overrides global configuration.
"""
model_config = ConfigDict(extra="allow") # pyright: ignore
reasoning: ReasoningConfiguration | None = Field(
default=None,
description="Configuration for reasoning functionality.",
)
peer_card: PeerCardConfiguration | None = Field(
default=None,
description="Configuration for peer card functionality. If reasoning is disabled, peer cards will also be disabled and these settings will be ignored.",
)
summary: SummaryConfiguration | None = Field(
default=None,
description="Configuration for summary functionality.",
)
dream: DreamConfiguration | None = Field(
default=None,
description="Configuration for dream functionality. If reasoning is disabled, dreams will also be disabled and these settings will be ignored.",
)
class SessionConfiguration(WorkspaceConfiguration):
"""
The set of options that can be in a session DB-level configuration dictionary.
All fields are optional. Session-level configuration overrides workspace-level configuration, which overrides global configuration.
"""
pass
class MessageConfiguration(BaseModel):
"""
The set of options that can be in a message DB-level configuration dictionary.
All fields are optional. Message-level configuration overrides all other configurations.
"""
reasoning: ReasoningConfiguration | None = Field(
default=None,
description="Configuration for reasoning functionality.",
)
class ResolvedReasoningConfiguration(BaseModel):
enabled: bool
class ResolvedPeerCardConfiguration(BaseModel):
use: bool
create: bool
class ResolvedSummaryConfiguration(BaseModel):
enabled: bool
messages_per_short_summary: int
messages_per_long_summary: int
class ResolvedDreamConfiguration(BaseModel):
enabled: bool
class ResolvedConfiguration(BaseModel):
"""
The final resolved configuration for a given message.
Hierarchy: message > session > workspace > global configuration
"""
reasoning: ResolvedReasoningConfiguration
peer_card: ResolvedPeerCardConfiguration
summary: ResolvedSummaryConfiguration
dream: ResolvedDreamConfiguration
@model_validator(mode="before")
@classmethod
def migrate_deriver_to_reasoning(cls, data: Any) -> Any:
"""Handle v3.0.0 migration: 'deriver' was renamed to 'reasoning'."""
if not isinstance(data, dict):
return data
config = cast(dict[str, Any], data)
if "deriver" in config and "reasoning" not in config:
config["reasoning"] = config.pop("deriver")
return config
class PeerConfig(BaseModel):
# TODO: Update description - should say "Whether honcho forms a representation of the peer itself"
observe_me: bool | None = Field(
default=None,
description="Whether Honcho will use reasoning to form a representation of this peer",
)
class SessionPeerConfig(PeerConfig):
# TODO: Update description - should say "Whether this peer forms representations of other peers in the session"
observe_others: bool | None = Field(
default=None,
description="Whether this peer should form a session-level theory-of-mind representation of other peers in the session",
)

184
src/schemas/internal.py Normal file
View File

@ -0,0 +1,184 @@
"""Internal schemas used by the deriver, dreamer, and other background systems.
These are not part of the public API contract and may change without notice.
"""
from enum import Enum
from typing import Annotated, Literal, Self
from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic_core import PydanticCustomError
from src.schemas.api import MessageCreate, strip_nul_bytes
from src.schemas.configuration import SessionPeerConfig
from src.utils.types import DocumentLevel
class ReconcilerType(str, Enum):
"""Types of reconciler tasks that can be performed."""
SYNC_VECTORS = "sync_vectors"
CLEANUP_QUEUE = "cleanup_queue"
# ---------------------------------------------------------------------------
# Document / observation schemas (vector storage internals)
# ---------------------------------------------------------------------------
class DocumentBase(BaseModel):
pass
class DocumentMetadata(BaseModel):
message_ids: list[int] = Field(
description="The ID range(s) of the messages that this document was derived from. Acts as a link to the primary source of the document. Note that as a document gets deduplicated, additional ranges will be added, because the same document could be derived from completely separate message ranges."
)
message_created_at: str = Field(
description="The timestamp of the message that this document was derived from. Note that this is not the same as the created_at timestamp of the document. This timestamp is usually only saved with second-level precision."
)
source_ids: list[str] | None = Field(
default=None,
description="Document IDs of source documents for tree traversal -- required for deductive and inductive documents",
)
premises: list[str] | None = Field(
default=None,
description="Human-readable premise text for display -- only applicable for deductive documents",
)
sources: list[str] | None = Field(
default=None,
description="Human-readable source text for display -- only applicable for inductive documents",
)
pattern_type: str | None = Field(
default=None,
description="Type of pattern identified (preference, behavior, personality, tendency, correlation) -- only applicable for inductive documents",
)
confidence: str | None = Field(
default=None,
description="Confidence level (high, medium, low) -- only applicable for inductive documents",
)
class DocumentCreate(DocumentBase):
content: Annotated[str, Field(min_length=1, max_length=100000)]
session_name: str | None = Field(
default=None,
description="The session from which the document was derived (NULL for global observations)",
)
level: DocumentLevel = Field(
default="explicit",
description="The level of the document (explicit, deductive, inductive, or contradiction)",
)
times_derived: int = Field(
default=1,
ge=1,
description="The number of times that a semantic duplicate document to this one has been derived",
)
metadata: DocumentMetadata = Field()
embedding: list[float] = Field()
# Tree linkage field
source_ids: list[str] | None = Field(
default=None,
description="Document IDs of source/premise documents -- for deductive and inductive documents",
)
class ObservationInput(BaseModel):
"""Validated observation input from LLM tool calls."""
content: Annotated[str, Field(min_length=1)]
level: DocumentLevel = "explicit"
source_ids: list[str] | None = None
premises: list[str] | None = None
sources: list[str] | None = None
pattern_type: (
Literal["preference", "behavior", "personality", "tendency", "correlation"]
| None
) = None
confidence: Literal["high", "medium", "low"] | None = None
@field_validator("content", mode="after")
@classmethod
def sanitize_content(cls, v: str) -> str:
sanitized = strip_nul_bytes(v)
if not sanitized:
raise PydanticCustomError(
"string_too_short",
"String should have at least 1 character",
)
return sanitized
@model_validator(mode="after")
def validate_level_fields(self) -> Self:
"""Validate that level-specific fields are present when required."""
if self.level == "deductive" and not self.source_ids:
raise ValueError(
"deductive observations require 'source_ids' field with document IDs of premises"
)
if self.level == "inductive" and not self.source_ids:
raise ValueError(
"inductive observations require 'source_ids' field with document IDs of sources"
)
if self.level == "contradiction" and (
not self.source_ids or len(self.source_ids) < 2
):
raise ValueError(
"contradiction observations require 'source_ids' field with at least 2 IDs of contradicting observations"
)
return self
# ---------------------------------------------------------------------------
# Queue internals
# ---------------------------------------------------------------------------
class SessionCounts(BaseModel):
"""Counts for a specific session in queue processing."""
completed: int
in_progress: int
pending: int
class QueueCounts(BaseModel):
"""Aggregated counts for queue processing status."""
total: int
completed: int
in_progress: int
pending: int
sessions: dict[str, SessionCounts]
class QueueStatusRow(BaseModel):
"""Represents a row from the queue status SQL query result."""
session_id: str | None
total: int
completed: int
in_progress: int
pending: int
session_total: int
session_completed: int
session_in_progress: int
session_pending: int
# ---------------------------------------------------------------------------
# Internal data containers
# ---------------------------------------------------------------------------
class SessionPeerData(BaseModel):
"""Data for managing session peer relationships."""
peer_names: dict[str, SessionPeerConfig]
class MessageBulkData(BaseModel):
"""Data for bulk message operations."""
messages: list[MessageCreate]
session_name: str
workspace_name: str

View File

@ -1,5 +1,6 @@
import asyncio
import logging
import weakref
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
@ -46,7 +47,14 @@ def _safe_int(value: Any, default: int) -> int:
# Module-level lock registry for thread-safe observation creation.
# Keyed by (workspace_name, observer, observed) to ensure all tool executors
# operating on the same data share the same lock.
_observation_locks: dict[tuple[str, str, str], asyncio.Lock] = {}
#
# Uses WeakValueDictionary so entries are automatically removed when no
# ToolContext holds a reference to the lock (i.e., all executors for that
# key have finished and been garbage collected). This prevents unbounded
# growth over the lifetime of a long-running deriver process.
_observation_locks: weakref.WeakValueDictionary[tuple[str, str, str], asyncio.Lock] = (
weakref.WeakValueDictionary()
)
_registry_lock = asyncio.Lock()
@ -59,6 +67,11 @@ async def get_observation_lock(
This ensures that concurrent tool executors operating on the same observation
space share a lock, preventing race conditions during document creation.
The lock is stored as a weak reference it stays alive as long as at least
one ToolContext (via create_tool_executor) holds a strong reference. Once all
executors for a key finish and are garbage collected, the entry is
automatically removed from the registry.
Args:
workspace_name: Workspace identifier
observer: The observing peer
@ -69,9 +82,11 @@ async def get_observation_lock(
"""
key = (workspace_name, observer, observed)
async with _registry_lock:
if key not in _observation_locks:
_observation_locks[key] = asyncio.Lock()
return _observation_locks[key]
lock = _observation_locks.get(key)
if lock is None:
lock = asyncio.Lock()
_observation_locks[key] = lock
return lock
@dataclass

View File

@ -10,7 +10,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
from src import schemas
from src.config import settings
from src.exceptions import FileProcessingError, UnsupportedFileTypeError
from src.exceptions import (
FileProcessingError,
UnsupportedFileTypeError,
ValidationException,
)
from src.schemas import Message
logger = logging.getLogger(__name__)
@ -58,7 +62,19 @@ class JSONProcessor:
async def extract_text(self, content: bytes) -> str:
import json
data = json.loads(content.decode("utf-8"))
try:
decoded_content = content.decode("utf-8")
except UnicodeDecodeError as exc:
raise ValidationException("JSON uploads must be UTF-8 encoded") from exc
if not decoded_content.strip():
return ""
try:
data = json.loads(decoded_content)
except json.JSONDecodeError as exc:
raise ValidationException("Uploaded JSON is invalid") from exc
# Convert JSON to readable text format
return json.dumps(data, ensure_ascii=False)

View File

@ -107,7 +107,7 @@ def apply_filter(
def _build_filter_conditions(
filter_dict: dict[str, Any], model_class: type[Any]
filter_dict: dict[str, Any], model_class: type[Any], *, _depth: int = 0
) -> ColumnElement[bool] | None:
"""
Recursively build filter conditions from a filter dictionary.
@ -119,6 +119,9 @@ def _build_filter_conditions(
Returns:
SQLAlchemy condition object or None
"""
if _depth > 5:
raise FilterError("Filter nesting exceeds maximum depth of 5")
conditions: list[ColumnElement[bool]] = []
# Handle logical operators
@ -129,7 +132,11 @@ def _build_filter_conditions(
)
and_conditions: list[ColumnElement[bool]] = []
for sub_filter in filter_dict["AND"]: # pyright: ignore
sub_condition = _build_filter_conditions(sub_filter, model_class) # pyright: ignore
sub_condition = _build_filter_conditions(
sub_filter, # pyright: ignore[reportUnknownArgumentType]
model_class,
_depth=_depth + 1,
)
if sub_condition is not None:
and_conditions.append(sub_condition)
if and_conditions:
@ -142,7 +149,11 @@ def _build_filter_conditions(
)
or_conditions: list[ColumnElement[bool]] = []
for sub_filter in filter_dict["OR"]: # pyright: ignore
sub_condition = _build_filter_conditions(sub_filter, model_class) # pyright: ignore
sub_condition = _build_filter_conditions(
sub_filter, # pyright: ignore[reportUnknownArgumentType]
model_class,
_depth=_depth + 1,
)
if sub_condition is not None:
or_conditions.append(sub_condition)
if or_conditions:
@ -157,7 +168,11 @@ def _build_filter_conditions(
)
not_conditions: list[ColumnElement[bool]] = []
for sub_filter in filter_dict["NOT"]: # pyright: ignore
sub_condition = _build_filter_conditions(sub_filter, model_class) # pyright: ignore
sub_condition = _build_filter_conditions(
sub_filter, # pyright: ignore[reportUnknownArgumentType]
model_class,
_depth=_depth + 1,
)
if sub_condition is not None:
not_conditions.append(
not_(sub_condition)

View File

@ -131,7 +131,7 @@ When you run the harness, it will start:
The harness uses environment variables to configure Honcho's database connection:
- `DB_CONNECTION_URI`: Set to `postgresql+psycopg://testuser:testpwd@localhost:{port}/honcho`
- `DB_CONNECTION_URI`: Derived from the database credentials in `docker-compose.yml.example` (e.g. `postgresql+psycopg://postgres:postgres@localhost:{port}/postgres`)
The script will print the actual configuration that Honcho is using after the FastAPI server starts. This gives you complete visibility into how Honcho's configuration system resolved the settings from environment variables, config files, and defaults.

View File

@ -20,6 +20,7 @@ import tempfile
import threading
import time
from pathlib import Path
from typing import Any
import yaml
@ -59,6 +60,35 @@ class HonchoHarness:
self.processes: list[tuple[str, subprocess.Popen[str]]] = []
self.env_file_backup: Path | None = None
self.output_threads: list[threading.Thread] = []
# DB credentials — populated from docker-compose.yml.example in create_temp_docker_compose
self.db_user: str = "postgres"
self.db_password: str = "postgres"
self.db_name: str = "postgres"
def _extract_db_credentials(self, compose_data: dict[str, Any]) -> None:
"""Extract POSTGRES_USER/PASSWORD/DB from the database service environment."""
services: dict[str, Any] = compose_data.get("services", {})
database: dict[str, Any] = services.get("database", {})
db_env: list[Any] = database.get("environment", [])
env_map: dict[str, str] = {}
for entry in db_env:
if isinstance(entry, str) and "=" in entry:
# Handle "- KEY=VALUE" format
key, _, value = entry.partition("=")
env_map[key.strip()] = value.strip()
self.db_user = env_map.get("POSTGRES_USER", self.db_user)
self.db_password = env_map.get("POSTGRES_PASSWORD", self.db_password)
self.db_name = env_map.get("POSTGRES_DB", self.db_name)
@property
def db_connection_uri(self) -> str:
"""SQLAlchemy-style connection URI for the test database."""
return f"postgresql+psycopg://{self.db_user}:{self.db_password}@localhost:{self.db_port}/{self.db_name}"
@property
def db_connection_uri_plain(self) -> str:
"""Plain psycopg connection URI (no +psycopg driver prefix)."""
return f"postgresql://{self.db_user}:{self.db_password}@localhost:{self.db_port}/{self.db_name}"
def create_temp_docker_compose(self) -> Path:
"""
@ -72,6 +102,9 @@ class HonchoHarness:
with open(example_file) as f:
compose_data = yaml.safe_load(f)
# Extract DB credentials from the compose file so the harness stays in sync
self._extract_db_credentials(compose_data)
# Update the database port
compose_data["services"]["database"]["ports"] = [f"{self.db_port}:5432"]
@ -165,7 +198,7 @@ class HonchoHarness:
Dictionary of environment variables for database connection, cache, and API keys
"""
return {
"DB_CONNECTION_URI": f"postgresql+psycopg://testuser:testpwd@localhost:{self.db_port}/honcho",
"DB_CONNECTION_URI": self.db_connection_uri,
"CACHE_ENABLED": "true",
"CACHE_URL": f"redis://localhost:{self.redis_port}/0",
}
@ -336,7 +369,7 @@ class HonchoHarness:
"-p",
str(self.db_port),
"-U",
"testuser",
self.db_user,
],
capture_output=True,
text=True,
@ -354,9 +387,7 @@ class HonchoHarness:
try:
import psycopg
conn = psycopg.connect(
f"postgresql://testuser:testpwd@localhost:{self.db_port}/honcho"
)
conn = psycopg.connect(self.db_connection_uri_plain)
conn.close()
print("Database is ready!")
return True
@ -401,9 +432,7 @@ class HonchoHarness:
import psycopg
# Connect to the database using instance-specific connection string
conn_string = (
f"postgresql://testuser:testpwd@localhost:{self.db_port}/honcho"
)
conn_string = self.db_connection_uri_plain
conn = psycopg.connect(conn_string)
with conn.cursor() as cursor:

View File

@ -134,6 +134,33 @@ async def test_create_messages_with_json_file(
assert message["session_id"] == session_name
@pytest.mark.asyncio
async def test_create_messages_with_empty_json_file(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
"""Test that empty JSON uploads do not crash and create empty content."""
test_workspace, test_peer = sample_data
test_session = await _create_test_session(db_session, test_workspace)
session_name = test_session.name
file_data = io.BytesIO(b"")
files = {"file": ("empty.json", file_data, "application/json")}
form_data = {"peer_id": test_peer.name}
url = _get_upload_url(test_workspace.name, session_name)
response = client.post(url, files=files, data=form_data)
assert response.status_code == 201
data = response.json()
assert len(data) == 1
assert data[0]["content"] == ""
assert data[0]["peer_id"] == test_peer.name
assert data[0]["session_id"] == session_name
@pytest.mark.asyncio
async def test_create_messages_with_unsupported_file_type(
client: TestClient,

View File

@ -13,12 +13,12 @@ from src.dependencies import tracked_db as real_tracked_db
class FakeSession:
def __init__(self, *, in_transaction: bool = False):
self._in_transaction: bool = in_transaction
self.execute_calls: list[Any] = []
self.execute_calls: list[tuple[Any, ...]] = []
self.rollback_calls: int = 0
self.close_calls: int = 0
async def execute(self, statement: Any) -> None:
self.execute_calls.append(statement)
async def execute(self, statement: Any, params: Any = None) -> None:
self.execute_calls.append((statement, params))
async def rollback(self) -> None:
self.rollback_calls += 1
@ -45,9 +45,9 @@ async def test_get_db_sets_application_name_when_tracing_enabled(
db = await anext(dep_gen)
assert db is fake_db
assert len(fake_db.execute_calls) == 1
assert "SET application_name = 'request:test-ctx'" in str(
fake_db.execute_calls[0]
)
stmt, params = fake_db.execute_calls[0]
assert "set_config" in str(stmt)
assert params == {"name": "request:test-ctx"}
finally:
await dep_gen.aclose()
request_context.reset(context_token)
@ -96,9 +96,9 @@ async def test_tracked_db_creates_and_resets_task_context(
assert request_context.get() is None
assert len(fake_db.execute_calls) == 1
assert "SET application_name = 'task:cleanup_job:12345678'" in str(
fake_db.execute_calls[0]
)
stmt, params = fake_db.execute_calls[0]
assert "set_config" in str(stmt)
assert params == {"name": "task:cleanup_job:12345678"}
assert fake_db.rollback_calls == 0
assert fake_db.close_calls == 1
@ -119,7 +119,9 @@ async def test_tracked_db_preserves_existing_request_context(
request_context.reset(context_token)
assert len(fake_db.execute_calls) == 1
assert "SET application_name = 'request:existing'" in str(fake_db.execute_calls[0])
stmt, params = fake_db.execute_calls[0]
assert "set_config" in str(stmt)
assert params == {"name": "request:existing"}
assert fake_db.rollback_calls == 0
assert fake_db.close_calls == 1

View File

@ -1160,3 +1160,147 @@ class TestToolExecutor:
if "Found" in result and "observations" in result:
# IDs should be included in the output
assert "[id:" in result or "observations" in result
# =============================================================================
# Observation Lock Registry Tests
# =============================================================================
@pytest.mark.asyncio
class TestObservationLockRegistry:
"""Tests for the WeakValueDictionary-based observation lock registry."""
async def test_same_key_returns_same_lock(self):
"""Concurrent callers with the same key get the same Lock instance."""
from src.utils.agent_tools import get_observation_lock
lock_a = await get_observation_lock("ws1", "obs1", "peer1")
lock_b = await get_observation_lock("ws1", "obs1", "peer1")
assert lock_a is lock_b
async def test_different_keys_return_different_locks(self):
"""Different keys produce independent Lock instances."""
from src.utils.agent_tools import get_observation_lock
lock_a = await get_observation_lock("ws_diff_a", "obs", "peer")
lock_b = await get_observation_lock("ws_diff_b", "obs", "peer")
assert lock_a is not lock_b
async def test_lock_evicted_after_all_references_dropped(self):
"""Lock is removed from registry once no strong references remain."""
import gc
from src.utils.agent_tools import (
_observation_locks, # pyright: ignore[reportPrivateUsage]
get_observation_lock,
)
key = ("ws_evict", "obs_evict", "peer_evict")
lock = await get_observation_lock(*key)
assert key in _observation_locks
# Drop the only strong reference and force GC
del lock
gc.collect()
assert key not in _observation_locks
async def test_lock_recreated_after_eviction(self):
"""A new lock is created for a key whose previous lock was evicted."""
import gc
import weakref
from src.utils.agent_tools import get_observation_lock
key = ("ws_recreate", "obs_recreate", "peer_recreate")
first_lock = await get_observation_lock(*key)
first_ref = weakref.ref(first_lock)
# Evict
del first_lock
gc.collect()
# Confirm the old lock was garbage-collected
assert first_ref() is None
# Recreate
second_lock = await get_observation_lock(*key)
assert isinstance(second_lock, asyncio.Lock)
async def test_lock_survives_while_any_reference_held(self):
"""Lock stays alive as long as at least one strong reference exists."""
import gc
from src.utils.agent_tools import (
_observation_locks, # pyright: ignore[reportPrivateUsage]
get_observation_lock,
)
key = ("ws_survive", "obs_survive", "peer_survive")
ref_a = await get_observation_lock(*key)
ref_b = await get_observation_lock(*key)
assert ref_a is ref_b
# Drop one reference — lock should survive via the other
del ref_a
gc.collect()
assert key in _observation_locks
# Drop the last reference — now it should be evicted
del ref_b
gc.collect()
assert key not in _observation_locks
async def test_concurrent_executors_share_lock_for_mutual_exclusion(self):
"""Two coroutines using the same key are serialized by the shared lock."""
from src.utils.agent_tools import get_observation_lock
key = ("ws_mutex", "obs_mutex", "peer_mutex")
shared_lock = await get_observation_lock(*key)
order: list[str] = []
async def task(name: str, delay: float):
async with shared_lock:
order.append(f"{name}_start")
await asyncio.sleep(delay)
order.append(f"{name}_end")
# task_a grabs the lock first, task_b must wait
task_a = asyncio.create_task(task("a", 0.05))
await asyncio.sleep(0.01) # let task_a acquire the lock
task_b = asyncio.create_task(task("b", 0.01))
await asyncio.gather(task_a, task_b)
# task_a must fully complete before task_b starts
assert order == ["a_start", "a_end", "b_start", "b_end"]
async def test_no_registry_growth_across_many_keys(self):
"""Registry does not retain locks after references are dropped."""
import gc
from src.utils.agent_tools import (
_observation_locks, # pyright: ignore[reportPrivateUsage]
get_observation_lock,
)
locks: list[asyncio.Lock] = []
for i in range(100):
locks.append(await get_observation_lock(f"ws_growth_{i}", "obs", "peer"))
count_before = sum(
1 for k in _observation_locks if k[0].startswith("ws_growth_")
)
assert count_before == 100
# Drop all strong references and force GC
locks.clear()
gc.collect()
# All 100 entries should be cleaned up
remaining = sum(1 for k in _observation_locks if k[0].startswith("ws_growth_"))
assert remaining == 0

39
tests/utils/test_files.py Normal file
View File

@ -0,0 +1,39 @@
import json
import pytest
from src.exceptions import ValidationException
from src.utils.files import JSONProcessor
@pytest.mark.asyncio
async def test_json_processor_returns_empty_string_for_blank_content():
processor = JSONProcessor()
assert await processor.extract_text(b"") == ""
assert await processor.extract_text(b" \n\t") == ""
@pytest.mark.asyncio
async def test_json_processor_preserves_valid_json_behavior():
processor = JSONProcessor()
result = await processor.extract_text(b'{"name": "test", "count": 1}')
assert json.loads(result) == {"name": "test", "count": 1}
@pytest.mark.asyncio
async def test_json_processor_rejects_non_utf8_content():
processor = JSONProcessor()
with pytest.raises(ValidationException, match="UTF-8"):
await processor.extract_text(b"\xff\xfe\x00{")
@pytest.mark.asyncio
async def test_json_processor_rejects_invalid_json_content():
processor = JSONProcessor()
with pytest.raises(ValidationException, match="invalid"):
await processor.extract_text(b'{"name": }')