From 7a9746ca8b4d5c66b18f7aaf0ddc8b03a26fb7cc Mon Sep 17 00:00:00 2001 From: doria <93405247+dr-frmr@users.noreply.github.com> Date: Tue, 12 Aug 2025 17:39:08 -0400 Subject: [PATCH] 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 --- .env.template | 9 +- config.toml.example | 5 + docs/v2/contributing/configuration.mdx | 12 + pyproject.toml | 4 +- src/config.py | 13 + src/crud/__init__.py | 9 +- src/crud/representation.py | 261 +++++-- src/deriver/consumer.py | 117 ++- src/deriver/deriver.py | 731 ++++++++---------- src/deriver/prompts.py | 106 ++- src/deriver/queue_manager.py | 27 +- src/deriver/queue_payload.py | 5 - src/dialectic/chat.py | 30 +- src/dialectic/prompts.py | 32 +- src/exceptions.py | 38 +- src/models.py | 3 + src/utils/clients.py | 44 +- src/utils/embedding_store.py | 23 +- src/utils/formatting.py | 19 +- src/utils/logging.py | 2 +- src/utils/shared_models.py | 12 + src/utils/summarizer.py | 6 +- tests/bench/README.md | 2 +- tests/bench/peer_card_bench.py | 435 +++++++++++ tests/bench/peer_card_tests/add_nickname.json | 11 + .../chitchat_only_no_change.json | 24 + ...adiction_update_location_amidst_noise.json | 17 + .../peer_card_tests/create_basic_card.json | 18 + .../peer_card_tests/large_card_no_change.json | 26 + ...rge_card_with_one_update_amidst_noise.json | 18 + tests/bench/peer_card_tests/limit_signal.json | 19 + .../peer_card_tests/multiple_changes.json | 14 + .../no_bio_signal_empty_observations.json | 13 + .../peer_card_tests/no_new_key_info.json | 19 + .../non_first_person_no_change.json | 21 + .../questions_only_no_change.json | 22 + .../sarcasm_hyperbole_no_change.json | 22 + .../third_party_only_no_change.json | 22 + tests/bench/peer_card_tests/update_age.json | 14 + .../peer_card_tests/update_location.json | 10 + .../urls_emojis_no_change.json | 21 + .../weather_news_no_change.json | 21 + tests/bench/run_tests.py | 205 +++-- tests/bench/tests/summary_and_query.json | 2 + tests/deriver/README.md | 10 - tests/deriver/conftest.py | 17 - tests/deriver/test_deriver_processing.py | 16 - tests/deriver/test_queue_processing.py | 19 - tests/deriver/test_representation_crud.py | 587 ++++++++++++++ tests/test_llm_mock.py | 4 +- uv.lock | 10 +- 51 files changed, 2421 insertions(+), 726 deletions(-) create mode 100644 tests/bench/peer_card_bench.py create mode 100644 tests/bench/peer_card_tests/add_nickname.json create mode 100644 tests/bench/peer_card_tests/chitchat_only_no_change.json create mode 100644 tests/bench/peer_card_tests/contradiction_update_location_amidst_noise.json create mode 100644 tests/bench/peer_card_tests/create_basic_card.json create mode 100644 tests/bench/peer_card_tests/large_card_no_change.json create mode 100644 tests/bench/peer_card_tests/large_card_with_one_update_amidst_noise.json create mode 100644 tests/bench/peer_card_tests/limit_signal.json create mode 100644 tests/bench/peer_card_tests/multiple_changes.json create mode 100644 tests/bench/peer_card_tests/no_bio_signal_empty_observations.json create mode 100644 tests/bench/peer_card_tests/no_new_key_info.json create mode 100644 tests/bench/peer_card_tests/non_first_person_no_change.json create mode 100644 tests/bench/peer_card_tests/questions_only_no_change.json create mode 100644 tests/bench/peer_card_tests/sarcasm_hyperbole_no_change.json create mode 100644 tests/bench/peer_card_tests/third_party_only_no_change.json create mode 100644 tests/bench/peer_card_tests/update_age.json create mode 100644 tests/bench/peer_card_tests/update_location.json create mode 100644 tests/bench/peer_card_tests/urls_emojis_no_change.json create mode 100644 tests/bench/peer_card_tests/weather_news_no_change.json create mode 100644 tests/deriver/test_representation_crud.py diff --git a/.env.template b/.env.template index f3cb0db8..abb4e5b3 100644 --- a/.env.template +++ b/.env.template @@ -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 diff --git a/config.toml.example b/config.toml.example index 0916269c..98d84f5a 100644 --- a/config.toml.example +++ b/config.toml.example @@ -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] diff --git a/docs/v2/contributing/configuration.mdx b/docs/v2/contributing/configuration.mdx index 26a4b9b2..cf903c97 100644 --- a/docs/v2/contributing/configuration.mdx +++ b/docs/v2/contributing/configuration.mdx @@ -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:** diff --git a/pyproject.toml b/pyproject.toml index ac6f8a0d..ca8a1ca2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", diff --git a/src/config.py b/src/config.py index 676746e3..68e9039d 100644 --- a/src/config.py +++ b/src/config.py @@ -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 diff --git a/src/crud/__init__.py b/src/crud/__init__.py index 3dc4eb51..4b96d862 100644 --- a/src/crud/__init__.py +++ b/src/crud/__init__.py @@ -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", diff --git a/src/crud/representation.py b/src/crud/representation.py index 17d069ad..ab2110ee 100644 --- a/src/crud/representation.py +++ b/src/crud/representation.py @@ -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}" diff --git a/src/deriver/consumer.py b/src/deriver/consumer.py index a62b0481..399904d7 100644 --- a/src/deriver/consumer.py +++ b/src/deriver/consumer.py @@ -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}") diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index c850da52..437d8781 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -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, + ) diff --git a/src/deriver/prompts.py b/src/deriver/prompts.py index 6615f254..426329c2 100644 --- a/src/deriver/prompts.py +++ b/src/deriver/prompts.py @@ -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: + +{chr(10).join(peer_card)} + +""" + if peer_card is not None + else "" + ) + + working_representation_section = ( + f""" +The current user understanding: + +{working_representation} + +""" + 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 - -{context} - +{peer_card_section} + +{working_representation_section} Recent conversation history for context: @@ -75,3 +97,63 @@ New conversation turn to analyze: """ ) + + +@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. + """ + ) diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 5c29378d..9f712b2d 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -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" ) diff --git a/src/deriver/queue_payload.py b/src/deriver/queue_payload.py index 1eb85332..c4f28245 100644 --- a/src/deriver/queue_payload.py +++ b/src/deriver/queue_payload.py @@ -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, diff --git a/src/dialectic/chat.py b/src/dialectic/chat.py index c03450cf..a0e1caf6 100644 --- a/src/dialectic/chat.py +++ b/src/dialectic/chat.py @@ -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 diff --git a/src/dialectic/prompts.py b/src/dialectic/prompts.py index 80184c71..81d4076c 100644 --- a/src/dialectic/prompts.py +++ b/src/dialectic/prompts.py @@ -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} diff --git a/src/exceptions.py b/src/exceptions.py index 82fe9ec8..1c22ba15 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -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 diff --git a/src/models.py b/src/models.py index cf885ce5..defe2110 100644 --- a/src/models.py +++ b/src/models.py @@ -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): diff --git a/src/utils/clients.py b/src/utils/clients.py index 7d0ea357..0fa7ddd8 100644 --- a/src/utils/clients.py +++ b/src/utils/clients.py @@ -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] = {} diff --git a/src/utils/embedding_store.py b/src/utils/embedding_store.py index 9bc303ce..34f0ba97 100644 --- a/src/utils/embedding_store.py +++ b/src/utils/embedding_store.py @@ -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( diff --git a/src/utils/formatting.py b/src/utils/formatting.py index 6b479e3b..e9c3b8ad 100644 --- a/src/utils/formatting.py +++ b/src/utils/formatting.py @@ -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() diff --git a/src/utils/logging.py b/src/utils/logging.py index e665842b..9d80f9a9 100644 --- a/src/utils/logging.py +++ b/src/utils/logging.py @@ -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: diff --git a/src/utils/shared_models.py b/src/utils/shared_models.py index 76d736b7..522227cc 100644 --- a/src/utils/shared_models.py +++ b/src/utils/shared_models.py @@ -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 diff --git a/src/utils/summarizer.py b/src/utils/summarizer.py index c3391e63..8c8394cd 100644 --- a/src/utils/summarizer.py +++ b/src/utils/summarizer.py @@ -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 diff --git a/tests/bench/README.md b/tests/bench/README.md index 52b047e0..2f124efb 100644 --- a/tests/bench/README.md +++ b/tests/bench/README.md @@ -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`) diff --git a/tests/bench/peer_card_bench.py b/tests/bench/peer_card_bench.py new file mode 100644 index 00000000..9a46cec6 --- /dev/null +++ b/tests/bench/peer_card_bench.py @@ -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()) diff --git a/tests/bench/peer_card_tests/add_nickname.json b/tests/bench/peer_card_tests/add_nickname.json new file mode 100644 index 00000000..156a7d5f --- /dev/null +++ b/tests/bench/peer_card_tests/add_nickname.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/chitchat_only_no_change.json b/tests/bench/peer_card_tests/chitchat_only_no_change.json new file mode 100644 index 00000000..ab4bb6ff --- /dev/null +++ b/tests/bench/peer_card_tests/chitchat_only_no_change.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/contradiction_update_location_amidst_noise.json b/tests/bench/peer_card_tests/contradiction_update_location_amidst_noise.json new file mode 100644 index 00000000..587cf26e --- /dev/null +++ b/tests/bench/peer_card_tests/contradiction_update_location_amidst_noise.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/create_basic_card.json b/tests/bench/peer_card_tests/create_basic_card.json new file mode 100644 index 00000000..872efa46 --- /dev/null +++ b/tests/bench/peer_card_tests/create_basic_card.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/large_card_no_change.json b/tests/bench/peer_card_tests/large_card_no_change.json new file mode 100644 index 00000000..e079f8b4 --- /dev/null +++ b/tests/bench/peer_card_tests/large_card_no_change.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/large_card_with_one_update_amidst_noise.json b/tests/bench/peer_card_tests/large_card_with_one_update_amidst_noise.json new file mode 100644 index 00000000..17175638 --- /dev/null +++ b/tests/bench/peer_card_tests/large_card_with_one_update_amidst_noise.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/limit_signal.json b/tests/bench/peer_card_tests/limit_signal.json new file mode 100644 index 00000000..9b121d1f --- /dev/null +++ b/tests/bench/peer_card_tests/limit_signal.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/multiple_changes.json b/tests/bench/peer_card_tests/multiple_changes.json new file mode 100644 index 00000000..483709ce --- /dev/null +++ b/tests/bench/peer_card_tests/multiple_changes.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/no_bio_signal_empty_observations.json b/tests/bench/peer_card_tests/no_bio_signal_empty_observations.json new file mode 100644 index 00000000..40d24431 --- /dev/null +++ b/tests/bench/peer_card_tests/no_bio_signal_empty_observations.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/no_new_key_info.json b/tests/bench/peer_card_tests/no_new_key_info.json new file mode 100644 index 00000000..24d31324 --- /dev/null +++ b/tests/bench/peer_card_tests/no_new_key_info.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/non_first_person_no_change.json b/tests/bench/peer_card_tests/non_first_person_no_change.json new file mode 100644 index 00000000..34487bdc --- /dev/null +++ b/tests/bench/peer_card_tests/non_first_person_no_change.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/questions_only_no_change.json b/tests/bench/peer_card_tests/questions_only_no_change.json new file mode 100644 index 00000000..8a4f4fed --- /dev/null +++ b/tests/bench/peer_card_tests/questions_only_no_change.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/sarcasm_hyperbole_no_change.json b/tests/bench/peer_card_tests/sarcasm_hyperbole_no_change.json new file mode 100644 index 00000000..f66353e0 --- /dev/null +++ b/tests/bench/peer_card_tests/sarcasm_hyperbole_no_change.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/third_party_only_no_change.json b/tests/bench/peer_card_tests/third_party_only_no_change.json new file mode 100644 index 00000000..844b8fb0 --- /dev/null +++ b/tests/bench/peer_card_tests/third_party_only_no_change.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/update_age.json b/tests/bench/peer_card_tests/update_age.json new file mode 100644 index 00000000..91751eb0 --- /dev/null +++ b/tests/bench/peer_card_tests/update_age.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/update_location.json b/tests/bench/peer_card_tests/update_location.json new file mode 100644 index 00000000..feb8b544 --- /dev/null +++ b/tests/bench/peer_card_tests/update_location.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/urls_emojis_no_change.json b/tests/bench/peer_card_tests/urls_emojis_no_change.json new file mode 100644 index 00000000..543e9947 --- /dev/null +++ b/tests/bench/peer_card_tests/urls_emojis_no_change.json @@ -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" + ] +} diff --git a/tests/bench/peer_card_tests/weather_news_no_change.json b/tests/bench/peer_card_tests/weather_news_no_change.json new file mode 100644 index 00000000..d910efc0 --- /dev/null +++ b/tests/bench/peer_card_tests/weather_news_no_change.json @@ -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" + ] +} diff --git a/tests/bench/run_tests.py b/tests/bench/run_tests.py index 72f7d7b8..9168fe7e 100644 --- a/tests/bench/run_tests.py +++ b/tests/bench/run_tests.py @@ -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 diff --git a/tests/bench/tests/summary_and_query.json b/tests/bench/tests/summary_and_query.json index c78af985..a0b2ac74 100644 --- a/tests/bench/tests/summary_and_query.json +++ b/tests/bench/tests/summary_and_query.json @@ -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" } ], diff --git a/tests/deriver/README.md b/tests/deriver/README.md index a814e335..d7f4ebfa 100644 --- a/tests/deriver/README.md +++ b/tests/deriver/README.md @@ -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) -``` diff --git a/tests/deriver/conftest.py b/tests/deriver/conftest.py index 95386c3e..3025a217 100644 --- a/tests/deriver/conftest.py +++ b/tests/deriver/conftest.py @@ -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""" diff --git a/tests/deriver/test_deriver_processing.py b/tests/deriver/test_deriver_processing.py index 28a4e123..deaa085a 100644 --- a/tests/deriver/test_deriver_processing.py +++ b/tests/deriver/test_deriver_processing.py @@ -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], diff --git a/tests/deriver/test_queue_processing.py b/tests/deriver/test_queue_processing.py index 080ac028..8f12f5cd 100644 --- a/tests/deriver/test_queue_processing.py +++ b/tests/deriver/test_queue_processing.py @@ -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]] ): diff --git a/tests/deriver/test_representation_crud.py b/tests/deriver/test_representation_crud.py new file mode 100644 index 00000000..422c6b12 --- /dev/null +++ b/tests/deriver/test_representation_crud.py @@ -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}" diff --git a/tests/test_llm_mock.py b/tests/test_llm_mock.py index 7bdab32a..6ce690d5 100644 --- a/tests/test_llm_mock.py +++ b/tests/test_llm_mock.py @@ -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", ) diff --git a/uv.lock b/uv.lock index 635733ac..a1f0a4be 100644 --- a/uv.lock +++ b/uv.lock @@ -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]]