From 68acf38134ff3134f06003dabe298a57dacecbe6 Mon Sep 17 00:00:00 2001 From: doria <93405247+dr-frmr@users.noreply.github.com> Date: Thu, 9 Oct 2025 16:53:41 -0400 Subject: [PATCH] Misc: bug fixes, multi-db test harness, DELETE workspace (#230) * feat: add optional JWT and webhook secrets to honcho instance creation * chore: ignore spurious warnings * feat: add response format if using gpt-5 model family * feat: add response models to all apis except anthropic * fix: raise NotImplementedError for response models in AsyncAnthropic client * chore: address review * [WIP] representation structure + deriver cleanup * chore: add tests, cleanup * feat: [WIP: semi-working] representation object * fix: alignment * fix: make observations hashable for dedup * fix: datetime formatting, observation counting * fix: switch to int for message id, clean up representation * feat: remove need for metadata working rep * chore: cleanup * fix: use tenacity instead of custom fns * feat: add representation and card to context if desired * feat: add semantically relevant observations * fix: pass all params to streaming, nonblocking streaming * feat: consolidate document saving, make working representation fetching much smarter * chore: add 100% test coverage of representation util * feat: basic dream infra * feat: dream queue item first pass * chore: fixes & cleanup from coderabbit * fix: dreams scheduled when new document count reaches a certain threshold * feat: wip: timed dreams (not working) * fix: test * fix: remove useless pyright ignore * fix: executing dreams * feat: dreaming * feat: [WIP] longmemeval bench * feat: add USE_PEER_CARD setting, fix longmem test driver * feat: get full working rep for dialectic in one swoop -- fix representation_from_documents to use the proper timestamp! * fix: timestamps for real, handle assistant qs in longmem * fix: remove old client, add batching to longmem * perf: remove duplicate detection, will move to background task * feat: track perf metrics on evals * feat: adjust deriver prompt to use peer_id, add question date to question, clean up deriver * fix: label metrics by task for better perf trace * chore: code review * feat: add efficiency score to longmem bench * chore: tuning and cleaning up eval * chore: bring in the big prompts * feat: add support for vllm client * feat: perf: bundle db calls in deriver and dialectic, increase max conns in docker db * feat: add merge-sessions flag to longmemeval, add SUMMARY_ENABLED flag * fix: COLLECT_METRICS default false * chore: display start/end message ids, don't include in metrics * fix: break large messages apart for eval * fix: only get/create collection when needed * feat: properly attribute documents with message id ranges and add session name column to documents * fix: revert move of get_or_create_collection (need for fkey) * fix: always get collection with peer name even if it's none * chore: coderabbit * fix: give peer card its own config, expand document schema, refactor get_context to be parallel, various cleanup chores and bugfixes * chore: refactor: reify observer/observed system across entire codebase, including db migration * refactor: cleanup code organization, make singletons where desired * refactor: replace embeddings store with representation manager * chore: coderabbit cleanup * feat: multi-db longmem harness * Merge branch 'main' into ben/multi-db-harness * [WIP] feat: add delete workspace endpoint, use in bench * chore: move excess logging to debug feat: improve metrics block logs to include more data fix: make longmem db deletion configurable * fix: [CRITICAL] use async genai client * chore: update core sdk, fix tests to use aio as well * fix: rollback prompt changes * chore: update version * fix: cleanup, coderabbit, wrap delete op in try/except --------- Co-authored-by: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> --- sdks/python/pyproject.toml | 4 +- sdks/python/src/honcho/async_client/client.py | 22 +- sdks/python/src/honcho/client.py | 22 +- sdks/typescript/bun.lock | 4 +- sdks/typescript/package.json | 4 +- sdks/typescript/src/client.ts | 14 + src/crud/__init__.py | 8 +- src/crud/peer.py | 2 +- src/crud/representation.py | 5 - src/crud/session.py | 8 +- src/crud/webhook.py | 4 +- src/crud/workspace.py | 120 ++++- src/deriver/consumer.py | 2 +- src/deriver/deriver.py | 25 +- src/deriver/enqueue.py | 2 +- src/deriver/queue_manager.py | 1 - src/dialectic/chat.py | 85 ++- src/dialectic/prompts.py | 17 +- src/embedding_client.py | 2 +- src/routers/messages.py | 7 +- src/routers/sessions.py | 16 +- src/routers/workspaces.py | 13 + src/utils/clients.py | 8 +- src/utils/logging.py | 38 +- src/utils/summarizer.py | 2 +- src/webhooks/webhook_delivery.py | 4 +- tests/bench/harness.py | 290 ++++++++-- tests/bench/longmem.py | 118 ++++- tests/conftest.py | 6 +- tests/crud/test_workspace.py | 498 ++++++++++++++++++ tests/routes/test_workspaces.py | 249 +++++++++ tests/utils/test_clients.py | 28 +- uv.lock | 24 +- 33 files changed, 1483 insertions(+), 169 deletions(-) create mode 100644 tests/crud/test_workspace.py diff --git a/sdks/python/pyproject.toml b/sdks/python/pyproject.toml index 9d8c7544..3132f09d 100644 --- a/sdks/python/pyproject.toml +++ b/sdks/python/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "honcho-ai" -version = "1.4.1" +version = "1.5.0" description = "Official DX Optimized Python SDK for Honcho" dynamic = ["readme"] license = "Apache-2.0" @@ -8,7 +8,7 @@ authors = [ { name = "Plastic Labs", email = "hello@plasticlabs.ai" }, ] dependencies = [ - "honcho-core>=1.5.0", + "honcho-core>=1.5.1", "httpx>=0.28.0, <1", "pydantic>=2.0.0, <3", "typing-extensions>=4.12.0; python_version < \"3.12\"", diff --git a/sdks/python/src/honcho/async_client/client.py b/sdks/python/src/honcho/async_client/client.py index bdc01fa8..2929d88f 100644 --- a/sdks/python/src/honcho/async_client/client.py +++ b/sdks/python/src/honcho/async_client/client.py @@ -8,7 +8,7 @@ from typing import Any, Literal import httpx from honcho_core import AsyncHoncho as AsyncHonchoCore from honcho_core import Honcho as HonchoCore -from honcho_core.types import DeriverStatus +from honcho_core.types import DeriverStatus, Workspace from honcho_core.types.workspaces.peer import Peer as PeerCore from honcho_core.types.workspaces.session import Session as SessionCore from honcho_core.types.workspaces.sessions.message import Message @@ -337,6 +337,26 @@ class AsyncHoncho(BaseModel): workspace_ids.append(workspace.id) return workspace_ids + @validate_call + async def delete_workspace( + self, + workspace_id: str = Field( + ..., min_length=1, description="ID of the workspace to delete" + ), + ) -> Workspace: + """ + Delete a workspace. + + Makes an async API call to delete the specified workspace. + + Args: + workspace_id: The ID of the workspace to delete + + Returns: + The deleted Workspace object + """ + return await self._client.workspaces.delete(workspace_id) + @validate_call async def search( self, diff --git a/sdks/python/src/honcho/client.py b/sdks/python/src/honcho/client.py index 777c5e4e..2c494ce6 100644 --- a/sdks/python/src/honcho/client.py +++ b/sdks/python/src/honcho/client.py @@ -6,7 +6,7 @@ from typing import Any, Literal import httpx from honcho_core import Honcho as HonchoCore -from honcho_core.types import DeriverStatus +from honcho_core.types import DeriverStatus, Workspace from honcho_core.types.workspaces.peer import Peer as PeerCore from honcho_core.types.workspaces.session import Session as SessionCore from honcho_core.types.workspaces.sessions.message import Message @@ -316,6 +316,26 @@ class Honcho(BaseModel): workspaces = self._client.workspaces.list(filters=filters) return [workspace.id for workspace in workspaces] + @validate_call + def delete_workspace( + self, + workspace_id: str = Field( + ..., min_length=1, description="ID of the workspace to delete" + ), + ) -> Workspace: + """ + Delete a workspace. + + Makes an API call to delete the specified workspace. + + Args: + workspace_id: The ID of the workspace to delete + + Returns: + The deleted Workspace object + """ + return self._client.workspaces.delete(workspace_id) + @validate_call def search( self, diff --git a/sdks/typescript/bun.lock b/sdks/typescript/bun.lock index b4ad630e..dcffc7b3 100644 --- a/sdks/typescript/bun.lock +++ b/sdks/typescript/bun.lock @@ -4,7 +4,7 @@ "": { "name": "@honcho-ai/sdk", "dependencies": { - "@honcho-ai/core": "^1.5.0", + "@honcho-ai/core": "^1.5.1", "@types/node": "^24.0.1", "zod": "4.0.0", }, @@ -108,7 +108,7 @@ "@cspotcode/source-map-support": ["@cspotcode/source-map-support@0.8.1", "", { "dependencies": { "@jridgewell/trace-mapping": "0.3.9" } }, "sha512-IchNf6dN4tHoMFIn/7OE8LWZ19Y6q/67Bmf6vnGREv8RSbBVb9LPJxEcnwrcwX6ixSvaiGoomAUvu4YSxXrVgw=="], - "@honcho-ai/core": ["@honcho-ai/core@1.5.0", "", { "dependencies": { "@types/node": "^18.11.18", "@types/node-fetch": "^2.6.4", "abort-controller": "^3.0.0", "agentkeepalive": "^4.2.1", "form-data-encoder": "1.7.2", "formdata-node": "^4.3.2", "node-fetch": "^2.6.7" } }, "sha512-kjYhCO0S9Ll3DfVpK8MetjY1PlD9FG6QGtudlMnSwfTWGzXFjL/OafpoE3SZ3c9BVY48JIbW9IttFUZFekv/xg=="], + "@honcho-ai/core": ["@honcho-ai/core@1.5.1", "", { "dependencies": { "@types/node": "^18.11.18", "@types/node-fetch": "^2.6.4", "abort-controller": "^3.0.0", "agentkeepalive": "^4.2.1", "form-data-encoder": "1.7.2", "formdata-node": "^4.3.2", "node-fetch": "^2.6.7" } }, "sha512-lbYtMTcL2AxdcIl5ZKenogeTlVMnE7buJWvAFOCLp0yQxcezyA/R9FPvLz2UGRApLsiplLNWhftML+3QBjIIJA=="], "@istanbuljs/load-nyc-config": ["@istanbuljs/load-nyc-config@1.1.0", "", { "dependencies": { "camelcase": "^5.3.1", "find-up": "^4.1.0", "get-package-type": "^0.1.0", "js-yaml": "^3.13.1", "resolve-from": "^5.0.0" } }, "sha512-VjeHSlIzpv/NyD3N0YuHfXOPDIixcA1q2ZV98wsMqcYlPmv2n3Yb2lYP9XMElnaFVXg5A7YLTeLu6V84uQDjmQ=="], diff --git a/sdks/typescript/package.json b/sdks/typescript/package.json index 803b9d23..8701165a 100644 --- a/sdks/typescript/package.json +++ b/sdks/typescript/package.json @@ -1,6 +1,6 @@ { "name": "@honcho-ai/sdk", - "version": "1.4.1", + "version": "1.5.0", "description": "Official DX Optimized TypeScript SDK for Honcho", "author": "Plastic Labs ", "license": "Apache-2.0", @@ -20,7 +20,7 @@ "test:coverage": "jest --coverage" }, "dependencies": { - "@honcho-ai/core": "^1.5.0", + "@honcho-ai/core": "^1.5.1", "@types/node": "^24.0.1", "zod": "4.0.0" }, diff --git a/sdks/typescript/src/client.ts b/sdks/typescript/src/client.ts index 0400381c..cd6ed695 100644 --- a/sdks/typescript/src/client.ts +++ b/sdks/typescript/src/client.ts @@ -314,6 +314,20 @@ export class Honcho { return ids } + /** + * Delete a workspace. + * + * Makes an API call to delete the specified workspace. + * + * @param workspaceId - The ID of the workspace to delete + * @returns Promise resolving to the deleted Workspace object + */ + async deleteWorkspace( + workspaceId: string + ): Promise>> { + return await this._client.workspaces.delete(workspaceId) + } + /** * Search for messages in the current workspace. * diff --git a/src/crud/__init__.py b/src/crud/__init__.py index f00a6ae1..69a4799e 100644 --- a/src/crud/__init__.py +++ b/src/crud/__init__.py @@ -44,7 +44,12 @@ from .webhook import ( get_or_create_webhook_endpoint, list_webhook_endpoints, ) -from .workspace import get_all_workspaces, get_or_create_workspace, update_workspace +from .workspace import ( + delete_workspace, + get_all_workspaces, + get_or_create_workspace, + update_workspace, +) __all__ = [ # Collection @@ -93,6 +98,7 @@ __all__ = [ "delete_webhook_endpoint", "list_webhook_endpoints", # Workspace + "delete_workspace", "get_or_create_workspace", "get_all_workspaces", "update_workspace", diff --git a/src/crud/peer.py b/src/crud/peer.py index 1e14aed7..cad234c8 100644 --- a/src/crud/peer.py +++ b/src/crud/peer.py @@ -166,7 +166,7 @@ async def update_peer( honcho_peer.configuration = peer.configuration await db.commit() - logger.info(f"Peer {peer_name} updated successfully") + logger.debug(f"Peer {peer_name} updated successfully") return honcho_peer diff --git a/src/crud/representation.py b/src/crud/representation.py index ed88bf4b..598d067f 100644 --- a/src/crud/representation.py +++ b/src/crud/representation.py @@ -322,11 +322,6 @@ class RepresentationManager: db, top_k=recent_observations, session_name=session_name ) - if not recent_docs: - logger.warning( - f"No observations for {self.observed} (observer: {self.observer}) found. Normal if brand-new peer." - ) - representation.merge_representation(Representation.from_documents(recent_docs)) return representation diff --git a/src/crud/session.py b/src/crud/session.py index 0e050a8b..b3adf591 100644 --- a/src/crud/session.py +++ b/src/crud/session.py @@ -117,7 +117,7 @@ async def get_or_create_session( except IntegrityError: await db.rollback() logger.debug( - f"Race condition detected for session: {session.name}, retrying get" + "Race condition detected for session: %s, retrying get", session.name ) if _retry: raise ConflictException( @@ -220,7 +220,7 @@ async def update_session( honcho_session.configuration = session.configuration await db.commit() - logger.info(f"Session {session_name} updated successfully") + logger.debug("Session %s updated successfully", session_name) return honcho_session @@ -257,7 +257,7 @@ async def delete_session( honcho_session.is_active = False await db.commit() - logger.info(f"Session {session_name} marked as inactive") + logger.debug("Session %s marked as inactive", session_name) return True @@ -358,7 +358,7 @@ async def clone_session( db.add(new_session_peer) await db.commit() - logger.info(f"Session {original_session_name} cloned successfully") + logger.debug("Session %s cloned successfully", original_session_name) return new_session diff --git a/src/crud/webhook.py b/src/crud/webhook.py index d2216cb2..1dcc29a8 100644 --- a/src/crud/webhook.py +++ b/src/crud/webhook.py @@ -58,7 +58,7 @@ async def get_or_create_webhook_endpoint( await db.commit() await db.refresh(webhook_endpoint) - logger.info(f"Webhook endpoint created: {webhook.url}") + logger.debug("Webhook endpoint created: %s", webhook.url) return schemas.WebhookEndpoint.model_validate(webhook_endpoint) @@ -112,4 +112,4 @@ async def delete_webhook_endpoint( await db.delete(endpoint) await db.commit() - logger.info(f"Webhook endpoint {endpoint_id} deleted") + logger.debug("Webhook endpoint %s deleted", endpoint_id) diff --git a/src/crud/workspace.py b/src/crud/workspace.py index 0f3e63b4..d9af943c 100644 --- a/src/crud/workspace.py +++ b/src/crud/workspace.py @@ -1,7 +1,7 @@ from logging import getLogger from typing import Any -from sqlalchemy import Select, select +from sqlalchemy import Select, delete, func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -38,7 +38,7 @@ async def get_or_create_workspace( if existing_workspace is not None: # Workspace already exists - logger.debug(f"Found existing workspace: {workspace.name}") + logger.debug("Found existing workspace: %s", workspace.name) return existing_workspace # Workspace doesn't exist, create a new one @@ -50,7 +50,7 @@ async def get_or_create_workspace( try: db.add(honcho_workspace) await db.commit() - logger.info(f"Workspace created successfully: {workspace.name}") + logger.debug("Workspace created successfully: %s", workspace.name) return honcho_workspace except IntegrityError: await db.rollback() @@ -134,5 +134,117 @@ async def update_workspace( honcho_workspace.configuration = workspace.configuration await db.commit() - logger.info(f"Workspace with id {honcho_workspace.id} updated successfully") + logger.debug("Workspace with id %s updated successfully", honcho_workspace.id) return honcho_workspace + + +async def delete_workspace(db: AsyncSession, workspace_name: str) -> schemas.Workspace: + """ + Delete a workspace. + + Args: + db: Database session + workspace_name: Name of the workspace + + Returns: + A snapshot of the deleted workspace as a Pydantic schema + """ + logger.warning("Deleting workspace %s", workspace_name) + stmt = select(models.Workspace).where(models.Workspace.name == workspace_name) + result = await db.execute(stmt) + honcho_workspace = result.scalar_one_or_none() + + if honcho_workspace is None: + logger.warning("Workspace %s not found", workspace_name) + raise ResourceNotFoundException() + + # Create a snapshot of the workspace data before deletion + workspace_snapshot = schemas.Workspace( + name=honcho_workspace.name, + h_metadata=honcho_workspace.h_metadata, + configuration=honcho_workspace.configuration, + created_at=honcho_workspace.created_at, + ) + + # order is important here. + # delete all active queue sessions referencing this workspace first (using work_unit_key parsing) + # then queue items referencing this workspace + + # then embeddings + # then documents + # then collections + # then messages + + # then webhook endpoints + # then session_peers + # then sessions + # then peers + # then workspace + + # Delete ActiveQueueSession entries first + # Work unit keys have format: {task_type}:{workspace_name}:{...} + # Extract workspace_name from position 2 (second component after splitting by ':') + try: + await db.execute( + delete(models.ActiveQueueSession).where( + func.split_part(models.ActiveQueueSession.work_unit_key, ":", 2) + == workspace_name + ) + ) + + # Then delete QueueItem entries + await db.execute( + delete(models.QueueItem).where( + func.split_part(models.QueueItem.work_unit_key, ":", 2) + == workspace_name + ) + ) + + await db.execute( + delete(models.MessageEmbedding).where( + models.MessageEmbedding.workspace_name == workspace_name + ) + ) + await db.execute( + delete(models.Document).where( + models.Document.workspace_name == workspace_name + ) + ) + await db.execute( + delete(models.Collection).where( + models.Collection.workspace_name == workspace_name + ) + ) + await db.execute( + delete(models.Message).where( + models.Message.workspace_name == workspace_name + ) + ) + + await db.execute( + delete(models.WebhookEndpoint).where( + models.WebhookEndpoint.workspace_name == workspace_name + ) + ) + await db.execute( + delete(models.SessionPeer).where( + models.SessionPeer.workspace_name == workspace_name + ) + ) + await db.execute( + delete(models.Session).where( + models.Session.workspace_name == workspace_name + ) + ) + await db.execute( + delete(models.Peer).where(models.Peer.workspace_name == workspace_name) + ) + await db.delete(honcho_workspace) + await db.commit() + logger.debug("Workspace %s deleted", workspace_name) + except Exception as e: + logger.error("Failed to delete workspace %s: %s", workspace_name, e) + await db.rollback() + raise e + + return workspace_snapshot diff --git a/src/deriver/consumer.py b/src/deriver/consumer.py index 59aab370..20e0b1b0 100644 --- a/src/deriver/consumer.py +++ b/src/deriver/consumer.py @@ -58,7 +58,7 @@ async def process_item(task_type: str, queue_payload: dict[str, Any]) -> None: message_public_id = validated.message_public_id if not message_public_id: - logger.info( + logger.debug( "Fetching message public ID for message %s", validated.message_id ) async with tracked_db(operation_name="summary_fallback") as db: diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index 4be32527..15900e86 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -17,7 +17,7 @@ from src.utils.logging import ( accumulate_metric, conditional_observe, log_performance_metrics, - log_representation, + # log_representation, ) from src.utils.peer_card import PeerCardQuery from src.utils.representation import PromptRepresentation, Representation @@ -65,6 +65,8 @@ async def critical_analysis_call( json_mode=True, stop_seqs=[" \n", "\n\n\n\n"], thinking_budget_tokens=settings.DERIVER.THINKING_BUDGET_TOKENS, + reasoning_effort="minimal", + verbosity="medium", enable_retry=True, retry_attempts=3, ) @@ -161,13 +163,6 @@ async def process_representation_tasks_batch( observer=observer, observed=observed, ) - if speaker_peer_card is None: - logger.warning( - "No peer card found for %s. Normal if brand-new peer.", - observed, - ) - else: - logger.info("Using peer card: %s", speaker_peer_card) else: speaker_peer_card = None @@ -228,7 +223,7 @@ async def process_representation_tasks_batch( "ms", ) - logger.info( + logger.debug( "Using working representation with %s explicit, %s deductive observations", len(working_representation.explicit), len(working_representation.deductive), @@ -262,7 +257,7 @@ async def process_representation_tasks_batch( ) # Display final observations in a beautiful tree - log_representation(final_observations) + # log_representation(final_observations) # Calculate and log overall timing overall_duration = (time.perf_counter() - overall_start) * 1000 @@ -431,10 +426,9 @@ class CertaintyReasoner: """ try: response = await peer_card_call(old_peer_card, new_observations) - logger.info("Jettisoned notes from peer card: %s", response.notes) new_peer_card = response.card if not new_peer_card: - logger.info("No changes to peer card") + # no changes return # even with a dedicated notes field, we still need to prune notes out of the card new_peer_card = [ @@ -442,7 +436,12 @@ class CertaintyReasoner: for observation in new_peer_card if not observation.lower().startswith(("note", "notes")) ] - logger.info("New peer card: %s", new_peer_card) + accumulate_metric( + f"deriver_{self.ctx[-1].id}_{self.observer}", + "new_peer_card", + "\n".join(new_peer_card), + "blob", + ) async with tracked_db("deriver.update_peer_card") as db: await crud.set_peer_card( db, diff --git a/src/deriver/enqueue.py b/src/deriver/enqueue.py index ee66a6be..a9eb736e 100644 --- a/src/deriver/enqueue.py +++ b/src/deriver/enqueue.py @@ -343,7 +343,7 @@ async def generate_queue_records( observed, ) - logger.info( + logger.debug( "message %s from %s created %s queue items", message_id, observed, diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 1b488b61..fbe463a0 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -161,7 +161,6 @@ class QueueManager: ) ) await db.commit() - logger.info("Cleanup completed successfully") except Exception as e: logger.error(f"Error during cleanup: {str(e)}") if settings.SENTRY.ENABLED: diff --git a/src/dialectic/chat.py b/src/dialectic/chat.py index cfca960b..b134b043 100644 --- a/src/dialectic/chat.py +++ b/src/dialectic/chat.py @@ -197,12 +197,11 @@ async def chat( "dialectic_model": settings.DIALECTIC.MODEL, } ) - logger.info( - "Received query:\n'%s'\nobserver: %s, observed: %s%s\n", - query, - observer, - observed, - f", session: {session_name}" if session_name else "", + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "query", + f"{query}\n\nobserver: {observer}\nobserved: {observed}\n{f'session: {session_name}' if session_name else ''}", + "blob", ) start_time = time.perf_counter() @@ -226,19 +225,28 @@ async def chat( working_rep_duration, "ms", ) - logger.info( - "Retrieved working representation with %s explicit, %s deductive observations", + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "working_rep_explicit", len(working_representation.explicit), + "count", + ) + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "working_rep_deductive", len(working_representation.deductive), + "count", ) working_representation_str = str(working_representation) context_window_size -= max(0, estimate_tokens(working_representation_str)) - logger.info( - "Constructed working representation:\n%s\n", + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "working_rep", working_representation_str, + "blob", ) # 2. Recent conversation history -------------------------------------------- @@ -252,14 +260,18 @@ async def chat( token_limit=context_window_size, include_summary=True, ) - logger.info("Retrieved recent conversation history") else: recent_history = None - logger.info( - "Query is not session-scoped, skipping recent conversation history" - ) - context_window_size -= max(0, estimate_tokens(recent_history or "")) + recent_history_tokens = estimate_tokens(recent_history or "") + context_window_size -= max(0, recent_history_tokens) + + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "recent_history_tokens", + recent_history_tokens, + "tokens", + ) accumulate_metric( f"dialectic_chat_{dialectic_chat_uuid}", @@ -282,9 +294,25 @@ async def chat( observed_peer_card = None if observed_peer_card: - logger.info("Retrieved peer cards:\n%s\n%s", peer_card, observed_peer_card) + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "peer_card", + "\n".join(peer_card) if peer_card else "", + "blob", + ) + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "observed_peer_card", + "\n".join(observed_peer_card), + "blob", + ) else: - logger.info("Retrieved peer card:\n%s", peer_card) + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "peer_card", + "\n".join(peer_card) if peer_card else "", + "blob", + ) else: peer_card = None observed_peer_card = None @@ -292,6 +320,20 @@ async def chat( # 4. Dialectic call -------------------------------------------------------- dialectic_call_start_time = time.perf_counter() if stream: + elapsed = (time.perf_counter() - start_time) * 1000 + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "response", + "(no logged response, streaming=true)", + "blob", + ) + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "duration_to_streaming", + elapsed, + "ms", + ) + log_performance_metrics("dialectic_chat", dialectic_chat_uuid) return await dialectic_stream( query, working_representation_str, @@ -312,6 +354,12 @@ async def chat( observed=observed, ) dialectic_call_duration = (time.perf_counter() - dialectic_call_start_time) * 1000 + accumulate_metric( + f"dialectic_chat_{dialectic_chat_uuid}", + "response", + response, + "blob", + ) accumulate_metric( f"dialectic_chat_{dialectic_chat_uuid}", "dialectic_call", @@ -326,5 +374,4 @@ async def chat( ) log_performance_metrics("dialectic_chat", dialectic_chat_uuid) - # Convert AnthropicCallResponse to string for compatibility - return str(response) + return response diff --git a/src/dialectic/prompts.py b/src/dialectic/prompts.py index db4a9392..7d1fbb55 100644 --- a/src/dialectic/prompts.py +++ b/src/dialectic/prompts.py @@ -48,6 +48,16 @@ The user's known biographical information: If the user's name or nickname is known, exclusively refer to them by that name. """ + recent_conversation_history_section = ( + f""" + +{recent_conversation_history} + +""" + if recent_conversation_history + else "" + ) + return c( f""" 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. @@ -122,11 +132,10 @@ Provide a natural language response that: {query_target} - -{recent_conversation_history} - - {query} + +{recent_conversation_history_section} + {working_representation} """ ) diff --git a/src/embedding_client.py b/src/embedding_client.py index 6e0a1195..08c61758 100644 --- a/src/embedding_client.py +++ b/src/embedding_client.py @@ -350,7 +350,7 @@ class EmbeddingClient: self._instance = _EmbeddingClient( api_key=api_key, provider=provider ) - logger.info( + logger.debug( f"Initialized embedding client with provider: {provider}" ) diff --git a/src/routers/messages.py b/src/routers/messages.py index 36f32b7a..9c858c8e 100644 --- a/src/routers/messages.py +++ b/src/routers/messages.py @@ -139,8 +139,9 @@ async def create_messages_with_file( ] background_tasks.add_task(enqueue, payloads) - logger.info( - f"Batch of {len(created_messages)} messages created from file uploads and queued for processing" + logger.debug( + "Batch of %s messages created from file uploads and queued for processing", + len(created_messages), ) prometheus.MESSAGES_CREATED.labels( workspace_name=workspace_id, @@ -218,7 +219,7 @@ async def update_message( session_name=session_id, message_id=message_id, ) - logger.info(f"Message {message_id} updated successfully") + logger.debug("Message %s updated successfully", message_id) return updated_message except ValueError as e: logger.warning(f"Failed to update message {message_id}: {str(e)}") diff --git a/src/routers/sessions.py b/src/routers/sessions.py index fafb310e..0633777d 100644 --- a/src/routers/sessions.py +++ b/src/routers/sessions.py @@ -202,7 +202,7 @@ async def update_session( updated_session = await crud.update_session( db, workspace_name=workspace_id, session_name=session_id, session=session ) - logger.info(f"Session {session_id} updated successfully") + logger.debug("Session %s updated successfully", session_id) return updated_session except ValueError as e: logger.warning(f"Failed to update session {session_id}: {str(e)}") @@ -225,7 +225,7 @@ async def delete_session( await crud.delete_session( db, workspace_name=workspace_id, session_name=session_id ) - logger.info(f"Session {session_id} deleted successfully") + logger.debug("Session %s deleted successfully", session_id) return {"message": "Session deleted successfully"} except ValueError as e: logger.warning(f"Failed to delete session {session_id}: {str(e)}") @@ -256,7 +256,7 @@ async def clone_session( original_session_name=session_id, cutoff_message_id=message_id, ) - logger.info(f"Session {session_id} cloned successfully") + logger.debug("Session %s cloned successfully", session_id) return cloned_session except ValueError as e: logger.warning(f"Failed to clone session {session_id}: {str(e)}") @@ -288,7 +288,7 @@ async def add_peers_to_session( ), workspace_name=workspace_id, ) - logger.info(f"Added peers to session {session_id} successfully") + logger.debug("Added peers to session %s successfully", session_id) return session except ValueError as e: logger.warning(f"Failed to add peers to session {session_id}: {str(e)}") @@ -324,7 +324,7 @@ async def set_session_peers( session=schemas.SessionCreate(name=session_id), workspace_name=workspace_id, ) - logger.info(f"Set peers for session {session_id} successfully") + logger.debug("Set peers for session %s successfully", session_id) return session except ValueError as e: logger.warning(f"Failed to set peers for session {session_id}: {str(e)}") @@ -360,7 +360,7 @@ async def remove_peers_from_session( session=schemas.SessionCreate(name=session_id), workspace_name=workspace_id, ) - logger.info(f"Removed peers from session {session_id} successfully") + logger.debug("Removed peers from session %s successfully", session_id) return session except ValueError as e: logger.warning(f"Failed to remove peers from session {session_id}: {str(e)}") @@ -411,8 +411,8 @@ async def set_peer_config( peer_name=peer_id, config=config, ) - logger.info( - f"Set peer config for {peer_id} in session {session_id} successfully" + logger.debug( + "Set peer config for %s in session %s successfully", peer_id, session_id ) return Response(status_code=200) except ValueError as e: diff --git a/src/routers/workspaces.py b/src/routers/workspaces.py index 2c10b5d0..7138f6fd 100644 --- a/src/routers/workspaces.py +++ b/src/routers/workspaces.py @@ -92,6 +92,19 @@ async def update_workspace( return honcho_workspace +@router.delete( + "/{workspace_id}", + response_model=schemas.Workspace, + dependencies=[Depends(require_auth(workspace_name="workspace_id"))], +) +async def delete_workspace( + workspace_id: str = Path(..., description="ID of the workspace to delete"), + db: AsyncSession = db, +): + """Delete a Workspace""" + return await crud.delete_workspace(db, workspace_name=workspace_id) + + @router.post( "/{workspace_id}/search", response_model=list[schemas.Message], diff --git a/src/utils/clients.py b/src/utils/clients.py index d5f1e235..dd30f02e 100644 --- a/src/utils/clients.py +++ b/src/utils/clients.py @@ -370,8 +370,6 @@ async def honcho_llm_call_inner( "model": params["model"], "messages": params["messages"], } - if stop_seqs: - openai_params["stop"] = stop_seqs if "gpt-5" in model: openai_params["max_completion_tokens"] = params["max_tokens"] if reasoning_effort: @@ -399,6 +397,8 @@ async def honcho_llm_call_inner( "schema": response_model.model_json_schema(), }, } + if stop_seqs: + openai_params["stop"] = stop_seqs response: ChatCompletion = await client.chat.completions.create( # pyright: ignore **openai_params ) @@ -505,7 +505,7 @@ async def honcho_llm_call_inner( case genai.Client(): if response_model is None: gemini_response: GenerateContentResponse = ( - client.models.generate_content( + await client.aio.models.generate_content( model=model, contents=prompt, config={ @@ -537,7 +537,7 @@ async def honcho_llm_call_inner( ) else: - gemini_response = client.models.generate_content( + gemini_response = await client.aio.models.generate_content( model=model, contents=prompt, config={ diff --git a/src/utils/logging.py b/src/utils/logging.py index 8d28e1db..ca615899 100644 --- a/src/utils/logging.py +++ b/src/utils/logging.py @@ -10,8 +10,10 @@ from typing import Any from fastapi import Request from rich import box -from rich.console import Console +from rich.console import Console, Group, RenderableType +from rich.panel import Panel from rich.table import Table +from rich.text import Text from rich.tree import Tree from src.config import settings @@ -157,17 +159,23 @@ def log_performance_metrics( if COLLECT_METRICS_LOCAL: append_metrics_to_file(task_slug, task_name, metrics) + # Remove metrics with "blob" unit type. They get printed separately below the table. + blob_metrics: list[tuple[str, str | int | float, str]] = [] + non_blob_metrics: list[tuple[str, str | int | float, str]] = [] + for metric in metrics: + (blob_metrics if metric[2] == "blob" else non_blob_metrics).append(metric) + table = Table( - title=f"{title} - {task_name}", show_header=True, header_style="bold green", - box=box.ROUNDED, + box=None, + padding=(0, 1), ) table.add_column("Metric", style="cyan", width=30) table.add_column("Value", justify="right", style="yellow", width=15) table.add_column("Unit", style="dim", width=8) - for metric, value, unit in metrics: + for metric, value, unit in non_blob_metrics: if unit == "ms": formatted_value = f"{value:.0f}" elif unit == "s": @@ -177,9 +185,25 @@ def log_performance_metrics( table.add_row(metric.replace("_", " ").title(), formatted_value, unit) - if metrics: - console.print(table) - console.print() + # Build content for the panel + content_items: list[RenderableType] = [table] + + if blob_metrics: + content_items.append(Text("")) # Empty line separator + for metric, value, _unit in blob_metrics: + content_items.append(Text(f"{metric}:", style="bold cyan")) + content_items.append(Text(str(value))) + + panel = Panel( + Group(*content_items), + title=f"[bold green]{title} - {task_name}[/]", + box=box.ROUNDED, + padding=(1, 2), + width=80, + ) + + console.print(panel) + console.print() def normalize_template_path(path: str) -> str: diff --git a/src/utils/summarizer.py b/src/utils/summarizer.py index e99bd14b..51c71629 100644 --- a/src/utils/summarizer.py +++ b/src/utils/summarizer.py @@ -639,7 +639,7 @@ async def get_session_context( messages_tokens = token_limit - latest_short_summary["token_count"] messages_start_id = latest_short_summary["message_id"] else: - logger.warning( + logger.debug( "No summary available for get_context call with token limit %s, returning empty string. Normal if brand-new session. long_summary_len: %s, short_summary_len: %s", token_limit, long_len, diff --git a/src/webhooks/webhook_delivery.py b/src/webhooks/webhook_delivery.py index 4573c79c..d9587009 100644 --- a/src/webhooks/webhook_delivery.py +++ b/src/webhooks/webhook_delivery.py @@ -23,7 +23,7 @@ async def deliver_webhook(db: AsyncSession, payload: WebhookPayload) -> None: try: webhook_urls = await _get_webhook_urls(db, payload.workspace_name) if not webhook_urls: - logger.info( + logger.debug( f"No webhook endpoints for workspace {payload.workspace_name}, skipping." ) return @@ -59,7 +59,7 @@ async def deliver_webhook(db: AsyncSession, payload: WebhookPayload) -> None: for url, result in zip(webhook_urls, results, strict=False): if isinstance(result, httpx.Response): if 200 <= result.status_code < 300: - logger.info( + logger.debug( f"Successfully delivered webhook {payload.event_type} to {url}" ) else: diff --git a/tests/bench/harness.py b/tests/bench/harness.py index 321c79a3..09bb7840 100755 --- a/tests/bench/harness.py +++ b/tests/bench/harness.py @@ -28,27 +28,28 @@ class HonchoHarness: Orchestrates running Honcho with a Docker database for development. """ - def __init__(self, db_port: int, project_root: Path) -> None: + def __init__( + self, db_port: int, api_port: int, project_root: Path, instance_id: int = 0 + ) -> None: """ - Initialize the harness with database port and project root. + Initialize the harness with database port, API port, and project root. Args: db_port: Port for the PostgreSQL database + api_port: Port for the FastAPI server project_root: Path to the Honcho project root + instance_id: Instance identifier for pool management """ self.db_port: int = db_port + self.api_port: int = api_port self.project_root: Path = project_root + self.instance_id: int = instance_id self.temp_dir: Path | None = None self.docker_compose_file: Path | None = None self.processes: list[tuple[str, subprocess.Popen[str]]] = [] self.env_file_backup: Path | None = None self.output_threads: list[threading.Thread] = [] - # Set environment variables in the current process - # This ensures they're inherited by all subprocesses - for key, value in self.get_database_env_vars().items(): - os.environ[key] = value - def create_temp_docker_compose(self) -> Path: """ Create a temporary docker-compose.yml with the specified database port. @@ -67,6 +68,14 @@ class HonchoHarness: # Add a unique project name to avoid conflicts compose_data["name"] = f"honcho_harness_{self.db_port}" + # remove init.sql mount since we use provision_db.py for setup + if "volumes" in compose_data["services"]["database"]: + compose_data["services"]["database"]["volumes"] = [ + vol + for vol in compose_data["services"]["database"]["volumes"] + if "init.sql" not in vol + ] + # Create temporary file self.temp_dir = Path(tempfile.mkdtemp(prefix="honcho_harness_")) self.docker_compose_file = self.temp_dir / "docker-compose.yml" @@ -200,32 +209,35 @@ class HonchoHarness: """ Provision the database using the provision_db.py script. """ - print("Provisioning database...") + print(f"[Instance {self.instance_id}] Provisioning database...") - # Run the provision script + # Run the provision script with explicit environment variables provision_script = self.project_root / "scripts" / "provision_db.py" + env = os.environ.copy() + env.update(self.get_database_env_vars()) + result = subprocess.run( [sys.executable, str(provision_script)], cwd=self.project_root, capture_output=True, text=True, + env=env, ) if result.returncode != 0: print(f"Failed to provision database: {result.stderr}") sys.exit(1) - print("Database provisioned successfully") + print(f"[Instance {self.instance_id}] Database provisioned successfully") def verify_empty_database(self) -> None: """ Verify that the database is empty with no workspaces and an empty queue. """ - try: import psycopg - # Connect to the database + # Connect to the database using instance-specific connection string conn_string = ( f"postgresql://testuser:testpwd@localhost:{self.db_port}/honcho" ) @@ -245,14 +257,20 @@ class HonchoHarness: # Report results if workspace_count != 0 or queue_count != 0: - print("❌ Database verification failed: Database is not empty") + print( + f"[Instance {self.instance_id}] ❌ Database verification failed: Database is not empty" + ) print( "This may indicate an issue with the database provisioning or cleanup." ) sys.exit(1) + print( + f"[Instance {self.instance_id}] ✅ Database verification passed: Database is empty" + ) + except Exception as e: - print(f"❌ Error verifying database: {e}") + print(f"[Instance {self.instance_id}] ❌ Error verifying database: {e}") print("Unable to verify database state. Continuing anyway...") def start_fastapi_server(self) -> subprocess.Popen[str]: @@ -262,7 +280,13 @@ class HonchoHarness: Returns: Process object for the FastAPI server """ - print("Starting FastAPI server...") + print( + f"[Instance {self.instance_id}] Starting FastAPI server on port {self.api_port}..." + ) + + # Create environment with instance-specific database connection + env = os.environ.copy() + env.update(self.get_database_env_vars()) process = subprocess.Popen( [ @@ -273,7 +297,7 @@ class HonchoHarness: "--host", "0.0.0.0", "--port", - "8000", + str(self.api_port), "--no-access-log", "--workers", "1", @@ -284,9 +308,10 @@ class HonchoHarness: text=True, bufsize=0, universal_newlines=True, + env=env, ) - self.processes.append(("FastAPI Server", process)) + self.processes.append((f"FastAPI [{self.instance_id}]", process)) return process def start_deriver(self) -> subprocess.Popen[str]: @@ -296,7 +321,11 @@ class HonchoHarness: Returns: Process object for the deriver """ - print("Starting deriver...") + print(f"[Instance {self.instance_id}] Starting deriver...") + + # Create environment with instance-specific database connection + env = os.environ.copy() + env.update(self.get_database_env_vars()) process = subprocess.Popen( [sys.executable, "-m", "src.deriver"], @@ -306,9 +335,10 @@ class HonchoHarness: text=True, bufsize=0, universal_newlines=True, + env=env, ) - self.processes.append(("Deriver", process)) + self.processes.append((f"[{self.instance_id}]", process)) return process def stream_process_output(self, name: str, process: subprocess.Popen[str]) -> None: @@ -357,16 +387,20 @@ class HonchoHarness: Returns: True if server is ready, False otherwise """ - print("Waiting for FastAPI server to be ready...") + print( + f"[Instance {self.instance_id}] Waiting for FastAPI server to be ready..." + ) start_time = time.time() while time.time() - start_time < timeout: try: import requests - response = requests.get("http://localhost:8000/docs", timeout=5) + response = requests.get( + f"http://localhost:{self.api_port}/docs", timeout=5 + ) if response.status_code == 200: - print("FastAPI server is ready!") + print(f"[Instance {self.instance_id}] FastAPI server is ready!") return True except Exception: pass @@ -459,12 +493,16 @@ except Exception as e: with open(script_file, "w") as f: f.write(config_script) - # Run the script + # Run the script with instance-specific environment + env = os.environ.copy() + env.update(self.get_database_env_vars()) + result = subprocess.run( [sys.executable, str(script_file)], cwd=self.project_root, capture_output=True, text=True, + env=env, ) if result.returncode == 0: @@ -594,10 +632,10 @@ except Exception as e: _deriver_process = self.start_deriver() print("\n" + "=" * 60) - print("🎉 Honcho is running!") + print(f"🎉 Honcho Instance {self.instance_id} is running!") print(f"📊 Database: localhost:{self.db_port}") - print("🌐 API Server: http://localhost:8000") - print("📚 API Docs: http://localhost:8000/docs") + print(f"🌐 API Server: http://localhost:{self.api_port}") + print(f"📚 API Docs: http://localhost:{self.api_port}/docs") print("🔄 Deriver: Running") print("=" * 60) print("Press Ctrl+C to stop all services") @@ -629,6 +667,164 @@ except Exception as e: self.cleanup() +class HonchoHarnessPool: + """ + Manages a pool of HonchoHarness instances for parallel testing. + """ + + def __init__( + self, pool_size: int, base_db_port: int, base_api_port: int, project_root: Path + ) -> None: + """ + Initialize a pool of Honcho harnesses. + + Args: + pool_size: Number of Honcho instances to create + base_db_port: Base port for PostgreSQL databases (each instance gets base + instance_id) + base_api_port: Base port for FastAPI servers (each instance gets base + instance_id) + project_root: Path to the Honcho project root + """ + self.pool_size: int = pool_size + self.base_db_port: int = base_db_port + self.base_api_port: int = base_api_port + self.project_root: Path = project_root + self.harnesses: list[HonchoHarness] = [] + + # Create all harness instances + for i in range(pool_size): + harness = HonchoHarness( + db_port=base_db_port + i, + api_port=base_api_port + i, + project_root=project_root, + instance_id=i, + ) + self.harnesses.append(harness) + + def run(self) -> None: + """ + Run all Honcho harnesses in the pool. + """ + try: + print(f"\n{'=' * 80}") + print(f"Starting Honcho Pool with {self.pool_size} instances") + print(f"{'=' * 80}\n") + + # Backup existing .env and copy test .env file + # This provides API keys while we override DB settings via environment variables + if self.harnesses: + self.harnesses[0].backup_env_file() + # Copy .env file from tests/bench to get API keys + shutil.copy( + self.project_root / "tests" / "bench" / ".env", + self.project_root / ".env", + ) + + # Remove DB_CONNECTION_URI from .env to ensure env vars take precedence + env_file = self.project_root / ".env" + if env_file.exists(): + with open(env_file) as f: + lines = f.readlines() + with open(env_file, "w") as f: + for line in lines: + # Skip DB_CONNECTION_URI lines + if not line.strip().startswith("DB_CONNECTION_URI"): + f.write(line) + + # Start all harnesses + for harness in self.harnesses: + print(f"\n--- Starting Instance {harness.instance_id} ---") + + # Create temporary docker-compose.yml + harness.create_temp_docker_compose() + + # Create an empty .env file in temp directory + if harness.temp_dir and harness.temp_dir.exists(): + (harness.temp_dir / ".env").touch() + else: + raise Exception( + f"Temporary directory does not exist for instance {harness.instance_id}" + ) + + # Start database + harness.start_database() + + # Wait for database to be ready + if not harness.wait_for_database(): + print( + f"Database failed to start for instance {harness.instance_id}. Exiting." + ) + sys.exit(1) + + # Provision database + harness.provision_database() + + # Verify database is empty + harness.verify_empty_database() + + # Start FastAPI server + harness.start_fastapi_server() + + # Wait for FastAPI to be ready + if not harness.wait_for_fastapi(): + print( + f"FastAPI server failed to start for instance {harness.instance_id}. Exiting." + ) + sys.exit(1) + + # Start deriver + harness.start_deriver() + + # Start output streaming threads + for name, process in harness.processes: + thread = threading.Thread( + target=harness.stream_process_output, + args=(name, process), + daemon=True, + ) + thread.start() + harness.output_threads.append(thread) + + print(f"✅ Instance {harness.instance_id} is ready!") + + # Print summary + print(f"\n{'=' * 80}") + print(f"🎉 All {self.pool_size} Honcho instances are running!") + print(f"{'=' * 80}") + for harness in self.harnesses: + print(f"\nInstance {harness.instance_id}:") + print(f" 📊 Database: localhost:{harness.db_port}") + print(f" 🌐 API Server: http://localhost:{harness.api_port}") + print(f" 📚 API Docs: http://localhost:{harness.api_port}/docs") + print(f"\n{'=' * 80}") + print("Press Ctrl+C to stop all services") + print(f"{'=' * 80}\n") + + # Monitor all processes for unexpected termination + while True: + for harness in self.harnesses: + for name, process in harness.processes: + if process.poll() is not None: + print(f"❌ {name} has stopped unexpectedly") + return + time.sleep(1) + + except KeyboardInterrupt: + print("\n🛑 Received interrupt signal") + except Exception as e: + print(f"❌ Error: {e}") + finally: + self.cleanup() + + def cleanup(self) -> None: + """ + Clean up all harnesses in the pool. + """ + print("\nCleaning up pool...") + for harness in self.harnesses: + print(f"\n--- Cleaning up Instance {harness.instance_id} ---") + harness.cleanup() + + def main(): """ Main entry point for the Honcho harness. @@ -639,6 +835,7 @@ def main(): epilog=""" Examples: %(prog)s --port 5433 # Run with database on port 5433 + %(prog)s --pool-size 4 # Run pool of 4 instances (ports 5433-5436, APIs 8000-8003) %(prog)s --port 5434 --project-root /path/to/honcho # Custom project root """, ) @@ -647,7 +844,21 @@ Examples: "--port", type=int, default=5433, - help="Port for the PostgreSQL database (default: 5433)", + help="Base port for the PostgreSQL database (default: 5433)", + ) + + parser.add_argument( + "--api-port", + type=int, + default=8000, + help="Base port for the FastAPI server (default: 8000)", + ) + + parser.add_argument( + "--pool-size", + type=int, + default=1, + help="Number of Honcho instances to run in parallel (default: 1)", ) parser.add_argument( @@ -659,6 +870,11 @@ Examples: args = parser.parse_args() + # Validate pool size + if args.pool_size <= 0: + print(f"Error: Pool size must be positive, got {args.pool_size}") + sys.exit(1) + # Validate project root if not (args.project_root / "src" / "main.py").exists(): print( @@ -679,9 +895,23 @@ Examples: print(f"Error: Required file {file_path} not found in {args.project_root}") sys.exit(1) - # Create and run the harness - harness = HonchoHarness(args.port, args.project_root) - harness.run() + # Create and run the harness or pool + if args.pool_size > 1: + pool = HonchoHarnessPool( + pool_size=args.pool_size, + base_db_port=args.port, + base_api_port=args.api_port, + project_root=args.project_root, + ) + pool.run() + else: + harness = HonchoHarness( + db_port=args.port, + api_port=args.api_port, + project_root=args.project_root, + instance_id=0, + ) + harness.run() if __name__ == "__main__": diff --git a/tests/bench/longmem.py b/tests/bench/longmem.py index 05d2b63e..ae789333 100644 --- a/tests/bench/longmem.py +++ b/tests/bench/longmem.py @@ -38,10 +38,12 @@ Optional arguments: ``` --anthropic-api-key: Anthropic API key for response judging (can be set in .env as LLM_ANTHROPIC_API_KEY or provided as an argument) --timeout: Timeout for deriver queue to empty in seconds (default: 10 minutes) ---honcho-url: URL of the running Honcho instance (default: http://localhost:8000) +--base-api-port: Base port for Honcho API instances (default: 8000) +--pool-size: Number of Honcho instances in the pool (default: 1) --batch-size: Number of questions to run concurrently in each batch (default: 10) --json-output: Path to write JSON summary results for analytics (if not provided, creates timestamped file in tests/bench/eval_results) --merge-sessions: Merge all sessions within a question into a single session (default: False) +--cleanup-workspace: Delete workspace after executing each question (default: False) ``` ## Other notes @@ -115,26 +117,32 @@ class LongMemEvalRunner: def __init__( self, - honcho_url: str = "http://localhost:8000", + base_api_port: int = 8000, + pool_size: int = 1, anthropic_api_key: str | None = None, timeout_seconds: int | None = None, merge_sessions: bool = False, + cleanup_workspace: bool = False, ): """ Initialize the test runner. Args: - honcho_url: URL of the running Honcho instance + base_api_port: Base port for Honcho API instances (default: 8000) + pool_size: Number of Honcho instances in the pool (default: 1) anthropic_api_key: Anthropic API key for judging responses timeout_seconds: Timeout for deriver queue in seconds merge_sessions: If True, merge all sessions within a question into one session + cleanup_workspace: If True, delete workspace after executing question (default: False) """ - self.honcho_url: str = honcho_url + self.base_api_port: int = base_api_port + self.pool_size: int = pool_size self.anthropic_api_key: str | None = anthropic_api_key self.timeout_seconds: int = ( timeout_seconds if timeout_seconds is not None else 10000 ) self.merge_sessions: bool = merge_sessions + self.cleanup_workspace: bool = cleanup_workspace # Initialize metrics collector self.metrics_collector: MetricsCollector = MetricsCollector() @@ -162,6 +170,20 @@ class LongMemEvalRunner: raise ValueError("LLM_ANTHROPIC_API_KEY is not set") self.anthropic_client = AsyncAnthropic(api_key=api_key) + def get_honcho_url_for_index(self, question_index: int) -> str: + """ + Get the Honcho URL for a given question index using round-robin distribution. + + Args: + question_index: Index of the question in the test file + + Returns: + URL of the Honcho instance to use for this question + """ + instance_id = question_index % self.pool_size + port = self.base_api_port + instance_id + return f"http://localhost:{port}" + def _format_duration(self, total_seconds: float) -> str: """Format a duration in seconds into a human-readable string. @@ -201,7 +223,20 @@ class LongMemEvalRunner: for session_messages in haystack_sessions: for msg in session_messages: content = msg.get("content", "") - total_tokens += len(tokenizer.encode(content)) + try: + total_tokens += len( + tokenizer.encode( + content, + disallowed_special=( + tokenizer.special_tokens_set - {"<|endoftext|>"} + ), + ) + ) + except Exception: + total_tokens += len(content) // 4 + self.logger.warning( + f"Error tokenizing content. Using rough estimate of {len(content) // 4} tokens" + ) return total_tokens @@ -281,12 +316,15 @@ class LongMemEvalRunner: with open(test_file) as f: return json.load(f) - async def create_honcho_client(self, workspace_id: str) -> AsyncHoncho: + async def create_honcho_client( + self, workspace_id: str, honcho_url: str + ) -> AsyncHoncho: """ Create a Honcho client for a specific workspace. Args: workspace_id: Workspace ID for the test + honcho_url: URL of the Honcho instance Returns: AsyncHoncho client instance @@ -294,7 +332,7 @@ class LongMemEvalRunner: return AsyncHoncho( environment="local", workspace_id=workspace_id, - base_url=self.honcho_url, + base_url=honcho_url, ) async def wait_for_deriver_queue_empty( @@ -402,12 +440,15 @@ Evaluate whether the actual response correctly answers the question based on the "reasoning": f"Fallback string matching due to error: {'Match found' if is_correct else 'No match found'}", } - async def execute_question(self, question_data: dict[str, Any]) -> TestResult: + async def execute_question( + self, question_data: dict[str, Any], honcho_url: str + ) -> TestResult: """ Execute a single longmemeval question. Args: question_data: Dictionary containing question data + honcho_url: URL of the Honcho instance to use Returns: Test execution results @@ -428,10 +469,11 @@ Evaluate whether the actual response correctly answers the question based on the ) output_lines.append(f"Question: {question_with_date}") output_lines.append(f"Expected: {expected_answer}") + output_lines.append(f"Using Honcho instance: {honcho_url}") # Create workspace for this question workspace_id = f"{question_id}_{question_type}" - honcho_client = await self.create_honcho_client(workspace_id) + honcho_client = await self.create_honcho_client(workspace_id, honcho_url) results: TestResult = { "question_id": question_id, @@ -663,7 +705,9 @@ Evaluate whether the actual response correctly answers the question based on the ) if honcho_messages: - await session.add_messages(honcho_messages) + for i in range(0, len(honcho_messages), 100): + batch = honcho_messages[i : i + 100] + await session.add_messages(batch) results["sessions_created"].append( SessionResult( @@ -696,6 +740,14 @@ Evaluate whether the actual response correctly answers the question based on the # For user questions, use the user peer (default behavior) actual_response = await user_peer.chat(question_with_date) + # Clean up workspace if requested + if self.cleanup_workspace: + try: + await honcho_client.delete_workspace(workspace_id) + print(f"[{workspace_id}] cleaned up workspace") + except Exception as e: + print(f"Failed to delete workspace: {e}") + actual_response = ( actual_response if isinstance(actual_response, str) else "" ) @@ -790,6 +842,10 @@ Evaluate whether the actual response correctly answers the question based on the print( f"found {len(questions)} {'question' if len(questions) == 1 else 'questions'} in {test_file}" ) + if self.pool_size > 1: + print( + f"distributing questions across {self.pool_size} Honcho instances (ports {self.base_api_port}-{self.base_api_port + self.pool_size - 1})" + ) overall_start = time.time() @@ -807,9 +863,12 @@ Evaluate whether the actual response correctly answers the question based on the ) print(f"{'=' * 60}") - # Run questions in current batch concurrently + # Run questions in current batch concurrently, distributing via round-robin batch_results: list[TestResult] = await asyncio.gather( - *[self.execute_question(q) for q in batch] + *[ + self.execute_question(q, self.get_honcho_url_for_index(i + idx)) + for idx, q in enumerate(batch) + ] ) # Print detailed per-question outputs for this batch @@ -983,7 +1042,8 @@ Evaluate whether the actual response correctly answers the question based on the "test_file": str(test_file), "execution_timestamp": datetime.now().isoformat(), "runner_version": "1.0.0", - "honcho_url": self.honcho_url, + "base_api_port": self.base_api_port, + "pool_size": self.pool_size, "timeout_seconds": self.timeout_seconds, "deriver_settings": settings.DERIVER.model_dump(), "dialectic_settings": settings.DIALECTIC.model_dump(), @@ -1034,7 +1094,8 @@ async def main() -> int: epilog=""" Examples: %(prog)s --test-file tests/bench/longmemeval_data/longmemeval_s.json # Run longmemeval tests - %(prog)s --honcho-url http://localhost:8000 # Custom Honcho URL + %(prog)s --test-file test.json --pool-size 4 # Use 4 Honcho instances + %(prog)s --test-file test.json --base-api-port 8000 --pool-size 4 # Custom base port with pool """, ) @@ -1046,10 +1107,17 @@ Examples: ) parser.add_argument( - "--honcho-url", - type=str, - default="http://localhost:8000", - help="URL of the running Honcho instance (default: http://localhost:8000)", + "--base-api-port", + type=int, + default=8000, + help="Base port for Honcho API instances (default: 8000)", + ) + + parser.add_argument( + "--pool-size", + type=int, + default=1, + help="Number of Honcho instances in the pool (default: 1)", ) parser.add_argument( @@ -1084,6 +1152,12 @@ Examples: help="Merge all sessions within a question into a single session (default: False)", ) + parser.add_argument( + "--cleanup-workspace", + action="store_true", + help="Delete workspace after executing each question (default: False)", + ) + args = parser.parse_args() # Validate arguments @@ -1095,12 +1169,18 @@ Examples: print(f"Error: Batch size must be positive, got {args.batch_size}") return 1 + if args.pool_size <= 0: + print(f"Error: Pool size must be positive, got {args.pool_size}") + return 1 + # Create test runner runner = LongMemEvalRunner( - honcho_url=args.honcho_url, + base_api_port=args.base_api_port, + pool_size=args.pool_size, anthropic_api_key=args.anthropic_api_key, timeout_seconds=args.timeout, merge_sessions=args.merge_sessions, + cleanup_workspace=args.cleanup_workspace, ) try: diff --git a/tests/conftest.py b/tests/conftest.py index 58da6027..844a2769 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -351,10 +351,8 @@ def mock_llm_call_functions(): mock_critical_analysis_result._response = mock_response mock_critical_analysis.return_value = mock_critical_analysis_result - # Create a proper async mock result for dialectic_call - mock_dialectic_result = MagicMock() - mock_dialectic_result.content = "Test dialectic response" - mock_dialectic_call.return_value = mock_dialectic_result + # Mock dialectic_call to return a string (matching actual return type) + mock_dialectic_call.return_value = "Test dialectic response" mock_dialectic_stream.return_value = AsyncMock() diff --git a/tests/crud/test_workspace.py b/tests/crud/test_workspace.py new file mode 100644 index 00000000..a9d32fb6 --- /dev/null +++ b/tests/crud/test_workspace.py @@ -0,0 +1,498 @@ +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.exceptions import ResourceNotFoundException + + +class TestWorkspaceCRUD: + """Test suite for workspace CRUD operations""" + + @pytest.mark.asyncio + async def test_delete_workspace_not_found(self, db_session: AsyncSession): + """Test delete_workspace with non-existent workspace raises ResourceNotFoundException""" + with pytest.raises(ResourceNotFoundException): + await crud.delete_workspace(db_session, "nonexistent_workspace") + + @pytest.mark.asyncio + async def test_delete_workspace_cascade_peers( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that deleting a workspace cascades to delete peers""" + test_workspace, _test_peer = sample_data + + # Create additional peer + peer2 = models.Peer( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(peer2) + await db_session.flush() + + # Verify peers exist + stmt = select(models.Peer).where( + models.Peer.workspace_name == test_workspace.name + ) + result = await db_session.execute(stmt) + peers = result.scalars().all() + assert len(peers) == 2 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify peers are deleted + result = await db_session.execute(stmt) + peers = result.scalars().all() + assert len(peers) == 0 + + @pytest.mark.asyncio + async def test_delete_workspace_cascade_sessions( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that deleting a workspace cascades to delete sessions""" + test_workspace, _test_peer = sample_data + + # Create sessions + session1 = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + session2 = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add_all([session1, session2]) + await db_session.flush() + + # Verify sessions exist + stmt = select(models.Session).where( + models.Session.workspace_name == test_workspace.name + ) + result = await db_session.execute(stmt) + sessions = result.scalars().all() + assert len(sessions) == 2 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify sessions are deleted + result = await db_session.execute(stmt) + sessions = result.scalars().all() + assert len(sessions) == 0 + + @pytest.mark.asyncio + async def test_delete_workspace_cascade_messages( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that deleting a workspace cascades to delete messages""" + test_workspace, test_peer = sample_data + + # Create session + session = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(session) + await db_session.flush() + + # Create messages + message1 = models.Message( + content="Test message 1", + workspace_name=test_workspace.name, + session_name=session.name, + peer_name=test_peer.name, + ) + message2 = models.Message( + content="Test message 2", + workspace_name=test_workspace.name, + session_name=session.name, + peer_name=test_peer.name, + ) + db_session.add_all([message1, message2]) + await db_session.flush() + + # Verify messages exist + stmt = select(models.Message).where( + models.Message.workspace_name == test_workspace.name + ) + result = await db_session.execute(stmt) + messages = result.scalars().all() + assert len(messages) == 2 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify messages are deleted + result = await db_session.execute(stmt) + messages = result.scalars().all() + assert len(messages) == 0 + + @pytest.mark.asyncio + async def test_delete_workspace_cascade_collections( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that deleting a workspace cascades to delete collections""" + test_workspace, test_peer = sample_data + + # Create collection + collection = models.Collection( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer.name, + ) + db_session.add(collection) + await db_session.flush() + + # Verify collection exists + stmt = select(models.Collection).where( + models.Collection.workspace_name == test_workspace.name + ) + result = await db_session.execute(stmt) + collections = result.scalars().all() + assert len(collections) == 1 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify collection is deleted + result = await db_session.execute(stmt) + collections = result.scalars().all() + assert len(collections) == 0 + + @pytest.mark.asyncio + async def test_delete_workspace_cascade_documents( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that deleting a workspace cascades to delete documents""" + test_workspace, test_peer = sample_data + + # Create collection + collection = models.Collection( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer.name, + ) + db_session.add(collection) + await db_session.flush() + + # Create session for document + session = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(session) + await db_session.flush() + + # Create document + document = models.Document( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer.name, + session_name=session.name, + content="Test document content", + embedding=[0.1] * 1536, # Mock embedding vector + ) + db_session.add(document) + await db_session.flush() + + # Verify document exists + stmt = select(models.Document).where( + models.Document.workspace_name == test_workspace.name + ) + result = await db_session.execute(stmt) + documents = result.scalars().all() + assert len(documents) == 1 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify document is deleted + result = await db_session.execute(stmt) + documents = result.scalars().all() + assert len(documents) == 0 + + @pytest.mark.asyncio + async def test_delete_workspace_cascade_session_peers( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that deleting a workspace cascades to delete session_peers associations""" + test_workspace, test_peer = sample_data + + # Create session + session = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(session) + await db_session.flush() + + # Add peer to session + from src.models import session_peers_table + + stmt = session_peers_table.insert().values( + workspace_name=test_workspace.name, + session_name=session.name, + peer_name=test_peer.name, + ) + await db_session.execute(stmt) + await db_session.flush() + + # Verify session_peer association exists + stmt = select(session_peers_table).where( + session_peers_table.c.workspace_name == test_workspace.name + ) + result = await db_session.execute(stmt) + session_peers = result.all() + assert len(session_peers) == 1 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify session_peer association is deleted + result = await db_session.execute(stmt) + session_peers = result.all() + assert len(session_peers) == 0 + + @pytest.mark.asyncio + async def test_delete_workspace_cascade_webhooks( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that deleting a workspace cascades to delete webhook endpoints""" + test_workspace, _test_peer = sample_data + + # Create webhook endpoint + webhook = models.WebhookEndpoint( + workspace_name=test_workspace.name, + url="https://example.com/webhook", + ) + db_session.add(webhook) + await db_session.flush() + + # Verify webhook exists + stmt = select(models.WebhookEndpoint).where( + models.WebhookEndpoint.workspace_name == test_workspace.name + ) + result = await db_session.execute(stmt) + webhooks = result.scalars().all() + assert len(webhooks) == 1 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify webhook is deleted + result = await db_session.execute(stmt) + webhooks = result.scalars().all() + assert len(webhooks) == 0 + + @pytest.mark.asyncio + async def test_delete_workspace_cascade_queue_items( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that deleting a workspace cascades to delete queue items""" + test_workspace, test_peer = sample_data + + # Create session + session = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(session) + await db_session.flush() + + # Create queue item with work_unit_key containing workspace name + # Format: {task_type}:{workspace_name}:{...} + queue_item = models.QueueItem( + work_unit_key=f"representation:{test_workspace.name}:{session.name}:{test_peer.name}:{test_peer.name}", + task_type="representation", + payload={"test": "data"}, + ) + db_session.add(queue_item) + await db_session.flush() + + # Verify queue item exists + stmt = select(models.QueueItem) + result = await db_session.execute(stmt) + queue_items = result.scalars().all() + assert len(queue_items) == 1 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify queue item is deleted + result = await db_session.execute(stmt) + queue_items = result.scalars().all() + assert len(queue_items) == 0 + + @pytest.mark.asyncio + async def test_delete_workspace_cascade_active_queue_sessions( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that deleting a workspace cascades to delete active queue sessions""" + test_workspace, test_peer = sample_data + + # Create session + session = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(session) + await db_session.flush() + + # Create active queue session with work_unit_key containing workspace name + # Format: {task_type}:{workspace_name}:{...} + active_queue = models.ActiveQueueSession( + work_unit_key=f"representation:{test_workspace.name}:{session.name}:{test_peer.name}:{test_peer.name}", + ) + db_session.add(active_queue) + await db_session.flush() + + # Verify active queue session exists + stmt = select(models.ActiveQueueSession) + result = await db_session.execute(stmt) + active_queues = result.scalars().all() + assert len(active_queues) == 1 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify active queue session is deleted + result = await db_session.execute(stmt) + active_queues = result.scalars().all() + assert len(active_queues) == 0 + + @pytest.mark.asyncio + async def test_delete_workspace_returns_deleted_workspace( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that delete_workspace returns the deleted workspace object""" + test_workspace, _test_peer = sample_data + + # Store workspace details before deletion + workspace_name = test_workspace.name + + # Delete workspace + deleted_workspace = await crud.delete_workspace(db_session, test_workspace.name) + + # Verify returned workspace matches the deleted workspace + assert deleted_workspace.name == workspace_name + assert isinstance(deleted_workspace, schemas.Workspace) + + @pytest.mark.asyncio + async def test_delete_workspace_complex_cascade( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test deleting a workspace with multiple related resources of different types""" + test_workspace, test_peer = sample_data + + # Create additional peer + peer2 = models.Peer( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(peer2) + + # Create sessions + session1 = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + session2 = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add_all([session1, session2]) + await db_session.flush() + + # Create messages + message1 = models.Message( + content="Test message 1", + workspace_name=test_workspace.name, + session_name=session1.name, + peer_name=test_peer.name, + ) + message2 = models.Message( + content="Test message 2", + workspace_name=test_workspace.name, + session_name=session2.name, + peer_name=peer2.name, + ) + db_session.add_all([message1, message2]) + + # Create collection and document + collection = models.Collection( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=peer2.name, + ) + db_session.add(collection) + await db_session.flush() + + document = models.Document( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=peer2.name, + session_name=session1.name, + content="Test document", + embedding=[0.1] * 1536, # Mock embedding vector + ) + db_session.add(document) + + # Create webhook + webhook = models.WebhookEndpoint( + workspace_name=test_workspace.name, + url="https://example.com/webhook", + ) + db_session.add(webhook) + await db_session.flush() + + # Count all resources before deletion + peer_stmt = select(models.Peer).where( + models.Peer.workspace_name == test_workspace.name + ) + session_stmt = select(models.Session).where( + models.Session.workspace_name == test_workspace.name + ) + message_stmt = select(models.Message).where( + models.Message.workspace_name == test_workspace.name + ) + collection_stmt = select(models.Collection).where( + models.Collection.workspace_name == test_workspace.name + ) + document_stmt = select(models.Document).where( + models.Document.workspace_name == test_workspace.name + ) + webhook_stmt = select(models.WebhookEndpoint).where( + models.WebhookEndpoint.workspace_name == test_workspace.name + ) + + # Verify all resources exist + assert len((await db_session.execute(peer_stmt)).scalars().all()) == 2 + assert len((await db_session.execute(session_stmt)).scalars().all()) == 2 + assert len((await db_session.execute(message_stmt)).scalars().all()) == 2 + assert len((await db_session.execute(collection_stmt)).scalars().all()) == 1 + assert len((await db_session.execute(document_stmt)).scalars().all()) == 1 + assert len((await db_session.execute(webhook_stmt)).scalars().all()) == 1 + + # Delete workspace + await crud.delete_workspace(db_session, test_workspace.name) + + # Verify all related resources are deleted + assert len((await db_session.execute(peer_stmt)).scalars().all()) == 0 + assert len((await db_session.execute(session_stmt)).scalars().all()) == 0 + assert len((await db_session.execute(message_stmt)).scalars().all()) == 0 + assert len((await db_session.execute(collection_stmt)).scalars().all()) == 0 + assert len((await db_session.execute(document_stmt)).scalars().all()) == 0 + assert len((await db_session.execute(webhook_stmt)).scalars().all()) == 0 diff --git a/tests/routes/test_workspaces.py b/tests/routes/test_workspaces.py index 7cdaa335..f43e4dbe 100644 --- a/tests/routes/test_workspaces.py +++ b/tests/routes/test_workspaces.py @@ -264,3 +264,252 @@ def test_search_workspace_nonexistent(client: TestClient): # Should return empty list for nonexistent workspace assert isinstance(data, list) assert len(data) == 0 + + +def test_delete_workspace(client: TestClient): + """Test deleting a workspace""" + name = str(generate_nanoid()) + + # Create a workspace + response = client.post("/v2/workspaces", json={"name": name}) + assert response.status_code == 200 + workspace = response.json() + assert workspace["id"] == name + + # Delete the workspace + response = client.delete(f"/v2/workspaces/{name}") + assert response.status_code == 200 + deleted_workspace = response.json() + assert deleted_workspace["id"] == name + + # Verify the workspace no longer exists by trying to update it + response = client.put( + f"/v2/workspaces/{name}", json={"metadata": {"test": "value"}} + ) + # Should create a new workspace since the old one was deleted + assert response.status_code == 200 + + +def test_delete_nonexistent_workspace(client: TestClient): + """Test deleting a workspace that doesn't exist""" + nonexistent_workspace_id = str(generate_nanoid()) + + response = client.delete(f"/v2/workspaces/{nonexistent_workspace_id}") + assert response.status_code == 404 + data = response.json() + assert "detail" in data + assert "not found" in data["detail"].lower() + + +def test_delete_workspace_with_peers(client: TestClient): + """Test deleting a workspace that has peers""" + workspace_name = str(generate_nanoid()) + + # Create workspace + response = client.post("/v2/workspaces", json={"name": workspace_name}) + assert response.status_code == 200 + + # Create peers + peer1_name = str(generate_nanoid()) + peer2_name = str(generate_nanoid()) + response = client.post( + f"/v2/workspaces/{workspace_name}/peers", json={"name": peer1_name} + ) + assert response.status_code == 200 + response = client.post( + f"/v2/workspaces/{workspace_name}/peers", json={"name": peer2_name} + ) + assert response.status_code == 200 + + # Delete workspace + response = client.delete(f"/v2/workspaces/{workspace_name}") + assert response.status_code == 200 + + +def test_delete_workspace_with_sessions(client: TestClient): + """Test deleting a workspace that has sessions""" + workspace_name = str(generate_nanoid()) + + # Create workspace + response = client.post("/v2/workspaces", json={"name": workspace_name}) + assert response.status_code == 200 + + # Create peer + peer_name = str(generate_nanoid()) + response = client.post( + f"/v2/workspaces/{workspace_name}/peers", json={"name": peer_name} + ) + assert response.status_code == 200 + + # Create sessions + session1_name = str(generate_nanoid()) + session2_name = str(generate_nanoid()) + response = client.post( + f"/v2/workspaces/{workspace_name}/sessions", json={"name": session1_name} + ) + assert response.status_code == 200 + response = client.post( + f"/v2/workspaces/{workspace_name}/sessions", json={"name": session2_name} + ) + assert response.status_code == 200 + + # Delete workspace + response = client.delete(f"/v2/workspaces/{workspace_name}") + assert response.status_code == 200 + + +def test_delete_workspace_with_messages(client: TestClient): + """Test deleting a workspace that has messages""" + workspace_name = str(generate_nanoid()) + + # Create workspace + response = client.post("/v2/workspaces", json={"name": workspace_name}) + assert response.status_code == 200 + + # Create peer + peer_name = str(generate_nanoid()) + response = client.post( + f"/v2/workspaces/{workspace_name}/peers", json={"name": peer_name} + ) + assert response.status_code == 200 + + # Create session + session_name = str(generate_nanoid()) + response = client.post( + f"/v2/workspaces/{workspace_name}/sessions", json={"name": session_name} + ) + assert response.status_code == 200 + + # Add peer to session + response = client.post( + f"/v2/workspaces/{workspace_name}/sessions/{session_name}/peers", + json={peer_name: {}}, + ) + assert response.status_code == 200 + + # Create messages + response = client.post( + f"/v2/workspaces/{workspace_name}/sessions/{session_name}/messages", + json={ + "messages": [ + {"content": "Test message 1", "peer_id": peer_name}, + {"content": "Test message 2", "peer_id": peer_name}, + ] + }, + ) + assert response.status_code == 200 + + # Delete workspace + response = client.delete(f"/v2/workspaces/{workspace_name}") + assert response.status_code == 200 + + +def test_delete_workspace_with_webhooks(client: TestClient): + """Test deleting a workspace that has webhooks""" + workspace_name = str(generate_nanoid()) + + # Create workspace + response = client.post("/v2/workspaces", json={"name": workspace_name}) + assert response.status_code == 200 + + # Create webhook + response = client.post( + f"/v2/workspaces/{workspace_name}/webhooks", + json={ + "url": "https://example.com/webhook", + }, + ) + assert response.status_code == 200 + + # Delete workspace + response = client.delete(f"/v2/workspaces/{workspace_name}") + assert response.status_code == 200 + + # Verify webhook is deleted by checking workspace doesn't exist + response = client.get(f"/v2/workspaces/{workspace_name}/webhooks") + # This should either return 404 or empty list depending on implementation + assert response.status_code in [404, 200] + + +def test_delete_workspace_cascade(client: TestClient): + """Test that deleting a workspace cascades to all related resources""" + workspace_name = str(generate_nanoid()) + + # Create workspace with complex structure + response = client.post( + "/v2/workspaces", + json={"name": workspace_name, "metadata": {"test": "cascade"}}, + ) + assert response.status_code == 200 + + # Create multiple peers + peer_names = [str(generate_nanoid()) for _ in range(3)] + for peer_name in peer_names: + response = client.post( + f"/v2/workspaces/{workspace_name}/peers", json={"name": peer_name} + ) + assert response.status_code == 200 + + # Create multiple sessions + session_names = [str(generate_nanoid()) for _ in range(2)] + for session_name in session_names: + response = client.post( + f"/v2/workspaces/{workspace_name}/sessions", json={"name": session_name} + ) + assert response.status_code == 200 + + # Add peers to sessions and create messages + for session_name in session_names: + for peer_name in peer_names[:2]: # Add 2 peers to each session + response = client.post( + f"/v2/workspaces/{workspace_name}/sessions/{session_name}/peers", + json={peer_name: {}}, + ) + assert response.status_code == 200 + + # Create messages in session + response = client.post( + f"/v2/workspaces/{workspace_name}/sessions/{session_name}/messages", + json={ + "messages": [ + { + "content": f"Test message in {session_name}", + "peer_id": peer_names[0], + } + ] + }, + ) + assert response.status_code == 200 + + # Delete the workspace + response = client.delete(f"/v2/workspaces/{workspace_name}") + assert response.status_code == 200 + deleted_workspace = response.json() + assert deleted_workspace["id"] == workspace_name + assert deleted_workspace["metadata"]["test"] == "cascade" + + +def test_delete_workspace_returns_workspace_data(client: TestClient): + """Test that delete workspace returns the deleted workspace data""" + name = str(generate_nanoid()) + metadata = {"key": "value", "number": 42} + configuration = {"feature": True} + + # Create workspace with metadata and configuration + response = client.post( + "/v2/workspaces", + json={"name": name, "metadata": metadata, "configuration": configuration}, + ) + assert response.status_code == 200 + created_workspace = response.json() + + # Delete workspace + response = client.delete(f"/v2/workspaces/{name}") + assert response.status_code == 200 + deleted_workspace = response.json() + + # Verify returned data matches original workspace + assert deleted_workspace["id"] == created_workspace["id"] + assert deleted_workspace["metadata"] == metadata + assert deleted_workspace["configuration"] == configuration + assert "created_at" in deleted_workspace diff --git a/tests/utils/test_clients.py b/tests/utils/test_clients.py index 93816c85..abc85f5d 100644 --- a/tests/utils/test_clients.py +++ b/tests/utils/test_clients.py @@ -545,7 +545,10 @@ class TestGoogleClient: mock_response.candidates = [ Mock(token_count=5, finish_reason=mock_finish_reason) ] - mock_client.models.generate_content.return_value = mock_response + # Mock the async aio interface + mock_aio = Mock() + mock_aio.models.generate_content = AsyncMock(return_value=mock_response) + mock_client.aio = mock_aio with patch.dict(CLIENTS, {"google": mock_client}): response = await honcho_llm_call_inner( @@ -572,7 +575,10 @@ class TestGoogleClient: mock_response.candidates = [ Mock(token_count=10, finish_reason=mock_finish_reason) ] - mock_client.models.generate_content.return_value = mock_response + # Mock the async aio interface + mock_aio = Mock() + mock_aio.models.generate_content = AsyncMock(return_value=mock_response) + mock_client.aio = mock_aio with patch.dict(CLIENTS, {"google": mock_client}): _response = await honcho_llm_call_inner( @@ -584,8 +590,8 @@ class TestGoogleClient: ) # Verify JSON mode was set in config - mock_client.models.generate_content.assert_called_once() - call_args = mock_client.models.generate_content.call_args + mock_aio.models.generate_content.assert_called_once() + call_args = mock_aio.models.generate_content.call_args assert ( call_args.kwargs["config"]["response_mime_type"] == "application/json" ) @@ -603,7 +609,10 @@ class TestGoogleClient: mock_response.candidates = [ Mock(token_count=15, finish_reason=mock_finish_reason) ] - mock_client.models.generate_content.return_value = mock_response + # Mock the async aio interface + mock_aio = Mock() + mock_aio.models.generate_content = AsyncMock(return_value=mock_response) + mock_client.aio = mock_aio with patch.dict(CLIENTS, {"google": mock_client}): response = await honcho_llm_call_inner( @@ -620,8 +629,8 @@ class TestGoogleClient: assert response.content.age == 25 # Verify structured output config - mock_client.models.generate_content.assert_called_once() - call_args = mock_client.models.generate_content.call_args + mock_aio.models.generate_content.assert_called_once() + call_args = mock_aio.models.generate_content.call_args config = call_args.kwargs["config"] assert config["response_mime_type"] == "application/json" assert config["response_schema"] == SampleTestModel @@ -682,7 +691,10 @@ class TestGoogleClient: mock_response = Mock() mock_response.text = "Response text" mock_response.candidates = [] # Empty candidates - mock_client.models.generate_content.return_value = mock_response + # Mock the async aio interface + mock_aio = Mock() + mock_aio.models.generate_content = AsyncMock(return_value=mock_response) + mock_client.aio = mock_aio with patch.dict(CLIENTS, {"google": mock_client}): response = await honcho_llm_call_inner( diff --git a/uv.lock b/uv.lock index 95765dbb..3569fc78 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.10" resolution-markers = [ "python_full_version >= '3.13'", @@ -764,7 +764,7 @@ dev = [ [[package]] name = "honcho-ai" -version = "1.4.1" +version = "1.5.0" source = { editable = "sdks/python" } dependencies = [ { name = "honcho-core" }, @@ -780,7 +780,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "honcho-core", specifier = ">=1.5.0" }, + { name = "honcho-core", specifier = ">=1.5.1" }, { name = "httpx", specifier = ">=0.28.0,<1" }, { name = "pydantic", specifier = ">=2.0.0,<3" }, { name = "typing-extensions", marker = "python_full_version < '3.12'", specifier = ">=4.12.0" }, @@ -791,7 +791,7 @@ dev = [{ name = "ruff", specifier = ">=0.11.13" }] [[package]] name = "honcho-core" -version = "1.5.0" +version = "1.5.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, @@ -801,9 +801,9 @@ dependencies = [ { name = "sniffio" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/75/cd/05d2a21afd037673637e390de411cd66ca56b07e524cd7ead65c655a2b49/honcho_core-1.5.0.tar.gz", hash = "sha256:4876195dad16db437117d40a1d5e34ff88974e8eca6d093f3653b4ac2bda3c6d", size = 132236, upload-time = "2025-10-08T18:30:15.777Z" } +sdist = { url = "https://files.pythonhosted.org/packages/a7/9f/6246f02a301a5fa1b9cd00784fafbb3f812feebfaa10a5135104885547da/honcho_core-1.5.1.tar.gz", hash = "sha256:d76da6657707df76ff464ac6874925f31c9e83fe8de51daeb0b10986385e02c7", size = 132626, upload-time = "2025-10-09T20:01:03.07Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/ef/4e/9ef193bba00c521be1152791d2ba37a98bdb58ece8f8ce7863b710a4f511/honcho_core-1.5.0-py3-none-any.whl", hash = "sha256:01db345371d6e80230b202c797a73f9f086dc244936522dbc39efb05f4cee306", size = 123210, upload-time = "2025-10-08T18:30:14.277Z" }, + { url = "https://files.pythonhosted.org/packages/a4/92/87e7c894175fa6aab5e49fac51a234290bc2e46d6723617edc154e0ec675/honcho_core-1.5.1-py3-none-any.whl", hash = "sha256:740cff160e2d9e6dc98ad39f566c1b96f5413ec1abdb415d2fd70f66151cf61a", size = 123292, upload-time = "2025-10-09T20:01:01.57Z" }, ] [[package]] @@ -1564,18 +1564,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b8/db/14bafcb4af2139e046d03fd00dea7873e48eafe18b7d2797e73d6681f210/prometheus_client-0.23.1-py3-none-any.whl", hash = "sha256:dd1913e6e76b59cfe44e7a4b83e01afc9873c1bdfd2ed8739f1e76aeca115f99", size = 61145, upload-time = "2025-09-18T20:47:23.875Z" }, ] -[[package]] -name = "proto-plus" -version = "1.26.1" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "protobuf" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/f4/ac/87285f15f7cce6d4a008f33f1757fb5a13611ea8914eb58c3d0d26243468/proto_plus-1.26.1.tar.gz", hash = "sha256:21a515a4c4c0088a773899e23c7bbade3d18f9c66c73edd4c7ee3816bc96a012", size = 56142, upload-time = "2025-03-10T15:54:38.843Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/4e/6d/280c4c2ce28b1593a19ad5239c8b826871fc6ec275c21afc8e1820108039/proto_plus-1.26.1-py3-none-any.whl", hash = "sha256:13285478c2dcf2abb829db158e1047e2f1e8d63a077d94263c2b88b043c75a66", size = 50163, upload-time = "2025-03-10T15:54:37.335Z" }, -] - [[package]] name = "protobuf" version = "5.29.5"