c3218f70c4
Domain layer (zero framework imports): - domain/models.py: pure dataclasses (RuleDocument, RuleSearchResult, Conversation, ChatMessage, LLMResponse, ChatResult) - domain/ports.py: ABC interfaces (RuleRepository, LLMPort, ConversationStore, IssueTracker) - domain/services.py: ChatService orchestrates Q&A flow using only ports Outbound adapters (implement domain ports): - adapters/outbound/openrouter.py: OpenRouterLLM with persistent httpx client, robust JSON parsing, regex citation fallback - adapters/outbound/sqlite_convos.py: SQLiteConversationStore with async_sessionmaker, timezone-aware datetimes, cleanup support - adapters/outbound/gitea_issues.py: GiteaIssueTracker with markdown injection protection (fenced code blocks) - adapters/outbound/chroma_rules.py: ChromaRuleRepository with clamped similarity scores Inbound adapter: - adapters/inbound/api.py: thin FastAPI router with input validation (max_length constraints), proper HTTP status codes (503 for missing LLM) Configuration & wiring: - config/settings.py: Pydantic v2 SettingsConfigDict (no module-level singleton) - config/container.py: create_app() factory with lifespan-managed DI - main.py: minimal entry point Test infrastructure (90 tests, all passing): - tests/fakes/: in-memory implementations of all 4 ports - tests/domain/: 26 tests for models and ChatService - tests/adapters/: 64 tests for all adapters using fakes/mocks - No real API calls, no model downloads, no disk I/O in fast tests Also fixes: aiosqlite version constraint (>=0.19.0), adds hatch build targets for new package layout. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
"""In-memory ConversationStore for testing — no SQLite, no SQLAlchemy."""
|
|
|
|
from typing import Optional
|
|
import uuid
|
|
|
|
from domain.ports import ConversationStore
|
|
|
|
|
|
class FakeConversationStore(ConversationStore):
|
|
"""Stores conversations and messages in dicts."""
|
|
|
|
def __init__(self):
|
|
self.conversations: dict[str, dict] = {}
|
|
self.messages: dict[str, list[dict]] = {}
|
|
|
|
async def get_or_create_conversation(
|
|
self, user_id: str, channel_id: str, conversation_id: Optional[str] = None
|
|
) -> str:
|
|
if conversation_id and conversation_id in self.conversations:
|
|
return conversation_id
|
|
|
|
new_id = conversation_id or str(uuid.uuid4())
|
|
self.conversations[new_id] = {
|
|
"user_id": user_id,
|
|
"channel_id": channel_id,
|
|
}
|
|
self.messages[new_id] = []
|
|
return new_id
|
|
|
|
async def add_message(
|
|
self,
|
|
conversation_id: str,
|
|
content: str,
|
|
is_user: bool,
|
|
parent_id: Optional[str] = None,
|
|
) -> str:
|
|
message_id = str(uuid.uuid4())
|
|
if conversation_id not in self.messages:
|
|
self.messages[conversation_id] = []
|
|
self.messages[conversation_id].append(
|
|
{
|
|
"id": message_id,
|
|
"content": content,
|
|
"is_user": is_user,
|
|
"parent_id": parent_id,
|
|
}
|
|
)
|
|
return message_id
|
|
|
|
async def get_conversation_history(
|
|
self, conversation_id: str, limit: int = 10
|
|
) -> list[dict[str, str]]:
|
|
msgs = self.messages.get(conversation_id, [])
|
|
history = []
|
|
for msg in msgs[-limit:]:
|
|
role = "user" if msg["is_user"] else "assistant"
|
|
history.append({"role": role, "content": msg["content"]})
|
|
return history
|