RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Peer Cards (#180) 

* chore: fill out missing metadata inputs in python sdk

* feat: add get_peer_config to python sdk, thoroughly document ts sdk and remove bad client usage

* feat: zod
chore: update tests
chore: bump version, changelog

* chore: python sdk version bump and changelog

* [WIP] feat: combine search methods and rework endpoint to include limit param

* chore: test new stainless config with library

* nits: coderabbit

* Merge branch 'ben/sdk-improvements' into ben/search-rrf

* chore: pre-commit hooks cleanup

* feat: thoroughly document observation config

* Update sdks/python/src/honcho/peer.py

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* chore: v1.3.0

* feat: update version to 2.2.0 and enhance search functionality with arbitrary filters

- Remove unused config variables
- Added arbitrary filters to all search endpoints.
- Pluralize `filters` everywhere in SDKs for consistency
- Updated documentation and changelog to reflect these changes.

* expose core client in TS and Python SDKs (#150)

* expose core client from sdks

* align text

* fix: resolve get_effective_observe me race condition, default peer config (#176)

* fix: resolve get_effective_observe me race condition, default peer config

* fix: preserve custom config even after leaving

* chore: test cases, enqueue types

* Update sdks/typescript/package.json

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

---------

Co-authored-by: doria <93405247+dr-frmr@users.noreply.github.com>
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* chore: formatting

* chore: revert undesired changes to v1 spec, clean up docs, coderabbit

* feat: better search docs, fix worker.ts

* fix: correctly make ts params optional in cases, update docs

* chore: coderabbit

* chore: remove spurious package-lock

* feat: [WIP] introduce peer cards

* chore: remove search.mdx

* chore: clean up deriver

* feat: peer cards working in deriver

* feat: add basic peer_card_bench

* feat: refine peer card prompt, add mini-benchmark, switch to gpt-5-nano

* refactor: update peer card handling in dialectic functions and improve error handling

- Enhanced `get_peer_card` function to handle `ResourceNotFoundException`.
- Updated `dialectic_call` and `dialectic_stream` to accept `peer_card` and `target_peer_card` parameters.
- Modified prompt generation to include peer card information.
- Cleaned up whitespace in several files for consistency.

* chore: update mirascope dependency version in configuration files

- Bumped mirascope version from 1.25.1 to 1.25.5 in pyproject.toml and uv.lock.
- Added a note in config.py regarding peer card output token handling.
- Removed unnecessary comments in clients.py for clarity.

* fix: [coderabbit] improve error handling in set_peer_card and enhance logging

- Added a check in `set_peer_card` to raise `ResourceNotFoundException` if the peer does not exist.
- Updated logging in `CertaintyReasoner` to capture exceptions with Sentry when enabled.
- Refined logging messages for clarity and consistency across various functions.
- Cleaned up whitespace and formatting in several files for improved readability.

* refactor: update working representation handling and improve metadata key usage

- Introduced constants for representation collection names to enhance clarity and maintainability.
- Updated function signatures in `get_working_representation` and `set_working_representation` to require `session_name`.
- Simplified metadata key determination logic by using constants instead of hardcoded strings.
- Removed legacy fallback logic for working representation data retrieval.
- Refactored `save_working_representation_to_peer` to utilize the new `set_working_representation` function for improved code reuse.

* chore: update configuration files and enhance working representation settings

- Added new peer card settings and context token limits to `.env.template`, `config.toml.example`, and documentation.
- Introduced `WORKING_REPRESENTATION_MAX_OBSERVATIONS` to `DeriverSettings` for better control over observation storage.
- Updated `set_working_representation` to merge new observations while respecting the maximum limit.
- Improved docstrings for clarity and consistency across functions.

* feat: introduce LLMError exception and enhance error handling in deriver

- Added LLMError exception to handle failures in LLM calls, normalizing inputs into a JSON-serializable format.
- Updated CertaintyReasoner to raise LLMError on exceptions during LLM function calls.
- Enhanced QueueManager to log LLMError occurrences and re-queue messages appropriately.
- Modified test runner to support asynchronous operations and improved output formatting for test results.
- Updated test cases to include session information for better context.

* feat: add __repr__ method to QueueItem for improved string representation

- Implemented a __repr__ method in the QueueItem class to provide a clear and informative string representation of its attributes.
- Updated timeout handling in TestRunner to default to 10000.0 seconds when timeout_seconds is not set, enhancing robustness in polling operations.

* refactor: update peer card data structure and improve handling in related functions

- Changed return type of `get_peer_card` and `set_peer_card` to use `list[str]` instead of `str | None`.
- Updated `peer_card_call` and related functions to accommodate the new list structure for peer cards.
- Introduced `PeerCardQuery` model to standardize responses from peer card queries.
- Adjusted prompt generation in `peer_card_prompt` to reflect the new data structure.
- Modified benchmark tests to align with the updated peer card handling.

* refactor: adjust peer card output token settings and update related functions

- Increased `PEER_CARD_MAX_OUTPUT_TOKENS` from 2000 to 4000 in `DeriverSettings`.
- Updated `critical_analysis_call` to use `json_mode` and removed unused parameters.
- Modified benchmark tests to utilize the new `PEER_CARD_MAX_OUTPUT_TOKENS` setting.
- Removed obsolete `add_dislike.json` test file.

* refactor: update peer card handling in critical analysis and dialectic prompts

- Changed `peer_card` parameter type from `str | None` to `list[str] | None` in `critical_analysis_call` and related functions.
- Simplified error handling in `process_representation_task` by removing redundant try-except block.
- Updated prompt generation in `critical_analysis_prompt` and `dialectic_prompt` to format `peer_card` as a string with newlines.
- Adjusted benchmark tests to reflect changes in peer card structure and output formatting.

* refactor: update peer card test cases to use list structure

- Modified test cases in `test_representation_crud.py` to reflect the change in `peer_card` parameter type from `str` to `list[str]`.
- Updated assertions to accommodate the new list format for setting and retrieving peer cards.
- Ensured that tests for missing peers correctly handle the list input format.

* fix: improve formatting of peer card output in prompts

- Updated `peer_card_prompt` to join `old_peer_card` list elements with newlines for better readability.
- Removed outdated comment in `dialectic_prompt` regarding handling of non-existent cards.

* chore: [coderabbit] enhance docstring and logging in prompts and queue manager

- Updated the docstring in `critical_analysis_prompt` to provide detailed type annotations for parameters.
- Improved logging in `chat` to differentiate between single and multiple retrieved peer cards.
- Adjusted logging format in `QueueManager` to use a more structured approach for shutdown messages.

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
Co-authored-by: Rajat Ahuja <rahuja445@gmail.com>
This commit is contained in:
doria 2025-08-12 17:39:08 -04:00 committed by GitHub
parent e914dbe334
commit 7a9746ca8b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
51 changed files with 2421 additions and 726 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
.env.template
View File

@ -81,9 +81,14 @@ LLM_ANTHROPIC_API_KEY=your-anthropic-api-key-here
# DERIVER_STALE_SESSION_TIMEOUT_MINUTES=5
# DERIVER_PROVIDER=google
# DERIVER_MODEL=gemini-2.0-flash-lite
# MAX_OUTPUT_TOKENS=2500
# DERIVER_MAX_OUTPUT_TOKENS=2500
# only applied when using Anthropic as provider
# THINKING_BUDGET_TOKENS=1024
# DERIVER_THINKING_BUDGET_TOKENS=1024
# DERIVER_PEER_CARD_PROVIDER=openai
# DERIVER_PEER_CARD_MODEL=gpt-5-nano-2025-08-07
# DERIVER_PEER_CARD_MAX_OUTPUT_TOKENS=2000
# DERIVER_CONTEXT_TOKEN_LIMIT=30000
# DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100
# =============================================================================
# Dialectic Settings

5
config.toml.example
View File

@ -64,6 +64,11 @@ PROVIDER = "google"
MODEL = "gemini-2.0-flash-lite"
MAX_OUTPUT_TOKENS = 2500
THINKING_BUDGET_TOKENS = 1024 # only applied when using Anthropic
PEER_CARD_PROVIDER = "openai"
PEER_CARD_MODEL = "gpt-5-nano-2025-08-07"
PEER_CARD_MAX_OUTPUT_TOKENS = 2000
CONTEXT_TOKEN_LIMIT = 30000
WORKING_REPRESENTATION_MAX_OBSERVATIONS = 100
# Dialectic settings
[dialectic]

12
docs/v2/contributing/configuration.mdx
View File

@ -254,6 +254,18 @@ DERIVER_MODEL=gemini-2.0-flash-lite
DERIVER_WORKERS=1
DERIVER_STALE_SESSION_TIMEOUT_MINUTES=5
DERIVER_POLLING_SLEEP_INTERVAL_SECONDS=1.0
# Peer card settings
DERIVER_PEER_CARD_PROVIDER=openai
DERIVER_PEER_CARD_MODEL=gpt-5-nano-2025-08-07
DERIVER_PEER_CARD_MAX_OUTPUT_TOKENS=2000
# Context token limit for get_context method
DERIVER_CONTEXT_TOKEN_LIMIT=30000
# Maximum number of observations to store in working representation
# This is applied to both explicit and deductive observations
DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100
```
**Summary Generation:**

4
pyproject.toml
View File

@ -22,8 +22,8 @@ dependencies = [
"alembic>=1.14.0",
"pyjwt>=2.10.0",
"tiktoken>=0.9.0",
"mirascope[anthropic,google,groq,langfuse]>=1.25.1",
"openai>=1.91.0",
"mirascope[anthropic,google,groq,langfuse]>=1.25.5",
"openai>=1.99.7",
"pydantic>=2.11.7",
"pydantic-settings>=2.10.1",
"google-generativeai>=0.8.5",

13
src/config.py
View File

@ -194,11 +194,24 @@ class DeriverSettings(HonchoSettings):
# Thinking budget tokens are only applied when using Anthropic as provider
THINKING_BUDGET_TOKENS: Annotated[int, Field(default=1024, gt=0, le=5000)] = 1024
PEER_CARD_PROVIDER: Providers = "openai"
PEER_CARD_MODEL: str = "gpt-5-nano-2025-08-07"
# Note: peer cards should be very short, but GPT-5 models need output tokens for thinking which cannot be turned off...
PEER_CARD_MAX_OUTPUT_TOKENS: Annotated[
int, Field(default=4000, gt=1000, le=10_000)
] = 4000
# Context token limit for get_context method
CONTEXT_TOKEN_LIMIT: Annotated[int, Field(default=30_000, gt=1000, le=100_000)] = (
30_000
)
# Maximum number of observations to store in working representation
# This is applied to both explicit and deductive observations
WORKING_REPRESENTATION_MAX_OBSERVATIONS: Annotated[
int, Field(default=100, gt=0, le=500)
] = 100
class DialecticSettings(HonchoSettings):
model_config = SettingsConfigDict(env_prefix="DIALECTIC_", extra="ignore") # pyright: ignore

9
src/crud/__init__.py
View File

@ -18,8 +18,10 @@ from .peer import (
)
from .representation import (
construct_collection_name,
get_peer_card,
get_working_representation,
get_working_representation_data,
set_peer_card,
set_working_representation,
)
from .session import (
@ -66,12 +68,13 @@ __all__ = [
"get_peers",
"update_peer",
"get_sessions_for_peer",
# Search
"representation",
# Representation
"construct_collection_name",
"get_peer_card",
"get_working_representation",
"get_working_representation_data",
"set_peer_card",
"set_working_representation",
"construct_collection_name",
# Session
"get_sessions",
"get_or_create_session",

261
src/crud/representation.py
View File

@ -1,20 +1,83 @@
from logging import getLogger
from typing import Any
from typing import Any, cast
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from src import models
from src import exceptions, models, schemas
from src.config import settings
from src.crud.peer import get_peer
from src.utils.shared_models import ObservationDict
logger = getLogger(__name__)
# The collection name for documents that make up a peer's global representation
GLOBAL_REPRESENTATION_COLLECTION_NAME = "global_representation"
# The key for the working representation in the session peer's internal_metadata
WORKING_REPRESENTATION_METADATA_KEY = "working_representation"
# Old working representation key--remove in 2.3.0?
WORKING_REPRESENTATION_LEGACY_METADATA_KEY = "global_representation"
Observation = str | ObservationDict
async def get_peer_card(
db: AsyncSession, workspace_name: str, peer_name: str
) -> list[str] | None:
"""
Get peer card from internal_metadata.
Args:
db: Database session
workspace_name: Name of the workspace
peer_name: Name of the peer
Returns:
The peer's card text if present, otherwise None (also None if peer not found).
"""
try:
peer = await get_peer(db, workspace_name, schemas.PeerCreate(name=peer_name))
return peer.internal_metadata.get("peer_card", None)
except exceptions.ResourceNotFoundException:
return None
async def set_peer_card(
db: AsyncSession, workspace_name: str, peer_name: str, peer_card: list[str] | None
) -> None:
"""
Set peer card for a peer.
Raises:
ResourceNotFoundException: If the peer does not exist
"""
stmt = (
update(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.name == peer_name)
.values(
internal_metadata=models.Peer.internal_metadata.op("||")(
{"peer_card": peer_card}
)
)
)
result = await db.execute(stmt)
if result.rowcount == 0:
raise exceptions.ResourceNotFoundException(
f"Peer {peer_name} not found in workspace {workspace_name}"
)
await db.commit()
async def get_working_representation(
db: AsyncSession,
workspace_name: str,
observer_name: str,
observed_name: str,
session_name: str | None = None,
session_name: str,
) -> str:
"""
Get working representation for observer/observed relationship.
@ -23,8 +86,8 @@ async def get_working_representation(
db: Database session
workspace_name: Name of the workspace
observer_name: Name of the peer doing the observing
observed_name: Name of the peer being observed (required for explicit global/local)
session_name: Optional session name (None for peer-level metadata)
observed_name: Name of the peer being observed
session_name: Name of the session
Returns:
Formatted working representation string
@ -60,8 +123,8 @@ async def get_working_representation_data(
db: AsyncSession,
workspace_name: str,
observer_name: str,
observed_name: str, # now required
session_name: str | None = None,
observed_name: str,
session_name: str,
) -> dict[str, Any] | str | None:
"""
Get raw working representation data from internal_metadata.
@ -70,23 +133,17 @@ async def get_working_representation_data(
"""
# Determine metadata key based on observer/observed relationship
if observer_name == observed_name:
metadata_key = "global_representation"
metadata_key = WORKING_REPRESENTATION_METADATA_KEY
else:
metadata_key = construct_collection_name(
observer=observer_name, observed=observed_name
)
if session_name:
stmt = select(models.SessionPeer.internal_metadata).where(
models.SessionPeer.peer_name == observer_name,
models.SessionPeer.workspace_name == workspace_name,
models.SessionPeer.session_name == session_name,
)
else:
stmt = select(models.Peer.internal_metadata).where(
models.Peer.name == observer_name,
models.Peer.workspace_name == workspace_name,
)
stmt = select(models.SessionPeer.internal_metadata).where(
models.SessionPeer.peer_name == observer_name,
models.SessionPeer.workspace_name == workspace_name,
models.SessionPeer.session_name == session_name,
)
result = await db.execute(stmt)
peer_metadata = result.scalar_one_or_none()
@ -94,32 +151,15 @@ async def get_working_representation_data(
if not peer_metadata:
return None
# Try new prefixed key first, then fallback to legacy keys
working_rep_data = peer_metadata.get(metadata_key)
if working_rep_data:
return working_rep_data
return cast(dict[str, Any] | str, working_rep_data)
# Fallback logic for migration period
legacy_data = peer_metadata.get("latest_working_representation")
if legacy_data:
logger.debug(
"Using legacy key 'latest_working_representation' for %s->%s",
observer_name,
observed_name,
)
return legacy_data
# Final fallback to old user_representation key
USER_REPRESENTATION_METADATA_KEY = "user_representation"
user_rep_data = peer_metadata.get(USER_REPRESENTATION_METADATA_KEY)
if user_rep_data:
logger.debug(
"Using legacy key '%s' for %s->%s",
USER_REPRESENTATION_METADATA_KEY,
observer_name,
observed_name,
)
return user_rep_data
# Try legacy key--remove in 2.3.0?
if observer_name == observed_name:
working_rep_data = peer_metadata.get(WORKING_REPRESENTATION_LEGACY_METADATA_KEY)
if working_rep_data:
return cast(dict[str, Any] | str, working_rep_data)
return None
@ -129,7 +169,8 @@ def _format_observations_by_level(final_observations: dict[str, Any]) -> str:
formatted_sections: list[str] = []
for level in ["explicit", "deductive"]:
observations: list[Any] = final_observations.get(level, [])
observations_raw: Any = final_observations.get(level, [])
observations: list[Any] = cast(list[Any], observations_raw or [])
if observations:
formatted_sections.append(f"{level.upper()} OBSERVATIONS:")
formatted_sections.extend(_format_observation_list(observations))
@ -138,7 +179,7 @@ def _format_observations_by_level(final_observations: dict[str, Any]) -> str:
return "\n".join(formatted_sections) if formatted_sections else ""
def _format_observation_list(observations: list[dict[str, Any] | str]) -> list[str]:
def _format_observation_list(observations: list[Observation]) -> list[str]:
"""Format a list of observations into consistent string format."""
formatted: list[str] = []
for obs in observations:
@ -163,70 +204,130 @@ def _format_observation_list(observations: list[dict[str, Any] | str]) -> list[s
return formatted
def _merge_working_representation(
existing: dict[str, Any] | str | None, new: dict[str, Any]
) -> dict[str, Any]:
"""Merge a new working representation into an existing one.
- Appends `explicit` and `deductive` observations in that order
- Trims each list to the most recent `WORKING_REPRESENTATION_MAX_OBSERVATIONS` entries (FIFO)
- Uses the latest `thinking`, `message_id`, and `created_at`
"""
max_observations = settings.DERIVER.WORKING_REPRESENTATION_MAX_OBSERVATIONS
new_final_raw: Any = new.get("final_observations") or {}
new_explicit: list[Observation] = cast(
list[Observation], (new_final_raw.get("explicit") or [])
)
new_deductive: list[ObservationDict] = cast(
list[ObservationDict], (new_final_raw.get("deductive") or [])
)
existing_explicit: list[Observation] = []
existing_deductive: list[ObservationDict] = []
if isinstance(existing, dict):
existing_final_raw: Any = existing.get("final_observations") or {}
existing_explicit = cast(
list[Observation], (existing_final_raw.get("explicit") or [])
)
existing_deductive = cast(
list[ObservationDict], (existing_final_raw.get("deductive") or [])
)
merged_explicit: list[Observation] = existing_explicit + new_explicit
merged_deductive: list[ObservationDict] = existing_deductive + new_deductive
if len(merged_explicit) > max_observations:
merged_explicit = merged_explicit[-max_observations:]
if len(merged_deductive) > max_observations:
merged_deductive = merged_deductive[-max_observations:]
return {
"final_observations": {
"explicit": merged_explicit,
"thinking": cast(str | None, new_final_raw.get("thinking")),
"deductive": merged_deductive,
},
"message_id": cast(str | None, new.get("message_id")),
"created_at": cast(str | None, new.get("created_at")),
}
async def set_working_representation(
db: AsyncSession,
representation: str | dict[str, Any],
workspace_name: str,
observer_name: str, # renamed from peer_name
observed_name: str, # now required - no default
session_name: str | None = None,
observer_name: str,
observed_name: str,
session_name: str,
) -> None:
"""
Set working representation for observer/observed relationship.
If the provided representation is structured (dict with `final_observations`),
append new observations to the existing ones for both `explicit` and `deductive`
kinds, update `message_id` and `created_at`, and cap each observations list to
the most recent `WORKING_REPRESENTATION_MAX_OBSERVATIONS` items (FIFO trimming
of oldest entries).
Args:
db: Database session
representation: Working representation data (string or structured dict)
workspace_name: Name of the workspace
observer_name: Name of the peer doing the observing
observed_name: Name of the peer being observed (required for explicit global/local)
session_name: Optional session name (None for peer-level metadata)
session_name: Name of the session
"""
# Determine metadata key based on observer/observed relationship
if observer_name == observed_name:
metadata_key = "global_representation"
metadata_key = WORKING_REPRESENTATION_METADATA_KEY
else:
metadata_key = construct_collection_name(
observer=observer_name, observed=observed_name
)
if session_name:
# Session-level: save all types (global and local)
stmt = (
update(models.SessionPeer)
.where(models.SessionPeer.workspace_name == workspace_name)
.where(models.SessionPeer.peer_name == observer_name)
.where(models.SessionPeer.session_name == session_name)
.values(
internal_metadata=models.SessionPeer.internal_metadata.op("||")(
{metadata_key: representation}
)
merged_value: str | dict[str, Any] = representation
if isinstance(representation, dict):
try:
existing = await get_working_representation_data(
db=db,
workspace_name=workspace_name,
observer_name=observer_name,
observed_name=observed_name,
session_name=session_name,
)
merged_value = _merge_working_representation(
existing,
representation,
)
except Exception:
logger.exception(
"Failed to merge working representation; storing as provided"
)
merged_value = representation
stmt = (
update(models.SessionPeer)
.where(models.SessionPeer.workspace_name == workspace_name)
.where(models.SessionPeer.peer_name == observer_name)
.where(models.SessionPeer.session_name == session_name)
.values(
internal_metadata=models.SessionPeer.internal_metadata.op("||")(
{metadata_key: merged_value}
)
)
else:
# Peer-level: only save global representations
if observer_name == observed_name:
stmt = (
update(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.name == observer_name)
.values(
internal_metadata=models.Peer.internal_metadata.op("||")(
{metadata_key: representation}
)
)
)
else:
logger.error(
"Skipping peer-level local representation save (this should never happen!): observer=%s, observed=%s",
observer_name,
observed_name,
)
return
)
await db.execute(stmt)
await db.commit()
logger.info(
"Saved working representation to session peer %s - %s with key %s",
session_name,
observer_name,
metadata_key,
)
def construct_collection_name(*, observer: str, observed: str) -> str:
return f"{observer}_{observed}"

117
src/deriver/consumer.py
View File

@ -1,50 +1,103 @@
import logging
from typing import Any
import sentry_sdk
from langfuse.decorators import langfuse_context
from pydantic import ValidationError
from rich.console import Console
from sqlalchemy.ext.asyncio import AsyncSession
from .deriver import Deriver
from .queue_payload import RepresentationPayload, SummaryPayload, WebhookPayload
from src.config import settings
from src.dependencies import tracked_db
from src.deriver import deriver
from src.utils import summarizer
from src.utils.logging import log_performance_metrics
from src.webhooks import webhook_delivery
from .queue_payload import (
RepresentationPayload,
SummaryPayload,
WebhookPayload,
)
logger = logging.getLogger(__name__)
logging.getLogger("sqlalchemy.engine.Engine").disabled = True
console = Console(markup=True)
deriver = Deriver()
async def process_item(task_type: str, payload: dict[str, Any]) -> None:
# Validate payload structure and types before processing
try:
if task_type == "representation":
validated_payload = RepresentationPayload(**payload)
elif task_type == "summary":
validated_payload = SummaryPayload(**payload)
elif task_type == "webhook":
validated_payload = WebhookPayload(**payload)
else:
raise ValueError(f"Invalid task_type: {task_type}")
except ValidationError as e:
logger.error("Invalid payload received: %s. Payload: %s", str(e), payload)
raise ValueError(f"Invalid payload structure: {str(e)}") from e
"""Validate an incoming queue payload and dispatch it to the appropriate handler.
logger.debug(
"process_item received payload for task type %s ",
task_type,
)
This function centralizes payload validation using a simple mapping from
task type to Pydantic model. After validation, it routes the request to
the correct processor without repeating type checks elsewhere.
"""
logger.debug("process_item received payload for task type %s", task_type)
if task_type == "webhook":
if not isinstance(validated_payload, WebhookPayload):
raise ValueError(f"Expected WebhookPayload, got {type(validated_payload)}")
await deriver.process_webhook(validated_payload)
logger.debug("Finished processing webhook %s", validated_payload.event_type)
else:
if not isinstance(validated_payload, RepresentationPayload | SummaryPayload):
raise ValueError(
f"Expected DeriverQueuePayload, got {type(validated_payload)}"
try:
validated = WebhookPayload(**payload)
except ValidationError as e:
logger.error(
"Invalid webhook payload received: %s. Payload: %s", str(e), payload
)
deriver_payload = validated_payload
await deriver.process_message(task_type, deriver_payload)
logger.debug("Finished processing message %s", deriver_payload.message_id)
raise ValueError(f"Invalid payload structure: {str(e)}") from e
await process_webhook(validated)
logger.debug("Finished processing webhook %s", validated.event_type)
if settings.LANGFUSE_PUBLIC_KEY:
langfuse_context.update_current_trace(
metadata={
"critical_analysis_model": settings.DERIVER.MODEL,
}
)
# Open a DB session only for the duration of the processing call
async with tracked_db("deriver") as db:
if task_type == "summary":
try:
validated = SummaryPayload(**payload)
except ValidationError as e:
logger.error(
"Invalid summary payload received: %s. Payload: %s", str(e), payload
)
raise ValueError(f"Invalid payload structure: {str(e)}") from e
await process_summary_task(db, validated)
elif task_type == "representation":
try:
validated = RepresentationPayload(**payload)
except ValidationError as e:
logger.error(
"Invalid representation payload received: %s. Payload: %s",
str(e),
payload,
)
raise ValueError(f"Invalid payload structure: {str(e)}") from e
await deriver.process_representation_task(db, validated)
@sentry_sdk.trace
async def process_webhook(
payload: WebhookPayload,
) -> None:
async with tracked_db() as db:
await webhook_delivery.deliver_webhook(db, payload)
@sentry_sdk.trace
async def process_summary_task(
db: AsyncSession,
payload: SummaryPayload,
) -> None:
"""
Process a summary task by generating summaries if needed.
"""
await summarizer.summarize_if_needed(
db,
payload.workspace_name,
payload.session_name,
payload.message_id,
payload.message_seq_in_session,
)
log_performance_metrics(f"summary_{payload.workspace_name}_{payload.message_id}")

731
src/deriver/deriver.py
View File

@ -1,4 +1,5 @@
import datetime
import json
import logging
import time
from typing import Any
@ -7,15 +8,14 @@ import sentry_sdk
from langfuse.decorators import langfuse_context
from sqlalchemy.ext.asyncio import AsyncSession
from src import crud
from src import crud, exceptions
from src.config import settings
from src.dependencies import tracked_db
from src.crud.representation import GLOBAL_REPRESENTATION_COLLECTION_NAME
from src.utils import summarizer
from src.utils.clients import honcho_llm_call
from src.utils.embedding_store import EmbeddingStore
from src.utils.formatting import (
REASONING_LEVELS,
extract_observation_content,
find_new_observations,
format_context_for_prompt,
format_new_turn_with_timestamp,
@ -33,18 +33,15 @@ from src.utils.logging import (
from src.utils.shared_models import (
DeductiveObservation,
ObservationContext,
PeerCardQuery,
ReasoningResponse,
ReasoningResponseWithThinking,
UnifiedObservation,
)
from src.webhooks import webhook_delivery
from .prompts import critical_analysis_prompt
from .prompts import critical_analysis_prompt, peer_card_prompt
from .queue_payload import (
DeriverQueuePayload,
RepresentationPayload,
SummaryPayload,
WebhookPayload,
)
logger = logging.getLogger(__name__)
@ -65,313 +62,239 @@ logging.getLogger("sqlalchemy.engine.Engine").disabled = True
retry_attempts=3,
)
async def critical_analysis_call(
peer_name: str,
peer_card: list[str] | None,
message_created_at: datetime.datetime,
context: str,
working_representation: str | None,
history: str,
new_turn: str,
):
return critical_analysis_prompt(
peer_name=peer_name,
peer_card=peer_card,
message_created_at=message_created_at,
context=context,
working_representation=working_representation,
history=history,
new_turn=new_turn,
)
@honcho_llm_call(
provider=settings.DERIVER.PEER_CARD_PROVIDER,
model=settings.DERIVER.PEER_CARD_MODEL,
track_name="Peer Card Call",
response_model=PeerCardQuery,
json_mode=True,
max_tokens=settings.DERIVER.PEER_CARD_MAX_OUTPUT_TOKENS
or settings.LLM.DEFAULT_MAX_TOKENS,
reasoning_effort="minimal",
enable_retry=True,
retry_attempts=1, # unstructured output means we shouldn't need to retry, 1 just in case
)
async def peer_card_call(
old_peer_card: list[str] | None,
new_observations: list[str],
):
return peer_card_prompt(
old_peer_card=old_peer_card,
new_observations=new_observations,
)
@conditional_observe
class Deriver:
"""Deriver class for processing messages and extracting insights."""
@sentry_sdk.trace
async def process_representation_task(
db: AsyncSession,
payload: RepresentationPayload,
) -> None:
"""
Process a representation task by extracting insights and updating working representations.
"""
# Start overall timing
overall_start = time.perf_counter()
@sentry_sdk.trace
async def process_webhook(
self,
payload: WebhookPayload,
) -> None:
async with tracked_db() as db:
await webhook_delivery.deliver_webhook(db, payload)
logger.debug("Starting insight extraction for user message: %s", payload.message_id)
@sentry_sdk.trace
async def process_message(
self,
task_type: str,
payload: DeriverQueuePayload,
) -> None:
"""
Process a user message by extracting insights and saving them to the vector store.
This runs as a background process after a user message is logged.
"""
# Use get_session_context_formatted with configurable token limit
formatted_history = await summarizer.get_session_context_formatted(
db,
payload.workspace_name,
payload.session_name,
token_limit=settings.DERIVER.CONTEXT_TOKEN_LIMIT,
cutoff=payload.message_id,
include_summary=True,
)
if settings.LANGFUSE_PUBLIC_KEY:
langfuse_context.update_current_trace(
metadata={
"critical_analysis_model": settings.DERIVER.MODEL,
}
)
# Open a DB session only for the duration of the processing call
async with tracked_db("deriver") as db:
if task_type == "summary":
if not isinstance(payload, SummaryPayload):
raise ValueError(f"Expected SummaryPayload, got {type(payload)}")
await self.process_summary_task(db, payload)
elif task_type == "representation":
if not isinstance(payload, RepresentationPayload):
raise ValueError(
f"Expected RepresentationPayload, got {type(payload)}"
)
await self.process_representation_task(db, payload)
else:
raise ValueError(f"Unknown task type: {task_type}")
@sentry_sdk.trace
async def process_summary_task(
self,
db: AsyncSession,
payload: SummaryPayload,
) -> None:
"""
Process a summary task by generating summaries if needed.
"""
await summarizer.summarize_if_needed(
db,
payload.workspace_name,
payload.session_name,
payload.message_id,
payload.message_seq_in_session,
# instantiate embedding store from collection
# if the sender is also the target, we're handling a global representation task.
# otherwise, we're handling a directional representation task where the sender is
# being observed by the target.
collection_name = (
crud.construct_collection_name(
observer=payload.target_name, observed=payload.sender_name
)
log_performance_metrics(f"deriver_message_{payload.message_id}")
if payload.sender_name != payload.target_name
else GLOBAL_REPRESENTATION_COLLECTION_NAME
)
@sentry_sdk.trace
async def process_representation_task(
self,
db: AsyncSession,
payload: RepresentationPayload,
) -> None:
"""
Process a representation task by extracting insights and updating working representations.
"""
# Start overall timing
overall_start = time.perf_counter()
# get_or_create_collection already handles IntegrityError with rollback and a retry
collection = await crud.get_or_create_collection(
db,
payload.workspace_name,
collection_name,
payload.sender_name,
)
# Extract variables from payload for cleaner access
content = payload.content
workspace_name = payload.workspace_name
session_name = payload.session_name
message_id = payload.message_id
sender_name = payload.sender_name
target_name = payload.target_name
created_at = payload.created_at
# Use the embedding store directly
embedding_store = EmbeddingStore(
workspace_name=payload.workspace_name,
peer_name=payload.sender_name,
collection_name=collection.name,
)
logger.debug("Starting insight extraction for user message: %s", message_id)
# Create reasoner instance
reasoner = CertaintyReasoner(embedding_store=embedding_store, ctx=payload)
# Use message timestamp instead of wall-clock time for reasoning/insight dating
# created_at is now always a datetime object from Pydantic validation
message_dt_obj = created_at
# Check for existing working representation first, fall back to global search
working_rep_data: (
dict[str, Any] | str | None
) = await crud.get_working_representation_data(
db,
payload.workspace_name,
payload.target_name,
payload.sender_name,
payload.session_name,
)
# Use get_session_context_formatted with configurable token limit
formatted_history = await summarizer.get_session_context_formatted(
db,
workspace_name,
session_name,
token_limit=settings.DERIVER.CONTEXT_TOKEN_LIMIT,
cutoff=message_id,
include_summary=True,
)
# instantiate embedding store from collection
collection_name = (
crud.construct_collection_name(observer=target_name, observed=sender_name)
if sender_name != target_name
else "global_representation"
)
try:
collection = await crud.get_or_create_collection(
db, workspace_name, collection_name, sender_name
)
except Exception as e:
# Handle race condition from concurrent processing
if "duplicate key" in str(e).lower():
# Rollback the failed transaction
await db.rollback()
# Collection already exists, fetch it
collection = await crud.get_collection(
db, workspace_name, collection_name, sender_name
# Time context preparation
context_prep_start = time.perf_counter()
if (
working_rep_data
and isinstance(working_rep_data, dict)
and working_rep_data.get("final_observations")
):
# Reconstruct ReasoningResponse from stored peer data
final_obs: dict[str, Any] = working_rep_data["final_observations"]
deductive_observations: list[DeductiveObservation] = []
for deductive_data in final_obs.get("deductive", []):
deductive_observations.append(
DeductiveObservation(
conclusion=deductive_data["conclusion"],
premises=deductive_data.get("premises", []),
)
else:
raise
)
# Use the embedding store directly
embedding_store = EmbeddingStore(
workspace_name=workspace_name,
peer_name=sender_name,
collection_name=collection.name,
working_representation = ReasoningResponseWithThinking(
thinking=final_obs.get("thinking"),
explicit=final_obs.get("explicit", []),
deductive=deductive_observations,
)
logger.info(
"Using existing working representation with %s explicit, %s deductive observations",
len(working_representation.explicit),
len(working_representation.deductive),
)
else:
# No existing working representation, use global search
working_representation = await embedding_store.get_relevant_observations(
query=payload.content,
conversation_context=formatted_history,
for_reasoning=True,
)
# Create reasoner instance
reasoner = CertaintyReasoner(embedding_store=embedding_store)
# Check for existing working representation first, fall back to global search
working_rep_data: (
dict[str, Any] | str | None
) = await crud.get_working_representation_data(
db, workspace_name, target_name, sender_name, session_name
working_representation = observation_context_to_reasoning_response(
working_representation
)
logger.info("No working representation found, using global semantic search")
context_prep_duration = (time.perf_counter() - context_prep_start) * 1000
accumulate_metric(
f"deriver_representation_{payload.message_id}_{payload.target_name}",
"context_preparation",
context_prep_duration,
"ms",
)
# Time context preparation
context_prep_start = time.perf_counter()
if (
working_rep_data
and isinstance(working_rep_data, dict)
and working_rep_data.get("final_observations")
):
# Reconstruct ReasoningResponse from stored peer data
final_obs: dict[str, Any] = working_rep_data["final_observations"]
deductive_observations: list[DeductiveObservation] = []
for deductive_data in final_obs.get("deductive", []):
deductive_observations.append(
DeductiveObservation(
conclusion=deductive_data["conclusion"],
premises=deductive_data.get("premises", []),
)
)
# Run consolidated reasoning that handles explicit and deductive levels
logger.debug(
"REASONING: Running unified insight derivation across explicit and deductive reasoning levels"
)
initial_reasoning_context = ReasoningResponseWithThinking(
thinking=final_obs.get("thinking"),
explicit=final_obs.get("explicit", []),
deductive=deductive_observations,
)
logger.info(
"Using existing working representation with %s explicit, %s deductive observations",
len(initial_reasoning_context.explicit),
len(initial_reasoning_context.deductive),
)
# We currently only use Peer Cards in Honcho-level representation derivation.
if payload.sender_name == payload.target_name:
sender_peer_card: list[str] | None = await crud.get_peer_card(
db, payload.workspace_name, payload.sender_name
)
if sender_peer_card is None:
logger.warning("No peer card found for %s", payload.sender_name)
else:
# No working representation, use global search
initial_context = await embedding_store.get_relevant_observations(
query=content,
conversation_context=formatted_history,
for_reasoning=True,
)
logger.info("Using peer card: %s", sender_peer_card)
else:
logger.info("No peer card used for directional representation derivation")
sender_peer_card = None
initial_reasoning_context = (
reasoner.observation_context_to_reasoning_response(initial_context)
)
logger.info("No working representation found, using global semantic search")
context_prep_duration = (time.perf_counter() - context_prep_start) * 1000
accumulate_metric(
f"deriver_message_{message_id}",
"context_preparation",
context_prep_duration,
"ms",
# Run single-pass reasoning
final_observations = await reasoner.reason(
db,
working_representation,
formatted_history,
sender_peer_card,
)
logger.debug("REASONING COMPLETION: Unified reasoning completed across all levels.")
# Display final observations in a beautiful tree
final_obs_dict = {
level: getattr(final_observations, level, []) for level in REASONING_LEVELS
}
log_observations_tree(final_obs_dict)
# Always save working representation to peer for dialectic access
await save_working_representation_to_peer(db, payload, final_observations)
# Calculate and log overall timing
overall_duration = (time.perf_counter() - overall_start) * 1000
accumulate_metric(
f"deriver_representation_{payload.message_id}_{payload.target_name}",
"total_processing_time",
overall_duration,
"ms",
)
total_observations = sum(len(obs_list) for obs_list in final_obs_dict.values())
accumulate_metric(
f"deriver_representation_{payload.message_id}_{payload.target_name}",
"final_observation_count",
total_observations,
"",
)
log_performance_metrics(
f"deriver_representation_{payload.message_id}_{payload.target_name}"
)
if settings.LANGFUSE_PUBLIC_KEY:
langfuse_context.update_current_trace(
output=format_reasoning_response_as_markdown(final_observations)
)
# Run consolidated reasoning that handles explicit and deductive levels
logger.debug(
"REASONING: Running unified insight derivation across explicit and deductive reasoning levels"
)
# Run single-pass reasoning
final_observations = await reasoner.reason(
initial_reasoning_context,
formatted_history,
content,
str(message_id), # Convert int to str
session_name,
message_dt_obj,
sender_name, # Pass the speaker name
)
logger.debug(
"REASONING COMPLETION: Unified reasoning completed across all levels."
)
# Display final observations in a beautiful tree
final_obs_dict = {
level: getattr(final_observations, level, []) for level in REASONING_LEVELS
}
log_observations_tree(final_obs_dict)
# Always save working representation to peer for dialectic access
await save_working_representation_to_peer(
db,
workspace_name,
target_name, # observer (whose metadata we update)
sender_name, # observed (for key calculation)
session_name,
final_observations,
message_id,
)
# Calculate and log overall timing
overall_duration = (time.perf_counter() - overall_start) * 1000
accumulate_metric(
f"deriver_message_{message_id}",
"total_processing_time",
overall_duration,
"ms",
)
total_observations = sum(len(obs_list) for obs_list in final_obs_dict.values())
accumulate_metric(
f"deriver_message_{message_id}",
"final_observation_count",
total_observations,
"",
)
log_performance_metrics(f"deriver_message_{message_id}")
if settings.LANGFUSE_PUBLIC_KEY:
langfuse_context.update_current_trace(
output=format_reasoning_response_as_markdown(final_observations)
)
class CertaintyReasoner:
"""Certainty reasoner for analyzing and deriving insights."""
embedding_store: EmbeddingStore
ctx: RepresentationPayload
def __init__(self, embedding_store: EmbeddingStore) -> None:
def __init__(
self, embedding_store: EmbeddingStore, ctx: RepresentationPayload
) -> None:
self.embedding_store = embedding_store
def observation_context_to_reasoning_response(
self, context: "ObservationContext"
) -> ReasoningResponseWithThinking:
"""Convert ObservationContext to ReasoningResponse for compatibility."""
thinking = context.thinking
# Convert explicit observations to new structure
explicit: list[str] = []
for obs in context.explicit:
explicit.append(obs.content)
# Convert deductive observations
deductive: list[DeductiveObservation] = []
for obs in context.deductive:
deductive_obs = DeductiveObservation(
conclusion=obs.content,
premises=obs.metadata.premises if obs.metadata else [],
)
deductive.append(deductive_obs)
return ReasoningResponseWithThinking(
thinking=thinking,
explicit=explicit,
deductive=deductive,
)
self.ctx = ctx
@conditional_observe
@sentry_sdk.trace
async def derive_new_insights(
self,
context: ReasoningResponseWithThinking,
working_representation: ReasoningResponseWithThinking,
history: str,
new_turn: str,
message_created_at: datetime.datetime,
speaker: str,
speaker_peer_card: list[str] | None,
) -> ReasoningResponseWithThinking:
"""
Critically analyzes and revises understanding, returning structured observations.
@ -380,35 +303,46 @@ class CertaintyReasoner:
if settings.LANGFUSE_PUBLIC_KEY:
langfuse_context.update_current_observation(
input=format_reasoning_inputs_as_markdown(
context, history, new_turn, message_created_at
working_representation,
history,
self.ctx.content,
self.ctx.created_at,
)
)
formatted_new_turn = format_new_turn_with_timestamp(
new_turn, message_created_at, speaker
self.ctx.content,
self.ctx.created_at,
self.ctx.sender_name,
)
formatted_working_representation = format_context_for_prompt(
working_representation
)
formatted_context = format_context_for_prompt(context)
logger.debug(
"CRITICAL ANALYSIS: message_created_at='%s', formatted_new_turn='%s'",
message_created_at,
self.ctx.created_at,
formatted_new_turn,
)
# Call the standalone LLM function (now with Tenacity retries)
response_obj = await critical_analysis_call(
peer_name=speaker,
message_created_at=message_created_at,
context=formatted_context,
history=history,
new_turn=formatted_new_turn,
)
try:
response_obj = await critical_analysis_call(
peer_card=speaker_peer_card,
message_created_at=self.ctx.created_at,
working_representation=formatted_working_representation,
history=history,
new_turn=formatted_new_turn,
)
except Exception as e:
raise exceptions.LLMError(
speaker_peer_card=speaker_peer_card,
working_representation=formatted_working_representation,
history=history,
new_turn=formatted_new_turn,
) from e
# Handle different response types
# If response is a string, try to parse as JSON
if isinstance(response_obj, str):
# If response is a string, try to parse as JSON
import json
try:
response_data = json.loads(response_obj)
new_insights = ReasoningResponse(
@ -419,7 +353,9 @@ class CertaintyReasoner:
],
)
except (json.JSONDecodeError, KeyError, TypeError) as e:
logger.warning(f"Failed to parse string response as JSON: {e}")
if settings.SENTRY.ENABLED:
sentry_sdk.capture_exception(e)
logger.warning("Failed to parse string response as JSON: %s", e)
new_insights = ReasoningResponse(explicit=[], deductive=[])
else:
# If response is already a ReasoningResponse object
@ -438,7 +374,7 @@ class CertaintyReasoner:
if thinking is None:
logger.debug("No thinking content found in response")
except (AttributeError, TypeError) as e:
logger.warning(f"Error accessing thinking content: {e}, setting to None")
logger.warning("Error accessing thinking content: %s, setting to None", e)
thinking = None
response = ReasoningResponseWithThinking(
@ -464,26 +400,22 @@ class CertaintyReasoner:
@sentry_sdk.trace
async def reason(
self,
context: ReasoningResponseWithThinking,
db: AsyncSession,
working_representation: ReasoningResponseWithThinking,
history: str,
new_turn: str,
message_id: str,
session_name: str | None = None,
message_created_at: datetime.datetime | None = None,
speaker: str = "user",
speaker_peer_card: list[str] | None,
) -> ReasoningResponseWithThinking:
"""
Single-pass reasoning function that critically analyzes and derives insights.
Performs one analysis pass and returns the final observations.
"""
if message_created_at is None:
message_created_at = datetime.datetime.now(datetime.timezone.utc)
analysis_start = time.perf_counter()
# Perform critical analysis to get observation lists
reasoning_response = await self.derive_new_insights(
context, history, new_turn, message_created_at, speaker
working_representation,
history,
speaker_peer_card,
)
# Output the thinking content for this analysis
@ -491,7 +423,7 @@ class CertaintyReasoner:
analysis_duration_ms = (time.perf_counter() - analysis_start) * 1000
accumulate_metric(
f"deriver_message_{message_id}",
f"deriver_representation_{self.ctx.message_id}_{self.ctx.target_name}",
"critical_analysis_duration",
analysis_duration_ms,
"ms",
@ -499,39 +431,55 @@ class CertaintyReasoner:
save_observations_start = time.perf_counter()
# Save only the NEW observations that weren't in the original context
await self._save_new_observations(
context,
reasoning_response,
message_id,
session_name,
message_created_at,
new_observations_by_level: dict[
str, list[str]
] = await self._save_new_observations(
working_representation, reasoning_response
)
save_observations_duration = (
time.perf_counter() - save_observations_start
) * 1000
accumulate_metric(
f"deriver_message_{message_id}",
f"deriver_representation_{self.ctx.message_id}_{self.ctx.target_name}",
"save_new_observations",
save_observations_duration,
"ms",
)
# Only update peer card if we are in Honcho-level representation derivation.
if self.ctx.sender_name == self.ctx.target_name:
update_peer_card_start = time.perf_counter()
# flatten new observations by level into a list
new_observations = [
observation
for level in new_observations_by_level.values()
for observation in level
]
if new_observations:
await self._update_peer_card(db, speaker_peer_card, new_observations)
update_peer_card_duration = (
time.perf_counter() - update_peer_card_start
) * 1000
accumulate_metric(
f"deriver_representation_{self.ctx.message_id}_{self.ctx.target_name}",
"update_peer_card",
update_peer_card_duration,
"ms",
)
return reasoning_response
@conditional_observe
@sentry_sdk.trace
async def _save_new_observations(
self,
original_context: ReasoningResponse,
original_working_representation: ReasoningResponse,
revised_observations: ReasoningResponse,
message_id: str,
session_name: str | None = None,
message_created_at: datetime.datetime | None = None,
) -> None:
) -> dict[str, list[str]]:
"""Save only the observations that are new compared to the original context."""
# Use the utility function to find new observations
new_observations_by_level = find_new_observations(
original_context, revised_observations
new_observations_by_level: dict[str, list[str]] = find_new_observations(
original_working_representation, revised_observations
)
all_unified_observations: list[UnifiedObservation] = []
@ -561,7 +509,7 @@ class CertaintyReasoner:
len(observation.premises),
)
elif isinstance(observation, str):
else:
# String observations (explicit) have no premises
unified_obs = UnifiedObservation.from_string(
observation, level=level
@ -569,51 +517,82 @@ class CertaintyReasoner:
all_unified_observations.append(unified_obs)
logger.debug("Added %s observation: %s...", level, observation[:50])
else:
# Handle unexpected types
content = extract_observation_content(observation)
unified_obs = UnifiedObservation.from_string(content, level=level)
all_unified_observations.append(unified_obs)
logger.warning(
f"Added unexpected observation type: {type(observation)} as {level}"
)
total_observations_count += 1
if not all_unified_observations:
if all_unified_observations:
await self.embedding_store.save_unified_observations(
all_unified_observations,
self.ctx.message_id,
self.ctx.session_name,
self.ctx.created_at,
)
else:
logger.debug("No new observations to save")
return
await self.embedding_store.save_unified_observations(
all_unified_observations,
message_id=message_id,
session_name=session_name,
message_created_at=message_created_at,
return new_observations_by_level
@conditional_observe
@sentry_sdk.trace
async def _update_peer_card(
self,
db: AsyncSession,
old_peer_card: list[str] | None,
new_observations: list[str],
) -> None:
"""
Update the peer card by calling LLM with the old peer card and new observations.
The new peer card is returned by the LLM and saved to peer internal metadata.
"""
try:
response = await peer_card_call(old_peer_card, new_observations)
new_peer_card = response.card
if new_peer_card is None:
logger.info("No changes to peer card")
return
logger.info("New peer card: %s", new_peer_card)
await crud.set_peer_card(
db, self.ctx.workspace_name, self.ctx.sender_name, new_peer_card
)
except Exception as e:
if settings.SENTRY.ENABLED:
sentry_sdk.capture_exception(e)
logger.error("Error updating peer card! Skipping... %s", e)
def observation_context_to_reasoning_response(
context: ObservationContext,
) -> ReasoningResponseWithThinking:
"""Convert ObservationContext to ReasoningResponse for compatibility."""
thinking = context.thinking
# Convert explicit observations to new structure
explicit: list[str] = []
for obs in context.explicit:
explicit.append(obs.content)
# Convert deductive observations
deductive: list[DeductiveObservation] = []
for obs in context.deductive:
deductive_obs = DeductiveObservation(
conclusion=obs.content,
premises=obs.metadata.premises if obs.metadata else [],
)
deductive.append(deductive_obs)
return ReasoningResponseWithThinking(
thinking=thinking,
explicit=explicit,
deductive=deductive,
)
@sentry_sdk.trace
async def save_working_representation_to_peer(
db: AsyncSession,
workspace_name: str,
observer_name: str, # renamed from peer_name for clarity
observed_name: str, # new parameter
session_name: str | None,
payload: RepresentationPayload,
final_observations: ReasoningResponseWithThinking,
message_id: int,
) -> None:
"""Save working representation to peer internal_metadata for dialectic access."""
from sqlalchemy import update
from src import models
# Determine metadata key based on observer/observed relationship
if observer_name == observed_name:
metadata_key = "global_representation"
else:
metadata_key = crud.construct_collection_name(
observer=observer_name, observed=observed_name
)
# Convert ReasoningResponse to serializable dict
final_obs_dict = {
@ -630,57 +609,15 @@ async def save_working_representation_to_peer(
working_rep_data = {
"final_observations": final_obs_dict,
"message_id": message_id,
"message_id": payload.message_id,
"created_at": utc_now_iso(),
}
# if session_name is supplied, save working representation to session peer
if session_name:
stmt = (
update(models.SessionPeer)
.where(
models.SessionPeer.workspace_name == workspace_name,
models.SessionPeer.session_name == session_name,
models.SessionPeer.peer_name == observer_name,
)
.values(
internal_metadata=models.SessionPeer.internal_metadata.op("||")(
{metadata_key: working_rep_data}
)
)
)
await db.execute(stmt)
await db.commit()
logger.info(
f"Saved working representation to session peer {session_name} - {observer_name} with key {metadata_key}"
)
else:
# For peer-level messages (session_name=None), only save global representations
if observer_name == observed_name:
stmt = (
update(models.Peer)
.where(
models.Peer.workspace_name == workspace_name,
models.Peer.name == observer_name,
)
.values(
internal_metadata=models.Peer.internal_metadata.op("||")(
{metadata_key: working_rep_data}
)
)
)
await db.execute(stmt)
await db.commit()
logger.debug(
"Saved working representation to peer %s with key %s",
observer_name,
metadata_key,
)
else:
logger.debug(
"Skipping peer-level local representation save: observer=%s, observed=%s",
observer_name,
observed_name,
)
await crud.set_working_representation(
db,
working_rep_data,
payload.workspace_name,
payload.target_name,
payload.sender_name,
payload.session_name,
)

106
src/deriver/prompts.py
View File

@ -13,9 +13,9 @@ from mirascope import prompt_template
@prompt_template()
def critical_analysis_prompt(
peer_name: str,
peer_card: list[str] | None,
message_created_at: datetime.datetime,
context: str,
working_representation: str | None,
history: str,
new_turn: str,
) -> str:
@ -23,18 +23,41 @@ def critical_analysis_prompt(
Generate the critical analysis prompt for the deriver.
Args:
peer_name: The name of the user being analyzed
message_created_at: Timestamp of the message being analyzed
context: Current user understanding context
history: Recent conversation history
new_turn: New conversation turn to analyze
peer_card (list[str] | None): The bio card of the user being analyzed.
message_created_at (datetime.datetime): Timestamp of the message.
working_representation (str | None): Current user understanding context.
history (str): Recent conversation history.
new_turn (str): New conversation turn to analyze.
Returns:
Formatted prompt string for critical analysis
"""
# Format the peer card as a string with newlines
peer_card_section = (
f"""
The user's known biographical information:
<peer_card>
{chr(10).join(peer_card)}
</peer_card>
"""
if peer_card is not None
else ""
)
working_representation_section = (
f"""
The current user understanding:
<current_context>
{working_representation}
</current_context>
"""
if working_representation is not None
else ""
)
return c(
f"""
You are an agent who critically analyzes user messages through rigorous logical reasoning to produce only conclusions about the user that are CERTAIN. The user's name is **{peer_name}**.
You are an agent who critically analyzes user messages through rigorous logical reasoning to produce only conclusions about the user that are CERTAIN.
IMPORTANT NAMING RULES
• When you write a conclusion about the current user, always start the sentence with the user's name (e.g. "Anthony is 25 years old").
@ -59,10 +82,9 @@ Here are strict definitions for the reasoning modes you are to employ:
- Current date and time (which is: {message_created_at})
- Timestamps for user messages, and previous premises and conclusions
Here's the current user understanding
<current_context>
{context}
</current_context>
{peer_card_section}
{working_representation_section}
Recent conversation history for context:
<history>
@ -75,3 +97,63 @@ New conversation turn to analyze:
</new_turn>
"""
)
@prompt_template()
def peer_card_prompt(
old_peer_card: list[str] | None,
new_observations: list[str],
) -> str:
"""
Generate the peer card prompt for the deriver.
Currently optimized for GPT-5 mini/nano.
"""
old_peer_card_section = (
f"""
Current user biographical card:
{chr(10).join(old_peer_card)}
"""
if old_peer_card is not None
else """
User does not have a card. Create one with any key observations.
"""
)
return c(
f"""
You are an agent that creates a concise "biographical card" based on new observations for a user. A biographical card summarizes essential information like name, nicknames, location, age, occupation, interests/hobbies, and likes/dislikes.
The goal is to capture only the most important observations about the user. Value permanent properties over transient ones, and value concision over detail, preferring to omit details that are not essential to the user's identity. The card should give a broad overview of who the user is while not including details that are unlikely to be relevant in most settings.
For example, "User is from Chicago" is worth inclusion. "User has an Instagram account" is not.
"User is a software engineer" is worth inclusion. "User wrote Python today" is not.
Never infer or generalize traits from one-off behaviors. Never manipulate the text of an observation to make an action or behavior into a "permanent" trait.
When a new observation contradicts an existing one, update it, favoring new information.
Example 1:
{{
"card": [
"Name: Bob",
"Age: 24",
"Location: New York"
]
}}
Example 2:
{{
"card": [
"Name: Alice",
"Occupation: Artist",
"Interests: Painting, biking, cooking"
]
}}
{old_peer_card_section}
New observations:
{chr(10).join(new_observations)}
If there's no new key info, you should return an empty card. **NEVER** include notes or temporary information in the card itself, instead use the notes field.
"""
)

27
src/deriver/queue_manager.py
View File

@ -13,6 +13,7 @@ from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import func
from src import exceptions
from src.config import settings
from src.models import QueueItem
@ -235,7 +236,6 @@ class QueueManager:
message = await self.get_next_message(db, work_unit_key)
if not message:
logger.debug(f"No more messages for work unit {work_unit_key}")
break
message_count += 1
@ -247,6 +247,13 @@ class QueueManager:
logger.debug(
f"Successfully processed queue item for task type {message.task_type} with id {message.id}"
)
except exceptions.LLMError as e:
logger.error(
f"LLM returned bad JSON for message {message}, re-queueing",
)
if settings.SENTRY.ENABLED:
sentry_sdk.capture_exception(e)
continue
except Exception as e:
logger.error(
f"Error processing queue item for task type {message.task_type} with id {message.id}: {str(e)}",
@ -254,16 +261,9 @@ class QueueManager:
)
if settings.SENTRY.ENABLED:
sentry_sdk.capture_exception(e)
finally:
# Prevent malformed messages from stalling queue indefinitely
message.processed = True
await db.commit()
if self.shutdown_event.is_set():
logger.debug(
f"Shutdown requested, stopping processing for work unit {work_unit_key}"
)
break
# Prevent malformed messages from stalling queue indefinitely
message.processed = True
# Update last_updated timestamp to show this work unit is still being processed
await db.execute(
@ -273,6 +273,13 @@ class QueueManager:
)
await db.commit()
if self.shutdown_event.is_set():
logger.debug(
"Shutdown requested, stopping processing for work unit %s",
work_unit_key,
)
break
logger.debug(
f"Completed processing work unit {work_unit_key}, processed {message_count} messages"
)

5
src/deriver/queue_payload.py
View File

@ -42,11 +42,6 @@ class WebhookPayload(BasePayload):
data: dict[str, Any]
# Union type for all possible queue payloads
DeriverQueuePayload = RepresentationPayload | SummaryPayload
QueuePayload = DeriverQueuePayload | WebhookPayload
def create_webhook_payload(
workspace_name: str,
event_type: str,

30
src/dialectic/chat.py
View File

@ -17,6 +17,7 @@ from mirascope.llm import Stream
from src import crud
from src.config import settings
from src.crud.representation import GLOBAL_REPRESENTATION_COLLECTION_NAME
from src.dependencies import tracked_db
from src.utils import summarizer
from src.utils.clients import honcho_llm_call
@ -53,7 +54,9 @@ async def dialectic_call(
recent_conversation_history: str | None,
additional_context: str | None,
peer_name: str,
peer_card: list[str] | None,
target_name: str | None = None,
target_peer_card: list[str] | None = None,
):
"""
Make a direct call to the dialectic model for context synthesis.
@ -74,7 +77,9 @@ async def dialectic_call(
recent_conversation_history,
additional_context,
peer_name,
peer_card,
target_name,
target_peer_card,
)
# Pretty print the prompt content
@ -109,7 +114,9 @@ async def dialectic_stream(
recent_conversation_history: str | None,
additional_context: str | None,
peer_name: str,
peer_card: list[str] | None,
target_name: str | None = None,
target_peer_card: list[str] | None = None,
):
"""
Make a streaming call to the dialectic model for context synthesis.
@ -130,7 +137,9 @@ async def dialectic_stream(
recent_conversation_history,
additional_context,
peer_name,
peer_card,
target_name,
target_peer_card,
)
# Pretty print the prompt content
@ -235,7 +244,7 @@ async def chat(
embedding_store = EmbeddingStore(
workspace_name=workspace_name,
peer_name=target_name if target_name else peer_name,
collection_name="global_representation"
collection_name=GLOBAL_REPRESENTATION_COLLECTION_NAME
if not target_name
else crud.construct_collection_name(observer=peer_name, observed=target_name),
)
@ -285,7 +294,20 @@ async def chat(
"tokens",
)
# 4. Dialectic call --------------------------------------------------------
# 4. Peer card(s) ----------------------------------------------------------
async with tracked_db("chat.get_peer_card") as db:
peer_card = await crud.get_peer_card(db, workspace_name, peer_name)
if target_name:
target_peer_card = await crud.get_peer_card(db, workspace_name, target_name)
else:
target_peer_card = None
if target_peer_card:
logger.info("Retrieved peer cards:\n%s\n%s", peer_card, target_peer_card)
else:
logger.info("Retrieved peer card:\n%s", peer_card)
# 5. Dialectic call --------------------------------------------------------
dialectic_call_start_time = asyncio.get_event_loop().time()
if stream:
return await dialectic_stream(
@ -294,7 +316,9 @@ async def chat(
recent_conversation_history,
additional_context,
peer_name,
peer_card,
target_name,
target_peer_card,
)
response = await dialectic_call(
@ -303,7 +327,9 @@ async def chat(
recent_conversation_history,
additional_context,
peer_name,
peer_card,
target_name,
target_peer_card,
)
dialectic_call_duration = (
asyncio.get_event_loop().time() - dialectic_call_start_time

32
src/dialectic/prompts.py
View File

@ -10,7 +10,9 @@ def dialectic_prompt(
recent_conversation_history: str | None,
additional_context: str | None,
peer_name: str,
peer_card: list[str] | None,
target_name: str | None = None,
target_peer_card: list[str] | None = None,
) -> str:
"""
Generate the main dialectic prompt for context synthesis.
@ -24,14 +26,30 @@ def dialectic_prompt(
Returns:
Formatted prompt string for the dialectic model
"""
query_target = (
f"user {peer_name}'s understanding of {target_name}"
if target_name
else f"user {peer_name}"
)
if target_name:
query_target = f"""The query is about user {peer_name}'s understanding of {target_name}.
The user's known biographical information:
{chr(10).join(peer_card) if peer_card else "(none)"}
The target's known biographical information:
{chr(10).join(target_peer_card) if target_peer_card else "(none)"}
If the user's name or nickname is known, exclusively refer to them by that name.
If the target's name or nickname is known, exclusively refer to them by that name.
"""
else:
query_target = f"""The query is about user {peer_name}.
The user's known biographical information:
{chr(10).join(peer_card) if peer_card else "(none)"}
If the user's name or nickname is known, exclusively refer to them by that name.
"""
return c(
f"""
The query is about {query_target}.
You are a context synthesis agent that operates as a natural language API for AI applications. Your role is to analyze application queries about users and synthesize relevant conclusions into coherent, actionable insights that directly address what the application needs to know.
## INPUT STRUCTURE
@ -56,6 +74,8 @@ Provide a natural language response that:
4. Maintains appropriate confidence levels based on conclusion types
5. Flags any limitations or gaps in available information
{query_target}
<recent_conversation_history>
{recent_conversation_history}
</recent_conversation_history>

38
src/exceptions.py
View File

@ -2,7 +2,8 @@
Custom exceptions for the Honcho application.
"""
from typing import final
import json
from typing import Any, final
from src.config import settings
@ -106,3 +107,38 @@ class FileTooLargeError(HonchoException):
class FileProcessingError(HonchoException):
status_code = 500
detail = "File processing error"
class LLMError(Exception):
"""Exception raised when an LLM call fails.
Accepts arbitrary positional and keyword inputs, normalizes them into a
JSON-serializable object, and uses the resulting JSON string as the
exception message. The normalized object is available via ``to_dict()``
and the ``data`` attribute.
Positional and keyword inputs are represented as a JSON object. If a
single positional argument is a mapping and there are no keyword
arguments, that mapping is used as the root object; otherwise the shape is
``{"args": [...], "kwargs": {...}}``. Values that are not natively
serializable are converted using ``repr``.
"""
data: dict[str, Any]
def __init__(self, *args: Any, **kwargs: Any) -> None:
normalized = {"args": list(args), "kwargs": kwargs}
message = json.dumps(
normalized, default=self._json_fallback, ensure_ascii=False
)
self.data = normalized
super().__init__(message)
@staticmethod
def _json_fallback(value: Any) -> str:
"""Fallback serializer that returns ``repr(value)`` for unsupported types."""
return repr(value)
def to_dict(self) -> dict[str, Any]:
"""Return the normalized JSON object for programmatic access."""
return self.data

3
src/models.py
View File

@ -377,6 +377,9 @@ class QueueItem(Base):
payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
processed: Mapped[bool] = mapped_column(Boolean, default=False)
def __repr__(self) -> str:
return f"QueueItem(id={self.id}, session_id={self.session_id}, work_unit_key={self.work_unit_key}, task_type={self.task_type}, payload={self.payload}, processed={self.processed})"
@final
class ActiveQueueSession(Base):

44
src/utils/clients.py
View File

@ -9,6 +9,17 @@ from typing import (
runtime_checkable,
)
# --- OpenAI compatibility shim must run BEFORE importing mirascope ---
# Some versions of the OpenAI SDK do not expose ChatCompletionMessageToolCall
# at openai.types.chat, but some integrations import it from there at runtime.
# We defensively define it if missing to avoid import-time failures.
#
# We can get rid of this by getting rid of mirascope...
from openai.types import chat as _openai_chat_types # type: ignore
if not hasattr(_openai_chat_types, "ChatCompletionMessageToolCall"):
_openai_chat_types.ChatCompletionMessageToolCall = object # pyright: ignore
from anthropic import AsyncAnthropic
from google import genai
from groq import AsyncGroq
@ -113,6 +124,8 @@ def honcho_llm_call(
response_model: type[BaseModel] | None = None,
json_mode: bool = False,
max_tokens: int | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
@ -131,6 +144,8 @@ def honcho_llm_call(
response_model: type[T],
json_mode: bool = False,
max_tokens: int | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
@ -149,6 +164,8 @@ def honcho_llm_call(
response_model: None = None,
json_mode: bool = False,
max_tokens: int | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
@ -168,6 +185,8 @@ def honcho_llm_call(
response_model: None = None,
json_mode: bool = False,
max_tokens: int | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
@ -187,6 +206,8 @@ def honcho_llm_call(
response_model: type[BaseModel] | None = None,
json_mode: bool = False,
max_tokens: int | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
@ -202,6 +223,8 @@ def honcho_llm_call(
response_model: type[BaseModel] | None = None,
json_mode: bool = False,
max_tokens: int | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
@ -227,6 +250,8 @@ def honcho_llm_call(
response_model: Optional Pydantic model for structured responses
json_mode: Whether to enable JSON mode (for providers that support it)
max_tokens: Maximum tokens for the response
reasoning_effort: Optional reasoning effort hint passed to OpenAI GPT-5 models only
verbosity: Optional verbosity hint passed to OpenAI GPT-5 models only
thinking_budget_tokens: Budget for thinking tokens (Anthropic only)
enable_retry: Whether to enable retry logic (default: True)
retry_attempts: Number of retry attempts (default: 3)
@ -298,16 +323,27 @@ def honcho_llm_call(
}
if max_tokens:
call_params["max_tokens"] = max_tokens
elif resolved_provider == "openai" and model and "gpt-5" in model:
call_params["max_completion_tokens"] = max_tokens
if reasoning_effort is not None:
call_params["reasoning_effort"] = reasoning_effort
if verbosity is not None:
call_params["verbosity"] = verbosity
else:
# Other providers just use max_tokens
if max_tokens:
call_params["max_tokens"] = max_tokens
# Merge with any extra call params
# Remove return_call_response from extra_call_params --
# that one is just for our type system.
# Merge with any user-supplied provider call params
# Accept an explicit "extra_call_params" dict kwarg and merge its contents
# Do NOT forward the key itself into provider params.
# Also drop any wrapper-only flags.
user_extra_call_params: dict[str, Any] | None = extra_call_params.pop(
"extra_call_params", None
)
extra_call_params.pop("return_call_response", None)
call_params.update(extra_call_params)
if isinstance(user_extra_call_params, dict):
call_params.update(user_extra_call_params)
# Build kwargs for llm.call
llm_kwargs: dict[str, Any] = {}

23
src/utils/embedding_store.py
View File

@ -39,11 +39,11 @@ class EmbeddingStore:
async def save_unified_observations(
self,
observations: list[UnifiedObservation],
message_id: int,
session_name: str,
message_created_at: datetime.datetime,
fallback_level: str = "explicit",
similarity_threshold: float = 0.85,
message_id: str | None = None,
level: str | None = None,
session_name: str | None = None,
message_created_at: datetime.datetime | None = None,
) -> None:
"""Save UnifiedObservation objects to the collection.
@ -53,11 +53,11 @@ class EmbeddingStore:
Args:
observations: List of UnifiedObservation objects or strings
similarity_threshold: Threshold for considering observations similar
message_id: Message ID to link with observations
level: Reasoning level for the observations
session_name: Session name to link with existing summary context
message_created_at: Timestamp when the message was created
fallback_level: Reasoning level for the observations if not provided
similarity_threshold: Threshold for considering observations similar
"""
async with tracked_db("ed_embedding_store.save_unified_observations") as db:
# Extract conclusions for deduplication and embedding
@ -104,11 +104,8 @@ class EmbeddingStore:
# Batch create document objects
document_objects: list[models.Document] = []
for obs, embedding in zip(unique_observations, embeddings, strict=True):
# Use the observation's own level, fall back to parameter level,
# or infer from premises
obs_level = obs.level or level
if obs_level is None:
obs_level = "deductive" if obs.has_premises else "explicit"
# Use the observation's own level or fall back to parameter level
obs_level = obs.level or fallback_level
# Build metadata including premises
metadata: dict[str, Any] = {
@ -116,9 +113,7 @@ class EmbeddingStore:
"message_id": message_id,
"session_name": session_name,
"premises": obs.premises, # Store premises in metadata
"created_at": format_datetime_utc(message_created_at)
if message_created_at
else None,
"created_at": format_datetime_utc(message_created_at),
}
doc = models.Document(

19
src/utils/formatting.py
View File

@ -162,24 +162,21 @@ def extract_observation_content(observation: str | dict[str, Any] | Any) -> str:
def format_new_turn_with_timestamp(
new_turn: str, current_time: str | datetime, speaker: str
new_turn: str, current_time: datetime, speaker: str
) -> str:
"""
Format new turn message with optional timestamp.
Args:
new_turn: The message content
current_time: Timestamp string or "unknown"
current_time: Message timestamp
speaker: The speaker's name
Returns:
Formatted string like "2023-05-08 13:56:00 speaker: hello" or "speaker: hello"
Formatted string like "2023-05-08 13:56:00 speaker: hello"
"""
if isinstance(current_time, datetime):
current_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
if current_time and current_time != "unknown":
return f"{current_time} {speaker}: {new_turn}"
return f"{speaker}: {new_turn}"
current_time_str = current_time.strftime("%Y-%m-%d %H:%M:%S")
return f"{current_time_str} {speaker}: {new_turn}"
def format_context_for_prompt(
@ -249,7 +246,7 @@ def normalize_observations_for_comparison(observations: list[Any]) -> set[str]:
@conditional_observe
def find_new_observations(
original_context: ReasoningResponse, revised_observations: ReasoningResponse
) -> dict[str, Any]:
) -> dict[str, list[str]]:
"""
Find observations that are new in revised_observations compared to original_context.
@ -260,7 +257,7 @@ def find_new_observations(
Returns:
Dictionary with new observations by level
"""
new_observations_by_level: dict[str, Any] = {}
new_observations_by_level: dict[str, list[str]] = {}
for level in REASONING_LEVELS:
original_observations = normalize_observations_for_comparison(
@ -269,7 +266,7 @@ def find_new_observations(
revised_list = getattr(revised_observations, level, [])
# Find genuinely new observations
new_observations: list[Any] = []
new_observations: list[str] = []
for observation in revised_list:
normalized_observation = (
extract_observation_content(observation).strip().lower()

2
src/utils/logging.py
View File

@ -265,7 +265,7 @@ def log_performance_metrics(
for metric, value, unit in metrics:
if unit == "ms":
formatted_value = f"{value:.1f}"
formatted_value = f"{value:.0f}"
elif unit == "s":
formatted_value = f"{value:.3f}"
else:

12
src/utils/shared_models.py
View File

@ -191,3 +191,15 @@ class ObservationDict(TypedDict, total=False):
content: str
premises: list[str]
created_at: str
class PeerCardQuery(BaseModel):
"""
Model for peer card query generation responses.
Contains the new peer card, or None if there are no new key observations.
The notes field is just a place for stupid models to dump useless info.
"""
card: list[str] | None
notes: str | None

6
src/utils/summarizer.py
View File

@ -321,7 +321,7 @@ async def _create_and_save_summary(
summary_duration = (time.perf_counter() - summary_start) * 1000
accumulate_metric(
f"deriver_message_{message_id}",
f"summary_{workspace_name}_{message_id}",
f"{summary_type.name}_summary_creation",
summary_duration,
"ms",
@ -381,14 +381,14 @@ async def _create_summary(
summary_tokens = 50
accumulate_metric(
f"deriver_message_{messages[-1].id}",
f"summary_{messages[-1].workspace_name}_{messages[-1].id}",
f"{summary_type.name}_summary_input",
response.usage.input_tokens if response and response.usage else "unknown",
"tokens",
)
accumulate_metric(
f"deriver_message_{messages[-1].id}",
f"summary_{messages[-1].workspace_name}_{messages[-1].id}",
f"{summary_type.name}_summary_size",
response.usage.output_tokens
if response and response.usage

2
tests/bench/README.md
View File

@ -16,7 +16,7 @@ The `harness.py` script orchestrates the complete Honcho development environment
## Prerequisites
- Python 3.8+
- Python 3.11+
- Docker and Docker Compose
- Honcho project dependencies installed (`uv sync`)

435
tests/bench/peer_card_bench.py Normal file
View File

@ -0,0 +1,435 @@
"""
Peer Card Benchmark
This benchmark exercises the peer card LLM call with varied inputs and compares
outputs across multiple provider/model candidates. Results are graded by an LLM
judge.
Usage example:
python -m tests.bench.peer_card_bench --candidates anthropic:claude-3-7-sonnet-20250219 --candidates openai:gpt-4o-mini-2024-07-18
Environment variables for providers:
- Anthropic: LLM_ANTHROPIC_API_KEY
- OpenAI: LLM_OPENAI_API_KEY or OPENAI_API_KEY
- Google (Gemini): LLM_GEMINI_API_KEY or GEMINI_API_KEY
- Groq: LLM_GROQ_API_KEY or GROQ_API_KEY
"""
import argparse
import asyncio
import json
import os
import time
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from pathlib import Path
from typing import Any, cast
from anthropic import AsyncAnthropic
from src.config import settings
from src.deriver.prompts import peer_card_prompt
from src.utils.clients import honcho_llm_call
from src.utils.shared_models import PeerCardQuery
COLOR_GREEN = "\033[32m"
COLOR_RED = "\033[31m"
COLOR_RESET = "\033[0m"
@dataclass(frozen=True)
class Candidate:
"""Represents a provider/model pair to benchmark."""
provider: str
model: str
@dataclass
class Case:
"""Represents a single benchmark case with expectations for grading.
Attributes:
name: Human-friendly identifier for the case.
old_peer_card: Existing card text to update, or None to create fresh.
new_observations: New input observations that may change the card.
expected_facts: Facts that must be semantically present in the result.
forbidden_facts: Facts that must NOT be present in the result.
"""
name: str
old_peer_card: list[str] | None
new_observations: list[str]
expected_facts: list[str]
forbidden_facts: list[str]
def load_case_file(path: Path) -> Case:
"""Load a single peer-card test case from a JSON file.
The JSON schema must include: name, old_peer_card (nullable), new_observations (list[str]), expected_facts (list[str]).
"""
with path.open() as f:
data = json.load(f)
return Case(
name=str(data["name"]),
old_peer_card=data.get("old_peer_card"),
new_observations=list(data.get("new_observations", [])),
expected_facts=list(data.get("expected_facts", [])),
forbidden_facts=list(data.get("forbidden_facts", [])),
)
def load_cases(tests_dir: Path, test_name: str | None) -> list[Case]:
"""Load all cases from a directory, or a specific case by filename.
Args:
tests_dir: Directory containing JSON case files.
test_name: Optional filename to load a single case (e.g., "create_basic_card.json").
Returns:
List of loaded Case objects.
"""
if test_name:
file_path = tests_dir / test_name
if not file_path.exists():
raise FileNotFoundError(f"Test file {file_path} does not exist")
return [load_case_file(file_path)]
files = sorted(p for p in tests_dir.glob("*.json") if p.is_file())
return [load_case_file(p) for p in files]
def parse_candidates(values: list[str]) -> list[Candidate]:
"""Parse provider:model strings into Candidate objects."""
result: list[Candidate] = []
for v in values:
v = v.strip()
if not v:
continue
if ":" not in v:
raise ValueError(f"Invalid candidate format: {v} (expected provider:model)")
provider, model = v.split(":", 1)
result.append(Candidate(provider=provider.strip(), model=model.strip()))
return result
def deduplicate_preserve_order(items: list[Candidate]) -> list[Candidate]:
"""Return a new list with duplicate provider:model pairs removed, preserving order."""
seen: set[tuple[str, str]] = set()
unique: list[Candidate] = []
for item in items:
key = (item.provider, item.model)
if key in seen:
continue
seen.add(key)
unique.append(item)
return unique
def build_peer_card_caller(
candidate: Candidate,
) -> Callable[[list[str] | None, list[str]], Coroutine[Any, Any, PeerCardQuery]]:
"""Create an async callable that invokes the peer card prompt with a specific provider/model."""
resolved_provider = (
"openai" if candidate.provider == "custom" else candidate.provider
)
@honcho_llm_call(
provider=cast(Any, resolved_provider),
model=candidate.model,
track_name="Peer Card Call",
response_model=PeerCardQuery,
json_mode=True,
max_tokens=settings.DERIVER.PEER_CARD_MAX_OUTPUT_TOKENS,
reasoning_effort="minimal",
enable_retry=True,
retry_attempts=1, # unstructured output means we shouldn't need to retry, 1 just in case
)
async def call(old_peer_card: list[str] | None, new_observations: list[str]) -> Any:
"""Return the prompt content for Mirascope to execute as a model call."""
return peer_card_prompt(
old_peer_card=old_peer_card, new_observations=new_observations
)
return call
def _extract_json_from_text(text: str) -> dict[str, Any]:
"""Extract a JSON object from potentially noisy LLM output.
Strategy in order:
- Try to parse the whole text as JSON
- Try to parse contents of any fenced code blocks (``` or ```json)
- Try the substring from the first '{' to the last '}'
- Scan for balanced-brace substrings and try them in order
Raises ValueError when no valid JSON object can be found.
"""
stripped: str = text.strip()
candidates: list[str] = []
# 1) Fenced code blocks
if "```" in stripped:
idx: int = 0
while True:
start = stripped.find("```", idx)
if start == -1:
break
lang_line_end = stripped.find("\n", start + 3)
if lang_line_end == -1:
break
end = stripped.find("```", lang_line_end + 1)
if end == -1:
break
block = stripped[lang_line_end + 1 : end].strip()
if block:
candidates.append(block)
idx = end + 3
# 2) From first '{' to last '}'
first_brace = stripped.find("{")
last_brace = stripped.rfind("}")
if first_brace != -1 and last_brace != -1 and last_brace > first_brace:
candidates.append(stripped[first_brace : last_brace + 1])
# 3) Balanced-brace scan
depth = 0
start_idx = -1
for i, ch in enumerate(stripped):
if ch == "{":
if depth == 0:
start_idx = i
depth += 1
elif ch == "}":
if depth > 0:
depth -= 1
if depth == 0 and start_idx != -1:
candidates.append(stripped[start_idx : i + 1])
# Try all candidates, prefer ones containing the expected keys
preferred_keys = {"passed", "reasoning"}
fallback_obj: dict[str, Any] | None = None
for cand in candidates:
try:
obj_candidate: object = json.loads(cand)
if isinstance(obj_candidate, dict):
casted_obj: dict[str, Any] = {
str(k): v # pyright: ignore
for k, v in obj_candidate.items() # pyright: ignore
}
if preferred_keys.issubset(set(casted_obj.keys())):
return casted_obj
if fallback_obj is None:
fallback_obj = casted_obj
except Exception:
continue
if fallback_obj is not None:
return fallback_obj
raise ValueError("Could not extract JSON from judge response")
async def judge_response(
anthropic: AsyncAnthropic,
case: Case,
actual_card: list[str],
) -> dict[str, Any]:
"""Use an LLM judge to evaluate whether the card contains the expected facts.
Returns a dict with keys: passed (bool) and reasoning (str).
"""
system_prompt = (
"You are an expert evaluator. Determine if a biographical card satisfies BOTH: "
"(1) it contains all expected facts (semantic match allowed) and "
"(2) it does NOT contain any forbidden facts (semantic match). "
"Allow flexible phrasing and synonyms for matching. A fact is present if its semantic content is clearly stated. "
"Fail if any expected fact is missing or any forbidden fact appears. Always return JSON: "
'{"passed": boolean, "reasoning": string}'
)
expected = "\n".join(f"- {f}" for f in case.expected_facts)
forbidden = "\n".join(f"- {f}" for f in case.forbidden_facts)
card_text = "\n".join(actual_card) if actual_card else "- (none)"
user_prompt = (
f"Case: {case.name}\n\n"
f"Expected facts (must appear, semantic):\n{expected or '- (none)'}\n\n"
f"Forbidden facts (must NOT appear, semantic):\n{forbidden or '- (none)'}\n\n"
f"Biographical card to evaluate:\n{card_text}\n\n"
f"Evaluation criteria: PASS only if all expected facts are present AND all forbidden facts are absent."
)
judgment_text: str | None = None
try:
response = await anthropic.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1000,
temperature=0.0,
system=system_prompt,
messages=[{"role": "user", "content": user_prompt}],
)
content_block = response.content[0]
judgment_text = getattr(content_block, "text", None)
if not judgment_text:
raise ValueError("Empty judge response")
return _extract_json_from_text(judgment_text)
except Exception as e:
print(judgment_text)
raise ValueError(f"!!!Error judging response for case {case.name}: {e}") from e
async def run_benchmark(candidates: list[Candidate], cases: list[Case]) -> int:
"""Execute cases against candidates and print a concise report.
Returns non-zero when any case fails for any candidate.
"""
anthropic_key = os.getenv("LLM_ANTHROPIC_API_KEY")
if not anthropic_key:
raise ValueError("LLM_ANTHROPIC_API_KEY is required for grading")
anthropic = AsyncAnthropic(api_key=anthropic_key)
any_fail = False
print(f"Running {len(cases)} cases across {len(candidates)} candidates\n")
for candidate in candidates:
print(f"=== Candidate: {candidate.provider}:{candidate.model} ===")
candidate_start_time = time.perf_counter()
try:
caller = build_peer_card_caller(candidate)
except Exception as e:
print(f" SKIP: cannot initialize provider/model ({e})")
any_fail = True
continue
async def run_case(
case: Case,
_caller: Callable[
[list[str] | None, list[str]], Coroutine[Any, Any, PeerCardQuery]
] = caller,
) -> tuple[Case, dict[str, Any]]:
card: PeerCardQuery = await _caller(
case.old_peer_card, case.new_observations
)
new_card = card.card
if new_card is None:
new_card = case.old_peer_card or []
judgment = await judge_response(anthropic, case, new_card)
return case, {"card": card, "judgment": judgment}
results = await asyncio.gather(
*(run_case(c) for c in cases), return_exceptions=True
)
passed_count: int = 0
for res in results:
if isinstance(res, BaseException):
print(f" ERROR running case: {res}")
any_fail = True
continue
case, payload = res
judgment = payload["judgment"]
passed = bool(judgment.get("passed"))
status_colored = (
f"{COLOR_GREEN}PASS{COLOR_RESET}"
if passed
else f"{COLOR_RED}FAIL{COLOR_RESET}"
)
if passed:
print(f" {case.name:24} {status_colored}")
passed_count += 1
else:
print(
f" {case.name:24} {status_colored} - {judgment.get('reasoning', '')}"
)
any_fail = True
print(" expected:")
for f in case.expected_facts:
print(f" - {f}")
if case.forbidden_facts:
print(" forbidden (must NOT appear):")
for f in case.forbidden_facts:
print(f" - {f}")
print(" got:")
[print(" " + line) for line in payload["card"].card]
print(" with 'notes' field:")
if payload["card"].notes:
print(" " + payload["card"].notes)
else:
print(" - (none)")
total_count: int = len(cases)
percentage: float = (passed_count / total_count * 100.0) if total_count else 0.0
print(f" Summary: {passed_count}/{total_count} passed ({percentage:.1f}%)")
elapsed = time.perf_counter() - candidate_start_time
print(f" Time: {elapsed:.2f}s\n")
print("Done.")
return 1 if any_fail else 0
def main() -> int:
"""CLI entry point for running the peer card benchmark."""
parser = argparse.ArgumentParser(
description="Benchmark peer card LLM behavior across models"
)
parser.add_argument(
"--candidates",
action="append",
default=None,
help=(
"Provider:model pairs. Repeat or comma-separate. "
"Default: anthropic:claude-3-7-sonnet-20250219"
),
)
parser.add_argument(
"--tests-dir",
type=Path,
default=Path("tests/bench/peer_card_tests"),
help=(
"Directory containing JSON peer-card cases "
"(default: tests/bench/peer_card_tests)"
),
)
parser.add_argument(
"--test",
type=str,
help="Run a specific test file by name (e.g., 'create_basic_card.json')",
)
args = parser.parse_args()
# Use the default candidate only when the flag is not provided at all
candidate_entries: list[str] = (
args.candidates
if args.candidates is not None
else ["anthropic:claude-3-7-sonnet-20250219"]
)
flat: list[str] = []
for entry in candidate_entries:
flat.extend([s.strip() for s in entry.split(",") if s.strip()])
candidates = parse_candidates(flat)
candidates = deduplicate_preserve_order(candidates)
# Load cases from JSON files
if not args.tests_dir.exists():
raise SystemExit(f"Error: Tests directory {args.tests_dir} does not exist")
cases = load_cases(args.tests_dir, args.test)
if not cases:
raise SystemExit(f"Error: No JSON test cases found in {args.tests_dir}")
return asyncio.run(run_benchmark(candidates, cases))
if __name__ == "__main__":
raise SystemExit(main())

11
tests/bench/peer_card_tests/add_nickname.json Normal file
View File

@ -0,0 +1,11 @@
{
"name": "add_nickname",
"old_peer_card": "Name: Daniel\nLocation: Austin\nOccupation: Graphic Designer",
"new_observations": [
"Daniel's nickname is Dan the Man."
],
"expected_facts": [
"Name: Daniel",
"Nickname: Dan the Man"
]
}

24
tests/bench/peer_card_tests/chitchat_only_no_change.json Normal file
View File

@ -0,0 +1,24 @@
{
"name": "chitchat_only_no_change",
"old_peer_card": "Name: Jordan\nAge: 33\nLocation: Denver\nOccupation: Product Manager\nInterests: skiing, trail running",
"new_observations": [
"The meeting could have been an email.",
"The user's day is hectic.",
"It is Monday."
],
"expected_facts": [
"Name: Jordan",
"Age: 33",
"Location: Denver",
"Occupation: Product Manager",
"Interests: skiing",
"Interests: trail running"
],
"forbidden_facts": [
"Interests: meetings",
"Occupation: Executive Assistant",
"Location: Boulder",
"Day: hectic",
"Today is Monday"
]
}

17
tests/bench/peer_card_tests/contradiction_update_location_amidst_noise.json Normal file
View File

@ -0,0 +1,17 @@
{
"name": "contradiction_update_location_amidst_noise",
"old_peer_card": "Name: Mei\nAge: 32\nLocation: Vancouver\nOccupation: Nurse\nInterests: hiking, painting\nLanguages: English, Mandarin",
"new_observations": [
"The Canucks lost.",
"Mei moved to Calgary for work.",
"A friend started a bakery.",
"It is snowing again."
],
"expected_facts": [
"Location: Calgary"
],
"forbidden_facts": [
"Location: Vancouver",
"Occupation: Baker"
]
}

18
tests/bench/peer_card_tests/create_basic_card.json Normal file
View File

@ -0,0 +1,18 @@
{
"name": "create_basic_card",
"old_peer_card": null,
"new_observations": [
"The user's name is Alice.",
"Alice is 29 years old.",
"Alice lives in San Francisco.",
"Alice works as a Product Manager.",
"Alice is interested in climbing and cooking."
],
"expected_facts": [
"Name: Alice",
"Age: 29",
"Location: San Francisco",
"Occupation: Product Manager",
"Interests: climbing, cooking"
]
}

26
tests/bench/peer_card_tests/large_card_no_change.json Normal file
View File

@ -0,0 +1,26 @@
{
"name": "large_card_no_change",
"old_peer_card": "Name: Isabella\nAge: 28\nLocation: San Francisco\nOccupation: Software Engineer\nCompany: TechNova\nEducation: B.S. Computer Science, Stanford\nInterests: hiking, photography, baking, climbing, yoga\nLanguages: English, Spanish\nDislikes: cilantro\nFavorite Foods: sushi, ramen\nFavorite Music: jazz\nPets: Dog named Luna\nNickname: Izzy",
"new_observations": [
"It is a foggy morning in the city.",
"Traffic is heavy.",
"The Warriors game last night was intense.",
"The user feels casual."
],
"expected_facts": [
"Name: Isabella",
"Age: 28",
"Location: San Francisco",
"Occupation: Software Engineer",
"Interests: hiking",
"Interests: photography",
"Languages: Spanish",
"Nickname: Izzy",
"Pets: Dog named Luna"
],
"forbidden_facts": [
"Interests: basketball",
"Location: Oakland",
"Age: 29"
]
}

18
tests/bench/peer_card_tests/large_card_with_one_update_amidst_noise.json Normal file
View File

@ -0,0 +1,18 @@
{
"name": "large_card_with_one_update_amidst_noise",
"old_peer_card": "Name: Thomas\nAge: 40\nLocation: Portland\nOccupation: Architect\nCompany: UrbanForm\nEducation: M.Arch MIT\nInterests: cycling, woodworking, gardening, baking\nLanguages: English, French\nDislikes: loud bars\nFavorite Foods: pizza\nFavorite Music: classical\nPets: Cat named Maple\nNickname: Tom",
"new_observations": [
"Traffic is heavy.",
"Thomas is 41 years old.",
"The user listened to a podcast.",
"It has been rainy lately.",
"The user felt amused."
],
"expected_facts": [
"Age: 41"
],
"forbidden_facts": [
"Location: Seattle",
"Interests: podcasts"
]
}

19
tests/bench/peer_card_tests/limit_signal.json Normal file
View File

@ -0,0 +1,19 @@
{
"name": "limit_signal",
"old_peer_card": null,
"new_observations": [
"The user's name is Henry.",
"The user is interested in chess.",
"The user is interested in chess a lot.",
"Henry's nickname is Hank."
],
"expected_facts": [
"Name: Henry",
"Nickname: Hank",
"Interests: chess"
],
"forbidden_facts": [
"Interests: chess a lot",
"Interests: checkers"
]
}

14
tests/bench/peer_card_tests/multiple_changes.json Normal file
View File

@ -0,0 +1,14 @@
{
"name": "multiple_changes",
"old_peer_card": "Name: Grace\nAge: 30\nLocation: Paris\nOccupation: Artist\nInterests: Painting",
"new_observations": [
"Grace is 31 years old.",
"Grace lives in Berlin.",
"Grace is interested in biking."
],
"expected_facts": [
"Age: 31",
"Location: Berlin",
"Interests: biking"
]
}

13
tests/bench/peer_card_tests/no_bio_signal_empty_observations.json Normal file
View File

@ -0,0 +1,13 @@
{
"name": "no_bio_signal_empty_observations",
"old_peer_card": "Name: Leo\nAge: 22\nLocation: Madrid\nOccupation: Student\nInterests: football, coding",
"new_observations": [],
"expected_facts": [
"Name: Leo",
"Age: 22",
"Location: Madrid",
"Occupation: Student",
"Interests: football",
"Interests: coding"
]
}

19
tests/bench/peer_card_tests/no_new_key_info.json Normal file
View File

@ -0,0 +1,19 @@
{
"name": "no_new_key_info",
"old_peer_card": "Name: Frank\nAge: 41\nLocation: Chicago\nOccupation: Sales Manager",
"new_observations": [
"The weather is sunny and warm.",
"Traffic is heavy on I-90."
],
"expected_facts": [
"Name: Frank",
"Age: 41",
"Location: Chicago",
"Occupation: Sales Manager"
],
"forbidden_facts": [
"Interests: weather",
"Interests: traffic",
"Location: I-90"
]
}

21
tests/bench/peer_card_tests/non_first_person_no_change.json Normal file
View File

@ -0,0 +1,21 @@
{
"name": "non_first_person_no_change",
"old_peer_card": "Name: Sofia\nAge: 25\nLocation: Miami\nOccupation: Marketing Specialist\nInterests: dancing, podcasts",
"new_observations": [
"You should try a new cafe downtown.",
"They say running reduces stress.",
"People love living near the beach."
],
"expected_facts": [
"Name: Sofia",
"Age: 25",
"Location: Miami",
"Occupation: Marketing Specialist",
"Interests: dancing",
"Interests: podcasts"
],
"forbidden_facts": [
"Occupation: Barista",
"Location: Downtown"
]
}

22
tests/bench/peer_card_tests/questions_only_no_change.json Normal file
View File

@ -0,0 +1,22 @@
{
"name": "questions_only_no_change",
"old_peer_card": "Name: Nina\nAge: 26\nLocation: Toronto\nOccupation: Analyst\nInterests: reading, tennis",
"new_observations": [
"The user asked about watching the new Dune movie.",
"The user asked about the standup time.",
"The user asked whether to order Thai or pizza."
],
"expected_facts": [
"Name: Nina",
"Age: 26",
"Location: Toronto",
"Occupation: Analyst",
"Interests: reading",
"Interests: tennis"
],
"forbidden_facts": [
"Occupation: Film Critic",
"Interests: Thai food",
"Interests: pizza"
]
}

22
tests/bench/peer_card_tests/sarcasm_hyperbole_no_change.json Normal file
View File

@ -0,0 +1,22 @@
{
"name": "sarcasm_hyperbole_no_change",
"old_peer_card": "Name: Omar\nAge: 31\nLocation: Boston\nOccupation: Researcher\nInterests: soccer, cooking",
"new_observations": [
"The user says they died laughing at a meme.",
"The user joked that they are a million years old.",
"The user joked that they live in the office."
],
"expected_facts": [
"Name: Omar",
"Age: 31",
"Location: Boston",
"Occupation: Researcher",
"Interests: soccer",
"Interests: cooking"
],
"forbidden_facts": [
"Age: 1000000",
"Location: the office",
"Deceased: true"
]
}

22
tests/bench/peer_card_tests/third_party_only_no_change.json Normal file
View File

@ -0,0 +1,22 @@
{
"name": "third_party_only_no_change",
"old_peer_card": "Name: Priya\nAge: 29\nLocation: Seattle\nOccupation: Data Scientist\nInterests: climbing, board games",
"new_observations": [
"Sam loves salsa dancing.",
"Alex got a puppy.",
"Taylor is really into pickling."
],
"expected_facts": [
"Name: Priya",
"Age: 29",
"Location: Seattle",
"Occupation: Data Scientist",
"Interests: climbing",
"Interests: board games"
],
"forbidden_facts": [
"Interests: salsa dancing",
"Pets: puppy",
"Interests: pickling"
]
}

14
tests/bench/peer_card_tests/update_age.json Normal file
View File

@ -0,0 +1,14 @@
{
"name": "update_age",
"old_peer_card": "Name: Carol\nAge: 34\nLocation: Seattle\nOccupation: Data Scientist",
"new_observations": [
"Carol is 35 years old."
],
"expected_facts": [
"Age: 35"
],
"forbidden_facts": [
"Age: 34",
"Age: 36"
]
}

10
tests/bench/peer_card_tests/update_location.json Normal file
View File

@ -0,0 +1,10 @@
{
"name": "update_location",
"old_peer_card": "Name: Bob\nAge: 24\nLocation: New York\nOccupation: Software Engineer\nInterests: Programming, hiking",
"new_observations": [
"Bob now lives in Boston."
],
"expected_facts": [
"Location: Boston"
]
}

21
tests/bench/peer_card_tests/urls_emojis_no_change.json Normal file
View File

@ -0,0 +1,21 @@
{
"name": "urls_emojis_no_change",
"old_peer_card": "Name: Quinn\nAge: 27\nLocation: Chicago\nOccupation: Designer\nInterests: photography, running",
"new_observations": [
"The user shared a link: https://example.com/cool-thing.",
"The user reacted with laughter emojis.",
"The user said 'brb'."
],
"expected_facts": [
"Name: Quinn",
"Age: 27",
"Location: Chicago",
"Occupation: Designer",
"Interests: photography",
"Interests: running"
],
"forbidden_facts": [
"Interests: memes",
"Occupation: Social Media Manager"
]
}

21
tests/bench/peer_card_tests/weather_news_no_change.json Normal file
View File

@ -0,0 +1,21 @@
{
"name": "weather_news_no_change",
"old_peer_card": "Name: Ryan\nAge: 36\nLocation: Austin\nOccupation: Engineer\nInterests: guitars, coffee",
"new_observations": [
"A storm is incoming tonight.",
"Election results were discussed.",
"Gas prices have increased."
],
"expected_facts": [
"Name: Ryan",
"Age: 36",
"Location: Austin",
"Occupation: Engineer",
"Interests: guitars",
"Interests: coffee"
],
"forbidden_facts": [
"Occupation: Meteorologist",
"Occupation: Politician"
]
}

205
tests/bench/run_tests.py
View File

@ -23,8 +23,8 @@ from typing import Any
import tiktoken
from anthropic import AsyncAnthropic
from dotenv import load_dotenv
from honcho import Honcho
from honcho.session import SessionPeerConfig
from honcho import AsyncHoncho
from honcho.async_client.session import SessionPeerConfig
from typing_extensions import TypedDict
load_dotenv()
@ -61,6 +61,7 @@ class TestResult(TypedDict):
start_time: float
end_time: float
duration_seconds: float
output_lines: list[str]
class TestRunner:
@ -72,6 +73,7 @@ class TestRunner:
self,
honcho_url: str = "http://localhost:8000",
anthropic_api_key: str | None = None,
timeout_seconds: int | None = None,
):
"""
Initialize the test runner.
@ -82,6 +84,7 @@ class TestRunner:
"""
self.honcho_url: str = honcho_url
self.anthropic_api_key: str | None = anthropic_api_key
self.timeout_seconds: int | None = timeout_seconds
# Configure logging
logging.basicConfig(
@ -103,6 +106,28 @@ class TestRunner:
raise ValueError("LLM_ANTHROPIC_API_KEY is not set")
self.anthropic_client = AsyncAnthropic(api_key=api_key)
def _format_duration(self, total_seconds: float) -> str:
"""Format a duration in seconds into a human-readable string.
If the duration is at least one minute, this returns a string in the
form "XmYYs" with zero-padded seconds. Otherwise, it returns the
duration in seconds with two decimal places, e.g., "12.34s".
Args:
total_seconds: The duration in seconds.
Returns:
A formatted duration string.
"""
minutes = int(total_seconds // 60)
if minutes > 0:
seconds_rounded = int(round(total_seconds - minutes * 60))
if seconds_rounded == 60:
minutes += 1
seconds_rounded = 0
return f"{minutes}m{seconds_rounded:02d}s"
return f"{total_seconds:.2f}s"
def load_test_file(self, test_file: Path) -> dict[str, Any]:
"""
Load a test definition from a JSON file.
@ -116,7 +141,7 @@ class TestRunner:
with open(test_file) as f:
return json.load(f)
def create_honcho_client(self, workspace_id: str) -> Honcho:
async def create_honcho_client(self, workspace_id: str) -> AsyncHoncho:
"""
Create a Honcho client for a specific workspace.
@ -124,14 +149,16 @@ class TestRunner:
workspace_id: Workspace ID for the test
Returns:
Honcho client instance
AsyncHoncho client instance
"""
return Honcho(
environment="local", workspace_id=workspace_id, base_url=self.honcho_url
return AsyncHoncho(
environment="local",
workspace_id=workspace_id,
base_url=self.honcho_url,
)
async def wait_for_deriver_queue_empty(
self, honcho_client: Honcho, session_id: str | None = None
self, honcho_client: AsyncHoncho, session_id: str | None = None
) -> bool:
"""
Wait for the deriver queue to be empty.
@ -143,16 +170,13 @@ class TestRunner:
Returns:
True if queue is empty, False if timeout exceeded
"""
time.sleep(1)
try:
while True:
status = honcho_client.get_deriver_status(session_id=session_id)
if (
status.in_progress_work_units == 0
and status.pending_work_units == 0
):
break
time.sleep(0.5)
await honcho_client.poll_deriver_status(
session_id=session_id,
timeout=float(self.timeout_seconds)
if self.timeout_seconds
else 10000.0,
)
return True
except Exception as e:
self.logger.warning(f"Error polling deriver status: {e}")
@ -198,7 +222,7 @@ Evaluate whether the actual response contains the core correct information from
"""
response = await self.anthropic_client.messages.create(
model="claude-3-7-sonnet-20250219",
model="claude-sonnet-4-20250514",
max_tokens=300,
temperature=0.0,
system=system_prompt,
@ -257,14 +281,15 @@ Evaluate whether the actual response contains the core correct information from
Test execution results
"""
test_name = test_file.stem
print(f"\033[1mExecuting test {test_name}\033[0m")
output_lines: list[str] = []
output_lines.append(f"\033[1mExecuting test {test_name}\033[0m")
# Load test definition
test_def = self.load_test_file(test_file)
# Create workspace for this test
workspace_id = f"test_{test_name}_{int(time.time())}"
honcho_client = self.create_honcho_client(workspace_id)
honcho_client = await self.create_honcho_client(workspace_id)
results: TestResult = {
"test_name": test_name,
@ -276,6 +301,7 @@ Evaluate whether the actual response contains the core correct information from
"start_time": time.time(),
"end_time": 0.0,
"duration_seconds": 0.0,
"output_lines": output_lines,
}
try:
@ -316,13 +342,13 @@ Evaluate whether the actual response contains the core correct information from
# Create all peers first
peers: dict[str, Any] = {}
for peer_name in all_peers:
peers[peer_name] = honcho_client.peer(id=peer_name)
peers[peer_name] = await honcho_client.peer(id=peer_name)
for session_name, session_data in sessions.items():
# Create session
session = honcho_client.session(id=str(session_name))
session = await honcho_client.session(id=str(session_name))
print(f"\n session: {session_name}")
output_lines.append(f"\n session: {session_name}")
# Create peer configurations based on requirements
peer_configs: list[tuple[Any, SessionPeerConfig]] = []
@ -334,27 +360,31 @@ Evaluate whether the actual response contains the core correct information from
observe_me=False, observe_others=True
)
peer_configs.append((peers[peer_name], config))
print(f" peer config: {peer_name} -> {config}")
output_lines.append(
f" peer config: {peer_name} -> {config}"
)
else:
config = SessionPeerConfig(
observe_me=True, observe_others=True
)
peer_configs.append((peers[peer_name], config))
print(f" peer config: {peer_name} -> {config}")
output_lines.append(
f" peer config: {peer_name} -> {config}"
)
elif peer_name in observed_peers:
config = SessionPeerConfig(
observe_me=True, observe_others=False
)
peer_configs.append((peers[peer_name], config))
print(f" peer config: {peer_name} -> {config}")
output_lines.append(f" peer config: {peer_name} -> {config}")
else:
config = SessionPeerConfig(
observe_me=False, observe_others=False
)
peer_configs.append((peers[peer_name], config))
print(f" peer config: {peer_name} -> {config}")
output_lines.append(f" peer config: {peer_name} -> {config}")
session.add_peers(peer_configs)
await session.add_peers(peer_configs)
# Add messages to session
messages = session_data.get("messages", [])
@ -365,10 +395,10 @@ Evaluate whether the actual response contains the core correct information from
truncated_content = (
content[:140] + "..." if len(content) > 140 else content
)
print(f" {peer_name}: {truncated_content}")
output_lines.append(f" {peer_name}: {truncated_content}")
# Add messages to session
session.add_messages(
await session.add_messages(
[peers[msg["peer"]].message(msg["content"]) for msg in messages]
)
@ -379,6 +409,9 @@ Evaluate whether the actual response contains the core correct information from
# Step 2: Execute queries
all_queries_passed = True
# sleep so the deriver queue is not checked immediately, before tasks get added
await asyncio.sleep(1)
for i, query_data in enumerate(queries):
query: str = query_data["query"]
expected_response: str = query_data["expected_response"]
@ -394,7 +427,7 @@ Evaluate whether the actual response contains the core correct information from
print(f"Deriver queue never emptied for session {session_name}!!!")
sys.exit(1)
print(f"\n query {i + 1}: {query}")
output_lines.append(f"\n query {i + 1}: {query}")
context_parts: list[str] = []
if session_name:
context_parts.append(f"session: {session_name}")
@ -403,7 +436,7 @@ Evaluate whether the actual response contains the core correct information from
if target_name:
context_parts.append(f"target: {target_name}")
if context_parts:
print(" " + ", ".join(context_parts))
output_lines.append(" " + ", ".join(context_parts))
try:
# Determine which peer to use for the query (observer)
@ -417,20 +450,24 @@ Evaluate whether the actual response contains the core correct information from
# Execute chat query
if session_name and target_name:
response = query_peer.chat(
query, session_id=session_name, target=peers[target_name]
response_text = await query_peer.chat(
query,
session_id=session_name,
target=peers[target_name],
)
elif session_name:
response = query_peer.chat(query, session_id=session_name)
response_text = await query_peer.chat(
query, session_id=session_name
)
elif target_name:
response = query_peer.chat(query, target=peers[target_name])
response_text = await query_peer.chat(
query, target=peers[target_name]
)
else:
response = query_peer.chat(query)
response_text = await query_peer.chat(query)
actual_response: str = (
response.content
if hasattr(response, "content")
else str(response)
response_text if response_text is not None else ""
)
# Judge the response
@ -450,22 +487,22 @@ Evaluate whether the actual response contains the core correct information from
results["queries_executed"].append(query_result)
print(
output_lines.append(
" judgment: \033[1m\033[32mPASS\033[0m"
if judgment["passed"]
else " judgment: \033[1m\033[31mFAIL\033[0m"
)
if not judgment["passed"]:
# if failed, print
print(f" got response: \033[3m{actual_response}\033[0m")
print(f" expected: {expected_response}")
output_lines.append(
f" got response: \033[3m{actual_response}\033[0m"
)
output_lines.append(f" expected: {expected_response}")
else:
# if passed, just log
self.logger.info(
f" got response: \033[3m{actual_response}\033[0m"
)
self.logger.info(f" expected: {expected_response}")
print(f" reasoning: {judgment['reasoning']}")
output_lines.append(f" reasoning: {judgment['reasoning']}")
# Track if all queries pass
if not judgment["passed"]:
@ -491,21 +528,23 @@ Evaluate whether the actual response contains the core correct information from
# Step 3: Execute get_context calls
get_context_calls = test_def.get("get_context_calls", [])
for i, get_context_call in enumerate(get_context_calls):
print(f"\n get_context call #{i + 1}")
output_lines.append(f"\n get_context call #{i + 1}")
session_name = str(get_context_call["session"])
summary = get_context_call["summary"]
max_tokens: int | None = get_context_call.get("max_tokens")
session = honcho_client.session(id=session_name)
session = await honcho_client.session(id=session_name)
# Wait for deriver queue to be empty for this session
queue_empty = await self.wait_for_deriver_queue_empty(
honcho_client, session_id=session_name
)
if not queue_empty:
print(f"Deriver queue never emptied for session {session_name}!!!")
output_lines.append(
f"Deriver queue never emptied for session {session_name}!!!"
)
sys.exit(1)
session_context = session.get_context(
session_context = await session.get_context(
summary=summary, tokens=max_tokens
)
@ -515,15 +554,25 @@ Evaluate whether the actual response contains the core correct information from
tokenizer = tiktoken.get_encoding("cl100k_base")
summary_tokens = len(tokenizer.encode(summary_content))
print(f" summary: {session_context.summary}")
output_lines.append(f" summary: {session_context.summary}")
got_tokens = summary_tokens
for message in session_context.messages:
got_tokens += message.token_count
print(f" max tokens: {max_tokens}")
print(
output_lines.append(f" max tokens: {max_tokens}")
output_lines.append(
f" got token count: {got_tokens} (summary: {summary_tokens}, messages: {got_tokens - summary_tokens} in {len(session_context.messages)} messages)"
)
if (
summary
and summary_tokens == 0
and len(session_context.messages) > 20
and max_tokens is None
):
output_lines.append(
" summary is empty when it should not be, test failed"
)
all_queries_passed = False
if max_tokens and got_tokens > max_tokens:
all_queries_passed = False
@ -532,8 +581,8 @@ Evaluate whether the actual response contains the core correct information from
results["end_time"] = time.time()
results["duration_seconds"] = results["end_time"] - results["start_time"]
print(
f"\nTest {test_name} completed. Status: {'PASS' if results['passed'] else 'FAIL'} (Duration: {results['duration_seconds']:.2f}s)"
output_lines.append(
f"\nTest {test_name} completed. Status: {'PASS' if results['passed'] else 'FAIL'} (Duration: {self._format_duration(results['duration_seconds'])})"
)
except Exception as e:
@ -542,10 +591,11 @@ Evaluate whether the actual response contains the core correct information from
results["passed"] = False
results["end_time"] = time.time()
results["duration_seconds"] = results["end_time"] - results["start_time"]
output_lines.append(f"Error executing test {test_name}: {e}")
return results
async def run_all_tests(self, tests_dir: Path) -> list[TestResult]:
async def run_all_tests(self, tests_dir: Path) -> tuple[list[TestResult], float]:
"""
Run all tests in a directory.
@ -558,22 +608,26 @@ Evaluate whether the actual response contains the core correct information from
test_files = list(tests_dir.glob("*.json"))
print(f"found {len(test_files)} test files in {tests_dir}")
all_results: list[TestResult] = []
overall_start = time.time()
# Run all tests concurrently
results: list[TestResult] = await asyncio.gather(
*[self.execute_test(tf) for tf in test_files]
)
overall_end = time.time()
overall_duration = overall_end - overall_start
for test_file in test_files:
result = await self.execute_test(test_file)
all_results.append(result)
# Print summary for this test
# Print detailed per-test outputs in order after completion
for result in results:
print(f"\n{'=' * 60}")
print(f"Test: {result['test_name']}")
print(f"Status: {'PASS' if result['passed'] else 'FAIL'}")
print(f"Duration: {result['duration_seconds']:.2f}s")
print(f"Executing test {result['test_name']}")
print("\n".join(result.get("output_lines", [])))
print(f"{'=' * 60}\n")
return all_results
return results, overall_duration
def print_summary(self, results: list[TestResult]) -> None:
def print_summary(
self, results: list[TestResult], total_elapsed_seconds: float | None = None
) -> None:
"""
Print a summary of all test results.
@ -587,13 +641,17 @@ Evaluate whether the actual response contains the core correct information from
total_tests = len(results)
passed_tests = sum(1 for r in results if r.get("passed", False))
failed_tests = total_tests - passed_tests
total_test_time = sum(r["duration_seconds"] for r in results)
total_test_time = (
total_elapsed_seconds
if total_elapsed_seconds is not None
else sum(r["duration_seconds"] for r in results)
)
print(f"Total Tests: {total_tests}")
print(f"Passed: {passed_tests}")
print(f"Failed: {failed_tests}")
print(f"Success Rate: {(passed_tests / total_tests) * 100:.1f}%")
print(f"Total Test Time: {total_test_time:.2f}s")
print(f"Total Test Time: {self._format_duration(total_test_time)}")
print("\nDetailed Results:")
print(f"{'Test Name':<20} {'Status':<8} {'Duration':<10} {'Workspace ID':<30}")
@ -602,7 +660,7 @@ Evaluate whether the actual response contains the core correct information from
for result in results:
test_name = result["test_name"]
status = "PASS" if result.get("passed", False) else "FAIL"
duration = f"{result['duration_seconds']:.2f}s"
duration = self._format_duration(result["duration_seconds"])
workspace = result["workspace_id"]
print(f"{test_name:<20} {status:<8} {duration:<10} {workspace:<30}")
@ -681,11 +739,16 @@ Examples:
# Run single test
test_file_path = args.tests_dir / args.test
result = await runner.execute_test(test_file_path)
# Print detailed output for the single test
print(f"\n{'=' * 60}")
print(f"Executing test {result['test_name']}")
print("\n".join(result.get("output_lines", [])))
print(f"{'=' * 60}\n")
runner.print_summary([result])
else:
# Run all tests
results = await runner.run_all_tests(args.tests_dir)
runner.print_summary(results)
results, total_elapsed = await runner.run_all_tests(args.tests_dir)
runner.print_summary(results, total_elapsed_seconds=total_elapsed)
return 0

2
tests/bench/tests/summary_and_query.json
View File

@ -169,11 +169,13 @@
{
"observer": "alice",
"query": "What are Alice's favorite numbers?",
"session": "session1",
"expected_response": "7, 74, 12, 8, 2, 3, 14, 24, 37"
},
{
"observer": "bob",
"query": "What is Bob's favorite number?",
"session": "session1",
"expected_response": "7"
}
],

10
tests/deriver/README.md
View File

@ -25,7 +25,6 @@ This directory contains tests for the deriver system, which handles background p
### Mocking Fixtures
- `mock_deriver_process` - Mocks the deriver process_message method
- `mock_critical_analysis_call` - Mocks the critical analysis LLM call
- `mock_queue_manager` - Mocks the queue manager for testing
- `mock_embedding_store` - Mocks the embedding store operations
@ -61,12 +60,3 @@ work_unit = WorkUnit(
# Test string representation
assert str(work_unit) == f"({session.id}, {sender.name}, {target.name}, representation)"
```
### Mocking Deriver Processing
```python
# Use the mock_deriver_process fixture to avoid actual LLM calls
async def test_with_mocked_deriver(mock_deriver_process):
# Deriver processing will use the mock
await process_item(queue_item.payload)
```

17
tests/deriver/conftest.py
View File

@ -14,23 +14,6 @@ from src.deriver.queue_payload import create_payload
from src.deriver.utils import get_work_unit_key
@pytest.fixture
def mock_deriver_process(monkeypatch: pytest.MonkeyPatch) -> Callable[..., Any]:
"""Mock the deriver process_message method to avoid actual LLM calls"""
from src.deriver.deriver import Deriver
async def mock_process_message(
_self: Any, _task_type: str, _payload: dict[str, Any]
) -> None:
# Simulate processing without making actual LLM calls
pass
monkeypatch.setattr(Deriver, "process_message", mock_process_message)
# Return the mock for further configuration if needed
return mock_process_message
@pytest.fixture
def mock_critical_analysis_call() -> Generator[Callable[..., Any], None, None]:
"""Mock the critical analysis call to avoid actual LLM calls"""

16
tests/deriver/test_deriver_processing.py
View File

@ -11,22 +11,6 @@ from src import models
class TestDeriverProcessing:
"""Test suite for deriver processing using the conftest fixtures"""
async def test_mock_deriver_process(
self,
mock_deriver_process: Callable[..., Any], # noqa: ARG001
sample_queue_items: list[models.QueueItem], # noqa: ARG001
):
"""Test that the deriver process is properly mocked"""
# The mock should be in place, so processing should not make real LLM calls
assert mock_deriver_process is not None
# Verify that we have queue items to process
assert len(sample_queue_items) > 0
# Verify the mock is working by checking the first queue item
first_item = sample_queue_items[0]
assert first_item.payload is not None
async def test_mock_critical_analysis_call(
self,
mock_critical_analysis_call: Generator[Callable[..., Any], None, None],

19
tests/deriver/test_queue_processing.py
View File

@ -1,10 +1,7 @@
from typing import Any
import pytest
from sqlalchemy.ext.asyncio import AsyncSession
from src import models
from src.deriver.consumer import process_item
from src.deriver.queue_manager import QueueManager
@ -95,22 +92,6 @@ class TestQueueProcessing:
# This test ensures the cleanup logic doesn't break, though we don't have stale entries yet
assert isinstance(work_units, list)
async def test_process_item_with_mocked_deriver(
self,
mock_deriver_process: Any, # noqa: ARG001 # pyright: ignore[reportUnusedParameter]
sample_queue_items: list[models.QueueItem],
):
"""Test that process_item works with mocked deriver"""
# Take a sample queue item and process it
queue_item = sample_queue_items[0]
# This should not raise an exception since the deriver is mocked
await process_item(queue_item.task_type, queue_item.payload)
# The mock should have been called
# Note: We can't easily verify this since we're mocking the class method directly
# In a real test, we might want to mock at a different level
async def test_work_unit_key_format(
self, sample_session_with_peers: tuple[models.Session, list[models.Peer]]
):

587
tests/deriver/test_representation_crud.py Normal file
View File

@ -0,0 +1,587 @@
import datetime
from typing import Any
import pytest
from nanoid import generate as generate_nanoid
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src import crud, models, schemas
from src.config import settings
from src.crud.representation import (
get_peer_card,
get_working_representation,
get_working_representation_data,
set_peer_card,
set_working_representation,
)
from src.exceptions import ResourceNotFoundException
@pytest.mark.asyncio
async def test_peer_card_get_set_roundtrip(
db_session: AsyncSession, sample_data: tuple[models.Workspace, models.Peer]
):
"""Roundtrip set/get for peer card on a valid peer."""
workspace, peer = sample_data
# Initially absent
assert await get_peer_card(db_session, workspace.name, peer.name) is None
# Set and read back
value_1 = ["Initial peer card text"]
await set_peer_card(db_session, workspace.name, peer.name, value_1)
assert await get_peer_card(db_session, workspace.name, peer.name) == value_1
# Update and read back
value_2 = ["Updated peer card text", "Another line"]
await set_peer_card(db_session, workspace.name, peer.name, value_2)
assert await get_peer_card(db_session, workspace.name, peer.name) == value_2
@pytest.mark.asyncio
async def test_set_peer_card_missing_peer_raises(
db_session: AsyncSession, sample_data: tuple[models.Workspace, models.Peer]
):
"""Setting a peer card for a non-existent peer should raise ResourceNotFoundException."""
workspace, _existing_peer = sample_data
with pytest.raises(ResourceNotFoundException):
await set_peer_card(db_session, workspace.name, "missing-peer", ["card"])
async def _create_session_with_peers(
db_session: AsyncSession, workspace: models.Workspace
) -> tuple[models.Session, models.Peer, models.Peer]:
"""Create a session with two peers and return (session, observer, observed)."""
observer = models.Peer(name=str(generate_nanoid()), workspace_name=workspace.name)
observed = models.Peer(name=str(generate_nanoid()), workspace_name=workspace.name)
db_session.add_all([observer, observed])
await db_session.flush()
session_name = str(generate_nanoid())
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=session_name,
peers={
observer.name: schemas.SessionPeerConfig(),
observed.name: schemas.SessionPeerConfig(),
},
),
workspace.name,
)
await db_session.commit()
return session, observer, observed
def _make_wr_payload(
*,
explicit: list[str],
deductive: list[dict[str, Any]] | None = None,
thinking: str | None = None,
message_id: str = "m-new",
created_at: str | None = None,
) -> dict[str, Any]:
"""Build a structured working representation dict payload."""
return {
"final_observations": {
"explicit": explicit,
"deductive": deductive or [],
"thinking": thinking,
},
"message_id": message_id,
"created_at": created_at
or datetime.datetime.now(datetime.timezone.utc).isoformat(),
}
@pytest.mark.asyncio
async def test_working_representation_self_merge_and_trim(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
monkeypatch: pytest.MonkeyPatch,
):
"""Merging a self working representation appends and trims to 25, and updates metadata fields."""
workspace, peer = sample_data
# Align limit with current implementation and future config usage
monkeypatch.setattr(
settings.DERIVER, "WORKING_REPRESENTATION_MAX_OBSERVATIONS", 25, raising=False
)
LIMIT = settings.DERIVER.WORKING_REPRESENTATION_MAX_OBSERVATIONS
# Create a session with the single peer
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=str(generate_nanoid()), peers={peer.name: schemas.SessionPeerConfig()}
),
workspace.name,
)
await db_session.commit()
# Existing explicit: 24 items 0..23
existing = _make_wr_payload(
explicit=[f"E{i}" for i in range(24)],
deductive=[{"conclusion": f"D{i}", "premises": [f"P{i}"]} for i in range(10)],
thinking="old-think",
message_id="m-old",
created_at=datetime.datetime(
2024, 1, 1, tzinfo=datetime.timezone.utc
).isoformat(),
)
await set_working_representation(
db_session, existing, workspace.name, peer.name, peer.name, session.name
)
# New explicit: 3 items 24..26, deductive two new items
new_payload = _make_wr_payload(
explicit=["E24", "E25", "E26"],
deductive=[
{"conclusion": "D_new1", "premises": []},
{"conclusion": "D_new2", "premises": ["PX"]},
],
thinking="new-think",
message_id="m-new",
created_at=datetime.datetime(
2025, 1, 1, tzinfo=datetime.timezone.utc
).isoformat(),
)
await set_working_representation(
db_session, new_payload, workspace.name, peer.name, peer.name, session.name
)
# Verify merged raw data
raw = await get_working_representation_data(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert isinstance(raw, dict)
final = raw["final_observations"]
# Explicit should be last LIMIT of 24+3 (drop oldest overflow)
explicit = final["explicit"]
total_explicit = 24 + 3
expected_len = min(LIMIT, total_explicit)
dropped = max(0, total_explicit - LIMIT)
assert len(explicit) == expected_len
assert explicit[0] == f"E{dropped}"
assert explicit[-1] == "E26"
# Deductive should be appended and capped to LIMIT
deductive = final["deductive"]
assert len(deductive) == min(LIMIT, 12)
assert deductive[-2]["conclusion"] == "D_new1"
assert deductive[-1]["conclusion"] == "D_new2"
# Thinking and metadata should reflect latest
assert final.get("thinking") == "new-think"
assert raw.get("message_id") == "m-new"
created_at_value = raw.get("created_at")
assert created_at_value is not None
assert created_at_value.startswith("2025-01-01")
# Formatted string getter returns sections and bullets
formatted = await get_working_representation(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert "EXPLICIT OBSERVATIONS:" in formatted
assert "DEDUCTIVE OBSERVATIONS:" in formatted
assert "- E26" in formatted
@pytest.mark.asyncio
async def test_working_representation_directional_merge_and_keys(
db_session: AsyncSession, sample_data: tuple[models.Workspace, models.Peer]
):
"""Directional working representations are stored under observer_observed key and merge correctly."""
workspace, _ = sample_data
session, observer, observed = await _create_session_with_peers(
db_session, workspace
)
# Store initial
first = _make_wr_payload(explicit=["A", "B"], thinking="first")
await set_working_representation(
db_session, first, workspace.name, observer.name, observed.name, session.name
)
# Merge second
second = _make_wr_payload(
explicit=["C"],
deductive=[{"conclusion": "Z", "premises": ["p1", "p2"]}],
thinking="second",
)
await set_working_representation(
db_session, second, workspace.name, observer.name, observed.name, session.name
)
# Fetch raw and assert structure
raw = await get_working_representation_data(
db_session, workspace.name, observer.name, observed.name, session.name
)
assert isinstance(raw, dict)
final = raw["final_observations"]
assert final["explicit"] == ["A", "B", "C"]
assert final["deductive"][-1]["conclusion"] == "Z"
assert final.get("thinking") == "second"
# Validate it's stored under the derived key in SessionPeer.internal_metadata
derived_key = f"{observer.name}_{observed.name}"
stmt = select(models.SessionPeer).where(
models.SessionPeer.peer_name == observer.name,
models.SessionPeer.session_name == session.name,
models.SessionPeer.workspace_name == workspace.name,
)
result = await db_session.execute(stmt)
sp = result.scalar_one()
assert derived_key in sp.internal_metadata
assert sp.internal_metadata[derived_key]["final_observations"]["explicit"] == [
"A",
"B",
"C",
]
@pytest.mark.asyncio
async def test_wr_string_roundtrip_then_structured_overrides(
db_session: AsyncSession, sample_data: tuple[models.Workspace, models.Peer]
):
"""String WR stores as-is and later structured merge does not incorporate old string content."""
workspace, peer = sample_data
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=str(generate_nanoid()), peers={peer.name: schemas.SessionPeerConfig()}
),
workspace.name,
)
await db_session.commit()
# Store legacy string representation first
raw_string = "legacy working rep text"
await set_working_representation(
db_session, raw_string, workspace.name, peer.name, peer.name, session.name
)
# Ensure get returns the same string
assert (
await get_working_representation(
db_session, workspace.name, peer.name, peer.name, session.name
)
== raw_string
)
# Now store structured representation; merge should ignore old string and just store new structured
structured = _make_wr_payload(explicit=["X", "Y"], deductive=[])
await set_working_representation(
db_session, structured, workspace.name, peer.name, peer.name, session.name
)
raw = await get_working_representation_data(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert isinstance(raw, dict)
assert raw["final_observations"]["explicit"] == ["X", "Y"]
@pytest.mark.asyncio
async def test_wr_missing_levels_and_empty_new_lists(
db_session: AsyncSession, sample_data: tuple[models.Workspace, models.Peer]
):
"""Merge handles missing levels and empty new lists; result formatting is empty string."""
workspace, peer = sample_data
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=str(generate_nanoid()), peers={peer.name: schemas.SessionPeerConfig()}
),
workspace.name,
)
await db_session.commit()
# Existing has only deductive; explicit missing
existing = {
"final_observations": {
"deductive": [{"conclusion": "old", "premises": ["p"]}],
},
"message_id": "m-old",
"created_at": datetime.datetime(
2024, 1, 2, tzinfo=datetime.timezone.utc
).isoformat(),
}
await set_working_representation(
db_session, existing, workspace.name, peer.name, peer.name, session.name
)
# New has empty lists and no thinking
new_payload = _make_wr_payload(
explicit=[], deductive=[], thinking=None, message_id="m-new"
)
await set_working_representation(
db_session, new_payload, workspace.name, peer.name, peer.name, session.name
)
raw = await get_working_representation_data(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert isinstance(raw, dict)
final = raw["final_observations"]
# Deductive remains as old (no additions), explicit remains missing/empty and thinking None
assert final["deductive"] == [{"conclusion": "old", "premises": ["p"]}]
assert final["explicit"] == []
assert final.get("thinking") is None
# Formatter includes the remaining deductive observation
formatted = await get_working_representation(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert "DEDUCTIVE OBSERVATIONS:" in formatted
assert "- old (based on: p)" in formatted
@pytest.mark.asyncio
async def test_wr_trim_boundary_exact_25_plus_one(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
monkeypatch: pytest.MonkeyPatch,
):
"""Exactly 25 existing + 1 new yields last 25, preserving order and including the new last element."""
workspace, peer = sample_data
monkeypatch.setattr(
settings.DERIVER, "WORKING_REPRESENTATION_MAX_OBSERVATIONS", 25, raising=False
)
LIMIT = settings.DERIVER.WORKING_REPRESENTATION_MAX_OBSERVATIONS
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=str(generate_nanoid()), peers={peer.name: schemas.SessionPeerConfig()}
),
workspace.name,
)
await db_session.commit()
existing = _make_wr_payload(
explicit=[f"E{i}" for i in range(LIMIT)],
deductive=[{"conclusion": f"D{i}", "premises": []} for i in range(LIMIT)],
)
await set_working_representation(
db_session, existing, workspace.name, peer.name, peer.name, session.name
)
new_payload = _make_wr_payload(
explicit=[f"E{LIMIT}"],
deductive=[{"conclusion": f"D{LIMIT}", "premises": []}],
thinking="t2",
message_id="m2",
)
await set_working_representation(
db_session, new_payload, workspace.name, peer.name, peer.name, session.name
)
raw = await get_working_representation_data(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert isinstance(raw, dict)
final = raw["final_observations"]
assert final["explicit"] == [f"E{i}" for i in range(1, LIMIT + 1)]
assert final["explicit"][-1] == f"E{LIMIT}"
assert len(final["deductive"]) == LIMIT
assert final["deductive"][-1]["conclusion"] == f"D{LIMIT}"
@pytest.mark.asyncio
async def test_wr_formatting_rules_mixed_observation_types(
db_session: AsyncSession, sample_data: tuple[models.Workspace, models.Peer]
):
"""Formatting includes section headers, bullets, premise display, and content fallback."""
workspace, peer = sample_data
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=str(generate_nanoid()), peers={peer.name: schemas.SessionPeerConfig()}
),
workspace.name,
)
await db_session.commit()
payload = {
"final_observations": {
"explicit": ["likes pizza", {"content": "runs daily"}],
"deductive": [
{
"conclusion": "is healthy",
"premises": ["runs daily", "eats veggies"],
},
{"content": "fallback without conclusion"},
],
"thinking": "t",
},
"message_id": "m1",
"created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
}
await set_working_representation(
db_session, payload, workspace.name, peer.name, peer.name, session.name
)
formatted = await get_working_representation(
db_session, workspace.name, peer.name, peer.name, session.name
)
# Section headers present
assert formatted.splitlines()[0] == "EXPLICIT OBSERVATIONS:"
assert "DEDUCTIVE OBSERVATIONS:" in formatted
# Bullets present for strings and dict content fallback
assert "- likes pizza" in formatted
assert "- runs daily" in formatted
# Deductive with premises shows based on
assert "is healthy (based on: runs daily; eats veggies)" in formatted
# Deductive dict without conclusion falls back to content
assert "- fallback without conclusion" in formatted
@pytest.mark.asyncio
async def test_wr_legacy_key_fallback_for_self_representation(
db_session: AsyncSession, sample_data: tuple[models.Workspace, models.Peer]
):
"""If only the legacy key is present, data retrieval falls back appropriately."""
workspace, peer = sample_data
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=str(generate_nanoid()), peers={peer.name: schemas.SessionPeerConfig()}
),
workspace.name,
)
await db_session.commit()
# Manually write legacy key on SessionPeer
stmt = select(models.SessionPeer).where(
models.SessionPeer.peer_name == peer.name,
models.SessionPeer.session_name == session.name,
models.SessionPeer.workspace_name == workspace.name,
)
result = await db_session.execute(stmt)
sp: models.SessionPeer = result.scalar_one()
# Assign a new dict so JSON mutation is tracked & persisted
sp.internal_metadata = {
**(sp.internal_metadata or {}),
"global_representation": "legacy-global",
}
await db_session.commit()
# Retrieval should see legacy value
raw = await get_working_representation_data(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert raw == "legacy-global"
@pytest.mark.asyncio
async def test_wr_empty_both_levels_formats_empty_string(
db_session: AsyncSession, sample_data: tuple[models.Workspace, models.Peer]
):
"""When both explicit and deductive are empty after merge, formatted string is empty."""
workspace, peer = sample_data
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=str(generate_nanoid()), peers={peer.name: schemas.SessionPeerConfig()}
),
workspace.name,
)
await db_session.commit()
empty_payload = _make_wr_payload(explicit=[], deductive=[], thinking=None)
await set_working_representation(
db_session, empty_payload, workspace.name, peer.name, peer.name, session.name
)
formatted = await get_working_representation(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert formatted == ""
@pytest.mark.asyncio
async def test_wr_existing_dict_without_final_observations(
db_session: AsyncSession, sample_data: tuple[models.Workspace, models.Peer]
):
"""Existing dict lacking final_observations merges cleanly with new structured payload."""
workspace, peer = sample_data
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=str(generate_nanoid()), peers={peer.name: schemas.SessionPeerConfig()}
),
workspace.name,
)
await db_session.commit()
await set_working_representation(
db_session,
{"some": "field"},
workspace.name,
peer.name,
peer.name,
session.name,
)
new_payload = _make_wr_payload(
explicit=["n1"], deductive=[{"conclusion": "c1", "premises": []}]
)
await set_working_representation(
db_session, new_payload, workspace.name, peer.name, peer.name, session.name
)
raw = await get_working_representation_data(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert isinstance(raw, dict)
final = raw["final_observations"]
assert final["explicit"] == ["n1"]
assert final["deductive"][0]["conclusion"] == "c1"
@pytest.mark.asyncio
async def test_wr_deductive_trim_when_no_new_items(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
monkeypatch: pytest.MonkeyPatch,
):
"""If existing deductive > 25 and no new entries, merge still trims to last 25."""
workspace, peer = sample_data
monkeypatch.setattr(
settings.DERIVER, "WORKING_REPRESENTATION_MAX_OBSERVATIONS", 25, raising=False
)
LIMIT = settings.DERIVER.WORKING_REPRESENTATION_MAX_OBSERVATIONS
session = await crud.get_or_create_session(
db_session,
schemas.SessionCreate(
name=str(generate_nanoid()), peers={peer.name: schemas.SessionPeerConfig()}
),
workspace.name,
)
await db_session.commit()
over = LIMIT + 5
existing = _make_wr_payload(
explicit=[],
deductive=[{"conclusion": f"D{i}", "premises": []} for i in range(over)],
)
await set_working_representation(
db_session, existing, workspace.name, peer.name, peer.name, session.name
)
# Merge empty new; should trim to last 25 existing
new_payload = _make_wr_payload(explicit=[], deductive=[])
await set_working_representation(
db_session, new_payload, workspace.name, peer.name, peer.name, session.name
)
raw = await get_working_representation_data(
db_session, workspace.name, peer.name, peer.name, session.name
)
assert isinstance(raw, dict)
final = raw["final_observations"]
assert len(final["deductive"]) == LIMIT
# Oldest retained index is over - LIMIT
oldest_kept = over - LIMIT
assert final["deductive"][0]["conclusion"] == f"D{oldest_kept}"
assert final["deductive"][-1]["conclusion"] == f"D{over - 1}"

4
tests/test_llm_mock.py
View File

@ -15,9 +15,9 @@ async def test_generic_honcho_llm_call_mock():
# Call the decorated function - this should use our mock
result = await critical_analysis_call(
peer_name="test_peer",
peer_card=["test_peer_card"],
message_created_at=datetime(2023, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
context="test context",
working_representation="test working representation",
history="test history",
new_turn="test new turn",
)

10
uv.lock
View File

@ -877,9 +877,9 @@ requires-dist = [
{ name = "google-generativeai", specifier = ">=0.8.5" },
{ name = "greenlet", specifier = ">=3.0.3" },
{ name = "httpx", specifier = ">=0.27.0" },
{ name = "mirascope", extras = ["anthropic", "google", "groq", "langfuse"], specifier = ">=1.25.1" },
{ name = "mirascope", extras = ["anthropic", "google", "groq", "langfuse"], specifier = ">=1.25.5" },
{ name = "nanoid", specifier = ">=2.0.0" },
{ name = "openai", specifier = ">=1.91.0" },
{ name = "openai", specifier = ">=1.99.7" },
{ name = "pdfplumber", specifier = ">=0.11.7" },
{ name = "pgvector", specifier = ">=0.2.5" },
{ name = "psycopg", extras = ["binary"], specifier = ">=3.1.19" },
@ -1480,7 +1480,7 @@ wheels = [
[[package]]
name = "openai"
version = "1.99.1"
version = "1.99.7"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
@ -1492,9 +1492,9 @@ dependencies = [
{ name = "tqdm" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/03/30/f0fb7907a77e733bb801c7bdcde903500b31215141cdb261f04421e6fbec/openai-1.99.1.tar.gz", hash = "sha256:2c9d8e498c298f51bb94bcac724257a3a6cac6139ccdfc1186c6708f7a93120f", size = 497075, upload-time = "2025-08-05T19:42:36.131Z" }
sdist = { url = "https://files.pythonhosted.org/packages/d8/e3/14812e91dee4d7a1d6aa365ec722b9f2c7ecca3b4f1fb5c56c2c1d83de82/openai-1.99.7.tar.gz", hash = "sha256:d2f4211642b9dbcd8e3cc6e6ef1180ac149f80d2e5ab1ee7f5afdd8d34c9b33b", size = 505598, upload-time = "2025-08-11T15:13:10.693Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/54/15/9c85154ffd283abfc43309ff3aaa63c3fd02f7767ee684e73670f6c5ade2/openai-1.99.1-py3-none-any.whl", hash = "sha256:8eeccc69e0ece1357b51ca0d9fb21324afee09b20c3e5b547d02445ca18a4e03", size = 767827, upload-time = "2025-08-05T19:42:34.192Z" },
{ url = "https://files.pythonhosted.org/packages/64/2d/a41c49550a69374111647be22b587e3311a6fc31fc7370e14448e78018bc/openai-1.99.7-py3-none-any.whl", hash = "sha256:ef4165cc4f8872dd4a967d109f12b0b9c98a1e20ae05940c28701729c2883891", size = 786809, upload-time = "2025-08-11T15:13:08.537Z" },
]
[[package]]