examples: make session monitor demo deterministic

Author: TimesAndPlaces

examples/session_monitor.py

@@ -11,12 +11,18 @@
 sessions where context compaction or summarization may silently change the
 agent's behavior.
 
+Because a short fresh session will not reliably trigger compaction on demand,
+the default runnable demo below uses a simulated token-usage boundary while the
+live integration helpers keep the exact public SDK wiring you would use in a
+real session.
+
 Reference: https://github.com/anthropics/claude-agent-sdk-python/issues/772
 """
 
 import asyncio
 import json
 import math
+import os
 import re
 import time
 from collections import Counter
@@ -263,7 +269,45 @@ async def run_monitored_turn(
     return monitor.record_turn(" ".join(text_parts), total_tokens)
 
 
-async def main() -> None:
+def run_simulated_boundary_demo() -> None:
+    """Run a deterministic boundary demo using the same scoring logic."""
+    monitor = SessionMonitor(
+        compaction_drop_ratio=0.20,
+        drift_threshold=0.30,
+        log_path=None,
+    )
+
+    synthetic_turns = [
+        (
+            "Use jwt validation with bcrypt hashes, redis-backed sessions, and "
+            "foreign_key-safe migrations for the auth schema.",
+            1200,
+        ),
+        (
+            "Keep jwt auth, bcrypt password storage, and redis session checks "
+            "intact while you add the profile endpoint.",
+            1480,
+        ),
+        (
+            "Add PATCH /profile rate limiting with concise validation and 429 "
+            "responses. Focus on the endpoint only.",
+            860,
+        ),
+    ]
+
+    print("=== Deterministic session boundary demo ===")
+    print("This uses simulated token snapshots so the monitor always shows a boundary event.\n")
+    for text, total_tokens in synthetic_turns:
+        event = monitor.record_turn(text, total_tokens)
+        if event:
+            print(json.dumps(event, indent=2))
+
+    print("\n=== Session summary ===")
+    print(json.dumps(monitor.summary(), indent=2))
+
+
+async def run_live_demo() -> None:
+    """Optional live SDK demo using the same monitor."""
     monitor = SessionMonitor(
         compaction_drop_ratio=0.20,
         drift_threshold=0.30,
@@ -302,5 +346,15 @@ async def main() -> None:
     print("plus get_context_usage() as the compaction-boundary heuristic.")
 
 
+async def main() -> None:
+    run_simulated_boundary_demo()
+
+    if os.getenv("CLAUDE_SESSION_MONITOR_LIVE") == "1":
+        print("\n=== Live SDK session demo ===")
+        await run_live_demo()
+    else:
+        print("\nSet CLAUDE_SESSION_MONITOR_LIVE=1 to also run the live SDK session demo.")
+
+
 if __name__ == "__main__":
     asyncio.run(main())