diff --git a/docs/operating.md b/docs/operating.md index 7460c60..84396d2 100644 --- a/docs/operating.md +++ b/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 diff --git a/server/insights.py b/server/insights.py index d3671b7..5139b2b 100644 --- a/server/insights.py +++ b/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. diff --git a/server/server.py b/server/server.py index 1eb3c20..cbfe90f 100644 --- a/server/server.py +++ b/server/server.py @@ -199,6 +199,16 @@ "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") @@ -206,7 +216,7 @@ "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" diff --git a/server/tests/test_auth_store.py b/server/tests/test_auth_store.py index be26263..8573d38 100644 --- a/server/tests/test_auth_store.py +++ b/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 diff --git a/server/tests/test_insights.py b/server/tests/test_insights.py index 30d2455..c28193a 100644 --- a/server/tests/test_insights.py +++ b/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: