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

Merge branch 'main' into vineeth/dev-1193

This commit is contained in:
Vineeth Voruganti 2025-10-14 11:27:57 -04:00
parent c026a4fb45 65503955a4
commit fbb9b14618
48 changed files with 5956 additions and 4811 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
CHANGELOG.md
View File

@ -5,7 +5,7 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](http://keepachangelog.com/)
and this project adheres to [Semantic Versioning](http://semver.org/).
## [2.4.0] - 2025-10-08
## [2.4.0] - 2025-10-09
### Added
@ -17,6 +17,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
- Prometheus Client for better Metrics
- Performance metrics instrumentation
- Error reporting to deriver
- Workspace Delete Method
- Multi-db option in test harness
### Changed
@ -26,6 +28,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
- Semantic across codebase to reference resources based on `observer` and `observed`
- Prompts for Deriver & Dialectic to reference peer_id and add examples
- `Get Context` route returns peer card and representation in addition to messages and summaries
- Refactoring logger.info calls to logger.debug where applicable
### Fixed
- Gemini client to use async methods
## [2.3.3] — 2025-10-01

26
docs/changelog/introduction.mdx
View File

@ -38,6 +38,8 @@ Welcome to the Honcho changelog! This section documents all notable changes to t
- Prometheus Client for better Metrics
- Performance metrics instrumentation
- Error reporting to deriver
- Workspace Delete Method
- Multi-db option in test harness
### Changed
@ -47,6 +49,12 @@ Welcome to the Honcho changelog! This section documents all notable changes to t
- Semantic across codebase to reference resources based on `observer` and `observed`
- Prompts for Deriver & Dialectic to reference peer_id and add examples
- `Get Context` route returns peer card and representation in addition to messages and summaries
- Refactoring logger.info calls to logger.debug where applicable
### Fixed
- Gemini client to use async methods
</Update>
<Update label="v2.3.3">
@ -328,9 +336,14 @@ Welcome to the Honcho changelog! This section documents all notable changes to t
<Tab title="Python SDK">
[Python SDK](https://pypi.org/project/honcho-ai/)
<Update label="v1.5.0 (Current)">
### Changed
### Added
- message_id of `Summary` model is a string nanoid
- Delete workspace method
### Changed
- message_id of `Summary` model is a string nanoid
- Get Context can return Peer Card & Peer Representation
</Update>
<Update label="v1.4.1">
### Added
@ -397,9 +410,14 @@ Welcome to the Honcho changelog! This section documents all notable changes to t
<Tab title="TypeScript SDK">
[TypeScript SDK](https://www.npmjs.com/package/@honcho-ai/sdk)
<Update label="v1.5.0 (Current)">
### Changed
### Added
- message_id of `Summary` model is a string nanoid
- Delete workspace method
### Changed
- message_id of `Summary` model is a string nanoid
- Get Context can return Peer Card & Peer Representation
</Update>
<Update label="v1.4.1">
### Added

1
docs/docs.json
View File

@ -88,6 +88,7 @@
"v2/api-reference/endpoint/workspaces/get-or-create-workspace",
"v2/api-reference/endpoint/workspaces/get-all-workspaces",
"v2/api-reference/endpoint/workspaces/update-workspace",
"v2/api-reference/endpoint/workspaces/delete-workspace",
"v2/api-reference/endpoint/workspaces/search-workspace",
"v2/api-reference/endpoint/workspaces/get-deriver-status"
]

3
docs/v2/api-reference/endpoint/workspaces/delete-workspace.mdx Normal file
View File

@ -0,0 +1,3 @@
---
openapi: delete /v2/workspaces/{workspace_id}
---

4514
docs/v2/openapi.documented.json
View File

File diff suppressed because it is too large Load Diff

3894
docs/v2/openapi.documented.yml Normal file
View File

File diff suppressed because it is too large Load Diff

83
migrations/versions/08894082221a_replace_collection_name_with_observer_.py
View File

@ -10,6 +10,7 @@ from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
from nanoid import generate as generate_nanoid
from sqlalchemy import text
from migrations.utils import column_exists, constraint_exists, fk_exists, index_exists
@ -25,25 +26,47 @@ depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Replace collections.name and documents.collection_name with observer and observed fields."""
schema = settings.DB.SCHEMA
inspector = sa.inspect(op.get_bind())
connection = op.get_bind()
# SESSION_NAME MIGRATION
# Replace NULL session_name values with empty strings and make column non-nullable
# Replace NULL session_name values with __global_observations__ and make column non-nullable
# This only applies to documents table
# Update documents table
connection.execute(
inspector = sa.inspect(connection)
# query documents table to get ALL workspace_names that have documents without a session_name
workspace_names = connection.execute(
text(
"""
UPDATE documents
SET session_name = ''
WHERE session_name IS NULL
f"""
SELECT DISTINCT workspace_name FROM {schema}.documents WHERE session_name IS NULL
"""
)
)
if column_exists("documents", "session_name", inspector):
op.alter_column("documents", "session_name", nullable=False, schema=schema)
).fetchall()
if workspace_names and column_exists("sessions", "name", inspector):
# Create __global_observations__ session for EACH workspace that needs it
for (workspace_name,) in workspace_names:
session_id = generate_nanoid()
connection.execute(
text(
f"""
INSERT INTO {schema}.sessions (id, name, workspace_name, is_active) VALUES (:session_id, '__global_observations__', :workspace_name, true) ON CONFLICT DO NOTHING
"""
),
{"session_id": session_id, "workspace_name": workspace_name},
)
# Update all documents with NULL session_name
connection.execute(
text(
f"""
UPDATE {schema}.documents
SET session_name = '__global_observations__'
WHERE session_name IS NULL
"""
),
)
op.alter_column("documents", "session_name", nullable=False, schema=schema)
# COLLECTIONS TABLE
# Step 1: Add new observer and observed columns to collections
@ -64,17 +87,21 @@ def upgrade() -> None:
# Step 2: Populate collections observer and observed from existing name field
# The logic is:
# - observer = peer_name (the exact peer ID)
# - If name is "global_representation", observed = peer_name
# - Otherwise, observed = name with the "observer_" prefix stripped
# - If name is "global_representation", observed = peer_name (self-observation)
# - If name starts with peer_name + "_", extract the observed part (pattern: observer_observed)
# - If name ends with "_" + peer_name, extract the first part (pattern: observed_observer)
# - Otherwise (legacy edge cases), observed = name itself
connection.execute(
text(
"""
UPDATE collections
f"""
UPDATE {schema}.collections
SET
observer = peer_name,
observed = CASE
WHEN name = 'global_representation' THEN peer_name
ELSE substring(name from length(peer_name) + 2)
WHEN name LIKE peer_name || '_%' THEN substring(name from length(peer_name) + 2)
WHEN name LIKE '%_' || peer_name THEN substring(name from 1 for length(name) - length(peer_name) - 1)
ELSE name
END
WHERE observer IS NULL OR observed IS NULL
"""
@ -108,18 +135,18 @@ def upgrade() -> None:
while True:
result = connection.execute(
text(
"""
f"""
WITH batch AS (
SELECT d.ctid
FROM documents d
FROM {schema}.documents d
WHERE d.observer IS NULL OR d.observed IS NULL
LIMIT :batch_size
)
UPDATE documents d
UPDATE {schema}.documents d
SET
observer = c.observer,
observed = c.observed
FROM collections c, batch
FROM {schema}.collections c, batch
WHERE d.ctid = batch.ctid
AND d.collection_name = c.name
AND d.peer_name = c.peer_name
@ -329,8 +356,8 @@ def downgrade() -> None:
# Step 2: Populate collections name from observer and observed
connection.execute(
text(
"""
UPDATE collections
f"""
UPDATE {schema}.collections
SET name = CASE
WHEN observer = observed THEN 'global_representation'
ELSE observer || '_' || observed
@ -354,8 +381,8 @@ def downgrade() -> None:
# Populate peer_name with observer value
connection.execute(
text(
"""
UPDATE collections
f"""
UPDATE {schema}.collections
SET peer_name = observer
WHERE peer_name IS NULL
"""
@ -391,8 +418,8 @@ def downgrade() -> None:
# Step 5: Populate documents collection_name from observer and observed
connection.execute(
text(
"""
UPDATE documents
f"""
UPDATE {schema}.documents
SET collection_name = CASE
WHEN observer = observed THEN 'global_representation'
ELSE observer || '_' || observed
@ -416,8 +443,8 @@ def downgrade() -> None:
# Populate peer_name with observed value
connection.execute(
text(
"""
UPDATE documents
f"""
UPDATE {schema}.documents
SET peer_name = observed
WHERE peer_name IS NULL
"""

9
sdks/python/CHANGELOG.md
View File

@ -5,13 +5,18 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](http://keepachangelog.com/)
and this project adheres to [Semantic Versioning](http://semver.org/).
## [1.5.0] - 2025-10-08
## [1.5.0] - 2025-10-09
### Added
- Delete workspace method
### Changed
- message_id of `Summary` model is a string nanoid
- Get Context can return Peer Card & Peer Representation
## [1.4.1] — 2025-10-09
## [1.4.1] — 2025-10-01
### Added

2
sdks/python/pyproject.toml
View File

@ -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\"",

22
sdks/python/src/honcho/async_client/client.py
View File

@ -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,

46
sdks/python/src/honcho/async_client/session.py
View File

@ -427,11 +427,23 @@ class AsyncSession(BaseModel):
tokens: int | None = Field(
None, gt=0, description="Maximum number of tokens to include in the context"
),
peer_target: str | None = Field(
None,
description="A peer ID to get context for. If given *without* `peer_perspective`, a representation and peer card will be included from the omniscient Honcho-level view of `peer_target`. If given *with* `peer_perspective`, will get the representation and card for `peer_target` *from the perspective of `peer_perspective`*.",
),
last_user_message: str | Message | None = Field(
None,
description="The most recent message (string or Message object), used to fetch semantically relevant observations and returned as part of the context object. Use this alongside `peer_target` to get a more focused context -- does nothing if `peer_target` is not provided.",
),
peer_perspective: str | None = Field(
None,
description="A peer ID to get context *from the perspective of*. If given, response will attempt to include representation and card from the perspective of `peer_perspective`. Must be provided with `peer_target`.",
),
) -> SessionContext:
"""
Get optimized context for this session within a token limit.
Makes an async API call to retrieve a curated list of messages that provides
Makes an API call to retrieve a curated list of messages that provides
optimal context for the conversation while staying within the specified
token limit. Uses tiktoken for token counting, so results should be
compatible with OpenAI models.
@ -440,6 +452,9 @@ class AsyncSession(BaseModel):
summary: Whether to include summary information
tokens: Maximum number of tokens to include in the context. Will default
to Honcho server configuration if not provided.
peer_target: A peer ID to get context for. If given *without* `peer_perspective`, a representation and peer card will be included from the omniscient Honcho-level view of `peer_target`. If given *with* `peer_perspective`, will get the representation and card for `peer_target` *from the perspective of `peer_perspective`*.
last_user_message: The most recent message (string or Message object), used to fetch semantically relevant observations and returned as part of the context object. Use this alongside `peer_target` to get a more focused context -- does nothing if `peer_target` is not provided.
peer_perspective: A peer ID to get context *from the perspective of*. If given, response will attempt to include representation and card from the perspective of `peer_perspective`. Must be provided with `peer_target`.
Returns:
A SessionContext object containing the optimized message history and
@ -450,11 +465,32 @@ class AsyncSession(BaseModel):
Token counting is performed using tiktoken. For models using different
tokenizers, you may need to adjust the token limit accordingly.
"""
if peer_target is None and peer_perspective is not None:
raise ValueError(
"You must provide a `peer_target` when `peer_perspective` is provided"
)
if peer_target is None and last_user_message is not None:
raise ValueError(
"You must provide a `peer_target` when `last_user_message` is provided"
)
last_user_message_id = (
last_user_message.id
if isinstance(last_user_message, Message)
else last_user_message
)
context = await self._client.workspaces.sessions.get_context(
session_id=self.id,
workspace_id=self.workspace_id,
tokens=tokens if tokens is not None else omit,
summary=summary,
last_message=last_user_message_id
if last_user_message_id is not None
else omit,
peer_target=peer_target if peer_target is not None else omit,
peer_perspective=peer_perspective if peer_perspective is not None else omit,
)
# Convert the honcho_core summary to our Summary if it exists
@ -469,7 +505,13 @@ class AsyncSession(BaseModel):
)
return SessionContext(
session_id=self.id, messages=context.messages, summary=session_summary
session_id=self.id,
messages=context.messages,
summary=session_summary,
peer_representation=str(context.peer_representation)
if context.peer_representation
else None,
peer_card=context.peer_card,
)
async def get_summaries(self) -> SessionSummaries:

22
sdks/python/src/honcho/client.py
View File

@ -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,

44
sdks/python/src/honcho/session.py
View File

@ -407,6 +407,18 @@ class Session(BaseModel):
tokens: int | None = Field(
None, gt=0, description="Maximum number of tokens to include in the context"
),
peer_target: str | None = Field(
None,
description="A peer ID to get context for. If given *without* `peer_perspective`, a representation and peer card will be included from the omniscient Honcho-level view of `peer_target`. If given *with* `peer_perspective`, will get the representation and card for `peer_target` *from the perspective of `peer_perspective`*.",
),
last_user_message: str | Message | None = Field(
None,
description="The most recent message (string or Message object), used to fetch semantically relevant observations and returned as part of the context object. Use this alongside `peer_target` to get a more focused context -- does nothing if `peer_target` is not provided.",
),
peer_perspective: str | None = Field(
None,
description="A peer ID to get context *from the perspective of*. If given, response will attempt to include representation and card from the perspective of `peer_perspective`. Must be provided with `peer_target`.",
),
) -> SessionContext:
"""
Get optimized context for this session within a token limit.
@ -420,6 +432,9 @@ class Session(BaseModel):
summary: Whether to include summary information
tokens: Maximum number of tokens to include in the context. Will default
to Honcho server configuration if not provided.
peer_target: A peer ID to get context for. If given *without* `peer_perspective`, a representation and peer card will be included from the omniscient Honcho-level view of `peer_target`. If given *with* `peer_perspective`, will get the representation and card for `peer_target` *from the perspective of `peer_perspective`*.
last_user_message: The most recent message (string or Message object), used to fetch semantically relevant observations and returned as part of the context object. Use this alongside `peer_target` to get a more focused context -- does nothing if `peer_target` is not provided.
peer_perspective: A peer ID to get context *from the perspective of*. If given, response will attempt to include representation and card from the perspective of `peer_perspective`. Must be provided with `peer_target`.
Returns:
A SessionContext object containing the optimized message history and
@ -430,11 +445,32 @@ class Session(BaseModel):
Token counting is performed using tiktoken. For models using different
tokenizers, you may need to adjust the token limit accordingly.
"""
if peer_target is None and peer_perspective is not None:
raise ValueError(
"You must provide a `peer_target` when `peer_perspective` is provided"
)
if peer_target is None and last_user_message is not None:
raise ValueError(
"You must provide a `peer_target` when `last_user_message` is provided"
)
last_user_message_id = (
last_user_message.id
if isinstance(last_user_message, Message)
else last_user_message
)
context = self._client.workspaces.sessions.get_context(
session_id=self.id,
workspace_id=self.workspace_id,
tokens=tokens if tokens is not None else omit,
summary=summary,
last_message=last_user_message_id
if last_user_message_id is not None
else omit,
peer_target=peer_target if peer_target is not None else omit,
peer_perspective=peer_perspective if peer_perspective is not None else omit,
)
# Convert the honcho_core summary to our Summary if it exists
@ -449,7 +485,13 @@ class Session(BaseModel):
)
return SessionContext(
session_id=self.id, messages=context.messages, summary=session_summary
session_id=self.id,
messages=context.messages,
summary=session_summary,
peer_representation=str(context.peer_representation)
if context.peer_representation
else None,
peer_card=context.peer_card,
)
def get_summaries(self) -> SessionSummaries:

58
sdks/python/src/honcho/session_context.py
View File

@ -58,6 +58,14 @@ class SessionContext(BaseModel):
summary: Summary | None = Field(
None, description="Summary of the session history prior to the message cutoff"
)
peer_representation: str | None = Field(
None,
description="The peer representation, if context is requested from a specific perspective",
)
peer_card: list[str] | None = Field(
None,
description="The peer card, if context is requested from a specific perspective",
)
@validate_call
def __init__(
@ -72,6 +80,14 @@ class SessionContext(BaseModel):
None,
description="Summary of the session history prior to the message cutoff",
),
peer_representation: str | None = Field(
None,
description="The peer representation, if context is requested from a specific perspective",
),
peer_card: list[str] | None = Field(
None,
description="The peer card, if context is requested from a specific perspective",
),
) -> None:
"""
Initialize a new SessionContext.
@ -84,6 +100,8 @@ class SessionContext(BaseModel):
session_id=session_id,
messages=messages,
summary=summary,
peer_representation=peer_representation,
peer_card=peer_card,
)
def to_openai(
@ -117,14 +135,30 @@ class SessionContext(BaseModel):
}
for message in self.messages
]
system_messages: list[dict[str, str]] = []
if self.peer_representation:
peer_representation_message = {
"role": "system",
"content": f"<peer_representation>{self.peer_representation}</peer_representation>",
}
system_messages.append(peer_representation_message)
if self.peer_card:
peer_card_message = {
"role": "system",
"content": f"<peer_card>{self.peer_card}</peer_card>",
}
system_messages.append(peer_card_message)
if self.summary:
summary_message = {
"role": "system",
"content": f"<summary>{self.summary.content}</summary>",
}
return [summary_message, *messages]
return messages
system_messages.append(summary_message)
return system_messages + messages
def to_anthropic(
self,
@ -164,14 +198,30 @@ class SessionContext(BaseModel):
}
for message in self.messages
]
system_messages: list[dict[str, str]] = []
if self.peer_representation:
peer_representation_message = {
"role": "user",
"content": f"<peer_representation>{self.peer_representation}</peer_representation>",
}
system_messages.append(peer_representation_message)
if self.peer_card:
peer_card_message = {
"role": "user",
"content": f"<peer_card>{self.peer_card}</peer_card>",
}
system_messages.append(peer_card_message)
if self.summary:
summary_message = {
"role": "user",
"content": f"<summary>{self.summary.content}</summary>",
}
return [summary_message, *messages]
return messages
system_messages.append(summary_message)
return system_messages + messages
def __len__(self) -> int:
"""

9
sdks/typescript/CHANGELOG.md
View File

@ -5,13 +5,18 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](http://keepachangelog.com/)
and this project adheres to [Semantic Versioning](http://semver.org/).
## [1.5.0] - 2025-10-08
## [1.5.0] - 2025-10-09
### Added
- Delete workspace method
### Changed
- message_id of `Summary` model is a string nanoid
- Get Context can return Peer Card & Peer Representation
## [1.4.1] — 2025-10-09
## [1.4.1] — 2025-10-01
### Added

4
sdks/typescript/bun.lock
View File

@ -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=="],

2
sdks/typescript/package.json
View File

@ -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"
},

14
sdks/typescript/src/client.ts
View File

@ -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<Awaited<ReturnType<typeof this._client.workspaces.delete>>> {
return await this._client.workspaces.delete(workspaceId)
}
/**
* Search for messages in the current workspace.
*

105
sdks/typescript/src/session.ts
View File

@ -451,10 +451,21 @@ export class Session {
* compatible with OpenAI models. The context optimization balances
* recency and relevance to provide the best conversational context.
*
* @param summary - Whether to include summary information in the context.
* When true, includes session summary if available. Defaults to true
* @param tokens - Maximum number of tokens to include in the context. If not provided,
* uses the server's default configuration
* @param options - Configuration options for context retrieval
* @param options.summary - Whether to include summary information in the context.
* When true, includes session summary if available. Defaults to true
* @param options.tokens - Maximum number of tokens to include in the context. If not provided,
* uses the server's default configuration
* @param options.peerTarget - The target of the perspective. If given without `peerPerspective`,
* will get the Honcho-level representation and peer card for this peer.
* If given with `peerPerspective`, will get the representation and card
* for this peer from the perspective of that peer.
* @param options.lastUserMessage - The most recent message, used to fetch semantically relevant
* observations and returned as part of the context object.
* Can be either a message ID string or a Message object.
* @param options.peerPerspective - A peer to get context for. If given, response will attempt to
* include representation and card from the perspective of that peer.
* Must be provided with `peerTarget`.
* @returns Promise resolving to a SessionContext object containing the optimized
* message history and summary (if available) that maximizes conversational
* context while respecting the token limit
@ -462,25 +473,103 @@ export class Session {
* @note Token counting is performed using tiktoken. For models using different
* tokenizers, you may need to adjust the token limit accordingly.
*/
async getContext(
summary?: boolean,
tokens?: number,
peerTarget?: string | Peer,
lastUserMessage?: string | Message,
peerPerspective?: string | Peer
): Promise<SessionContext>
async getContext(options?: {
summary?: boolean
tokens?: number
}): Promise<SessionContext> {
peerTarget?: string | Peer
lastUserMessage?: string | Message
peerPerspective?: string | Peer
}): Promise<SessionContext>
async getContext(
summaryOrOptions?:
| boolean
| {
summary?: boolean
tokens?: number
peerTarget?: string | Peer
lastUserMessage?: string | Message
peerPerspective?: string | Peer
},
tokens?: number,
peerTarget?: string | Peer,
lastUserMessage?: string | Message,
peerPerspective?: string | Peer
): Promise<SessionContext> {
// Normalize positional arguments into options object
let options: {
summary?: boolean
tokens?: number
peerTarget?: string
lastUserMessage?: string
peerPerspective?: string
}
if (
typeof summaryOrOptions === 'boolean' ||
(summaryOrOptions === undefined && arguments.length > 1)
) {
// Positional arguments pattern
options = {
summary: summaryOrOptions as boolean | undefined,
tokens,
peerTarget: typeof peerTarget === 'object' ? peerTarget.id : peerTarget,
lastUserMessage:
typeof lastUserMessage === 'string'
? lastUserMessage
: lastUserMessage?.id,
peerPerspective:
typeof peerPerspective === 'object'
? peerPerspective.id
: peerPerspective,
}
} else {
// Options object pattern
options = (summaryOrOptions as typeof options) || {}
}
const contextParams = ContextParamsSchema.parse({
summary: options?.summary,
tokens: options?.tokens,
summary: options.summary,
tokens: options.tokens,
peerTarget: options.peerTarget,
lastUserMessage: options.lastUserMessage,
peerPerspective: options.peerPerspective,
})
// Extract message ID if lastUserMessage is a Message object
const lastMessageId =
typeof contextParams.lastUserMessage === 'string'
? contextParams.lastUserMessage
: contextParams.lastUserMessage?.id
const context = await this._client.workspaces.sessions.getContext(
this.workspaceId,
this.id,
{
tokens: contextParams.tokens,
summary: contextParams.summary,
last_message: lastMessageId,
peer_target: contextParams.peerTarget,
peer_perspective: contextParams.peerPerspective,
}
)
// Convert the summary response to Summary object if present
const summary = context.summary ? new Summary(context.summary) : null
return new SessionContext(this.id, context.messages, summary)
return new SessionContext(
this.id,
context.messages,
summary,
context.peer_representation
? JSON.stringify(context.peer_representation)
: null,
context.peer_card ?? null
)
}
/**

82
sdks/typescript/src/session_context.ts
View File

@ -102,21 +102,37 @@ export class SessionContext {
*/
readonly summary: Summary | null
/**
* The peer representation, if context is requested from a specific perspective.
*/
readonly peerRepresentation: string | null
/**
* The peer card, if context is requested from a specific perspective.
*/
readonly peerCard: string[] | null
/**
* Initialize a new SessionContext.
*
* @param sessionId ID of the session this context belongs to
* @param messages List of Message objects to include in the context
* @param summary Summary of the session history prior to the message cutoff
* @param peerRepresentation The peer representation, if context is requested from a specific perspective
* @param peerCard The peer card, if context is requested from a specific perspective
*/
constructor(
sessionId: string,
messages: Message[],
summary: Summary | null = null
summary: Summary | null = null,
peerRepresentation: string | null = null,
peerCard: string[] | null = null
) {
this.sessionId = sessionId
this.messages = messages
this.summary = summary
this.peerRepresentation = peerRepresentation
this.peerCard = peerCard
}
/**
@ -136,18 +152,36 @@ export class SessionContext {
assistant: string | Peer
): Array<{ role: string; content: string; name?: string }> {
const assistantId = typeof assistant === 'string' ? assistant : assistant.id
const summaryMessage = this.summary
? {
role: 'system',
content: `<summary>${this.summary.content}</summary>`,
}
: null
const messages = this.messages.map((message) => ({
role: message.peer_id === assistantId ? 'assistant' : 'user',
name: message.peer_id,
content: message.content,
}))
return summaryMessage ? [summaryMessage, ...messages] : messages
const systemMessages: Array<{ role: string; content: string }> = []
if (this.peerRepresentation) {
systemMessages.push({
role: 'system',
content: `<peer_representation>${this.peerRepresentation}</peer_representation>`,
})
}
if (this.peerCard) {
systemMessages.push({
role: 'system',
content: `<peer_card>${this.peerCard}</peer_card>`,
})
}
if (this.summary) {
systemMessages.push({
role: 'system',
content: `<summary>${this.summary.content}</summary>`,
})
}
return [...systemMessages, ...messages]
}
/**
@ -170,12 +204,6 @@ export class SessionContext {
assistant: string | Peer
): Array<{ role: string; content: string }> {
const assistantId = typeof assistant === 'string' ? assistant : assistant.id
const summaryMessage = this.summary
? {
role: 'user',
content: `<summary>${this.summary.content}</summary>`,
}
: null
const messages = this.messages.map((message) =>
message.peer_id === assistantId
? {
@ -187,7 +215,31 @@ export class SessionContext {
content: `${message.peer_id}: ${message.content}`,
}
)
return summaryMessage ? [summaryMessage, ...messages] : messages
const systemMessages: Array<{ role: string; content: string }> = []
if (this.peerRepresentation) {
systemMessages.push({
role: 'user',
content: `<peer_representation>${this.peerRepresentation}</peer_representation>`,
})
}
if (this.peerCard) {
systemMessages.push({
role: 'user',
content: `<peer_card>${this.peerCard}</peer_card>`,
})
}
if (this.summary) {
systemMessages.push({
role: 'user',
content: `<summary>${this.summary.content}</summary>`,
})
}
return [...systemMessages, ...messages]
}
/**

55
sdks/typescript/src/validation.ts
View File

@ -1,3 +1,4 @@
import type { Message } from '@honcho-ai/core/resources/workspaces/sessions/messages'
import { z } from 'zod'
/**
@ -123,16 +124,56 @@ export const ChatQuerySchema = z.object({
sessionId: z.string().optional(),
})
/**
* Schema for validating Message objects from the core SDK.
*/
const MessageSchema: z.ZodType<Message> = z.object({
id: z.string(),
content: z.string(),
created_at: z.string(),
peer_id: z.string(),
session_id: z.string(),
token_count: z.number(),
workspace_id: z.string(),
metadata: z.record(z.string(), z.unknown()).optional(),
}) as z.ZodType<Message>
/**
* Schema for context retrieval parameters.
*/
export const ContextParamsSchema = z.object({
summary: z.boolean().optional(),
tokens: z
.number()
.positive('Token limit must be a positive number')
.optional(),
})
export const ContextParamsSchema = z
.object({
summary: z.boolean().optional(),
tokens: z
.number()
.positive('Token limit must be a positive number')
.optional(),
lastUserMessage: z
.union([
z.string().min(1, 'Last user message must be a non-empty string'),
MessageSchema,
])
.optional(),
peerTarget: PeerIdSchema.optional(),
peerPerspective: PeerIdSchema.optional(),
})
.superRefine((data, ctx) => {
if (data.lastUserMessage && !data.peerTarget) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: 'peerTarget is required when lastUserMessage is provided',
path: ['lastUserMessage'],
})
}
if (data.peerPerspective && !data.peerTarget) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: 'peerTarget is required when peerPerspective is provided',
path: ['peerPerspective'],
})
}
})
/**
* Schema for deriver status options.

8
src/crud/__init__.py
View File

@ -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",

2
src/crud/peer.py
View File

@ -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

15
src/crud/representation.py
View File

@ -300,9 +300,7 @@ class RepresentationManager:
db,
query=include_semantic_query,
top_k=semantic_observations,
max_distance=semantic_search_max_distance
if semantic_search_max_distance is not None
else 0.3,
max_distance=semantic_search_max_distance,
)
representation.merge_representation(
Representation.from_documents(semantic_docs)
@ -322,11 +320,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
@ -336,7 +329,7 @@ class RepresentationManager:
db: AsyncSession,
query: str,
top_k: int,
max_distance: float,
max_distance: float | None = None,
level: str | None = None,
conversation_context: str = "",
) -> list[models.Document]:
@ -348,8 +341,8 @@ class RepresentationManager:
query,
level,
conversation_context,
max_distance,
top_k,
max_distance,
)
else:
documents = await crud.query_documents(
@ -433,8 +426,8 @@ class RepresentationManager:
query: str,
level: str,
conversation_context: str,
max_distance: float,
count: int,
max_distance: float | None = None,
) -> list[models.Document]:
"""Query documents for a specific level."""
documents = await crud.query_documents(

8
src/crud/session.py
View File

@ -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

4
src/crud/webhook.py
View File

@ -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)

120
src/crud/workspace.py
View File

@ -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

2
src/deriver/consumer.py
View File

@ -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:

25
src/deriver/deriver.py
View File

@ -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,

2
src/deriver/enqueue.py
View File

@ -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,

1
src/deriver/queue_manager.py
View File

@ -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:

85
src/dialectic/chat.py
View File

@ -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

17
src/dialectic/prompts.py
View File

@ -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>
{recent_conversation_history}
</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>
{recent_conversation_history}
</recent_conversation_history>
<query>{query}</query>
{recent_conversation_history_section}
<working_representation>{working_representation}</working_representation>
"""
)

2
src/embedding_client.py
View File

@ -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}"
)

7
src/routers/messages.py
View File

@ -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)}")

29
src/routers/sessions.py
View File

@ -7,7 +7,7 @@ from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import apaginate
from sqlalchemy.ext.asyncio import AsyncSession
from src import config, crud, models, schemas
from src import config, crud, schemas
from src.dependencies import db, tracked_db
from src.exceptions import (
AuthenticationException,
@ -87,7 +87,7 @@ async def _get_session_context_task(
session_id: str,
token_limit: int,
include_summary: bool,
) -> tuple[schemas.Summary | None, list[models.Message]]:
) -> tuple[schemas.Summary | None, list[schemas.Message]]:
"""
Atomic task to get session context using tracked_db.
@ -101,13 +101,16 @@ async def _get_session_context_task(
Tuple of (summary, messages)
"""
async with tracked_db("get_session_context") as db:
return await summarizer.get_session_context(
summary, messages = await summarizer.get_session_context(
db,
workspace_name=workspace_id,
session_name=session_id,
token_limit=token_limit,
include_summary=include_summary,
)
# Convert SQLAlchemy models to Pydantic schemas while session is active
message_schemas = [schemas.Message.model_validate(msg) for msg in messages]
return summary, message_schemas
@router.post(
@ -202,7 +205,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 +228,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 +259,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 +291,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 +327,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 +363,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 +414,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:
@ -501,7 +504,7 @@ async def get_session_context(
)
return schemas.SessionContext(
name=session_id,
messages=messages, # pyright: ignore -- db message type and schema message type are different, but excess gets removed by schema
messages=messages,
summary=summary,
)
@ -540,7 +543,7 @@ async def get_session_context(
return schemas.SessionContext(
name=session_id,
messages=messages, # pyright: ignore -- db message type and schema message type are different, but excess gets removed by schema
messages=messages,
summary=summary,
peer_representation=representation,
peer_card=card,

13
src/routers/workspaces.py
View File

@ -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],

16
src/utils/clients.py
View File

@ -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={
@ -519,8 +519,8 @@ async def honcho_llm_call_inner(
# Safely extract response data
text_content = gemini_response.text if gemini_response.text else ""
token_count = (
gemini_response.candidates[0].token_count or 0
if gemini_response.candidates
gemini_response.usage_metadata.candidates_token_count or 0
if gemini_response.usage_metadata
else 0
)
finish_reason = (
@ -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={
@ -547,8 +547,8 @@ async def honcho_llm_call_inner(
)
token_count = (
gemini_response.candidates[0].token_count or 0
if gemini_response.candidates
gemini_response.usage_metadata.candidates_token_count or 0
if gemini_response.usage_metadata
else 0
)
finish_reason = (

38
src/utils/logging.py
View File

@ -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:

68
src/utils/summarizer.py
View File

@ -49,7 +49,7 @@ def to_schema_summary(s: Summary) -> schemas.Summary:
summary_type=s["summary_type"],
created_at=s["created_at"],
token_count=s["token_count"],
message_public_id=s["message_public_id"],
message_public_id=s.get("message_public_id", ""),
)
@ -195,9 +195,6 @@ async def summarize_if_needed(
session_name: The session name
message_id: The message ID
"""
logger.debug("Checking if summaries should be created for session %s", session_name)
should_create_long: bool = message_seq_in_session % MESSAGES_PER_LONG_SUMMARY == 0
should_create_short: bool = message_seq_in_session % MESSAGES_PER_SHORT_SUMMARY == 0
@ -214,11 +211,11 @@ async def summarize_if_needed(
SummaryType.LONG,
message_public_id,
)
logger.info(
"Saved long summary for session %s covering up to message %s (%s in session)",
session_name,
message_id,
accumulate_metric(
f"summary_{workspace_name}_{message_id}",
"long_summary_up_to_message",
message_seq_in_session,
"count",
)
async def create_short_summary():
@ -231,11 +228,11 @@ async def summarize_if_needed(
SummaryType.SHORT,
message_public_id,
)
logger.info(
"Saved short summary for session %s covering up to message %s (%s in session)",
session_name,
message_id,
accumulate_metric(
f"summary_{workspace_name}_{message_id}",
"short_summary_up_to_message",
message_seq_in_session,
"count",
)
await asyncio.gather(
@ -255,11 +252,11 @@ async def summarize_if_needed(
SummaryType.LONG,
message_public_id,
)
logger.info(
"Saved long summary for session %s covering up to message %s (%s in session)",
session_name,
message_id,
accumulate_metric(
f"summary_{workspace_name}_{message_id}",
"long_summary_up_to_message",
message_seq_in_session,
"count",
)
elif should_create_short:
await _create_and_save_summary(
@ -270,11 +267,11 @@ async def summarize_if_needed(
SummaryType.SHORT,
message_public_id,
)
logger.info(
"Saved short summary for session %s covering up to message %s (%s in session)",
session_name,
message_id,
accumulate_metric(
f"summary_{workspace_name}_{message_id}",
"short_summary_up_to_message",
message_seq_in_session,
"count",
)
@ -294,7 +291,7 @@ async def _create_and_save_summary(
4. Save the new summary to the database
"""
logger.info("Creating new %s summary", summary_type.name)
logger.debug("Creating new %s summary", summary_type.name)
# Time summarization step
summary_start = time.perf_counter()
@ -329,6 +326,19 @@ async def _create_and_save_summary(
session_name,
)
accumulate_metric(
f"summary_{workspace_name}_{message_id}",
f"{summary_type.name}_summary_text",
new_summary["content"],
"blob",
)
accumulate_metric(
f"summary_{workspace_name}_{message_id}",
f"{summary_type.name}_summary_size",
new_summary["token_count"],
"tokens",
)
summary_duration = (time.perf_counter() - summary_start) * 1000
accumulate_metric(
f"summary_{workspace_name}_{message_id}",
@ -374,9 +384,6 @@ async def _create_summary(
logger.error(
"Generated summary is empty! This may indicate a token limit issue."
)
logger.info("Summary text: %s", summary_text)
logger.info("Summary size: %s tokens", summary_tokens)
except Exception:
logger.exception("Error generating summary!")
# Fallback to a basic summary in case of error
@ -387,13 +394,6 @@ async def _create_summary(
)
summary_tokens = 50
accumulate_metric(
f"summary_{messages[-1].workspace_name}_{messages[-1].id}",
f"{summary_type.name}_summary_size",
response.output_tokens if response else f"{summary_tokens} (est.)",
"tokens",
)
return Summary(
content=summary_text,
message_id=messages[-1].id if messages else 0,
@ -621,7 +621,7 @@ async def get_session_context(
summary_type=latest_long_summary["summary_type"],
created_at=latest_long_summary["created_at"],
token_count=latest_long_summary["token_count"],
message_public_id=latest_long_summary["message_public_id"],
message_public_id=latest_long_summary.get("message_public_id", ""),
)
messages_tokens = token_limit - latest_long_summary["token_count"]
messages_start_id = latest_long_summary["message_id"]
@ -634,12 +634,12 @@ async def get_session_context(
summary_type=latest_short_summary["summary_type"],
created_at=latest_short_summary["created_at"],
token_count=latest_short_summary["token_count"],
message_public_id=latest_short_summary["message_public_id"],
message_public_id=latest_short_summary.get("message_public_id", ""),
)
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,

4
src/webhooks/webhook_delivery.py
View File

@ -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:

297
tests/bench/harness.py
View File

@ -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
@ -436,11 +470,8 @@ try:
print_settings(value, full_key, max_depth, current_depth + 1)
else:
# Mask sensitive information
if isinstance(value, str) and any(sensitive in value.lower() for sensitive in ['password', 'secret', 'key', 'token']):
if 'testpwd' in value:
masked_value = value.replace('testpwd', '***')
else:
masked_value = '***'
if isinstance(full_key, str) and any(sensitive in full_key.lower() for sensitive in ['password', 'secret', 'key', 'uri']):
masked_value = '*' * len(value) if value else 'None'
else:
masked_value = value
print(f" {{key}}: {{masked_value}}")
@ -462,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:
@ -597,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")
@ -632,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.
@ -642,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
""",
)
@ -650,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(
@ -662,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(
@ -682,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__":

185
tests/bench/longmem.py
View File

@ -38,10 +38,13 @@ 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)
--use-get-context: Use get_context + judge LLM instead of dialectic .chat endpoint (default: False)
```
## Other notes
@ -57,10 +60,11 @@ import os
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from typing import Any, cast
import tiktoken
from anthropic import AsyncAnthropic
from anthropic.types import MessageParam
from dotenv import load_dotenv
from honcho import AsyncHoncho
from honcho.async_client.session import SessionPeerConfig
@ -115,26 +119,35 @@ 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,
use_get_context: 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)
use_get_context: If True, use get_context + judge LLM instead of dialectic .chat endpoint
"""
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
self.use_get_context: bool = use_get_context
# Initialize metrics collector
self.metrics_collector: MetricsCollector = MetricsCollector()
@ -162,6 +175,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 +228,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 +321,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 +337,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(
@ -358,7 +401,7 @@ Actual response: "{actual_response}"
Evaluate whether the actual response correctly answers the question based on the expected answer. Focus on factual accuracy and evidence that the AI accessed the correct memory."""
response = await self.anthropic_client.messages.create(
model="claude-sonnet-4-20250514",
model="claude-sonnet-4-5",
max_tokens=300,
temperature=0.0,
system=system_prompt,
@ -402,12 +445,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 +474,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,
@ -579,6 +626,7 @@ Evaluate whether the actual response correctly answers the question based on the
)
)
else:
merged_session_id = None
# create separate sessions
# Zip together dates, session IDs, and session content
for session_date, session_id, session_messages in zip(
@ -663,7 +711,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(
@ -688,13 +738,59 @@ Evaluate whether the actual response correctly answers the question based on the
output_lines.append(f"\nAsking question: {question_with_date}")
try:
# Use the appropriate peer based on question type
if is_assistant_type:
# For assistant questions, use the assistant peer
actual_response = await assistant_peer.chat(question_with_date)
if self.use_get_context:
# Use get_context instead of dialectic .chat endpoint
# Get the session to retrieve context from
if not self.merge_sessions or merged_session_id is None:
raise ValueError(
"Merged session ID is required when using get_context. Set --merge-sessions to True."
)
session = await honcho_client.session(id=merged_session_id)
# Get context for the appropriate peer
peer_id = "assistant" if is_assistant_type else "user"
context = await session.get_context(
summary=True,
peer_target=peer_id,
last_user_message=question,
)
# Format context using to_anthropic method
context_messages = context.to_anthropic(assistant="assistant")
# Add the question as the final user message
context_messages.append(
{"role": "user", "content": question_with_date}
)
# Call Anthropic API to generate response
response = await self.anthropic_client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
messages=cast(list[MessageParam], context_messages),
)
if not response.content:
raise ValueError("Anthropic returned empty response")
content_block = response.content[0]
actual_response = getattr(content_block, "text", "")
else:
# For user questions, use the user peer (default behavior)
actual_response = await user_peer.chat(question_with_date)
# Use the appropriate peer based on question type
if is_assistant_type:
# For assistant questions, use the assistant peer
actual_response = await assistant_peer.chat(question_with_date)
else:
# 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 +886,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 +907,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 +1086,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 +1138,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 +1151,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 +1196,18 @@ 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)",
)
parser.add_argument(
"--use-get-context",
action="store_true",
help="Use get_context + judge LLM instead of dialectic .chat endpoint (default: False)",
)
args = parser.parse_args()
# Validate arguments
@ -1095,12 +1219,19 @@ 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,
use_get_context=args.use_get_context,
)
try:

6
tests/conftest.py
View File

@ -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()

498
tests/crud/test_workspace.py Normal file
View File

@ -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

249
tests/routes/test_workspaces.py
View File

@ -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

54
tests/utils/test_clients.py
View File

@ -542,10 +542,15 @@ class TestGoogleClient:
mock_response.text = "Hello from Gemini"
mock_finish_reason = Mock()
mock_finish_reason.name = "STOP"
mock_response.candidates = [
Mock(token_count=5, finish_reason=mock_finish_reason)
]
mock_client.models.generate_content.return_value = mock_response
mock_response.candidates = [Mock(finish_reason=mock_finish_reason)]
# Mock the usage_metadata with candidates_token_count
mock_usage_metadata = Mock()
mock_usage_metadata.candidates_token_count = 5
mock_response.usage_metadata = mock_usage_metadata
# 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(
@ -569,10 +574,15 @@ class TestGoogleClient:
mock_response.text = '{"result": "success"}'
mock_finish_reason = Mock()
mock_finish_reason.name = "STOP"
mock_response.candidates = [
Mock(token_count=10, finish_reason=mock_finish_reason)
]
mock_client.models.generate_content.return_value = mock_response
mock_response.candidates = [Mock(finish_reason=mock_finish_reason)]
# Mock the usage_metadata with candidates_token_count
mock_usage_metadata = Mock()
mock_usage_metadata.candidates_token_count = 10
mock_response.usage_metadata = mock_usage_metadata
# 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 +594,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"
)
@ -600,10 +610,15 @@ class TestGoogleClient:
mock_response.parsed = mock_parsed
mock_finish_reason = Mock()
mock_finish_reason.name = "STOP"
mock_response.candidates = [
Mock(token_count=15, finish_reason=mock_finish_reason)
]
mock_client.models.generate_content.return_value = mock_response
mock_response.candidates = [Mock(finish_reason=mock_finish_reason)]
# Mock the usage_metadata with candidates_token_count
mock_usage_metadata = Mock()
mock_usage_metadata.candidates_token_count = 15
mock_response.usage_metadata = mock_usage_metadata
# 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 +635,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 +697,12 @@ 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 usage_metadata as None to test fallback
mock_response.usage_metadata = None
# 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(

8
uv.lock
View File

@ -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]]