Persist published takeaways in the system database

docs/operating.md

@@ -179,8 +179,10 @@ start paid model runs:
   queries the agent ran. The server keeps no conversation state.
 - `GET /api/insights/takeaways` returns the cached 30-day analysis for the
   org, or `status: pending` while one generates in a background thread; the
-  page polls. `?fresh=1` starts a new run. The cache is in-process, so a
-  restart (or a second App Runner instance) regenerates on first view.
+  page polls. `?fresh=1` starts a new run. A finished analysis is written to
+  the system database (`insights_takeaways`, one row per org database), so a
+  restart, a deploy or a second App Runner instance shows the last published
+  takeaways on first view and only regenerates once they are six hours old.
 
 Both count against one per-IP limit (`Handler.CHAT_RATE_LIMIT`, 60 runs an
 hour), and at most four agent runs execute at once per process so a burst of

server/insights.py

@@ -270,7 +270,11 @@ class InsightsEngine:
     agent, and the takeaways cache. All public methods are thread-safe."""
 
     def __init__(self, model=None, chat_effort="medium", analysis_effort="high",
-                 max_concurrent=4):
+                 max_concurrent=4, store=None):
+        # `store` is the system database (AuthStore.sysdb): published
+        # takeaways are written there so they survive the process. None
+        # keeps everything in memory, which the tests use.
+        self.store = store
         self.base_url = os.environ.get("HOTOTEL_INSIGHTS_BASE_URL", "").rstrip("/")
         self.provider = "endpoint" if self.base_url else "anthropic"
         default = DEFAULT_ENDPOINT_MODEL if self.base_url else DEFAULT_ANTHROPIC_MODEL
@@ -318,13 +322,13 @@ def missing_requirements():
         return None
 
     @classmethod
-    def from_env(cls):
+    def from_env(cls, store=None):
         """An engine, or None with the reason on stderr."""
         why = cls.missing_requirements()
         if why:
             print(f"warn: insights page is OFF: {why}", file=sys.stderr)
             return None
-        return cls()
+        return cls(store=store)
 
     def status(self):
         return {"enabled": True, "model": self.model_name, "provider": self.provider,
@@ -484,6 +488,15 @@ def takeaways(self, database_id, org_name, fresh=False):
         now = time.time()
         with self._lock:
             entry = self._takeaways.get(database_id)
+        if entry is None and self.store is not None:
+            # first ask since this process started: the last published
+            # analysis, if any, is in the system database
+            entry = self._load_saved(database_id)
+            if entry:
+                with self._lock:
+                    self._takeaways.setdefault(database_id, entry)
+        with self._lock:
+            entry = self._takeaways.get(database_id, entry)
             stale = not entry or now - entry["at"] >= TAKEAWAYS_TTL
             running = database_id in self._generating
             if (fresh or stale) and not running:
@@ -514,6 +527,7 @@ def _generate(self, database_id, org_name):
             data = parse_takeaways(answer)
             entry = {"at": time.time(), "generatedAt": _now_iso(), "data": data,
                      "steps": steps, "seconds": round(time.time() - started)}
+            self._save(database_id, entry)
         except Exception as e:
             print(f"error: takeaways for {database_id}: {str(e)[:300]}", file=sys.stderr)
             # A failed run is cached briefly too, so a page that keeps polling
@@ -524,6 +538,54 @@ def _generate(self, database_id, org_name):
             self._takeaways[database_id] = entry
             self._generating.discard(database_id)
 
+    # -- the published analysis, across processes -----------------------------
+
+    TABLE = "insights_takeaways"
+
+    def _save(self, database_id, entry):
+        """Best effort: a write failure costs persistence, not the analysis."""
+        if self.store is None:
+            return
+        try:
+            self.store.load(self.TABLE, [{
+                "database_id": database_id,
+                "generated_at": entry["generatedAt"],
+                "model": self.model_name,
+                "seconds": int(entry.get("seconds") or 0),
+                "payload": json.dumps({"data": entry["data"], "steps": entry["steps"]}),
+            }], "upsert")
+        except Exception as e:
+            print(f"warn: takeaways for {database_id} not saved: {str(e)[:200]}",
+                  file=sys.stderr)
+
+    def _load_saved(self, database_id):
+        """The saved entry in the in-memory shape, or None."""
+        try:
+            rows = self.store.rows(
+                f"SELECT generated_at, model, seconds, payload "
+                f"FROM {core.SYSTEM_CATALOG}.public.{self.TABLE} "
+                f"WHERE database_id = '{database_id}'")
+        except Exception as e:
+            print(f"warn: takeaways for {database_id} not read: {str(e)[:200]}",
+                  file=sys.stderr)
+            return None
+        if not rows:
+            return None
+        r = rows[0]
+        try:
+            payload = json.loads(r["payload"]) if isinstance(r["payload"], str) else r["payload"]
+            when = r["generated_at"]
+            if not isinstance(when, datetime):
+                when = datetime.fromisoformat(str(when).replace("Z", "+00:00"))
+            if when.tzinfo is None:
+                when = when.replace(tzinfo=timezone.utc)
+            return {"at": when.timestamp(), "generatedAt": when.isoformat(),
+                    "data": payload["data"], "steps": payload.get("steps") or [],
+                    "seconds": int(r.get("seconds") or 0), "model": r.get("model")}
+        except (KeyError, TypeError, ValueError) as e:
+            print(f"warn: saved takeaways for {database_id} unreadable: {e}", file=sys.stderr)
+            return None
+
 
 def _unpack(messages):
     """The final text and the tool calls that led to it.

server/server.py

@@ -199,14 +199,24 @@
         "key": ["email"],
         "cols": [("email", "VARCHAR"), ("verified_at", "TIMESTAMPTZ")],
     },
+    # The Insights page's published takeaways, one row per org database. The
+    # analysis costs minutes of model time, so a finished one outlives the
+    # process that made it: a deploy, a restart or a second App Runner
+    # instance shows the last published analysis instead of a blank card.
+    # `payload` is the JSON the page renders (highlights + the queries run).
+    "insights_takeaways": {
+        "key": ["database_id"],
+        "cols": [("database_id", "VARCHAR"), ("generated_at", "TIMESTAMPTZ"),
+                 ("model", "VARCHAR"), ("seconds", "INTEGER"), ("payload", "VARCHAR")],
+    },
 }
 
 USAGE_TABLES = ("sessions", "requests", "daily_usage")
 SYSTEM_TABLES = ("orgs", "users", "auth_sessions", "invites", "team_invites",
                  "device_codes", "device_scopes", "collector_tokens",
                  "collector_token_usage", "token_scopes",
                  "org_admins", "system_admins", "org_memberships",
-                 "email_verifications", "verified_emails")
+                 "email_verifications", "verified_emails", "insights_takeaways")
 
 
 # ---------------------------------------------------------------------------
@@ -3503,7 +3513,7 @@ def seed_retry():
               "is OFF: any address can register and no mailbox is ever proved",
               file=sys.stderr)
     Handler.token = token
-    Handler.insights = insights.InsightsEngine.from_env()
+    Handler.insights = insights.InsightsEngine.from_env(store=Handler.auth.sysdb)
     if Handler.insights:
         print(f"hototel server: insights page ON, model {Handler.insights.model_name}")
     Handler.allow_shared = os.environ.get("HOTUSAGE_ALLOW_SHARED_INGEST") == "1"

server/tests/test_auth_store.py

@@ -87,7 +87,8 @@ class StatefulSysDb:
             "collector_token_usage": ("token",), "auth_sessions": ("token",),
             "system_admins": ("email",), "invites": ("token",),
             "team_invites": ("token",),
-            "email_verifications": ("token",), "verified_emails": ("email",)}
+            "email_verifications": ("token",), "verified_emails": ("email",),
+            "insights_takeaways": ("database_id",)}
 
     def __init__(self, db="dbidsystem00000000000000"):
         self.db = db

server/tests/test_insights.py

@@ -13,6 +13,7 @@
 import os
 import sys
 import threading
+import time
 from http.server import ThreadingHTTPServer
 
 HERE = os.path.dirname(os.path.abspath(__file__))
@@ -291,6 +292,43 @@ def post(pw):
         server.Handler.insights = None
 
 
+def test_published_takeaways_survive_the_process():
+    """A finished analysis is written to the system database and read back
+    by a fresh engine, so a deploy or a second instance shows the last
+    published takeaways instead of starting over."""
+    import threading as th
+    db = fakes.StatefulSysDb()
+    canned = json.dumps(GOOD)
+
+    def engine():
+        e = insights.InsightsEngine.__new__(insights.InsightsEngine)
+        e.store, e.model_name, e.base_url = db, "fake", ""
+        e.analysis_effort = "high"
+        e._lock, e._takeaways, e._generating = th.Lock(), {}, set()
+        e._health = (None, 0.0)
+        e._run = lambda *a, **k: (canned, [{"tool": "hotdata_execute_sql", "sql": "SELECT 1",
+                                           "rows": 1, "error": None, "table": None}])
+        return e
+
+    a = engine()
+    first = a.takeaways("dbid1", "Acme")
+    assert first["status"] == "pending", first
+    for _ in range(50):               # the generation thread is quick with a canned run
+        if a.takeaways("dbid1", "Acme")["status"] == "ready":
+            break
+        time.sleep(0.05)
+    saved = db.tables["insights_takeaways"]
+    assert len(saved) == 1 and saved[0]["database_id"] == "dbid1", saved
+    assert json.loads(saved[0]["payload"])["data"]["summary"] == "A busy month."
+
+    b = engine()                      # a new process: empty memory, same database
+    out = b.takeaways("dbid1", "Acme")
+    assert out["status"] == "ready" and out["refreshing"] is False, out
+    assert out["data"]["summary"] == "A busy month." and out["steps"][0]["sql"] == "SELECT 1"
+    assert "dbid1" not in b._generating, "a fresh, saved analysis must not be regenerated"
+    print("takeaways: saved on finish, read back by a fresh engine without a new run")
+
+
 def test_chat_refuses_everyone_but_a_cookie_session():
     server.Handler.insights = FakeEngine()
     try: