Add Hearthkeeper preview endpoint

.env.example

@@ -1,24 +0,0 @@
-# Application Settings
-APP_NAME=Sanctum Chronicler
-APP_ENV=development
-DEBUG=false
-
-# Database
-DATABASE_URL=postgresql+asyncpg://sanctum:password@localhost:5432/sanctum
-DB_PASSWORD=password
-
-# Twitch Configuration
-TWITCH_CLIENT_ID=
-TWITCH_CLIENT_SECRET=
-TWITCH_BOT_USERNAME=
-TWITCH_CHANNEL_NAME=
-
-# LLM Configuration
-# Supported providers: openai, ollama, lm_studio (or leave empty for mock)
-LLM_PROVIDER=
-LLM_BASE_URL=
-LLM_API_KEY=
-LLM_MODEL=gpt-3.5-turbo
-
-# Export Configuration
-EXPORT_PATH=./exports

.gitignore

@@ -44,7 +44,7 @@ pgdata/
 # =========================================
 # Runtime / Exports
 # =========================================
-exports/
+/exports/
 data/
 logs/
 
@@ -78,4 +78,4 @@ Thumbs.db
 # =========================================
 # Misc
 # =========================================
-*.log
+*.log

Dockerfile

@@ -42,7 +42,7 @@ EXPOSE 8000
 
 # Health check
 HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
-    CMD python -c "import httpx; httpx.get('http://localhost:8000/health')"
+    CMD python -c "import httpx; httpx.get('http://localhost:8000/health').raise_for_status()"
 
 # Run application
 CMD ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

app/agent/modes/hearthkeeper.py

@@ -34,9 +34,18 @@ class HearthkeeperMode:
         """Determine if Hearthkeeper should generate a prompt."""
         return minutes_since_activity >= self.activity_threshold
 
-    async def generate_prompt(self, theme: str | None = None) -> str:
+    async def generate_prompt(
+        self,
+        theme: str | None = None,
+        dashboard: dict | None = None,
+        recent_discussion: list[str] | None = None,
+    ) -> str:
         """Generate a gentle prompt for the stream."""
-        prompt = PromptTemplates.gentle_prompt(theme)
+        prompt = PromptTemplates.gentle_prompt(
+            current_theme=theme,
+            dashboard=dashboard,
+            recent_discussion=recent_discussion or [],
+        )
         response = await self.llm_client.generate(prompt)
         logger.info("Hearthkeeper generated gentle prompt")
         return response

app/agent/orchestrator.py

@@ -1,9 +1,8 @@
 """Agent Orchestrator - Routes messages and manages agent modes."""
 
 import logging
-import uuid
-from datetime import datetime
-from sqlalchemy.ext.asyncio import AsyncSession
+from collections.abc import Awaitable, Callable
+from datetime import datetime, timedelta
 
 from app.agent.policies import (
     ChatActivityPolicy,
@@ -15,10 +14,12 @@ from app.agent.modes.steward import StewardMode
 from app.agent.modes.warden import WardenMode
 from app.agent.modes.librarian import LibrarianMode
 from app.agent.modes.scribe import ScribeMode
+from app.config import settings
 from app.llm.client import LLMClient
-from app.memory.database import async_session_factory
+from app.memory.database import get_session
 from app.memory.models import AgentActionType
 from app.memory.repository import Repository
+from app.twitch.chat import send_chat_message
 
 logger = logging.getLogger(__name__)
 
@@ -32,9 +33,13 @@ class AgentOrchestrator:
     and how to flag suspicious content.
     """
 
-    def __init__(self):
+    def __init__(self, loop_interval_seconds: float = 60.0):
         """Initialize the orchestrator and all modes."""
         self.llm_client = LLMClient()
+        self.loop_interval_seconds = loop_interval_seconds
+        self.hearthkeeper_prompt_interval = timedelta(
+            minutes=settings.HEARTHKEEPER_PROMPT_INTERVAL_MINUTES
+        )
         
         # Initialize modes
         self.hearthkeeper = HearthkeeperMode(self.llm_client)
@@ -50,9 +55,27 @@ class AgentOrchestrator:
         
         # Track active sessions
         self.active_sessions: dict[str, dict] = {}
+        self.chat_interaction_gate: Callable[[str], Awaitable[bool]] | None = None
         
         logger.info("AgentOrchestrator initialized with all modes and policies")
 
+    def set_chat_interaction_gate(
+        self,
+        gate: Callable[[str], Awaitable[bool]] | None,
+    ) -> None:
+        """Set an async gate that must pass before the agent can post to chat."""
+        self.chat_interaction_gate = gate
+
+    async def can_interact_with_chat(self, channel_name: str) -> bool:
+        """Return whether outbound chat interaction is currently allowed."""
+        if not self.chat_interaction_gate:
+            return True
+        try:
+            return await self.chat_interaction_gate(channel_name)
+        except Exception as e:
+            logger.warning("Chat interaction gate failed closed: %s", e)
+            return False
+
     async def start_session(self, channel_name: str) -> str:
         """
         Start a new stream session.
@@ -63,22 +86,110 @@ class AgentOrchestrator:
         Returns:
             Session ID
         """
-        session_id = str(uuid.uuid4())
-        
-        async with async_session_factory() as db_session:
+        session_id: str | None = None
+        async for db_session in get_session():
             repo = Repository(db_session)
-            await repo.create_session(channel_name)
+            session_id = await repo.create_session(channel_name)
+
+        if session_id is None:
+            raise RuntimeError("Failed to create stream session")
         
         self.active_sessions[session_id] = {
             "channel_name": channel_name,
             "started_at": datetime.utcnow(),
             "message_count": 0,
             "theme": None,
+            "dashboard": None,
+            "last_hearthkeeper_prompt_at": None,
         }
+        self.chat_activity.record_activity(session_id)
         
         logger.info(f"Started session {session_id} for {channel_name}")
         return session_id
 
+    async def ensure_single_active_session_for_channel(self, channel_name: str) -> str:
+        """Return one active session for a channel and end duplicate active sessions."""
+        normalized_channel = channel_name.lower()
+        matching_sessions = [
+            (session_id, session)
+            for session_id, session in self.active_sessions.items()
+            if session.get("channel_name", "").lower() == normalized_channel
+        ]
+
+        if not matching_sessions:
+            return await self.start_session(channel_name)
+
+        keep_session_id, _ = max(
+            matching_sessions,
+            key=lambda item: item[1].get("started_at", datetime.min),
+        )
+        duplicate_session_ids = [
+            session_id
+            for session_id, _ in matching_sessions
+            if session_id != keep_session_id
+        ]
+
+        if duplicate_session_ids:
+            async for db_session in get_session():
+                repo = Repository(db_session)
+                for session_id in duplicate_session_ids:
+                    await repo.end_session(session_id)
+
+            for session_id in duplicate_session_ids:
+                self.active_sessions.pop(session_id, None)
+                self.chat_activity.clear_activity(session_id)
+
+            logger.warning(
+                "Ended %s duplicate active session(s) for channel %s; keeping %s",
+                len(duplicate_session_ids),
+                channel_name,
+                keep_session_id,
+            )
+
+        return keep_session_id
+
+    async def restore_active_sessions(self) -> int:
+        """Restore active sessions from the database after app startup."""
+        restored_count = 0
+
+        async for db_session in get_session():
+            repo = Repository(db_session)
+            sessions = await repo.get_active_sessions()
+
+            for session in sessions:
+                recent_messages = await repo.get_recent_human_messages(
+                    session.id,
+                    limit=1,
+                )
+                message_count = await repo.count_messages(session.id)
+                dashboard = await repo.get_dashboard(session.id)
+                last_hearthkeeper_action = await repo.get_latest_action(
+                    session_id=session.id,
+                    action_type=AgentActionType.RESPONSE,
+                    mode="hearthkeeper",
+                )
+                last_activity_at = (
+                    recent_messages[0].timestamp if recent_messages else session.started_at
+                )
+
+                self.active_sessions[session.id] = {
+                    "channel_name": session.channel_name,
+                    "started_at": session.started_at,
+                    "message_count": message_count,
+                    "theme": session.theme,
+                    "dashboard": Repository.serialize_dashboard(dashboard),
+                    "last_hearthkeeper_prompt_at": (
+                        last_hearthkeeper_action.timestamp
+                        if last_hearthkeeper_action
+                        else None
+                    ),
+                }
+                self.chat_activity.record_activity(session.id, occurred_at=last_activity_at)
+                restored_count += 1
+
+        logger.info(f"Restored {restored_count} active sessions")
+        return restored_count
+
     async def end_session(self, session_id: str) -> None:
         """
         End a stream session and trigger ledger generation.
@@ -90,7 +201,7 @@ class AgentOrchestrator:
             logger.warning(f"Session {session_id} not found")
             return
         
-        async with async_session_factory() as db_session:
+        async for db_session in get_session():
             repo = Repository(db_session)
             await repo.end_session(session_id)
         
@@ -122,7 +233,7 @@ class AgentOrchestrator:
         actions = []
         agent_response = None
 
-        async with async_session_factory() as db_session:
+        async for db_session in get_session():
             repo = Repository(db_session)
             
             # Store the message
@@ -136,12 +247,13 @@ class AgentOrchestrator:
         # Record activity
         self.chat_activity.record_activity(session_id)
         session_info["message_count"] += 1
+        session_info["last_hearthkeeper_prompt_at"] = None
 
         # 1. Warden always analyzes (passive mode)
         warden_result = await self.warden.analyze_message(message)
         if warden_result["is_suspicious"]:
             actions.append(f"WARDEN_FLAG: {warden_result['severity']}")
-            async with async_session_factory() as db_session:
+            async for db_session in get_session():
                 repo = Repository(db_session)
                 await repo.record_action(
                     session_id=session_id,
@@ -153,9 +265,12 @@ class AgentOrchestrator:
 
         # 2. Check if we should suppress responses due to active chat
         recent_messages = []
-        async with async_session_factory() as db_session:
+        async for db_session in get_session():
             repo = Repository(db_session)
-            recent_messages = await repo.get_recent_messages(session_id, limit=10)
+            recent_messages = await repo.get_human_messages_since(
+                session_id=session_id,
+                since=datetime.utcnow() - timedelta(minutes=1),
+            )
 
         if self.response_suppression.should_suppress_response(len(recent_messages)):
             logger.debug("Response suppressed due to active chat")
@@ -164,25 +279,7 @@ class AgentOrchestrator:
                 "actions_taken": actions,
             }
 
-        # 3. Hearthkeeper: Generate prompt if chat inactive
-        if self.chat_activity.should_hearthkeeper_prompt(session_id):
-            try:
-                agent_response = await self.hearthkeeper.generate_prompt(
-                    theme=session_info.get("theme")
-                )
-                actions.append("HEARTHKEEPER_PROMPT")
-                async with async_session_factory() as db_session:
-                    repo = Repository(db_session)
-                    await repo.record_action(
-                        session_id=session_id,
-                        action_type=AgentActionType.RESPONSE,
-                        mode="hearthkeeper",
-                        description=agent_response,
-                    )
-            except Exception as e:
-                logger.error(f"Error in Hearthkeeper: {e}")
-
-        # 4. Librarian: Archive important messages (passive)
+        # 3. Librarian: Archive important messages (passive)
         if len(message) > 50:  # Archive longer messages
             await self.librarian.archive_message(message_id, message, username)
 
@@ -195,6 +292,313 @@ class AgentOrchestrator:
             "actions_taken": actions,
         }
 
+    async def emit_agent_response(
+        self,
+        session_id: str,
+        message: str,
+        mode: str,
+        triggered_by_message_id: str | None = None,
+    ) -> dict:
+        """Send an agent response through the outbound chat boundary."""
+        session_info = self.active_sessions.get(session_id)
+        if not session_info:
+            logger.warning(f"Session {session_id} not found")
+            return {"sent": False, "reason": "session_not_found"}
+
+        channel_name = session_info["channel_name"]
+        if not await self.can_interact_with_chat(channel_name):
+            logger.info(
+                "Agent response suppressed because stream is not live. Session: %s",
+                session_id,
+            )
+            return {"sent": False, "reason": "stream_not_live"}
+
+        sent = await send_chat_message(channel_name=channel_name, message=message)
+        bot_username = settings.TWITCH_BOT_USERNAME or "sanctum_chronicler"
+
+        async for db_session in get_session():
+            repo = Repository(db_session)
+            bot_message_id = await repo.add_chat_message(
+                session_id=session_id,
+                username=bot_username,
+                content=message,
+                is_bot=True,
+            )
+            action_id = await repo.record_action(
+                session_id=session_id,
+                action_type=AgentActionType.RESPONSE,
+                mode=mode,
+                description=message,
+                triggered_by_message_id=triggered_by_message_id,
+            )
+
+        session_info["message_count"] += 1
+        logger.info(
+            f"Agent response emitted. Session: {session_id}, Mode: {mode}, Sent: {sent}"
+        )
+        return {
+            "sent": sent,
+            "channel": channel_name,
+            "message_id": bot_message_id,
+            "action_id": action_id,
+        }
+
+    async def run_hearthkeeper_loop_test(
+        self,
+        session_id: str,
+        inactive_minutes: int = 16,
+    ) -> dict:
+        """Exercise the quiet-chat loop and verify it prompts exactly once."""
+        session_info = self.active_sessions.get(session_id)
+        if not session_info:
+            return {"passed": False, "reason": "session_not_found"}
+
+        simulated_activity_at = datetime.utcnow() - timedelta(minutes=inactive_minutes)
+        self.chat_activity.record_activity(
+            session_id=session_id,
+            occurred_at=simulated_activity_at,
+        )
+        session_info["last_hearthkeeper_prompt_at"] = None
+
+        before_count = await self._count_response_actions(
+            session_id=session_id,
+            mode="hearthkeeper",
+        )
+        first_tick = await self._tick_session(session_id)
+        second_tick = await self._tick_session(session_id)
+        after_count = await self._count_response_actions(
+            session_id=session_id,
+            mode="hearthkeeper",
+        )
+        session_info["last_hearthkeeper_prompt_at"] = (
+            datetime.utcnow() - self.hearthkeeper_prompt_interval
+        )
+        third_tick = await self._tick_session(session_id)
+        final_count = await self._count_response_actions(
+            session_id=session_id,
+            mode="hearthkeeper",
+        )
+        prompts_created = after_count - before_count
+        prompts_created_after_interval = final_count - before_count
+
+        return {
+            "passed": (
+                prompts_created == 1
+                and first_tick is not None
+                and second_tick is None
+                and third_tick is not None
+                and prompts_created_after_interval == 2
+            ),
+            "session_id": session_id,
+            "inactive_minutes": inactive_minutes,
+            "prompts_created": prompts_created,
+            "prompts_created_after_interval": prompts_created_after_interval,
+            "first_tick": first_tick,
+            "second_tick": second_tick,
+            "third_tick_after_interval": third_tick,
+        }
+
+    async def preview_hearthkeeper_prompt(self, session_id: str) -> dict:
+        """Generate a Hearthkeeper prompt preview without sending or recording it."""
+        session_info = self.active_sessions.get(session_id)
+        if not session_info:
+            return {"generated": False, "reason": "session_not_found"}
+
+        recent_discussion_messages = []
+        async for db_session in get_session():
+            repo = Repository(db_session)
+            recent_discussion_messages = await repo.get_recent_human_messages(
+                session_id=session_id,
+                limit=5,
+            )
+
+        recent_discussion = [
+            message.content for message in recent_discussion_messages[:5]
+        ]
+        agent_response = await self.hearthkeeper.generate_prompt(
+            theme=session_info.get("theme"),
+            dashboard=session_info.get("dashboard"),
+            recent_discussion=recent_discussion,
+        )
+
+        return {
+            "generated": True,
+            "session_id": session_id,
+            "agent_response": agent_response,
+            "theme": session_info.get("theme"),
+            "dashboard": session_info.get("dashboard"),
+            "recent_discussion": recent_discussion,
+        }
+
+    async def _count_response_actions(self, session_id: str, mode: str) -> int:
+        """Count response actions for a mode in a session."""
+        async for db_session in get_session():
+            repo = Repository(db_session)
+            actions = await repo.get_session_actions(session_id)
+            return sum(
+                1
+                for action in actions
+                if action.action_type == AgentActionType.RESPONSE
+                and action.mode == mode
+            )
+        return 0
+
+    def set_loop_interval(self, interval_seconds: float) -> None:
+        """Update how frequently the background agent loop runs."""
+        if interval_seconds < 1:
+            raise ValueError("Loop interval must be at least 1 second")
+        self.loop_interval_seconds = interval_seconds
+
+    def get_loop_status(self) -> dict:
+        """Get background loop configuration and current session count."""
+        return {
+            "interval_seconds": self.loop_interval_seconds,
+            "hearthkeeper_prompt_interval_minutes": int(
+                self.hearthkeeper_prompt_interval.total_seconds() / 60
+            ),
+            "active_session_count": len(self.active_sessions),
+        }
+
+    def get_hearthkeeper_runtime_status(self, session_id: str) -> dict:
+        """Get Hearthkeeper timing status for an active session."""
+        session_info = self.active_sessions.get(session_id)
+        if not session_info:
+            return {}
+
+        now = datetime.utcnow()
+        last_activity_at = self.chat_activity.last_activity_at(session_id)
+        last_prompt_at = session_info.get("last_hearthkeeper_prompt_at")
+
+        next_from_activity = None
+        if last_activity_at:
+            next_from_activity = last_activity_at + self.chat_activity.inactivity_threshold
+
+        next_from_prompt = None
+        if last_prompt_at:
+            next_from_prompt = last_prompt_at + self.hearthkeeper_prompt_interval
+
+        next_eligible_at = max(
+            [candidate for candidate in (next_from_activity, next_from_prompt) if candidate],
+            default=None,
+        )
+        seconds_until_next_prompt = None
+        if next_eligible_at:
+            seconds_until_next_prompt = max(
+                0,
+                int((next_eligible_at - now).total_seconds()),
+            )
+        quiet_enough = bool(
+            last_activity_at
+            and now - last_activity_at >= self.chat_activity.inactivity_threshold
+        )
+        prompt_interval_elapsed = bool(
+            not last_prompt_at
+            or now - last_prompt_at >= self.hearthkeeper_prompt_interval
+        )
+
+        return {
+            "last_activity_at": last_activity_at.isoformat() if last_activity_at else None,
+            "last_hearthkeeper_prompt_at": (
+                last_prompt_at.isoformat() if last_prompt_at else None
+            ),
+            "next_eligible_prompt_at": (
+                next_eligible_at.isoformat() if next_eligible_at else None
+            ),
+            "seconds_until_next_prompt": seconds_until_next_prompt,
+            "inactivity_threshold_minutes": int(
+                self.chat_activity.inactivity_threshold.total_seconds() / 60
+            ),
+            "prompt_interval_minutes": int(
+                self.hearthkeeper_prompt_interval.total_seconds() / 60
+            ),
+            "can_prompt_now": quiet_enough and prompt_interval_elapsed,
+        }
+
+    async def tick(self) -> list[dict]:
+        """Evaluate active sessions for time-based agent behavior."""
+        results = []
+        for session_id in list(self.active_sessions.keys()):
+            result = await self._tick_session(session_id)
+            if result:
+                results.append(result)
+        return results
+
+    async def _tick_session(self, session_id: str) -> dict | None:
+        """Evaluate a single active session during the background loop."""
+        session_info = self.active_sessions.get(session_id)
+        if not session_info:
+            return None
+
+        active_chat_messages = []
+        recent_discussion_messages = []
+        async for db_session in get_session():
+            repo = Repository(db_session)
+            active_chat_messages = await repo.get_human_messages_since(
+                session_id=session_id,
+                since=datetime.utcnow() - timedelta(minutes=1),
+            )
+            recent_discussion_messages = await repo.get_recent_human_messages(
+                session_id=session_id,
+                limit=5,
+            )
+
+        if self.response_suppression.should_suppress_response(len(active_chat_messages)):
+            return {
+                "session_id": session_id,
+                "actions_taken": [],
+                "agent_response": None,
+                "reason": "active_chat",
+            }
+
+        if not self.chat_activity.should_hearthkeeper_prompt(session_id):
+            return None
+
+        if not await self.can_interact_with_chat(session_info["channel_name"]):
+            return None
+
+        last_activity_at = self.chat_activity.last_activity_at(session_id)
+        last_prompt_at = session_info.get("last_hearthkeeper_prompt_at")
+        if last_activity_at and last_prompt_at and last_activity_at > last_prompt_at:
+            session_info["last_hearthkeeper_prompt_at"] = None
+            last_prompt_at = None
+        if (
+            last_prompt_at
+            and datetime.utcnow() - last_prompt_at < self.hearthkeeper_prompt_interval
+        ):
+            return None
+
+        try:
+            recent_discussion = [
+                message.content for message in recent_discussion_messages[:5]
+            ]
+            agent_response = await self.hearthkeeper.generate_prompt(
+                theme=session_info.get("theme"),
+                dashboard=session_info.get("dashboard"),
+                recent_discussion=recent_discussion,
+            )
+            session_info["last_hearthkeeper_prompt_at"] = datetime.utcnow()
+            delivery = await self.emit_agent_response(
+                session_id=session_id,
+                message=agent_response,
+                mode="hearthkeeper",
+            )
+
+            return {
+                "session_id": session_id,
+                "actions_taken": ["HEARTHKEEPER_PROMPT"],
+                "agent_response": agent_response,
+                "delivery": delivery,
+                "reason": "inactive_chat",
+            }
+        except Exception as e:
+            logger.error(f"Error in Hearthkeeper loop: {e}")
+            return {
+                "session_id": session_id,
+                "actions_taken": [],
+                "agent_response": None,
+                "reason": "hearthkeeper_error",
+            }
+
     async def get_session_status(self, session_id: str) -> dict:
         """Get status of a session."""
         if session_id not in self.active_sessions:
@@ -208,4 +612,5 @@ class AgentOrchestrator:
             "message_count": session["message_count"],
             "uptime_seconds": (datetime.utcnow() - session["started_at"]).total_seconds(),
             "theme": session.get("theme"),
+            "dashboard": session.get("dashboard"),
         }

app/agent/policies.py

@@ -19,9 +19,17 @@ class ChatActivityPolicy:
         self.inactivity_threshold = timedelta(minutes=inactivity_threshold_minutes)
         self.last_message_time: dict[str, datetime] = {}
 
-    def record_activity(self, session_id: str) -> None:
+    def record_activity(self, session_id: str, occurred_at: datetime | None = None) -> None:
         """Record that chat activity occurred."""
-        self.last_message_time[session_id] = datetime.utcnow()
+        self.last_message_time[session_id] = occurred_at or datetime.utcnow()
+
+    def last_activity_at(self, session_id: str) -> datetime | None:
+        """Get the most recent chat activity time for a session."""
+        return self.last_message_time.get(session_id)
+
+    def clear_activity(self, session_id: str) -> None:
+        """Stop tracking activity for an ended session."""
+        self.last_message_time.pop(session_id, None)
 
     def minutes_since_activity(self, session_id: str) -> int:
         """Get minutes since last chat message."""

app/config.py

@@ -1,5 +1,6 @@
 """Configuration management using pydantic-settings."""
 
+from pydantic import field_validator
 from pydantic_settings import BaseSettings
 from typing import Optional
 
@@ -11,6 +12,31 @@ class Settings(BaseSettings):
     APP_NAME: str = "Sanctum Chronicler"
     APP_ENV: str = "development"
     DEBUG: bool = False
+    ADMIN_API_KEY: Optional[str] = None
+
+    @field_validator("DEBUG", mode="before")
+    @classmethod
+    def parse_debug(cls, value: object) -> bool:
+        """Parse permissive runtime DEBUG values from shell environments."""
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in {"1", "true", "t", "yes", "y", "on", "debug"}:
+                return True
+            if normalized in {
+                "0",
+                "false",
+                "f",
+                "no",
+                "n",
+                "off",
+                "release",
+                "prod",
+                "production",
+            }:
+                return False
+        return False
 
     # Database
     DATABASE_URL: str = "postgresql+asyncpg://sanctum:password@localhost:5432/sanctum"
@@ -20,6 +46,11 @@ class Settings(BaseSettings):
     TWITCH_CLIENT_SECRET: Optional[str] = None
     TWITCH_BOT_USERNAME: Optional[str] = None
     TWITCH_CHANNEL_NAME: Optional[str] = None
+    TWITCH_ACCESS_TOKEN: Optional[str] = None
+    TWITCH_CHAT_ENABLED: bool = True
+    TWITCH_REQUIRE_LIVE_STREAM: bool = True
+    TWITCH_LIVE_STATUS_CACHE_SECONDS: int = 60
+    TWITCH_OFFLINE_CHAT_TEST_MODE: bool = False
 
     # LLM
     LLM_PROVIDER: Optional[str] = None  # "openai", "ollama", "lm_studio", or None
@@ -27,6 +58,10 @@ class Settings(BaseSettings):
     LLM_API_KEY: Optional[str] = None
     LLM_MODEL: str = "gpt-3.5-turbo"
 
+    # Agent loop
+    AGENT_LOOP_INTERVAL_SECONDS: float = 60.0
+    HEARTHKEEPER_PROMPT_INTERVAL_MINUTES: int = 15
+
     # Export
     EXPORT_PATH: str = "exports"
 

app/exports/__init__.py

@@ -0,0 +1 @@
+"""Exports module."""

app/exports/markdown.py

@@ -0,0 +1,142 @@
+"""Markdown export functionality for stream ledgers."""
+
+import logging
+from datetime import datetime
+from pathlib import Path
+
+from app.config import settings
+from app.memory.database import get_session
+from app.memory.repository import Repository
+
+logger = logging.getLogger(__name__)
+
+
+class MarkdownExporter:
+    """Exports stream session data as markdown ledgers."""
+
+    def __init__(self, export_path: str | None = None):
+        """
+        Initialize exporter.
+        
+        Args:
+            export_path: Directory to export ledgers to (defaults to settings.EXPORT_PATH)
+        """
+        self.export_path = Path(export_path or settings.EXPORT_PATH)
+        self.export_path.mkdir(parents=True, exist_ok=True)
+
+    async def export_session(self, session_id: str) -> str:
+        """
+        Export a session as a markdown ledger.
+        
+        Args:
+            session_id: Session ID to export
+            
+        Returns:
+            Markdown content
+        """
+        async for db_session in get_session():
+            repo = Repository(db_session)
+            
+            session = await repo.get_session(session_id)
+            if not session:
+                logger.warning(f"Session {session_id} not found")
+                return ""
+            
+            # Gather data
+            messages = await repo.get_recent_messages(session_id, limit=1000)
+            actions = await repo.get_session_actions(session_id)
+            clips = await repo.get_clip_candidates(session_id)
+            seeds = await repo.get_blog_seeds(session_id)
+            dashboard = await repo.get_dashboard(session_id)
+
+        dashboard_data = Repository.serialize_dashboard(dashboard)
+
+        # Build markdown
+        date = session.started_at.strftime("%Y-%m-%d")
+        ledger = f"# Sanctum Ledger — {date}\n\n"
+        ledger += f"**Channel:** {session.channel_name}\n"
+        ledger += f"**Started:** {session.started_at.isoformat()}\n"
+        if session.ended_at:
+            ledger += f"**Ended:** {session.ended_at.isoformat()}\n"
+        ledger += "\n"
+        
+        # Stream Theme
+        ledger += "## Stream Theme\n"
+        if dashboard_data:
+            if dashboard_data.get("stream_title"):
+                ledger += f"**Title:** {dashboard_data['stream_title']}\n"
+            if dashboard_data.get("game"):
+                ledger += f"**Game:** {dashboard_data['game']}\n"
+            if dashboard_data.get("mood"):
+                ledger += f"**Mood:** {dashboard_data['mood']}\n"
+            if dashboard_data.get("content_angle"):
+                ledger += f"**Content Angle:** {dashboard_data['content_angle']}\n"
+            if dashboard_data.get("session_goals"):
+                ledger += "\n**Session Goals**\n"
+                for goal in dashboard_data["session_goals"]:
+                    ledger += f"- {goal}\n"
+        elif session.theme:
+            ledger += f"{session.theme}\n"
+        else:
+            ledger += "*No theme recorded*\n"
+        ledger += "\n"
+        
+        # Notable Discussion
+        ledger += "## Notable Discussion\n"
+        if messages:
+            for msg in messages[:20]:  # Latest 20 messages
+                ledger += f"- **{msg.username}:** {msg.content[:100]}\n"
+        else:
+            ledger += "*No messages recorded*\n"
+        ledger += "\n"
+        
+        # Agent Actions
+        ledger += "## Agent Actions\n"
+        if actions:
+            for action in actions:
+                ledger += f"- **{action.mode}** ({action.action_type}): {action.description}\n"
+        else:
+            ledger += "*No agent actions recorded*\n"
+        ledger += "\n"
+        
+        # Clip Candidates
+        ledger += "## Clip Candidates\n"
+        if clips:
+            for clip in clips:
+                ledger += f"- {clip.reason}\n"
+        else:
+            ledger += "*No clip candidates identified*\n"
+        ledger += "\n"
+        
+        # Blog Seeds
+        ledger += "## Blog Seeds\n"
+        if seeds:
+            for seed in seeds:
+                ledger += f"- **{seed.topic}:** {seed.description}\n"
+        else:
+            ledger += "*No blog seeds proposed*\n"
+        ledger += "\n"
+        
+        logger.info(f"Generated ledger for session {session_id}")
+        return ledger
+
+    async def save_session_ledger(self, session_id: str) -> Path:
+        """
+        Export session and save to file.
+        
+        Args:
+            session_id: Session ID
+            
+        Returns:
+            Path to saved file
+        """
+        ledger = await self.export_session(session_id)
+        
+        date = datetime.utcnow().strftime("%Y-%m-%d")
+        filename = f"ledger_{date}_{session_id[:8]}.md"
+        filepath = self.export_path / filename
+        
+        filepath.write_text(ledger, encoding="utf-8")
+        logger.info(f"Saved ledger to {filepath}")
+        
+        return filepath

app/llm/client.py

@@ -105,6 +105,12 @@ class LLMClient:
         Used when no provider is configured or for testing.
         """
         logger.debug(f"Mock generation for prompt: {prompt[:50]}...")
+
+        if "content angle:" in prompt.lower():
+            for line in prompt.splitlines():
+                if line.lower().startswith("content angle:"):
+                    angle = line.split(":", 1)[1].strip()
+                    return f"The quiet here keeps circling back to {angle}."
         
         # Simple deterministic responses for testing
         if "hello" in prompt.lower():

app/llm/prompts.py

@@ -7,11 +7,37 @@ class PromptTemplates:
     """Collection of prompt templates for different modes."""
 
     @staticmethod
-    def gentle_prompt(current_theme: Optional[str] = None) -> str:
+    def gentle_prompt(
+        current_theme: Optional[str] = None,
+        dashboard: Optional[dict] = None,
+        recent_discussion: Optional[list[str]] = None,
+    ) -> str:
         """Generate a gentle prompt when chat has been inactive."""
+        dashboard = dashboard or {}
+        recent_discussion = recent_discussion or []
+
+        context_lines = [
+            "Generate one brief Hearthkeeper prompt for a calm, reflective livestream.",
+            "The prompt must be restrained, thematic, and not engagement bait.",
+        ]
+        if dashboard.get("stream_title"):
+            context_lines.append(f"Stream title: {dashboard['stream_title']}")
+        if dashboard.get("game"):
+            context_lines.append(f"Game: {dashboard['game']}")
+        if dashboard.get("mood"):
+            context_lines.append(f"Mood: {dashboard['mood']}")
+        if dashboard.get("content_angle"):
+            context_lines.append(f"Content angle: {dashboard['content_angle']}")
+        if dashboard.get("session_goals"):
+            goals = "; ".join(dashboard["session_goals"])
+            context_lines.append(f"Session goals: {goals}")
         if current_theme:
-            return f"Gently prompt the chat about: {current_theme}"
-        return "Generate a gentle, inviting prompt to encourage discussion in the stream."
+            context_lines.append(f"Current theme: {current_theme}")
+        if recent_discussion:
+            context_lines.append("Recent human discussion:")
+            context_lines.extend(f"- {message}" for message in recent_discussion[:5])
+
+        return "\n".join(context_lines)
 
     @staticmethod
     def steward_response(message: str, context: Optional[str] = None) -> str:

app/main.py

@@ -1,14 +1,22 @@
 """FastAPI main application."""
 
-from fastapi import FastAPI, HTTPException
-from fastapi.responses import JSONResponse
+import asyncio
+import secrets
+from contextlib import suppress
+from pydantic import BaseModel, Field
+from fastapi import Depends, FastAPI, Form, Header, HTTPException
 from datetime import datetime
 import logging
 
 from app.config import settings
 from app.agent.orchestrator import AgentOrchestrator
 from app.memory.database import init_db
+from app.memory.database import get_session as get_db_session
+from app.memory.models import AgentActionType
+from app.memory.repository import Repository
 from app.exports.markdown import MarkdownExporter
+from app.twitch.chat import TwitchChatMessage, TwitchIRCClient, set_active_client
+from app.twitch.live import TwitchLiveStatusService
 
 logger = logging.getLogger(__name__)
 
@@ -18,17 +26,148 @@ app = FastAPI(
     version="0.1.0",
 )
 
+
+class DashboardRequest(BaseModel):
+    """Request body for saving a stream dashboard."""
+
+    session_id: str | None = None
+    raw_markdown: str
+    stream_title: str | None = None
+    game: str | None = None
+    mood: str | None = None
+    go_live_notification: str | None = None
+    social_post: str | None = None
+    session_goals: list[str] = Field(default_factory=list)
+    content_angle: str | None = None
+
 # Global orchestrator instance
 orchestrator: AgentOrchestrator | None = None
+agent_loop_task: asyncio.Task | None = None
+twitch_chat_client: TwitchIRCClient | None = None
+twitch_chat_task: asyncio.Task | None = None
+twitch_session_id: str | None = None
+twitch_live_status: TwitchLiveStatusService | None = None
+offline_chat_test_mode: bool = settings.TWITCH_OFFLINE_CHAT_TEST_MODE
+
+
+async def require_admin(
+    admin_token: str | None = Header(default=None, alias="X-Admin-Token"),
+) -> None:
+    """Require the configured admin token for mutable/control endpoints."""
+    if not settings.ADMIN_API_KEY:
+        raise HTTPException(status_code=503, detail="Admin API key is not configured")
+    if not admin_token or not secrets.compare_digest(
+        admin_token,
+        settings.ADMIN_API_KEY,
+    ):
+        raise HTTPException(status_code=401, detail="Invalid admin token")
+
+
+async def agent_loop() -> None:
+    """Run periodic time-based agent behavior for active sessions."""
+    if not orchestrator:
+        return
+
+    while True:
+        try:
+            results = await orchestrator.tick()
+            if results:
+                logger.info(f"Agent loop actions: {results}")
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            logger.error(f"Agent loop tick failed: {e}")
+
+        await asyncio.sleep(orchestrator.loop_interval_seconds)
+
+
+def twitch_configured() -> bool:
+    """Return whether Twitch chat has enough runtime configuration to start."""
+    return bool(
+        settings.TWITCH_CHAT_ENABLED
+        and settings.TWITCH_CHANNEL_NAME
+        and settings.TWITCH_BOT_USERNAME
+        and settings.TWITCH_ACCESS_TOKEN
+    )
+
+
+async def channel_is_live_for_chat(channel_name: str) -> bool:
+    """Return whether the agent may interact with chat for a channel."""
+    if offline_chat_test_mode:
+        return True
+    if not settings.TWITCH_REQUIRE_LIVE_STREAM:
+        return True
+    if not twitch_live_status:
+        return False
+    status = await twitch_live_status.get_status(channel_name)
+    return status.is_live
+
+
+async def get_or_create_twitch_session(channel_name: str) -> str:
+    """Use an active session for the Twitch channel, or create one."""
+    if not orchestrator:
+        raise RuntimeError("Orchestrator not initialized")
+
+    return await orchestrator.ensure_single_active_session_for_channel(channel_name)
+
+
+async def handle_twitch_chat_message(message: TwitchChatMessage) -> None:
+    """Route a Twitch chat message into the orchestrator."""
+    if not orchestrator or not twitch_session_id:
+        return
+
+    if settings.TWITCH_BOT_USERNAME and (
+        message.username.lower() == settings.TWITCH_BOT_USERNAME.lower()
+    ):
+        return
+
+    await orchestrator.handle_chat_message(
+        session_id=twitch_session_id,
+        username=message.display_name or message.username,
+        message=message.content,
+    )
+
+
+async def start_twitch_chat() -> None:
+    """Start Twitch chat monitoring when configured."""
+    global twitch_chat_client, twitch_chat_task, twitch_session_id
+
+    if not orchestrator or not twitch_configured():
+        logger.info("Twitch chat listener not started; configuration is incomplete")
+        return
+
+    twitch_session_id = await get_or_create_twitch_session(settings.TWITCH_CHANNEL_NAME)
+    twitch_chat_client = TwitchIRCClient(
+        channel_name=settings.TWITCH_CHANNEL_NAME,
+        bot_username=settings.TWITCH_BOT_USERNAME,
+        access_token=settings.TWITCH_ACCESS_TOKEN,
+        on_message=handle_twitch_chat_message,
+    )
+    set_active_client(twitch_chat_client)
+    twitch_chat_task = asyncio.create_task(twitch_chat_client.run())
+    logger.info(
+        "Twitch chat listener starting for %s on session %s",
+        settings.TWITCH_CHANNEL_NAME,
+        twitch_session_id,
+    )
 
 
 @app.on_event("startup")
 async def startup_event():
     """Initialize database and services on startup."""
-    global orchestrator
+    global orchestrator, agent_loop_task, twitch_live_status
     try:
         await init_db()
-        orchestrator = AgentOrchestrator()
+        twitch_live_status = TwitchLiveStatusService(
+            cache_seconds=settings.TWITCH_LIVE_STATUS_CACHE_SECONDS
+        )
+        orchestrator = AgentOrchestrator(
+            loop_interval_seconds=settings.AGENT_LOOP_INTERVAL_SECONDS
+        )
+        orchestrator.set_chat_interaction_gate(channel_is_live_for_chat)
+        await orchestrator.restore_active_sessions()
+        await start_twitch_chat()
+        agent_loop_task = asyncio.create_task(agent_loop())
         logger.info("Application started successfully")
     except Exception as e:
         logger.error(f"Failed to start application: {e}")
@@ -38,6 +177,21 @@ async def startup_event():
 @app.on_event("shutdown")
 async def shutdown_event():
     """Clean up resources on shutdown."""
+    global twitch_chat_client, twitch_live_status
+
+    if twitch_chat_task:
+        twitch_chat_task.cancel()
+        with suppress(asyncio.CancelledError):
+            await twitch_chat_task
+    if twitch_chat_client:
+        await twitch_chat_client.disconnect()
+        set_active_client(None)
+        twitch_chat_client = None
+    twitch_live_status = None
+    if agent_loop_task:
+        agent_loop_task.cancel()
+        with suppress(asyncio.CancelledError):
+            await agent_loop_task
     logger.info("Application shutting down")
 
 
@@ -52,8 +206,8 @@ async def health_check() -> dict:
     }
 
 
-@app.post("/admin/session/start")
-async def start_session(channel_name: str) -> dict:
+@app.post("/admin/session/start", dependencies=[Depends(require_admin)])
+async def start_session(channel_name: str = Form(...)) -> dict:
     """Start a new stream session."""
     if not orchestrator:
         raise HTTPException(status_code=503, detail="Orchestrator not initialized")
@@ -67,8 +221,8 @@ async def start_session(channel_name: str) -> dict:
     }
 
 
-@app.post("/admin/session/end")
-async def end_session(session_id: str) -> dict:
+@app.post("/admin/session/end", dependencies=[Depends(require_admin)])
+async def end_session(session_id: str = Form(...)) -> dict:
     """End the current stream session."""
     if not orchestrator:
         raise HTTPException(status_code=503, detail="Orchestrator not initialized")
@@ -81,8 +235,8 @@ async def end_session(session_id: str) -> dict:
     }
 
 
-@app.post("/admin/test-message")
-async def test_message(session_id: str, message: str, username: str = "test_user") -> dict:
+@app.post("/admin/test-message", dependencies=[Depends(require_admin)])
+async def test_message(session_id: str = Form(...), message: str = Form(...), username: str = Form("test_user")) -> dict:
     """Send a test message to the orchestrator."""
     if not orchestrator:
         raise HTTPException(status_code=503, detail="Orchestrator not initialized")
@@ -100,7 +254,225 @@ async def test_message(session_id: str, message: str, username: str = "test_user
     }
 
 
-@app.get("/admin/ledger")
+@app.post("/admin/test-agent-response", dependencies=[Depends(require_admin)])
+async def test_agent_response(
+    session_id: str = Form(...),
+    message: str = Form(...),
+    mode: str = Form("admin"),
+) -> dict:
+    """Send a test agent response through the outbound boundary."""
+    if not orchestrator:
+        raise HTTPException(status_code=503, detail="Orchestrator not initialized")
+
+    delivery = await orchestrator.emit_agent_response(
+        session_id=session_id,
+        message=message,
+        mode=mode,
+    )
+    if not delivery.get("sent"):
+        raise HTTPException(status_code=404, detail=delivery.get("reason", "send_failed"))
+
+    return {
+        "status": "agent_response_emitted",
+        "delivery": delivery,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.post("/admin/test-loop-inactivity", dependencies=[Depends(require_admin)])
+async def test_loop_inactivity(
+    session_id: str = Form(...),
+    inactive_minutes: int = Form(16),
+) -> dict:
+    """Verify the quiet-chat loop records exactly one Hearthkeeper prompt."""
+    if not orchestrator:
+        raise HTTPException(status_code=503, detail="Orchestrator not initialized")
+
+    result = await orchestrator.run_hearthkeeper_loop_test(
+        session_id=session_id,
+        inactive_minutes=inactive_minutes,
+    )
+    if result.get("reason") == "session_not_found":
+        raise HTTPException(status_code=404, detail="Active session not found")
+
+    return {
+        "status": "passed" if result["passed"] else "failed",
+        "result": result,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.post("/admin/hearthkeeper/preview", dependencies=[Depends(require_admin)])
+async def preview_hearthkeeper_prompt(session_id: str | None = Form(None)) -> dict:
+    """Generate a Hearthkeeper prompt preview without posting to Twitch."""
+    if not orchestrator:
+        raise HTTPException(status_code=503, detail="Orchestrator not initialized")
+
+    resolved_session_id = session_id or get_current_stream_session_id()
+    if not resolved_session_id:
+        raise HTTPException(status_code=404, detail="No active stream session found")
+
+    result = await orchestrator.preview_hearthkeeper_prompt(resolved_session_id)
+    if result.get("reason") == "session_not_found":
+        raise HTTPException(status_code=404, detail="Active session not found")
+
+    live_status = None
+    session_info = orchestrator.active_sessions.get(resolved_session_id)
+    if twitch_live_status and session_info:
+        live_status = (
+            await twitch_live_status.get_status(session_info["channel_name"])
+        ).to_dict()
+
+    return {
+        "status": "preview_generated" if result.get("generated") else "failed",
+        "preview": result,
+        "would_post_now": await orchestrator.can_interact_with_chat(
+            session_info["channel_name"]
+        ) if session_info else False,
+        "live_status": live_status,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+def serialize_dashboard(dashboard) -> dict:
+    """Serialize a dashboard database model into an API response."""
+    return Repository.serialize_dashboard(dashboard)
+
+
+def dashboard_status(dashboard: dict | None) -> dict:
+    """Return a compact dashboard status for operator checks."""
+    if not dashboard:
+        return {
+            "present": False,
+            "stream_title": None,
+            "game": None,
+            "mood": None,
+            "content_angle": None,
+            "session_goal_count": 0,
+            "updated_at": None,
+        }
+
+    return {
+        "present": True,
+        "stream_title": dashboard.get("stream_title"),
+        "game": dashboard.get("game"),
+        "mood": dashboard.get("mood"),
+        "content_angle": dashboard.get("content_angle"),
+        "session_goal_count": len(dashboard.get("session_goals") or []),
+        "updated_at": dashboard.get("updated_at"),
+    }
+
+
+def get_current_stream_session_id() -> str | None:
+    """Resolve the session currently bound to Twitch or the only active session."""
+    if not orchestrator:
+        return None
+    if twitch_session_id and twitch_session_id in orchestrator.active_sessions:
+        return twitch_session_id
+    if len(orchestrator.active_sessions) == 1:
+        return next(iter(orchestrator.active_sessions))
+    return None
+
+
+async def save_dashboard_for_session(
+    session_id: str,
+    request: DashboardRequest,
+) -> dict:
+    """Persist a dashboard and refresh live orchestrator context."""
+    async for db_session in get_db_session():
+        repo = Repository(db_session)
+        stream_session = await repo.get_session(session_id)
+        if not stream_session:
+            raise HTTPException(status_code=404, detail="Session not found")
+
+        dashboard = await repo.upsert_dashboard(
+            session_id=session_id,
+            raw_markdown=request.raw_markdown,
+            stream_title=request.stream_title,
+            game=request.game,
+            mood=request.mood,
+            go_live_notification=request.go_live_notification,
+            social_post=request.social_post,
+            session_goals=request.session_goals,
+            content_angle=request.content_angle,
+        )
+
+    dashboard_data = serialize_dashboard(dashboard)
+    if orchestrator and session_id in orchestrator.active_sessions:
+        orchestrator.active_sessions[session_id]["dashboard"] = dashboard_data
+        orchestrator.active_sessions[session_id]["theme"] = (
+            request.content_angle or request.stream_title
+        )
+
+    return dashboard_data
+
+
+@app.post("/admin/session/dashboard", dependencies=[Depends(require_admin)])
+async def save_session_dashboard(request: DashboardRequest) -> dict:
+    """Create or update the approved dashboard for a stream session."""
+    if not request.session_id:
+        raise HTTPException(status_code=400, detail="session_id is required")
+
+    dashboard = await save_dashboard_for_session(request.session_id, request)
+    return {
+        "status": "dashboard_saved",
+        "dashboard": dashboard,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.post("/admin/stream/dashboard", dependencies=[Depends(require_admin)])
+async def save_current_stream_dashboard(request: DashboardRequest) -> dict:
+    """Create or update the approved dashboard for the current stream session."""
+    session_id = get_current_stream_session_id()
+    if not session_id:
+        raise HTTPException(status_code=404, detail="No active stream session found")
+
+    dashboard = await save_dashboard_for_session(session_id, request)
+    return {
+        "status": "dashboard_saved",
+        "session_id": session_id,
+        "dashboard": dashboard,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.get("/admin/session/dashboard", dependencies=[Depends(require_admin)])
+async def get_session_dashboard(session_id: str) -> dict:
+    """Get the approved dashboard for a stream session."""
+    async for db_session in get_db_session():
+        repo = Repository(db_session)
+        dashboard = await repo.get_dashboard(session_id)
+        if not dashboard:
+            raise HTTPException(status_code=404, detail="Dashboard not found")
+
+    return {
+        "dashboard": serialize_dashboard(dashboard),
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.get("/admin/stream/dashboard", dependencies=[Depends(require_admin)])
+async def get_current_stream_dashboard() -> dict:
+    """Get the approved dashboard for the current stream session."""
+    session_id = get_current_stream_session_id()
+    if not session_id:
+        raise HTTPException(status_code=404, detail="No active stream session found")
+
+    async for db_session in get_db_session():
+        repo = Repository(db_session)
+        dashboard = await repo.get_dashboard(session_id)
+        if not dashboard:
+            raise HTTPException(status_code=404, detail="Dashboard not found")
+
+    return {
+        "session_id": session_id,
+        "dashboard": serialize_dashboard(dashboard),
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.get("/admin/ledger", dependencies=[Depends(require_admin)])
 async def get_ledger(session_id: str) -> dict:
     """Get the markdown ledger for a session."""
     if not orchestrator:
@@ -116,6 +488,182 @@ async def get_ledger(session_id: str) -> dict:
     }
 
 
+@app.get("/admin/session/status", dependencies=[Depends(require_admin)])
+async def get_session_status(session_id: str) -> dict:
+    """Get status for an active stream session."""
+    if not orchestrator:
+        raise HTTPException(status_code=503, detail="Orchestrator not initialized")
+
+    status = await orchestrator.get_session_status(session_id)
+    if not status:
+        raise HTTPException(status_code=404, detail="Active session not found")
+
+    return {
+        **status,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.get("/admin/loop/status", dependencies=[Depends(require_admin)])
+async def get_loop_status() -> dict:
+    """Get the background agent loop runtime configuration."""
+    if not orchestrator:
+        raise HTTPException(status_code=503, detail="Orchestrator not initialized")
+
+    return {
+        "status": "running" if agent_loop_task and not agent_loop_task.done() else "stopped",
+        **orchestrator.get_loop_status(),
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.get("/admin/twitch/status", dependencies=[Depends(require_admin)])
+async def get_twitch_status() -> dict:
+    """Get Twitch chat connection status."""
+    configured = twitch_configured()
+    client_status = twitch_chat_client.status() if twitch_chat_client else {}
+    live_status = None
+    if twitch_live_status and settings.TWITCH_CHANNEL_NAME:
+        live_status = (
+            await twitch_live_status.get_status(settings.TWITCH_CHANNEL_NAME)
+        ).to_dict()
+    return {
+        "configured": configured,
+        "running": bool(twitch_chat_task and not twitch_chat_task.done()),
+        "require_live_stream": settings.TWITCH_REQUIRE_LIVE_STREAM,
+        "offline_chat_test_mode": offline_chat_test_mode,
+        "live_status": live_status,
+        "session_id": twitch_session_id,
+        **client_status,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.post("/admin/twitch/offline-test-mode", dependencies=[Depends(require_admin)])
+async def set_offline_chat_test_mode(enabled: bool = Form(...)) -> dict:
+    """Allow or block Twitch chat posting while the stream is offline."""
+    global offline_chat_test_mode
+
+    offline_chat_test_mode = enabled
+    return {
+        "status": "offline_chat_test_mode_updated",
+        "offline_chat_test_mode": offline_chat_test_mode,
+        "warning": (
+            "Agent may post to Twitch chat while the stream is offline"
+            if offline_chat_test_mode
+            else None
+        ),
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.get("/admin/stream/status", dependencies=[Depends(require_admin)])
+async def get_current_stream_status() -> dict:
+    """Get one operator view of the current Twitch stream runtime state."""
+    if not orchestrator:
+        raise HTTPException(status_code=503, detail="Orchestrator not initialized")
+
+    session_id = twitch_session_id
+    if not session_id and len(orchestrator.active_sessions) == 1:
+        session_id = next(iter(orchestrator.active_sessions))
+    if not session_id or session_id not in orchestrator.active_sessions:
+        raise HTTPException(status_code=404, detail="No active stream session found")
+
+    session_info = orchestrator.active_sessions[session_id]
+    latest_human_message = None
+    latest_hearthkeeper_action = None
+    dashboard = None
+    async for db_session in get_db_session():
+        repo = Repository(db_session)
+        dashboard = await repo.get_dashboard(session_id)
+        latest_human_message = await repo.get_latest_human_message(session_id)
+        latest_hearthkeeper_action = await repo.get_latest_action(
+            session_id=session_id,
+            action_type=AgentActionType.RESPONSE,
+            mode="hearthkeeper",
+        )
+
+    dashboard_data = serialize_dashboard(dashboard)
+    runtime = orchestrator.get_hearthkeeper_runtime_status(session_id)
+    twitch_status = twitch_chat_client.status() if twitch_chat_client else {}
+    live_status = None
+    if twitch_live_status:
+        live_status = (
+            await twitch_live_status.get_status(session_info["channel_name"])
+        ).to_dict()
+    chat_interaction_allowed = (
+        offline_chat_test_mode
+        or not settings.TWITCH_REQUIRE_LIVE_STREAM
+        or bool(live_status and live_status["is_live"])
+    )
+
+    return {
+        "session": {
+            "id": session_id,
+            "channel": session_info["channel_name"],
+            "started_at": session_info["started_at"].isoformat(),
+            "uptime_seconds": (
+                datetime.utcnow() - session_info["started_at"]
+            ).total_seconds(),
+            "message_count": session_info["message_count"],
+            "theme": session_info.get("theme"),
+        },
+        "twitch": {
+            "configured": twitch_configured(),
+            "running": bool(twitch_chat_task and not twitch_chat_task.done()),
+            "require_live_stream": settings.TWITCH_REQUIRE_LIVE_STREAM,
+            "offline_chat_test_mode": offline_chat_test_mode,
+            "live_status": live_status,
+            "session_id": twitch_session_id,
+            **twitch_status,
+        },
+        "dashboard": dashboard_status(dashboard_data),
+        "loop": {
+            "running": bool(agent_loop_task and not agent_loop_task.done()),
+            "interval_seconds": orchestrator.loop_interval_seconds,
+            "active_session_count": len(orchestrator.active_sessions),
+        },
+        "hearthkeeper": {
+            **runtime,
+            "chat_interaction_allowed": chat_interaction_allowed,
+            "last_human_chat_at": (
+                latest_human_message.timestamp.isoformat()
+                if latest_human_message
+                else None
+            ),
+            "last_posted_at": (
+                latest_hearthkeeper_action.timestamp.isoformat()
+                if latest_hearthkeeper_action
+                else None
+            ),
+            "last_posted_message": (
+                latest_hearthkeeper_action.description
+                if latest_hearthkeeper_action
+                else None
+            ),
+        },
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@app.post("/admin/loop/frequency", dependencies=[Depends(require_admin)])
+async def set_loop_frequency(interval_seconds: float = Form(...)) -> dict:
+    """Set how frequently the background agent loop runs."""
+    if not orchestrator:
+        raise HTTPException(status_code=503, detail="Orchestrator not initialized")
+
+    try:
+        orchestrator.set_loop_interval(interval_seconds)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+
+    return {
+        "status": "loop_frequency_updated",
+        "interval_seconds": orchestrator.loop_interval_seconds,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(

app/memory/models.py

@@ -35,6 +35,24 @@ class ChatMessage(Base):
     is_moderator = Column(Boolean, default=False)
 
 
+class StreamDashboard(Base):
+    """Stores the approved stream dashboard for a session."""
+
+    __tablename__ = "stream_dashboards"
+
+    session_id = Column(String, primary_key=True)
+    raw_markdown = Column(Text, nullable=False)
+    stream_title = Column(String, nullable=True)
+    game = Column(String, nullable=True)
+    mood = Column(String, nullable=True)
+    go_live_notification = Column(Text, nullable=True)
+    social_post = Column(Text, nullable=True)
+    session_goals = Column(Text, nullable=True)  # JSON array of strings
+    content_angle = Column(Text, nullable=True)
+    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
+    updated_at = Column(DateTime, default=datetime.utcnow, nullable=False)
+
+
 class AgentActionType(str, Enum):
     """Types of actions the agent can take."""
 

app/memory/repository.py

@@ -1,13 +1,15 @@
 """Data access layer for database operations."""
 
+import json
 import logging
 import uuid
 from datetime import datetime
-from sqlalchemy import select, update
+from sqlalchemy import func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.memory.models import (
     StreamSession,
+    StreamDashboard,
     ChatMessage,
     AgentAction,
     ClipCandidate,
@@ -56,6 +58,98 @@ class Repository:
         result = await self.session.execute(stmt)
         return result.scalars().first()
 
+    async def get_active_sessions(self) -> list[StreamSession]:
+        """Retrieve sessions that are still marked active."""
+        stmt = (
+            select(StreamSession)
+            .where(StreamSession.is_active.is_(True))
+            .order_by(StreamSession.started_at.asc())
+        )
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())
+
+    # Stream Dashboard operations
+
+    async def upsert_dashboard(
+        self,
+        session_id: str,
+        raw_markdown: str,
+        stream_title: str | None = None,
+        game: str | None = None,
+        mood: str | None = None,
+        go_live_notification: str | None = None,
+        social_post: str | None = None,
+        session_goals: list[str] | None = None,
+        content_angle: str | None = None,
+    ) -> StreamDashboard:
+        """Create or update a stream dashboard for a session."""
+        dashboard = await self.get_dashboard(session_id)
+        now = datetime.utcnow()
+        goals_json = json.dumps(session_goals or [])
+
+        if dashboard is None:
+            dashboard = StreamDashboard(
+                session_id=session_id,
+                raw_markdown=raw_markdown,
+                stream_title=stream_title,
+                game=game,
+                mood=mood,
+                go_live_notification=go_live_notification,
+                social_post=social_post,
+                session_goals=goals_json,
+                content_angle=content_angle,
+                created_at=now,
+                updated_at=now,
+            )
+            self.session.add(dashboard)
+        else:
+            dashboard.raw_markdown = raw_markdown
+            dashboard.stream_title = stream_title
+            dashboard.game = game
+            dashboard.mood = mood
+            dashboard.go_live_notification = go_live_notification
+            dashboard.social_post = social_post
+            dashboard.session_goals = goals_json
+            dashboard.content_angle = content_angle
+            dashboard.updated_at = now
+
+        await self.session.commit()
+        logger.info(f"Saved dashboard for session {session_id}")
+        return dashboard
+
+    async def get_dashboard(self, session_id: str) -> StreamDashboard | None:
+        """Retrieve the dashboard for a session."""
+        stmt = select(StreamDashboard).where(StreamDashboard.session_id == session_id)
+        result = await self.session.execute(stmt)
+        return result.scalars().first()
+
+    @staticmethod
+    def serialize_dashboard(dashboard: StreamDashboard | None) -> dict | None:
+        """Serialize a dashboard model into a plain dict."""
+        if dashboard is None:
+            return None
+
+        session_goals = []
+        if dashboard.session_goals:
+            try:
+                session_goals = json.loads(dashboard.session_goals)
+            except json.JSONDecodeError:
+                session_goals = []
+
+        return {
+            "session_id": dashboard.session_id,
+            "raw_markdown": dashboard.raw_markdown,
+            "stream_title": dashboard.stream_title,
+            "game": dashboard.game,
+            "mood": dashboard.mood,
+            "go_live_notification": dashboard.go_live_notification,
+            "social_post": dashboard.social_post,
+            "session_goals": session_goals,
+            "content_angle": dashboard.content_angle,
+            "created_at": dashboard.created_at.isoformat(),
+            "updated_at": dashboard.updated_at.isoformat(),
+        }
+
     # Chat Message operations
 
     async def add_chat_message(
@@ -95,6 +189,66 @@ class Repository:
         result = await self.session.execute(stmt)
         return list(result.scalars().all())
 
+    async def get_recent_human_messages(
+        self, session_id: str, limit: int = 50
+    ) -> list[ChatMessage]:
+        """Get recent non-bot chat messages from a session."""
+        stmt = (
+            select(ChatMessage)
+            .where(
+                ChatMessage.session_id == session_id,
+                ChatMessage.is_bot.is_(False),
+            )
+            .order_by(ChatMessage.timestamp.desc())
+            .limit(limit)
+        )
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())
+
+    async def get_latest_human_message(self, session_id: str) -> ChatMessage | None:
+        """Get the most recent non-bot chat message from a session."""
+        messages = await self.get_recent_human_messages(session_id=session_id, limit=1)
+        return messages[0] if messages else None
+
+    async def count_messages(self, session_id: str) -> int:
+        """Count chat messages stored for a session."""
+        stmt = select(func.count()).select_from(ChatMessage).where(
+            ChatMessage.session_id == session_id
+        )
+        result = await self.session.execute(stmt)
+        return result.scalar_one()
+
+    async def get_messages_since(
+        self, session_id: str, since: datetime
+    ) -> list[ChatMessage]:
+        """Get messages recorded since a specific timestamp."""
+        stmt = (
+            select(ChatMessage)
+            .where(
+                ChatMessage.session_id == session_id,
+                ChatMessage.timestamp >= since,
+            )
+            .order_by(ChatMessage.timestamp.desc())
+        )
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())
+
+    async def get_human_messages_since(
+        self, session_id: str, since: datetime
+    ) -> list[ChatMessage]:
+        """Get non-bot messages recorded since a specific timestamp."""
+        stmt = (
+            select(ChatMessage)
+            .where(
+                ChatMessage.session_id == session_id,
+                ChatMessage.is_bot.is_(False),
+                ChatMessage.timestamp >= since,
+            )
+            .order_by(ChatMessage.timestamp.desc())
+        )
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())
+
     # Agent Action operations
 
     async def record_action(
@@ -131,6 +285,28 @@ class Repository:
         result = await self.session.execute(stmt)
         return list(result.scalars().all())
 
+    async def get_latest_action(
+        self,
+        session_id: str,
+        action_type: AgentActionType | None = None,
+        mode: str | None = None,
+    ) -> AgentAction | None:
+        """Get the most recent action for a session, optionally filtered."""
+        conditions = [AgentAction.session_id == session_id]
+        if action_type:
+            conditions.append(AgentAction.action_type == action_type)
+        if mode:
+            conditions.append(AgentAction.mode == mode)
+
+        stmt = (
+            select(AgentAction)
+            .where(*conditions)
+            .order_by(AgentAction.timestamp.desc())
+            .limit(1)
+        )
+        result = await self.session.execute(stmt)
+        return result.scalars().first()
+
     # Clip Candidate operations
 
     async def add_clip_candidate(

app/twitch/chat.py

@@ -1,10 +1,242 @@
-"""Twitch chat client for sending and receiving messages."""
+"""Twitch chat client for sending and receiving messages over IRC."""
 
+import asyncio
+import contextlib
 import logging
+import ssl
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
 from typing import Optional
 
+from app.config import settings
+
 logger = logging.getLogger(__name__)
 
+TWITCH_IRC_HOST = "irc.chat.twitch.tv"
+TWITCH_IRC_PORT = 6697
+
+
+@dataclass
+class TwitchChatMessage:
+    """Parsed Twitch chat message."""
+
+    username: str
+    display_name: str
+    content: str
+    is_moderator: bool = False
+
+
+def _normalize_token(access_token: str) -> str:
+    """Return a Twitch IRC PASS token with the expected oauth: prefix."""
+    token = access_token.strip()
+    if token.startswith("oauth:"):
+        return token
+    return f"oauth:{token}"
+
+
+def _channel_name(channel_name: str) -> str:
+    """Normalize a channel name for IRC commands."""
+    return channel_name.strip().lower().lstrip("#")
+
+
+def _parse_tags(raw_tags: str) -> dict[str, str]:
+    """Parse IRCv3 tags from a Twitch message."""
+    tags = {}
+    for raw_tag in raw_tags.split(";"):
+        key, _, value = raw_tag.partition("=")
+        tags[key] = (
+            value.replace(r"\s", " ")
+            .replace(r"\:", ";")
+            .replace(r"\\", "\\")
+            .replace(r"\r", "\r")
+            .replace(r"\n", "\n")
+        )
+    return tags
+
+
+def parse_privmsg(line: str) -> TwitchChatMessage | None:
+    """Parse a Twitch IRC PRIVMSG line into a chat message."""
+    tags: dict[str, str] = {}
+    if line.startswith("@"):
+        raw_tags, _, line = line.partition(" ")
+        tags = _parse_tags(raw_tags[1:])
+
+    if not line.startswith(":"):
+        return None
+
+    prefix, _, rest = line[1:].partition(" ")
+    command, _, params = rest.partition(" ")
+    if command != "PRIVMSG":
+        return None
+
+    _, _, content = params.partition(" :")
+    username = prefix.split("!", 1)[0]
+    display_name = tags.get("display-name") or username
+    badges = tags.get("badges", "")
+    is_moderator = tags.get("mod") == "1" or "moderator/" in badges
+
+    if not username or not content:
+        return None
+
+    return TwitchChatMessage(
+        username=username,
+        display_name=display_name,
+        content=content,
+        is_moderator=is_moderator,
+    )
+
+
+class TwitchIRCClient:
+    """Small Twitch IRC client for one channel."""
+
+    def __init__(
+        self,
+        channel_name: str,
+        bot_username: str,
+        access_token: str,
+        on_message: Callable[[TwitchChatMessage], Awaitable[None]] | None = None,
+    ):
+        self.channel_name = _channel_name(channel_name)
+        self.bot_username = bot_username.strip().lower()
+        self.access_token = _normalize_token(access_token)
+        self.on_message = on_message
+        self.connected = False
+        self._reader: asyncio.StreamReader | None = None
+        self._writer: asyncio.StreamWriter | None = None
+        self._stop_event = asyncio.Event()
+        self._last_error: str | None = None
+
+    @property
+    def last_error(self) -> str | None:
+        """Return the most recent connection/listener error."""
+        return self._last_error
+
+    async def connect(self) -> None:
+        """Open an IRC connection and join the configured channel."""
+        ssl_context = ssl.create_default_context()
+        self._reader, self._writer = await asyncio.open_connection(
+            TWITCH_IRC_HOST,
+            TWITCH_IRC_PORT,
+            ssl=ssl_context,
+        )
+        await self._send_raw(f"PASS {self.access_token}")
+        await self._send_raw(f"NICK {self.bot_username}")
+        await self._send_raw("CAP REQ :twitch.tv/tags twitch.tv/commands")
+        await self._send_raw(f"JOIN #{self.channel_name}")
+        await self._wait_for_login()
+        self.connected = True
+        self._last_error = None
+        logger.info("Connected to Twitch chat for #%s", self.channel_name)
+
+    async def disconnect(self) -> None:
+        """Close the active IRC connection."""
+        self._stop_event.set()
+        self.connected = False
+        if self._writer:
+            self._writer.close()
+            with contextlib.suppress(Exception):
+                await self._writer.wait_closed()
+        self._reader = None
+        self._writer = None
+
+    async def run(self) -> None:
+        """Run the chat listener with retry on connection loss."""
+        backoff_seconds = 5
+        while not self._stop_event.is_set():
+            try:
+                await self.connect()
+                await self._listen()
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                self.connected = False
+                self._last_error = str(e)
+                logger.warning("Twitch chat listener disconnected: %s", e)
+            finally:
+                await self._close_writer()
+
+            if not self._stop_event.is_set():
+                await asyncio.sleep(backoff_seconds)
+
+    async def send_message(self, message: str) -> bool:
+        """Send a chat message to the configured channel."""
+        if not self.connected or not self._writer:
+            return False
+        await self._send_raw(f"PRIVMSG #{self.channel_name} :{message}")
+        logger.info("Sent Twitch chat message to #%s", self.channel_name)
+        return True
+
+    def status(self) -> dict:
+        """Return connection status safe for API responses."""
+        return {
+            "configured": True,
+            "connected": self.connected,
+            "channel": self.channel_name,
+            "bot_username": self.bot_username,
+            "last_error": self._last_error,
+        }
+
+    async def _listen(self) -> None:
+        if not self._reader:
+            raise RuntimeError("Twitch IRC reader is not connected")
+
+        while not self._stop_event.is_set():
+            line = await self._read_line()
+            if await self._handle_control_line(line):
+                continue
+            message = parse_privmsg(line)
+            if message and self.on_message:
+                await self.on_message(message)
+
+    async def _wait_for_login(self) -> None:
+        """Wait briefly for Twitch to accept or reject IRC authentication."""
+        while True:
+            line = await asyncio.wait_for(self._read_line(), timeout=10)
+            if await self._handle_control_line(line):
+                continue
+            if "Login authentication failed" in line:
+                raise PermissionError("Twitch IRC authentication failed")
+            if " 001 " in line or "GLOBALUSERSTATE" in line or " JOIN #" in line:
+                return
+
+    async def _read_line(self) -> str:
+        if not self._reader:
+            raise RuntimeError("Twitch IRC reader is not connected")
+        raw_line = await self._reader.readline()
+        if not raw_line:
+            raise ConnectionError("Twitch IRC connection closed")
+        return raw_line.decode("utf-8", errors="replace").strip()
+
+    async def _handle_control_line(self, line: str) -> bool:
+        if line.startswith("PING"):
+            await self._send_raw("PONG :tmi.twitch.tv")
+            return True
+        return False
+
+    async def _send_raw(self, command: str) -> None:
+        if not self._writer:
+            raise RuntimeError("Twitch IRC writer is not connected")
+        self._writer.write(f"{command}\r\n".encode("utf-8"))
+        await self._writer.drain()
+
+    async def _close_writer(self) -> None:
+        self.connected = False
+        if self._writer:
+            self._writer.close()
+            with contextlib.suppress(Exception):
+                await self._writer.wait_closed()
+        self._reader = None
+        self._writer = None
+
+
+_active_client: TwitchIRCClient | None = None
+
+
+def set_active_client(client: TwitchIRCClient | None) -> None:
+    """Register the process-wide Twitch chat client used for outbound messages."""
+    global _active_client
+    _active_client = client
+
 
 async def send_chat_message(
     channel_name: str,
@@ -12,26 +244,34 @@ async def send_chat_message(
     access_token: Optional[str] = None,
 ) -> bool:
     """
-    Send a message to Twitch chat.
-    
-    Args:
-        channel_name: Twitch channel to send message to
-        message: Message content
-        access_token: OAuth token with chat:edit scope
-        
-    Returns:
-        True if message sent successfully
-        
-    TODO: Implement Twitch Send Chat Message API
-    Reference: https://dev.twitch.tv/docs/api/reference#send-chat-message
-    
-    TODO: Handle rate limiting (20 messages per 30 seconds for verified bots)
-    TODO: Implement message queue for reliable delivery
-    TODO: Add retry logic with exponential backoff
+    Send a message to Twitch chat through the active IRC client when available.
+
+    The fallback creates a short-lived IRC connection for admin/test paths.
     """
-    logger.info(f"Sending message to {channel_name}: {message[:50]}...")
-    # Stub implementation
-    return True
+    if _active_client and _active_client.channel_name == _channel_name(channel_name):
+        sent = await _active_client.send_message(message)
+        if sent:
+            return True
+
+    token = access_token or settings.TWITCH_ACCESS_TOKEN
+    bot_username = settings.TWITCH_BOT_USERNAME
+    if not token or not bot_username:
+        logger.info("Twitch chat send skipped because credentials are incomplete")
+        return False
+
+    client = TwitchIRCClient(
+        channel_name=channel_name,
+        bot_username=bot_username,
+        access_token=token,
+    )
+    try:
+        await client.connect()
+        return await client.send_message(message)
+    except Exception as e:
+        logger.warning("Failed to send Twitch chat message: %s", e)
+        return False
+    finally:
+        await client.disconnect()
 
 
 class ChatMessageBuffer:

app/twitch/live.py

@@ -0,0 +1,184 @@
+"""Twitch live-stream status checks."""
+
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+import httpx
+
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+TWITCH_STREAMS_URL = "https://api.twitch.tv/helix/streams"
+TWITCH_TOKEN_URL = "https://id.twitch.tv/oauth2/token"
+
+
+@dataclass
+class TwitchLiveStatus:
+    """Cached Twitch live status for a channel."""
+
+    channel_name: str
+    is_live: bool
+    checked_at: datetime
+    title: str | None = None
+    game_name: str | None = None
+    started_at: str | None = None
+    viewer_count: int | None = None
+    reason: str | None = None
+
+    def to_dict(self) -> dict:
+        """Return a response-safe representation."""
+        return {
+            "channel": self.channel_name,
+            "is_live": self.is_live,
+            "checked_at": self.checked_at.isoformat(),
+            "title": self.title,
+            "game_name": self.game_name,
+            "started_at": self.started_at,
+            "viewer_count": self.viewer_count,
+            "reason": self.reason,
+        }
+
+
+def _bearer_token(access_token: str) -> str:
+    """Normalize a Twitch token for Helix Authorization."""
+    token = access_token.strip()
+    if token.startswith("oauth:"):
+        return token.removeprefix("oauth:")
+    return token
+
+
+class TwitchLiveStatusService:
+    """Small cached client for Twitch Helix stream status."""
+
+    def __init__(self, cache_seconds: int = 60):
+        self.cache_duration = timedelta(seconds=max(1, cache_seconds))
+        self._cache: dict[str, TwitchLiveStatus] = {}
+        self._app_access_token: str | None = None
+        self._app_access_token_expires_at: datetime | None = None
+
+    async def get_status(
+        self,
+        channel_name: str,
+        force_refresh: bool = False,
+    ) -> TwitchLiveStatus:
+        """Return current live status for a channel, using a short cache."""
+        normalized_channel = channel_name.strip().lower().lstrip("#")
+        now = datetime.utcnow()
+        cached = self._cache.get(normalized_channel)
+        if (
+            cached
+            and not force_refresh
+            and now - cached.checked_at < self.cache_duration
+        ):
+            return cached
+
+        status = await self._fetch_status(normalized_channel, now)
+        self._cache[normalized_channel] = status
+        return status
+
+    async def _fetch_status(
+        self,
+        channel_name: str,
+        checked_at: datetime,
+    ) -> TwitchLiveStatus:
+        """Fetch live status from Twitch Helix."""
+        if not settings.TWITCH_CLIENT_ID or not settings.TWITCH_ACCESS_TOKEN:
+            return TwitchLiveStatus(
+                channel_name=channel_name,
+                is_live=False,
+                checked_at=checked_at,
+                reason="twitch_api_not_configured",
+            )
+
+        headers = {
+            "Client-ID": settings.TWITCH_CLIENT_ID,
+            "Authorization": f"Bearer {_bearer_token(settings.TWITCH_ACCESS_TOKEN)}",
+        }
+        params = {"user_login": channel_name}
+
+        try:
+            async with httpx.AsyncClient(timeout=10) as client:
+                response = await client.get(
+                    TWITCH_STREAMS_URL,
+                    headers=headers,
+                    params=params,
+                )
+                if response.status_code == 401:
+                    app_access_token = await self._get_app_access_token(client)
+                    if app_access_token:
+                        headers["Authorization"] = f"Bearer {app_access_token}"
+                        response = await client.get(
+                            TWITCH_STREAMS_URL,
+                            headers=headers,
+                            params=params,
+                        )
+                response.raise_for_status()
+        except Exception as e:
+            logger.warning("Failed to check Twitch live status: %s", e)
+            return TwitchLiveStatus(
+                channel_name=channel_name,
+                is_live=False,
+                checked_at=checked_at,
+                reason="twitch_api_error",
+            )
+
+        streams = response.json().get("data", [])
+        if not streams:
+            return TwitchLiveStatus(
+                channel_name=channel_name,
+                is_live=False,
+                checked_at=checked_at,
+                reason="stream_offline",
+            )
+
+        stream = streams[0]
+        return TwitchLiveStatus(
+            channel_name=channel_name,
+            is_live=True,
+            checked_at=checked_at,
+            title=stream.get("title"),
+            game_name=stream.get("game_name"),
+            started_at=stream.get("started_at"),
+            viewer_count=stream.get("viewer_count"),
+        )
+
+    async def _get_app_access_token(self, client: httpx.AsyncClient) -> str | None:
+        """Get a Twitch app access token for Helix reads."""
+        now = datetime.utcnow()
+        if (
+            self._app_access_token
+            and self._app_access_token_expires_at
+            and now < self._app_access_token_expires_at
+        ):
+            return self._app_access_token
+
+        if not settings.TWITCH_CLIENT_ID or not settings.TWITCH_CLIENT_SECRET:
+            return None
+
+        try:
+            response = await client.post(
+                TWITCH_TOKEN_URL,
+                data={
+                    "client_id": settings.TWITCH_CLIENT_ID,
+                    "client_secret": settings.TWITCH_CLIENT_SECRET,
+                    "grant_type": "client_credentials",
+                },
+            )
+            response.raise_for_status()
+        except Exception as e:
+            logger.warning("Failed to get Twitch app access token: %s", e)
+            return None
+
+        payload = response.json()
+        access_token = payload.get("access_token")
+        expires_in = int(payload.get("expires_in", 0))
+        if not access_token:
+            return None
+
+        self._app_access_token = access_token
+        self._app_access_token_expires_at = now + timedelta(
+            seconds=max(60, expires_in - 60)
+        )
+        return access_token

dashboard-prompt.md

@@ -0,0 +1,56 @@
+# Role
+
+You are a thoughtful RPG analyst, calm and introspective streamer, and content strategist for the _Withered Sanctum_ brand.
+# Context
+
+- Today’s Game: [INSERT GAME]
+- Current Mood/Energy Level: [Low / Medium / High / Reflective / etc.]
+- Familiarity with Game: [New / Some Experience / Very Familiar]
+- Recent Experience (optional): [e.g., “Struggled to get into it”, “Had a great session last time”]
+- Stream Start Time: [INSERT TIME]
+# Output Requirements
+
+Create a **Stream Dashboard** with the following sections:
+### 1. Stream Title
+
+- Should feel grounded, thoughtful, and slightly literary
+- Avoid hype-heavy or clickbait tone
+- Reflect the _experience_ or _journey_, not just the game
+
+### 2. Twitch Go Live Notification
+
+- Short (1–2 sentences)
+- Calm, inviting tone
+- Should feel like an invitation, not an advertisement
+
+### 3. Twitter (X) Announcement Post
+
+- Written 10 minutes before going live
+- Slightly more outward-facing, but still restrained and authentic
+- Optional light hook, but no hype language
+- Include 2–4 relevant hashtags
+
+### 4. Session Goals
+
+- 3–5 simple, achievable goals
+- Can include:
+    - Progress goals (quests, systems, areas)
+    - Comfort goals (learn mechanics, reduce friction)
+    - Personal goals (stay present, avoid rushing)
+
+### 5. One Content Angle
+
+- A single lens or theme for the stream
+- Should elevate the stream beyond “just gameplay”
+- Examples:
+    - First impressions vs. long-term potential
+    - Systems friction vs. player reward
+    - Narrative tone and immersion
+    - “Why this game hasn’t hooked me (yet)”
+- Keep it subtle, not forced—something to return to during quiet moments
+
+### Tone Guidance
+
+- Calm, reflective, slightly philosophical
+- Low-energy friendly (never forced enthusiasm)
+- Feels like a seasoned player sharing perspective, not performing

docker-compose.yml

@@ -32,24 +32,32 @@ services:
       APP_NAME: "Sanctum Chronicler"
       APP_ENV: ${APP_ENV:-development}
       DEBUG: ${DEBUG:-false}
+      ADMIN_API_KEY: ${ADMIN_API_KEY:-}
       DATABASE_URL: postgresql+asyncpg://sanctum:${DB_PASSWORD:-password}@sanctum-db:5432/sanctum
       TWITCH_CLIENT_ID: ${TWITCH_CLIENT_ID:-}
       TWITCH_CLIENT_SECRET: ${TWITCH_CLIENT_SECRET:-}
       TWITCH_BOT_USERNAME: ${TWITCH_BOT_USERNAME:-}
       TWITCH_CHANNEL_NAME: ${TWITCH_CHANNEL_NAME:-}
+      TWITCH_ACCESS_TOKEN: ${TWITCH_ACCESS_TOKEN:-}
+      TWITCH_CHAT_ENABLED: ${TWITCH_CHAT_ENABLED:-true}
+      TWITCH_REQUIRE_LIVE_STREAM: ${TWITCH_REQUIRE_LIVE_STREAM:-true}
+      TWITCH_LIVE_STATUS_CACHE_SECONDS: ${TWITCH_LIVE_STATUS_CACHE_SECONDS:-60}
+      TWITCH_OFFLINE_CHAT_TEST_MODE: ${TWITCH_OFFLINE_CHAT_TEST_MODE:-false}
       LLM_PROVIDER: ${LLM_PROVIDER:-}
       LLM_BASE_URL: ${LLM_BASE_URL:-}
       LLM_API_KEY: ${LLM_API_KEY:-}
       LLM_MODEL: ${LLM_MODEL:-gpt-3.5-turbo}
+      AGENT_LOOP_INTERVAL_SECONDS: ${AGENT_LOOP_INTERVAL_SECONDS:-60}
+      HEARTHKEEPER_PROMPT_INTERVAL_MINUTES: ${HEARTHKEEPER_PROMPT_INTERVAL_MINUTES:-15}
       EXPORT_PATH: /app/exports
     volumes:
       - ./exports:/app/exports
-    ports:
-      - "8000:8000"
     networks:
-      - sanctum-net
+      sanctum-net: {}
+      sanctum-lan:
+        ipv4_address: 192.168.5.92
     healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
+      test: ["CMD", "python", "-c", "import httpx; httpx.get('http://localhost:8000/health').raise_for_status()"]
       interval: 30s
       timeout: 3s
       retries: 3
@@ -62,3 +70,6 @@ volumes:
 networks:
   sanctum-net:
     driver: bridge
+  sanctum-lan:
+    external: true
+    name: blog_network

prompt.md

@@ -0,0 +1,173 @@
+Build a Dockerized Python MVP called sanctum-agent.
+
+Goal:
+Create scaffolding for an AI stream assistant called “The Sanctum Chronicler.” It should eventually connect to Twitch chat, monitor stream discussion, lightly guide conversation, store stream events, and export a post-stream markdown ledger.
+
+Tech stack:
+- Python 3.12
+- FastAPI
+- PostgreSQL
+- Docker Compose
+- Async architecture
+- Markdown export
+- Environment variables via .env
+- Placeholder LLM client that can later support OpenAI, Ollama, or LM Studio
+
+Project structure:
+
+sanctum-agent/
+  app/
+    main.py
+    config.py
+
+    twitch/
+      __init__.py
+      eventsub.py
+      chat.py
+
+    agent/
+      __init__.py
+      orchestrator.py
+      policies.py
+      modes/
+        __init__.py
+        hearthkeeper.py
+        steward.py
+        warden.py
+        librarian.py
+        scribe.py
+
+    memory/
+      __init__.py
+      database.py
+      models.py
+      repository.py
+
+    llm/
+      __init__.py
+      client.py
+      prompts.py
+
+    exports/
+      __init__.py
+      markdown.py
+
+  exports/
+  data/
+  Dockerfile
+  docker-compose.yml
+  requirements.txt
+  .env.example
+  README.md
+
+Requirements:
+
+1. FastAPI app
+Create endpoints:
+- GET /health
+- POST /admin/session/start
+- POST /admin/session/end
+- GET /admin/ledger
+- POST /admin/test-message
+
+2. Configuration
+Create app/config.py using pydantic-settings.
+Support these environment variables:
+- APP_NAME
+- APP_ENV
+- DATABASE_URL
+- TWITCH_CLIENT_ID
+- TWITCH_CLIENT_SECRET
+- TWITCH_BOT_USERNAME
+- TWITCH_CHANNEL_NAME
+- LLM_PROVIDER
+- LLM_BASE_URL
+- LLM_API_KEY
+- EXPORT_PATH
+
+3. Database
+Use SQLAlchemy async if reasonable.
+Create models for:
+- StreamSession
+- ChatMessage
+- AgentAction
+- ClipCandidate
+- BlogSeed
+
+The database layer can be functional scaffolding. It does not need full production migrations yet.
+
+4. Agent Orchestrator
+Create an AgentOrchestrator class that:
+- receives chat messages
+- stores them
+- decides whether the agent should respond
+- suppresses responses when human chat is active
+- routes behavior to internal modes
+
+Add a simple policy:
+- If no human chat for 15 minutes, Hearthkeeper may generate a gentle prompt.
+- If chat is active, agent stays silent.
+- If message contains suspicious Discord-growth language, Warden flags it.
+
+5. Modes
+Create placeholder classes:
+- HearthkeeperMode
+- StewardMode
+- WardenMode
+- LibrarianMode
+- ScribeMode
+
+Each should have clear docstrings explaining its purpose.
+
+6. LLM Client
+Create an LLMClient abstraction with a generate() method.
+For now, return deterministic placeholder text if no provider is configured.
+
+7. Markdown Export
+Create a markdown exporter that generates:
+
+# Sanctum Ledger — YYYY-MM-DD
+
+## Stream Theme
+
+## Notable Discussion
+
+## Agent Actions
+
+## Clip Candidates
+
+## Blog Seeds
+
+8. Twitch Layer
+Create placeholder Twitch modules:
+- eventsub.py should define a TwitchEventSubClient class with connect(), disconnect(), and listen() stubs.
+- chat.py should define send_chat_message() as a placeholder.
+Do not implement real OAuth yet. Add TODO comments with where EventSub and Send Chat Message API integration will go.
+
+9. Docker
+Create:
+- Dockerfile for FastAPI app
+- docker-compose.yml with:
+  - sanctum-agent
+  - sanctum-db using postgres:16
+
+10. README
+Write a README with:
+- project purpose
+- architecture overview
+- setup steps
+- docker compose commands
+- current limitations
+- next implementation steps
+
+Style:
+- Keep code clean and readable.
+- Use type hints.
+- Add comments where future Twitch, Discord, and LLM integrations will be inserted.
+- Do not overbuild.
+- This is scaffolding, not a finished production bot.
+
+After generating the files, also provide:
+1. a file tree
+2. commands to run the app
+3. a short explanation of the next practical implementation step

requirements.txt

@@ -7,3 +7,4 @@ asyncpg==0.29.0
 psycopg2-binary==2.9.9
 httpx==0.25.2
 python-dotenv==1.0.0
+python-multipart==0.0.6

usecases.md

@@ -0,0 +1,337 @@
+# Sanctum Chronicler Use Cases
+
+## Project
+Withered Sanctum — Sanctum Chronicler
+
+---
+
+# Purpose
+
+Hearthkeeper Mode exists to preserve conversational continuity and reflective atmosphere during live streams.
+
+The goal of Hearthkeeper is **not** to simulate viewers, inflate engagement, or dominate discussion.
+
+Its purpose is to:
+
+- reduce conversational dead air
+- maintain thematic continuity
+- help the streamer sustain reflective dialogue
+- preserve stream atmosphere
+- encourage authentic human participation
+- act as a gentle conversational steward
+
+Hearthkeeper should function more like:
+- a host
+- a steward
+- a monastery caretaker
+- a quiet tavernkeeper
+- a keeper of the fire
+
+and less like:
+- a hype bot
+- an engagement optimizer
+- a chatbot competing for attention
+
+---
+
+# Core Philosophy
+
+The Withered Sanctum stream environment is intentionally:
+- calm
+- reflective
+- mythic
+- atmospheric
+- slow-paced
+- discussion-oriented
+
+Silence is not inherently a failure state.
+
+Hearthkeeper must understand the distinction between:
+- contemplative silence
+- disengaged silence
+
+The agent should intervene lightly and rarely.
+
+The system should create:
+> permission for conversation
+
+rather than:
+> artificial conversation density
+
+Human discussion always takes priority over agent participation.
+
+---
+
+# High-Level Workflow
+
+## Daily Stream Preparation
+
+Before each stream, the streamer generates a "Stream Dashboard" document using a structured prompt template.
+
+The dashboard establishes:
+- stream title
+- stream theme
+- game/topic
+- philosophical framing
+- current mood/energy
+- intended discussion topics
+- relevant mythology/philosophy/design concepts
+- session goals
+
+Example themes:
+- pirate freedom
+- negative space in games
+- mythology of exploration
+- loneliness in open worlds
+- Tolkien’s concept of Secondary Worlds
+
+The Sanctum Chronicler must have access to the current Stream Dashboard before or during stream startup.
+
+---
+
+# Runtime Behavior
+
+## During Stream
+
+Hearthkeeper observes:
+- Twitch chat
+- chat activity frequency
+- current discussion topics
+- stream title
+- active game/category
+- prior Hearthkeeper prompts
+- stream dashboard themes
+- optionally prior stream summaries
+
+The agent should determine:
+- whether the stream currently needs conversational support
+- whether silence is natural/healthy
+- whether thematic prompts would help maintain flow
+
+---
+
+# Trigger Conditions
+
+Hearthkeeper may consider generating a prompt when:
+
+- chat has been quiet for a configurable duration
+- the streamer has stopped speaking for an extended period
+- discussion has drifted completely away from intended themes
+- there is visible conversational uncertainty
+- new viewers arrive during prolonged silence
+
+The agent should remain silent when:
+- humans are actively discussing
+- the streamer is engaged in active commentary
+- emotional or contemplative silence appears intentional
+- chat momentum is healthy
+
+---
+
+# Prompt Generation Sources
+
+Hearthkeeper prompts may be derived from:
+
+## 1. Current Stream Dashboard
+Primary source of thematic grounding.
+
+Example:
+- today’s themes
+- intended philosophical lens
+- stream purpose
+
+---
+
+## 2. Current Stream Discussion
+Topics actively discussed during the session.
+
+Example:
+- player comments
+- streamer reflections
+- emergent themes
+
+---
+
+## 3. Prior Stream Memory
+Previously discussed ideas or recurring themes.
+
+Example:
+- references to earlier Windrose discussions
+- callbacks to prior Tolkien observations
+- recurring discussions about ritual, immersion, or mythology
+
+---
+
+## 4. Content Knowledge
+General knowledge relevant to:
+- mythology
+- philosophy
+- game design
+- sociology
+- fantasy literature
+- symbolic analysis
+
+The goal is not academic performance, but thematic continuity.
+
+---
+
+# Prompt Style Requirements
+
+Hearthkeeper prompts must:
+
+- be brief
+- be calm
+- avoid hype
+- avoid excessive frequency
+- avoid sounding like engagement bait
+- avoid sounding like marketing
+- avoid excessive positivity or “content creator energy”
+
+The voice should feel:
+- thoughtful
+- restrained
+- reflective
+- atmospheric
+- human-compatible
+
+Good prompts should feel like:
+> a thoughtful observation tossed onto the fire
+
+rather than:
+> a social media optimization tactic
+
+---
+
+# Examples of Acceptable Prompts
+
+## Example 1
+> “Does the sea in Windrose feel empty, or contemplative?”
+
+## Example 2
+> “Today’s discussion keeps returning to freedom as exile rather than liberation.”
+
+## Example 3
+> “This reminds me somewhat of Tolkien’s distinction between Primary and Secondary Worlds.”
+
+## Example 4
+> “The silence between locations may matter as much as the locations themselves.”
+
+---
+
+# Examples of Unacceptable Prompts
+
+## Bad Example 1
+> “CHAT WHAT DO WE THINK???”
+
+## Bad Example 2
+> “Don’t forget to like and follow!”
+
+## Bad Example 3
+> “This stream is AMAZING today!”
+
+## Bad Example 4
+> “Drop your thoughts in chat right now!”
+
+---
+
+# Behavioral Constraints
+
+## Frequency Limits
+
+The agent should:
+- speak rarely
+- avoid repetition
+- avoid flooding chat
+- avoid interrupting active conversation
+
+Initial recommended limits:
+- minimum 10–15 minutes between prompts
+- configurable cooldowns
+- reduced activity during healthy human discussion
+
+---
+
+## Human Priority Rule
+
+The system must always prioritize authentic human interaction.
+
+If humans are actively engaging:
+- Hearthkeeper becomes quieter
+- prompts become less frequent
+- intervention threshold increases
+
+The agent exists to support conversation, not replace it.
+
+---
+
+# Ethical Constraints
+
+Hearthkeeper must:
+- identify itself openly as an AI steward
+- never pretend to be a human viewer
+- never fabricate audience engagement
+- never simulate fake community activity
+- never impersonate emotional attachment
+
+Its role is environmental support and continuity.
+
+Not deception.
+
+---
+
+# Future Expansion Possibilities
+
+Potential future integrations:
+
+- clip candidate detection
+- stream timestamping
+- discussion summaries
+- blog article generation
+- Obsidian export
+- lore indexing
+- Discord continuity discussions
+- semantic memory retrieval
+- long-term theme tracking
+
+---
+
+# MVP Requirements
+
+## Minimum Viable Hearthkeeper
+
+The MVP should:
+
+1. Read the current Stream Dashboard
+2. Observe Twitch chat activity
+3. Track periods of silence
+4. Generate rare thematic prompts
+5. Respect cooldowns
+6. Store prompts and timestamps
+7. Export a post-stream markdown ledger
+
+---
+
+# Success Criteria
+
+Hearthkeeper is successful if:
+
+- the stream feels less psychologically empty
+- discussion continuity improves
+- prompts feel natural and thematic
+- the streamer is helped rather than interrupted
+- viewers engage authentically with prompts
+- atmosphere is preserved
+
+Failure occurs if:
+- the bot dominates chat
+- prompts feel artificial
+- the system becomes noisy
+- viewers mistake synthetic activity for audience size inflation
+- the atmosphere becomes performative rather than reflective
+
+---
+
+# Guiding Principle
+
+> “The agent tends the space.
+> The humans give it life.”