From 97f3409280aebf085cd59007fa4c9f005cbd6981 Mon Sep 17 00:00:00 2001 From: Benjamin McCormick Date: Thu, 29 Jan 2026 13:57:00 -0500 Subject: [PATCH] feat: phase 3: introspective meta-dreaming --- src/dreamer/introspection.py | 454 +++++++++++++++++++++++++++++++++++ src/dreamer/orchestrator.py | 11 + src/schemas.py | 41 +++- tests/test_introspection.py | 423 ++++++++++++++++++++++++++++++++ 4 files changed, 928 insertions(+), 1 deletion(-) create mode 100644 src/dreamer/introspection.py create mode 100644 tests/test_introspection.py diff --git a/src/dreamer/introspection.py b/src/dreamer/introspection.py new file mode 100644 index 00000000..4b2a7e68 --- /dev/null +++ b/src/dreamer/introspection.py @@ -0,0 +1,454 @@ +""" +Introspection module for meta-cognitive analysis of Honcho's performance. + +This module provides functionality to analyze how Honcho is performing for a workspace +and generate configuration suggestions based on observed patterns. +""" + +from __future__ import annotations + +import datetime +import json +import logging + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from src import crud, models +from src.config import settings +from src.schemas import ( + IntrospectionReport, + IntrospectionSignals, + IntrospectionSuggestion, +) +from src.utils.clients import honcho_llm_call + +logger = logging.getLogger(__name__) + + +# Reserved peer names for storing introspection reports +SYSTEM_OBSERVER = "_system" +INTROSPECTION_OBSERVED = "_introspection" + + +async def gather_introspection_context( + db: AsyncSession, + workspace_name: str, +) -> IntrospectionSignals: + """ + Gather performance signals from a workspace for introspection analysis. + + Args: + db: Database session + workspace_name: Name of the workspace to analyze + + Returns: + IntrospectionSignals containing performance metrics and configuration + """ + # Calculate time window (last 30 days) + since = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=30) + + # Get dialectic trace statistics + try: + trace_stats = await crud.get_dialectic_trace_stats( + db, workspace_name, since=since + ) + except Exception as e: + logger.warning(f"Failed to get dialectic trace stats: {e}") + trace_stats = { + "total_queries": 0, + "avg_duration_ms": 0.0, + "abstention_count": 0, + "abstention_rate": 0.0, + } + + # Get recent dialectic queries for sample analysis + recent_queries: list[str] = [] + try: + traces = await crud.get_dialectic_traces(db, workspace_name, limit=20) + recent_queries = [t.query for t in traces] + except Exception as e: + logger.warning(f"Failed to get recent dialectic traces: {e}") + + # Get observation counts by level + observations_by_level: dict[str, int] = {} + total_observations = 0 + contradiction_count = 0 + try: + # Query documents grouped by level + stmt = ( + select( + models.Document.level, + func.count(models.Document.id).label("count"), + ) + .where(models.Document.workspace_name == workspace_name) + .where(models.Document.deleted_at.is_(None)) + .group_by(models.Document.level) + ) + result = await db.execute(stmt) + for row in result.all(): + level, count = row + observations_by_level[level] = count + total_observations += count + if level == "contradiction": + contradiction_count = count + except Exception as e: + logger.warning(f"Failed to get observation counts: {e}") + + # Get peer count + total_peers = 0 + try: + stmt = ( + select(func.count(models.Peer.id)) + .where(models.Peer.workspace_name == workspace_name) + .where(~models.Peer.name.startswith("_")) # Exclude system peers + ) + result = await db.execute(stmt) + total_peers = result.scalar() or 0 + except Exception as e: + logger.warning(f"Failed to get peer count: {e}") + + # Get session count + total_sessions = 0 + try: + stmt = select(func.count(models.Session.id)).where( + models.Session.workspace_name == workspace_name + ) + result = await db.execute(stmt) + total_sessions = result.scalar() or 0 + except Exception as e: + logger.warning(f"Failed to get session count: {e}") + + # Get current agent configuration + current_deriver_rules = "" + current_dialectic_rules = "" + try: + agent_config = await crud.get_workspace_agent_config(db, workspace_name) + current_deriver_rules = agent_config.deriver_rules + current_dialectic_rules = agent_config.dialectic_rules + except Exception as e: + logger.warning(f"Failed to get workspace agent config: {e}") + + return IntrospectionSignals( + total_dialectic_queries=int(trace_stats["total_queries"]), + avg_dialectic_duration_ms=float(trace_stats["avg_duration_ms"]), + abstention_count=int(trace_stats["abstention_count"]), + abstention_rate=float(trace_stats["abstention_rate"]), + recent_queries=recent_queries, + total_observations=total_observations, + observations_by_level=observations_by_level, + contradiction_count=contradiction_count, + total_peers=total_peers, + total_sessions=total_sessions, + current_deriver_rules=current_deriver_rules, + current_dialectic_rules=current_dialectic_rules, + ) + + +def build_introspection_prompt(signals: IntrospectionSignals) -> str: + """ + Build a prompt for the LLM to analyze workspace signals and suggest improvements. + + Args: + signals: Performance signals gathered from the workspace + + Returns: + A formatted prompt string for the LLM + """ + queries_sample = ( + "\n".join(f" - {q}" for q in signals.recent_queries[:10]) + if signals.recent_queries + else " (No recent queries)" + ) + + observations_breakdown = ( + "\n".join( + f" - {level}: {count}" + for level, count in signals.observations_by_level.items() + ) + if signals.observations_by_level + else " (No observations)" + ) + + return f"""You are analyzing the performance of a Honcho workspace to identify issues and suggest configuration improvements. + +## Workspace Performance Signals + +### Dialectic API Usage (Last 30 Days) +- Total queries: {signals.total_dialectic_queries} +- Average response time: {signals.avg_dialectic_duration_ms:.1f}ms +- Abstention count: {signals.abstention_count} +- Abstention rate: {signals.abstention_rate:.1%} + +### Recent Query Samples +{queries_sample} + +### Memory Statistics +- Total observations: {signals.total_observations} +- Observations by level: +{observations_breakdown} +- Contradictions detected: {signals.contradiction_count} + +### Workspace Scale +- Total peers: {signals.total_peers} +- Total sessions: {signals.total_sessions} + +### Current Configuration +Deriver rules: +``` +{signals.current_deriver_rules or "(empty - using defaults)"} +``` + +Dialectic rules: +``` +{signals.current_dialectic_rules or "(empty - using defaults)"} +``` + +## Your Task + +1. **Analyze the application type**: Based on the query samples and usage patterns, what kind of application is this workspace likely supporting? (e.g., customer support, personal assistant, educational tool, etc.) + +2. **Identify performance issues**: Look for: + - High abstention rate (>30% suggests memory gaps or query-memory mismatch) + - High contradiction count (suggests conflicting information being stored) + - Low observation count relative to query volume (suggests underutilization of memory) + - Query patterns that suggest certain topics aren't being captured + +3. **Suggest specific rule changes**: For each suggestion, specify: + - Whether it's for `deriver_rules` or `dialectic_rules` + - The exact suggested rule text + - Why this change would help + +Respond with a JSON object matching this schema: +{{ + "performance_summary": "A 2-3 sentence summary of the workspace's performance", + "identified_issues": ["Issue 1", "Issue 2", ...], + "suggestions": [ + {{ + "target": "deriver_rules" | "dialectic_rules", + "current_value": "current rule text", + "suggested_value": "new rule text to add or replace with", + "rationale": "why this change would help", + "confidence": "high" | "medium" | "low" + }} + ] +}} + +If there's insufficient data to make recommendations, return an empty suggestions array with an appropriate performance_summary.""" + + +class IntrospectionLLMResponse(IntrospectionReport): + """Schema for LLM response, excluding fields we'll fill in ourselves.""" + + # Override these to make them optional since LLM won't provide them + workspace_name: str = "" + generated_at: datetime.datetime = datetime.datetime.now(datetime.timezone.utc) + signals: IntrospectionSignals = IntrospectionSignals() + + +async def run_introspection( + db: AsyncSession, + workspace_name: str, +) -> IntrospectionReport | None: + """ + Run introspection analysis for a workspace and generate a report. + + Args: + db: Database session + workspace_name: Name of the workspace to analyze + + Returns: + IntrospectionReport with analysis and suggestions, or None if insufficient data + """ + logger.info(f"Starting introspection for workspace {workspace_name}") + + # Gather signals + signals = await gather_introspection_context(db, workspace_name) + + # Check for minimum data + if signals.total_dialectic_queries == 0 and signals.total_observations == 0: + logger.info( + f"Insufficient data for introspection in workspace {workspace_name}" + ) + report = IntrospectionReport( + workspace_name=workspace_name, + generated_at=datetime.datetime.now(datetime.timezone.utc), + performance_summary="Insufficient data for analysis. No dialectic queries or observations found.", + identified_issues=[], + suggestions=[], + signals=signals, + ) + await store_introspection_report(db, workspace_name, report) + return report + + # Build prompt and call LLM + prompt = build_introspection_prompt(signals) + + try: + llm_response = await honcho_llm_call( + llm_settings=settings.DREAM, + prompt=prompt, + max_tokens=4096, + track_name="introspection", + json_mode=True, + temperature=0.3, + ) + + # Parse the response + response_text = ( + llm_response.content + if hasattr(llm_response, "content") + else str(llm_response) + ) + + # Parse JSON and validate + try: + response_data: dict[str, object] = json.loads(response_text) + except json.JSONDecodeError as e: + logger.error(f"Failed to parse LLM response as JSON: {e}") + response_data = { + "performance_summary": f"Error parsing LLM response: {str(e)[:100]}", + "identified_issues": [], + "suggestions": [], + } + + # Build suggestions list + suggestions: list[IntrospectionSuggestion] = [] + raw_suggestions = response_data.get("suggestions", []) + if isinstance(raw_suggestions, list): + for raw_suggestion in raw_suggestions: + if isinstance(raw_suggestion, dict): + try: + # Validate target field + target_raw = raw_suggestion.get("target", "deriver_rules") + if target_raw not in ("deriver_rules", "dialectic_rules"): + target_raw = "deriver_rules" + + # Validate confidence field + confidence_raw = raw_suggestion.get("confidence", "low") + if confidence_raw not in ("high", "medium", "low"): + confidence_raw = "low" + + suggestions.append( + IntrospectionSuggestion( + target=target_raw, # type: ignore[arg-type] + current_value=str(raw_suggestion.get("current_value", "")), + suggested_value=str(raw_suggestion.get("suggested_value", "")), + rationale=str(raw_suggestion.get("rationale", "")), + confidence=confidence_raw, # type: ignore[arg-type] + ) + ) + except Exception as e: + logger.warning(f"Failed to parse suggestion: {e}") + + # Extract fields with type coercion + performance_summary = response_data.get("performance_summary", "Analysis completed.") + if not isinstance(performance_summary, str): + performance_summary = "Analysis completed." + + identified_issues_raw = response_data.get("identified_issues", []) + identified_issues: list[str] = [] + if isinstance(identified_issues_raw, list): + for issue in identified_issues_raw: + identified_issues.append(str(issue)) + + report = IntrospectionReport( + workspace_name=workspace_name, + generated_at=datetime.datetime.now(datetime.timezone.utc), + performance_summary=performance_summary, + identified_issues=identified_issues, + suggestions=suggestions, + signals=signals, + ) + + except Exception as e: + logger.error(f"LLM call failed during introspection: {e}") + report = IntrospectionReport( + workspace_name=workspace_name, + generated_at=datetime.datetime.now(datetime.timezone.utc), + performance_summary=f"Error during analysis: {str(e)[:200]}", + identified_issues=[], + suggestions=[], + signals=signals, + ) + + # Store the report + await store_introspection_report(db, workspace_name, report) + + logger.info( + f"Introspection completed for {workspace_name}: {len(report.suggestions)} suggestions" + ) + return report + + +async def store_introspection_report( + db: AsyncSession, + workspace_name: str, + report: IntrospectionReport, +) -> None: + """ + Store an introspection report as a document in a reserved collection. + + Reports are stored with: + - observer: _system + - observed: _introspection + - content: JSON-serialized report + - embedding: None (not for semantic search) + + Args: + db: Database session + workspace_name: Name of the workspace + report: The introspection report to store + """ + from src import schemas + + try: + # Ensure system peers exist + await crud.get_or_create_peers( + db, + workspace_name, + [ + schemas.PeerCreate(name=SYSTEM_OBSERVER), + schemas.PeerCreate(name=INTROSPECTION_OBSERVED), + ], + ) + + # Ensure collection exists + await crud.get_or_create_collection( + db, + workspace_name, + observer=SYSTEM_OBSERVER, + observed=INTROSPECTION_OBSERVED, + ) + + # Serialize report to JSON + report_json = report.model_dump_json() + + # Create document (without embedding - not for semantic search) + doc = models.Document( + workspace_name=workspace_name, + observer=SYSTEM_OBSERVER, + observed=INTROSPECTION_OBSERVED, + content=report_json, + level="explicit", + times_derived=1, + internal_metadata={ + "report_type": "introspection", + "generated_at": report.generated_at.isoformat(), + }, + session_name=None, + embedding=None, + sync_state="synced", # No need to sync - no embedding + ) + + db.add(doc) + await db.commit() + + logger.debug( + f"Stored introspection report for {workspace_name}, doc_id={doc.id}" + ) + + except Exception as e: + logger.error(f"Failed to store introspection report: {e}") + await db.rollback() + # Don't re-raise - storing the report is secondary to generating it diff --git a/src/dreamer/orchestrator.py b/src/dreamer/orchestrator.py index 91517cd8..bf7d18ab 100644 --- a/src/dreamer/orchestrator.py +++ b/src/dreamer/orchestrator.py @@ -360,6 +360,17 @@ DREAM: {payload.dream_type} documents for {workspace_name}/{payload.observer}/{p + f"duration={result.total_duration_ms:.0f}ms" ) + case DreamType.INTROSPECTION: + from src.dreamer.introspection import run_introspection + + async with tracked_db("introspection") as db: + result = await run_introspection(db, workspace_name) + + if result is not None: + logger.info( + f"Introspection completed: {len(result.suggestions)} suggestions" + ) + except Exception as e: logger.error( f"Error processing dream task {payload.dream_type} for {payload.observer}/{payload.observed}: {str(e)}", diff --git a/src/schemas.py b/src/schemas.py index d084d98c..21382bc5 100644 --- a/src/schemas.py +++ b/src/schemas.py @@ -1,7 +1,7 @@ import datetime import ipaddress from enum import Enum -from typing import Annotated, Any, Self, cast +from typing import Annotated, Any, Literal, Self, cast from urllib.parse import urlparse import tiktoken @@ -25,6 +25,7 @@ class DreamType(str, Enum): """Types of dreams that can be triggered.""" OMNI = "omni" + INTROSPECTION = "introspection" class WorkspaceAgentConfig(BaseModel): @@ -44,6 +45,44 @@ class WorkspaceAgentConfig(BaseModel): ) +class IntrospectionSignals(BaseModel): + """Signals gathered for introspection analysis.""" + + total_dialectic_queries: int = 0 + avg_dialectic_duration_ms: float = 0.0 + abstention_count: int = 0 + abstention_rate: float = 0.0 + recent_queries: list[str] = Field(default_factory=list) + total_observations: int = 0 + observations_by_level: dict[str, int] = Field(default_factory=dict) + contradiction_count: int = 0 + total_peers: int = 0 + total_sessions: int = 0 + current_deriver_rules: str = "" + current_dialectic_rules: str = "" + + +class IntrospectionSuggestion(BaseModel): + """A single configuration suggestion.""" + + target: Literal["deriver_rules", "dialectic_rules"] + current_value: str + suggested_value: str + rationale: str + confidence: Literal["high", "medium", "low"] + + +class IntrospectionReport(BaseModel): + """Complete introspection report for a workspace.""" + + workspace_name: str + generated_at: datetime.datetime + performance_summary: str + identified_issues: list[str] = Field(default_factory=list) + suggestions: list[IntrospectionSuggestion] = Field(default_factory=list) + signals: IntrospectionSignals + + class ReconcilerType(str, Enum): """Types of reconciler tasks that can be performed.""" diff --git a/tests/test_introspection.py b/tests/test_introspection.py new file mode 100644 index 00000000..a4e8ea2d --- /dev/null +++ b/tests/test_introspection.py @@ -0,0 +1,423 @@ +"""Tests for the introspection module.""" + +import datetime +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from nanoid import generate as generate_nanoid +from sqlalchemy.ext.asyncio import AsyncSession + +from src import crud, models +from src.dreamer.introspection import ( + INTROSPECTION_OBSERVED, + SYSTEM_OBSERVER, + build_introspection_prompt, + gather_introspection_context, + run_introspection, + store_introspection_report, +) +from src.schemas import ( + DialecticTraceCreate, + IntrospectionReport, + IntrospectionSignals, + IntrospectionSuggestion, +) + + +class TestGatherIntrospectionContext: + """Tests for gathering introspection signals.""" + + @pytest.mark.asyncio + async def test_gather_introspection_context_empty_workspace( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test gathering signals from an empty workspace returns zero values.""" + workspace, _ = sample_data + + signals = await gather_introspection_context(db_session, workspace.name) + + assert signals.total_dialectic_queries == 0 + assert signals.avg_dialectic_duration_ms == 0.0 + assert signals.abstention_count == 0 + assert signals.abstention_rate == 0.0 + assert signals.recent_queries == [] + assert signals.total_observations == 0 + assert signals.observations_by_level == {} + assert signals.contradiction_count == 0 + # Peer count may be 1 (the sample peer), but should not include system peers + assert signals.total_sessions == 0 + assert signals.current_deriver_rules == "" + assert signals.current_dialectic_rules == "" + + @pytest.mark.asyncio + async def test_gather_introspection_context_with_data( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test gathering signals from a workspace with data.""" + workspace, peer = sample_data + + # Create some dialectic traces + for i in range(5): + trace_data = DialecticTraceCreate( + workspace_name=workspace.name, + observer=peer.name, + observed=peer.name, + query=f"What does the user like? Query {i}", + response="The user likes coffee." if i < 3 else "I don't have information.", + reasoning_level="low", + total_duration_ms=100.0 * (i + 1), + input_tokens=100, + output_tokens=50, + ) + await crud.create_dialectic_trace(db_session, trace_data) + + # Create a session + session = models.Session(name=generate_nanoid(), workspace_name=workspace.name) + db_session.add(session) + await db_session.flush() + + signals = await gather_introspection_context(db_session, workspace.name) + + assert signals.total_dialectic_queries == 5 + assert signals.avg_dialectic_duration_ms == 300.0 # (100+200+300+400+500)/5 + assert signals.abstention_count == 2 # Last 2 responses are abstentions + assert signals.abstention_rate == 0.4 + assert len(signals.recent_queries) == 5 + assert "What does the user like?" in signals.recent_queries[0] + assert signals.total_sessions == 1 + + +class TestBuildIntrospectionPrompt: + """Tests for building the introspection prompt.""" + + def test_build_introspection_prompt_empty_signals(self): + """Test prompt building with empty signals.""" + signals = IntrospectionSignals() + + prompt = build_introspection_prompt(signals) + + assert "Total queries: 0" in prompt + assert "Abstention rate: 0.0%" in prompt + assert "(No recent queries)" in prompt + assert "(No observations)" in prompt + assert "(empty - using defaults)" in prompt + + def test_build_introspection_prompt_with_data(self): + """Test prompt building with populated signals.""" + signals = IntrospectionSignals( + total_dialectic_queries=100, + avg_dialectic_duration_ms=250.5, + abstention_count=30, + abstention_rate=0.3, + recent_queries=["What is the user's name?", "What do they prefer?"], + total_observations=500, + observations_by_level={"explicit": 300, "deductive": 150, "inductive": 50}, + contradiction_count=5, + total_peers=10, + total_sessions=25, + current_deriver_rules="Focus on preferences", + current_dialectic_rules="Be concise", + ) + + prompt = build_introspection_prompt(signals) + + assert "Total queries: 100" in prompt + assert "250.5ms" in prompt + assert "Abstention rate: 30.0%" in prompt + assert "What is the user's name?" in prompt + assert "explicit: 300" in prompt + assert "deductive: 150" in prompt + assert "Contradictions detected: 5" in prompt + assert "Total peers: 10" in prompt + assert "Total sessions: 25" in prompt + assert "Focus on preferences" in prompt + assert "Be concise" in prompt + + +class TestRunIntrospection: + """Tests for the main introspection runner.""" + + @pytest.mark.asyncio + async def test_run_introspection_insufficient_data( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test introspection with insufficient data returns appropriate report.""" + workspace, _ = sample_data + + with patch( + "src.dreamer.introspection.store_introspection_report", + new_callable=AsyncMock, + ) as mock_store: + report = await run_introspection(db_session, workspace.name) + + assert report is not None + assert report.workspace_name == workspace.name + assert "Insufficient data" in report.performance_summary + assert report.suggestions == [] + assert report.identified_issues == [] + mock_store.assert_called_once() + + @pytest.mark.asyncio + async def test_run_introspection_with_data( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test introspection with data calls LLM and parses response.""" + workspace, peer = sample_data + + # Create some dialectic traces + for i in range(3): + trace_data = DialecticTraceCreate( + workspace_name=workspace.name, + observer=peer.name, + observed=peer.name, + query=f"Query {i}", + response="Response", + reasoning_level="low", + total_duration_ms=100.0, + input_tokens=100, + output_tokens=50, + ) + await crud.create_dialectic_trace(db_session, trace_data) + + # Mock the LLM call + mock_llm_response = MagicMock() + mock_llm_response.content = json.dumps({ + "performance_summary": "The workspace is performing well.", + "identified_issues": ["High abstention rate"], + "suggestions": [ + { + "target": "deriver_rules", + "current_value": "", + "suggested_value": "Focus on capturing user preferences", + "rationale": "Too many queries about preferences are being missed", + "confidence": "medium", + } + ], + }) + + with ( + patch( + "src.dreamer.introspection.honcho_llm_call", + new_callable=AsyncMock, + return_value=mock_llm_response, + ), + patch( + "src.dreamer.introspection.store_introspection_report", + new_callable=AsyncMock, + ) as mock_store, + ): + report = await run_introspection(db_session, workspace.name) + + assert report is not None + assert report.workspace_name == workspace.name + assert report.performance_summary == "The workspace is performing well." + assert "High abstention rate" in report.identified_issues + assert len(report.suggestions) == 1 + assert report.suggestions[0].target == "deriver_rules" + assert report.suggestions[0].confidence == "medium" + mock_store.assert_called_once() + + @pytest.mark.asyncio + async def test_run_introspection_llm_failure( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test introspection handles LLM failures gracefully.""" + workspace, peer = sample_data + + # Create some dialectic traces + trace_data = DialecticTraceCreate( + workspace_name=workspace.name, + observer=peer.name, + observed=peer.name, + query="Query", + response="Response", + reasoning_level="low", + total_duration_ms=100.0, + input_tokens=100, + output_tokens=50, + ) + await crud.create_dialectic_trace(db_session, trace_data) + + with ( + patch( + "src.dreamer.introspection.honcho_llm_call", + new_callable=AsyncMock, + side_effect=Exception("LLM API error"), + ), + patch( + "src.dreamer.introspection.store_introspection_report", + new_callable=AsyncMock, + ) as mock_store, + ): + report = await run_introspection(db_session, workspace.name) + + assert report is not None + assert "Error during analysis" in report.performance_summary + assert report.suggestions == [] + mock_store.assert_called_once() + + +class TestStoreIntrospectionReport: + """Tests for storing introspection reports.""" + + @pytest.mark.asyncio + async def test_store_introspection_report_creates_document( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that storing a report creates a document in the reserved collection.""" + from src.crud.collection import get_or_create_collection as real_get_or_create_collection + + workspace, _ = sample_data + # Capture workspace name before any potential session issues + workspace_name = workspace.name + + # Create system peers and collection directly (bypassing mock) + system_peer = models.Peer(name=SYSTEM_OBSERVER, workspace_name=workspace_name) + introspection_peer = models.Peer( + name=INTROSPECTION_OBSERVED, workspace_name=workspace_name + ) + db_session.add_all([system_peer, introspection_peer]) + await db_session.flush() + + # Create the collection + collection = models.Collection( + workspace_name=workspace_name, + observer=SYSTEM_OBSERVER, + observed=INTROSPECTION_OBSERVED, + ) + db_session.add(collection) + await db_session.flush() + + report = IntrospectionReport( + workspace_name=workspace_name, + generated_at=datetime.datetime.now(datetime.timezone.utc), + performance_summary="Test summary", + identified_issues=["Issue 1"], + suggestions=[ + IntrospectionSuggestion( + target="deriver_rules", + current_value="", + suggested_value="New rule", + rationale="Test rationale", + confidence="high", + ) + ], + signals=IntrospectionSignals(total_dialectic_queries=10), + ) + + await store_introspection_report(db_session, workspace_name, report) + + # Verify document was created + from sqlalchemy import select + + stmt = ( + select(models.Document) + .where(models.Document.workspace_name == workspace_name) + .where(models.Document.observer == SYSTEM_OBSERVER) + .where(models.Document.observed == INTROSPECTION_OBSERVED) + ) + result = await db_session.execute(stmt) + doc = result.scalar_one_or_none() + + assert doc is not None + assert doc.level == "explicit" + assert doc.embedding is None # No embedding for reports + assert doc.sync_state == "synced" + + # Verify content is valid JSON + content = json.loads(doc.content) + assert content["workspace_name"] == workspace_name + assert content["performance_summary"] == "Test summary" + assert len(content["suggestions"]) == 1 + + @pytest.mark.asyncio + async def test_store_introspection_report_creates_system_peers( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that storing a report creates the _system and _introspection peers.""" + workspace, _ = sample_data + # Capture workspace name before any potential session issues + workspace_name = workspace.name + + report = IntrospectionReport( + workspace_name=workspace_name, + generated_at=datetime.datetime.now(datetime.timezone.utc), + performance_summary="Test", + signals=IntrospectionSignals(), + ) + + await store_introspection_report(db_session, workspace_name, report) + + # Verify system peers were created + from sqlalchemy import select + + stmt = ( + select(models.Peer) + .where(models.Peer.workspace_name == workspace_name) + .where(models.Peer.name.in_([SYSTEM_OBSERVER, INTROSPECTION_OBSERVED])) + ) + result = await db_session.execute(stmt) + peers = list(result.scalars().all()) + + assert len(peers) == 2 + peer_names = {p.name for p in peers} + assert SYSTEM_OBSERVER in peer_names + assert INTROSPECTION_OBSERVED in peer_names + + +class TestIntrospectionDreamDispatch: + """Tests for introspection via the dream dispatch mechanism.""" + + @pytest.mark.asyncio + async def test_introspection_dream_type_dispatch( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test that INTROSPECTION dream type is properly dispatched.""" + from src.dreamer.orchestrator import process_dream + from src.schemas import DreamType + from src.utils.queue_payload import DreamPayload + + workspace, peer = sample_data + + payload = DreamPayload( + dream_type=DreamType.INTROSPECTION, + observer=peer.name, + observed=peer.name, + session_name=None, + ) + + with patch( + "src.dreamer.introspection.run_introspection", + new_callable=AsyncMock, + ) as mock_run: + mock_run.return_value = IntrospectionReport( + workspace_name=workspace.name, + generated_at=datetime.datetime.now(datetime.timezone.utc), + performance_summary="Test", + suggestions=[], + signals=IntrospectionSignals(), + ) + + await process_dream(payload, workspace.name) + + mock_run.assert_called_once() + # Verify it was called with the workspace name + call_args = mock_run.call_args + assert call_args[0][1] == workspace.name