Add quiet loop verification endpoint

app/agent/orchestrator.py

@@ -247,6 +247,7 @@ class AgentOrchestrator:
                 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}"
         )
@@ -257,6 +258,61 @@ class AgentOrchestrator:
             "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",
+        )
+        prompts_created = after_count - before_count
+
+        return {
+            "passed": (
+                prompts_created == 1
+                and first_tick is not None
+                and second_tick is None
+            ),
+            "session_id": session_id,
+            "inactive_minutes": inactive_minutes,
+            "prompts_created": prompts_created,
+            "first_tick": first_tick,
+            "second_tick": second_tick,
+        }
+
+    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:

app/main.py

@@ -153,6 +153,29 @@ async def test_agent_response(
     }
 
 
+@app.post("/admin/test-loop-inactivity")
+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.get("/admin/ledger")
 async def get_ledger(session_id: str) -> dict:
     """Get the markdown ledger for a session."""