Make the AI user profile backend-owned for every client; delete the Dream state pass

backend/database/dream_store.py

@@ -1,4 +1,4 @@
 """Transactional Dream admission and conservative stale-lease settlement.
 
 Abandoned workers retain their reservation and pending dirty versions. New
 admission and reclamation share one transaction, fenced by the current run_id.
@@ -208,7 +208,7 @@
     return usage
 
 
 def acquire(
     uid,
     caps: Caps,
     *,
@@ -315,8 +315,6 @@
             'hunt': not has_dirty,
             'dry_hunts': _dry_hunts(data),
             'backoff_would_defer': backoff_would_defer and not has_dirty,
-            # Weekly whole-corpus state pass cadence; derived from the state doc already read.
-            'state_pass_due': _state_pass_due(data, now),
             'canary': canary,
             'mode': 'shadow' if canary else mode(),
             'size_bucket': data.get('size_bucket', 'light'),
@@ -467,16 +465,6 @@
         raise RuntimeError('dream_lease_lost')
 
 
-def _state_pass_due(data, now):
-    # Kept here (not utils.dream_state_pass) so acquire stays import-light.
-    try:
-        days = max(1, int(os.getenv('DREAM_STATE_PASS_WEEK_DAYS', '7') or 7))
-    except ValueError:
-        days = 7
-    last = data.get('last_state_pass_at')
-    return not isinstance(last, datetime) or now - last >= timedelta(days=days)
-
-
 def _dry_hunts(data):
     try:
         return max(0, int(data.get('dry_hunts', 0) or 0))
@@ -484,7 +472,7 @@
         return 0
 
 
 def finish(
     uid,
     lease,
     report,
@@ -519,7 +507,7 @@
     deferred_ids = {dirty_id(*ref.split('/', 1)) for ref in deferred - set(seeded)}
 
     @firestore.transactional
     def complete(tx):
         data = state.get(transaction=tx).to_dict() or {}
         current_lease = data.get('lease') or {}
         if current_lease.get('run_id') != lease['run_id']:
@@ -730,8 +718,6 @@
                 patch['cleanup_cursor'] = cleanup_cursor
             if not zero_refund:
                 patch['last_pass_at'] = lease['started_at']
-                if report.get('state_pass_done'):
-                    patch['last_state_pass_at'] = lease['started_at']
                 for ref in sorted(set(reviewed) - deferred)[:8]:
                     tx.set(
                         state.collection('reviews').document(review_id(ref)),

backend/database/redis_db.py

@@ -1,4 +1,4 @@
 import ast
 import base64
 import json
 from typing import Any, Callable, Dict, List, Optional, TypeVar, Union, cast
@@ -1406,6 +1406,16 @@
     return result is not None
 
 
+def try_acquire_ai_profile_refresh_lock(uid: str, ttl: int = 60 * 60) -> bool:
+    """One backend AI-profile refresh per owner per ``ttl``; also throttles retries after a failure."""
+    result = r.set(f'users:{uid}:ai_profile_refresh_lock', '1', ex=ttl, nx=True)
+    return result is not None
+
+
+def release_ai_profile_refresh_lock(uid: str) -> None:
+    r.delete(f'users:{uid}:ai_profile_refresh_lock')
+
+
 def try_acquire_daily_wear_lock(uid: str, date: str, ttl: int = 60 * 60 * 24) -> bool:
     """At most one wear FCM per uid per UTC day. True iff this caller may send."""
     result = r.set(f'users:{uid}:daily_wear_lock:{date}', '1', ex=ttl, nx=True)

backend/database/users.py

@@ -1,4 +1,4 @@
 from database.people_generation import bump_people_generation, write_person
 from database.person_speech_samples import add_sample_transaction
 from database.people_projection import project_people
 from config.refactor_modes import serving_for
@@ -2501,21 +2501,66 @@
     return get_or_fetch(_USER_AI_PROFILE_CACHE, uid, lambda: _get_ai_user_profile_from_firestore(uid))
 
 
+def get_ai_user_profile_fresh(uid: str) -> Optional[dict]:
+    """Uncached read for read-modify-write and refresh cadence decisions."""
+    return _get_ai_user_profile_from_firestore(uid)
+
+
 def update_ai_user_profile(
-    uid: str, profile_text: str = None, generated_at=None, data_sources_used: int = None
+    uid: str,
+    profile_text: str = None,
+    generated_at=None,
+    data_sources_used: int = None,
+    source: Optional[str] = None,
 ) -> dict:
-    """Update AI user profile.  Only writes non-None fields (partial update)."""
+    """Update AI user profile.  Only writes non-None fields (partial update).
+
+    ``source`` records who wrote the text (``user_edit``/``exploration`` from clients);
+    a legacy PATCH without one is recorded as ``client``.
+    """
     # Read existing profile directly from Firestore — never from cache — because
     # this is a read-modify-write path. Using a stale cached projection here
     # could overwrite newer profile fields.
     existing = _get_ai_user_profile_from_firestore(uid) or {}
     if profile_text is not None:
         existing['profile_text'] = profile_text
+        existing['source'] = source or 'client'
     if generated_at is not None:
         existing['generated_at'] = generated_at
+    elif profile_text is not None and source:
+        existing['generated_at'] = datetime.now(timezone.utc).isoformat()
     if data_sources_used is not None:
         existing['data_sources_used'] = data_sources_used
     user_ref = db.collection('users').document(uid)
     user_ref.update({'ai_user_profile': existing})
     invalidate(_USER_AI_PROFILE_CACHE, uid)
     return existing
+
+
+@transactional
+def _write_ai_user_profile_if_unchanged_txn(transaction, user_ref, expected_text: str, profile: dict) -> bool:
+    snapshot = user_ref.get(transaction=transaction)
+    if not snapshot.exists:
+        return False
+    current = (snapshot.to_dict() or {}).get('ai_user_profile') or {}
+    if str(current.get('profile_text') or '').strip() != expected_text:
+        return False
+    transaction.update(user_ref, {'ai_user_profile': profile})
+    return True
+
+
+def write_ai_user_profile_if_unchanged(uid: str, expected_text: str, profile: dict) -> bool:
+    """Replace the generated profile only if the stored text is still the synthesis input.
+
+    A user edit or delete that lands while the model runs wins; returns False then.
+    """
+    client = get_firestore_client()
+    user_ref = client.collection('users').document(uid)
+    written = _write_ai_user_profile_if_unchanged_txn(client.transaction(), user_ref, expected_text, profile)
+    invalidate(_USER_AI_PROFILE_CACHE, uid)
+    return written
+
+
+def delete_ai_user_profile(uid: str) -> None:
+    db.collection('users').document(uid).update({'ai_user_profile': firestore.DELETE_FIELD})
+    invalidate(_USER_AI_PROFILE_CACHE, uid)

backend/deploy/runtime_env.yaml

@@ -4909,7 +4909,6 @@ environments:
               category: reliability
             REVIEW_SURFACE_MODE: {value: 'on', category: rollout}
             DREAM_AGENT_PASSES_PER_DAY: {value: '48', category: rollout}
-            DREAM_STATE_PASS_MODE: {value: shadow, category: rollout}
             DREAM_AGENT_HUNT_INTERVAL_HOURS: {value: '2', category: rollout}
             DREAM_AGENT_MANUAL_RUNS_PER_DAY: {value: '3', category: rollout}
             DREAM_AGENT_DAILY_USD: {value: '20', category: rollout}
@@ -5422,7 +5421,6 @@ environments:
               value: '8'
               category: reliability
             DREAM_AGENT_PASSES_PER_DAY: {value: '48', category: rollout}
-            DREAM_STATE_PASS_MODE: {value: shadow, category: rollout}
             MENTOR_PIPELINE: {value: cohort, category: rollout}
             SPEAKER_GROUPING_SHADOW: {value: cohort, category: rollout}
             SPEAKER_GROUPING_SHADOW_PERCENT: {value: '10', category: rollout}

backend/deploy/runtime_env/prod.overlay.yaml

@@ -1168,9 +1168,6 @@ overlay:
           DREAM_AGENT_PASSES_PER_DAY:
             value: '48'
             category: rollout
-          DREAM_STATE_PASS_MODE:
-            value: shadow
-            category: rollout
           DREAM_AGENT_HUNT_INTERVAL_HOURS:
             value: '2'
             category: rollout
@@ -1572,9 +1569,6 @@ overlay:
           DREAM_AGENT_PASSES_PER_DAY:
             value: '48'
             category: rollout
-          DREAM_STATE_PASS_MODE:
-            value: shadow
-            category: rollout
           MENTOR_PIPELINE:
             value: cohort
             category: rollout