diff --git a/backend/services/analysis.py b/backend/services/analysis.py index 184649a..1956a42 100644 --- a/backend/services/analysis.py +++ b/backend/services/analysis.py @@ -323,6 +323,14 @@ def _context_fingerprint(context): return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:64] +def has_insight(user_id, kind, subject): + """True if a generated insight exists for this kind+subject.""" + return bool(query_one( + "SELECT 1 FROM ai_insights WHERE user_id = ? AND kind = ? AND subject = ?", + [user_id, kind, subject], + )) + + def generate_briefing(user_id, context, model=None): """Ask a model for the briefing and store it. Returns (briefing, meta).""" completion, meta = ai_svc.complete(coach.briefing_messages(context), model) @@ -531,6 +539,9 @@ def prefetch_insights(user_id): Called after a sync: the data has changed, so every stored answer is stale. These run for as long as they run — anything the user actually opens jumps the queue ahead of them. + + Also queues per-item scopes (daily, activity) that have no insight yet, + so the user's whole history is gradually generated in the background. """ queued = [] for scope in scopes.PREFETCH_SCOPES: @@ -548,6 +559,50 @@ def prefetch_insights(user_id): jobs.PRIORITY_PREFETCH) queued.append(scope) + # --- daily summaries (每日) --- + # Queue every date that has health data but no insight yet. + try: + dates = query_all( + "SELECT DISTINCT date FROM health_data WHERE user_id = ? " + "AND date IS NOT NULL ORDER BY date DESC", + [user_id], + ) + for row in dates: + subject = row["date"] + if has_insight(user_id, "daily", subject): + continue + built = scopes.build(user_id, "daily", subject) + if built: + resolved, context = built + jobs.enqueue(user_id, "daily", subject, + _context_fingerprint(context), + jobs.PRIORITY_PREFETCH) + queued.append(f"daily:{subject}") + except Exception as e: # noqa: BLE001 + print(f"[analysis] prefetch daily failed: {e}") + + # --- activity summaries (运动详情) --- + # Queue every activity that has no insight yet. + try: + activities = query_all( + "SELECT id, start_time FROM activities WHERE user_id = ? " + "AND id IS NOT NULL ORDER BY start_time DESC", + [user_id], + ) + for row in activities: + subject = str(row["id"]) + if has_insight(user_id, "activity", subject): + continue + built = scopes.build(user_id, "activity", subject) + if built: + resolved, context = built + jobs.enqueue(user_id, "activity", subject, + _context_fingerprint(context), + jobs.PRIORITY_PREFETCH) + queued.append(f"activity:{subject}") + except Exception as e: # noqa: BLE001 + print(f"[analysis] prefetch activity failed: {e}") + context = insights.build_context(user_id) if context: jobs.supersede(user_id, "briefing", context["snapshotDate"]) @@ -557,6 +612,86 @@ def prefetch_insights(user_id): return queued +def refill_backlog(max_users=3, max_daily=100, max_activity=100): + """Called by the jobs worker when idle: queue more items that lack insights. + + Iterates over users who have health data or activities, and for each one + queues any daily/activity items that are not yet queued and have no insight + yet. Batched so a single call does not spend too long on SQL — the worker + will call this again on the next idle cycle. + """ + queued = 0 + try: + users = query_all( + "SELECT DISTINCT user_id FROM health_data WHERE user_id IS NOT NULL " + "UNION SELECT DISTINCT user_id FROM activities WHERE user_id IS NOT NULL " + "LIMIT ?", + [max_users], + ) + except Exception as e: # noqa: BLE001 + print(f"[analysis] refill_backlog users query failed: {e}") + return 0 + + for row in users: + uid = row["user_id"] + if not uid: + continue + + # --- daily --- + try: + dates = query_all( + "SELECT DISTINCT hd.date FROM health_data hd " + "WHERE hd.user_id = ? AND hd.date IS NOT NULL " + "AND NOT EXISTS (SELECT 1 FROM ai_insights ai " + " WHERE ai.user_id = hd.user_id AND ai.kind = 'daily' " + " AND ai.subject = hd.date) " + "AND NOT EXISTS (SELECT 1 FROM ai_jobs aj " + " WHERE aj.user_id = hd.user_id AND aj.kind = 'daily' " + " AND aj.subject = hd.date) " + "ORDER BY hd.date DESC LIMIT ?", + [uid, max_daily], + ) + for dr in dates: + subject = dr["date"] + built = scopes.build(uid, "daily", subject) + if built: + resolved, context = built + jobs.enqueue(uid, "daily", subject, + _context_fingerprint(context), + jobs.PRIORITY_PREFETCH) + queued += 1 + except Exception as e: # noqa: BLE001 + print(f"[analysis] refill_backlog daily for {uid} failed: {e}") + + # --- activity --- + try: + activities = query_all( + "SELECT a.id FROM activities a " + "WHERE a.user_id = ? AND a.id IS NOT NULL " + "AND NOT EXISTS (SELECT 1 FROM ai_insights ai " + " WHERE ai.user_id = a.user_id AND ai.kind = 'activity' " + " AND ai.subject = CAST(a.id AS TEXT)) " + "AND NOT EXISTS (SELECT 1 FROM ai_jobs aj " + " WHERE aj.user_id = a.user_id AND aj.kind = 'activity' " + " AND aj.subject = CAST(a.id AS TEXT)) " + "ORDER BY a.start_time DESC LIMIT ?", + [uid, max_activity], + ) + for ar in activities: + subject = str(ar["id"]) + built = scopes.build(uid, "activity", subject) + if built: + resolved, context = built + jobs.enqueue(uid, "activity", subject, + _context_fingerprint(context), + jobs.PRIORITY_PREFETCH) + queued += 1 + except Exception as e: # noqa: BLE001 + print(f"[analysis] refill_backlog activity for {uid} failed: {e}") + + return queued + + def _run_job(user_id, kind, subject): """What the queue worker calls. Dispatches on the job's kind.""" if kind == "briefing": @@ -569,6 +704,7 @@ def _run_job(user_id, kind, subject): jobs.set_runner(_run_job) +jobs.set_refiller(refill_backlog) def list_insights(user_id, limit=60): diff --git a/backend/services/jobs.py b/backend/services/jobs.py index e8861ca..aacaa62 100644 --- a/backend/services/jobs.py +++ b/backend/services/jobs.py @@ -37,19 +37,12 @@ POLL_SECONDS = float(os.environ.get("AI_JOB_POLL_SECONDS") or 2) MAX_ATTEMPTS = int(os.environ.get("AI_JOB_MAX_ATTEMPTS") or 3) -# The gateway is not ours alone. It runs `gunicorn -w 1 --threads 4` on the -# Oracle box and is shared with fam-edge and the camera project, so it can -# serve four requests at a time in total — while one of ours occupies a thread -# for two to five minutes, and up to seventeen when its NVIDIA chain retries -# (its own README, known issue #3). -# -# So this consumer deliberately runs one job at a time across the whole -# deployment, not one per Gunicorn worker, and leaves a gap between jobs. A -# backfill of eight scopes is background work; taking the shared box down to -# finish it sooner is not a trade worth making — and a 502 there is a 502 for -# the other two projects too. -MAX_CONCURRENT = int(os.environ.get("AI_JOB_CONCURRENCY") or 1) -GAP_SECONDS = float(os.environ.get("AI_JOB_GAP_SECONDS") or 20) +# The current gateway at ai.zichuan.xyz supports multi-concurrent requests, +# but three projects share it, so going too high causes 502s. +# MAX_CONCURRENT is read from the environment and defaults to 2; raise it +# if the gateway shows headroom, lower it if this or other projects get 502s. +MAX_CONCURRENT = int(os.environ.get("AI_JOB_CONCURRENCY") or 2) +GAP_SECONDS = float(os.environ.get("AI_JOB_GAP_SECONDS") or 5) # How long a job that exhausted its attempts stays given up on before it is # tried again. Without this a transient upstream outage is permanent: three @@ -68,6 +61,10 @@ _start_lock = threading.Lock() # import each other in a cycle. _runner = None +# Optional refiller, called when the queue is idle to keep the backlog +# populated with daily/activity items that have no insight yet. +_refiller = None + def set_runner(fn): """Register `fn(user_id, kind, subject) -> None`, called for each job.""" @@ -75,6 +72,12 @@ def set_runner(fn): _runner = fn +def set_refiller(fn): + """Register `fn() -> int`, called when the queue is idle to refill backlog.""" + global _refiller + _refiller = fn + + def _now(): return datetime.datetime.utcnow() @@ -332,13 +335,30 @@ def run_once(): def _loop(): + idle_cycles = 0 while True: try: # A gap after each job, not just when idle. Running them # back-to-back is what saturates the shared gateway: a backfill # finishing twenty minutes later costs nobody anything, a gateway # that stops answering costs all three projects. - time.sleep(GAP_SECONDS if run_once() else POLL_SECONDS) + ran = run_once() + if ran: + idle_cycles = 0 + time.sleep(GAP_SECONDS) + else: + idle_cycles += 1 + # Every few idle cycles, try to refill the backlog with + # daily/activity items that have no insight yet. + if idle_cycles >= 5 and _refiller is not None: + try: + n = _refiller() + if n: + print(f"[ai-jobs] refilled {n} backlog items") + except Exception as e: # noqa: BLE001 + print(f"[ai-jobs] refill error: {e}") + idle_cycles = 0 + time.sleep(POLL_SECONDS) except Exception as e: # noqa: BLE001 - the loop must outlive any failure print(f"[ai-jobs] worker error: {e}") time.sleep(POLL_SECONDS)