From b504a27c7f9658e9e36945b8df1fb4ad249a8b3c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 19:44:14 +0000 Subject: [PATCH] Make the hourly sweep reflection's only driver, with a drain loop Reflection previously rode the chat turn's waitUntil tail as its primary driver, with the hourly cron sweep as a one-thread-per-tick catch-up. That coupled memory-formation throughput to how much the user happens to converse, and made the moment of formation unpredictable relative to the rest of the background fleet. Remove the tail call and its per-user claim path entirely, and turn runReflectionSweepTick into a sequential drain loop: claim the most-overdue eligible thread across all users, reflect it, claim the next - until the queue is empty, the per-tick cap (5 threads) is hit, or the time budget (180s of new claims) is spent. The next hourly tick resumes the drain, so throughput no longer depends on chat activity, and the strictly sequential loop keeps at most one reflection agent writing to the memory store per tick. The eligibility gates (prior-calendar-day, >= 2 user messages) are unchanged; the fixed cadence gives the user a predictable window to edit or retry a settled conversation before reflection reads it. The per-user claim_next_thread_for_reflection RPC loses its only caller and is dropped from the schema; mark_thread_reflected_if_claimed stays (the sweep's mark path). Docs, the QA walkthrough, and the dev cron shim comments follow the new drive shape. --- docs/dev/architecture.md | 7 +- docs/dev/chat.md | 16 +- docs/dev/memory.md | 85 ++++--- docs/dev/samskara.md | 14 +- docs/dev/second-thoughts.md | 4 +- docs/qa/use-cases/context-recall-priming.md | 2 +- docs/qa/use-cases/reflection-drain.md | 46 ++-- docs/user/memory.md | 9 +- scripts/dev-backfill-cron.mjs | 4 +- src/lib/supabase/realtime.ts | 2 +- src/screens/Chat.svelte | 2 +- .../functions/venice/agents/reflection.ts | 239 +++++++++--------- .../venice/agents/second_thoughts.ts | 2 +- .../functions/venice/getStreamingResponse.ts | 45 ++-- supabase/functions/venice/index.ts | 13 +- supabase/schema.sql | 180 +++---------- 16 files changed, 278 insertions(+), 392 deletions(-) diff --git a/docs/dev/architecture.md b/docs/dev/architecture.md index b00a2707..cf4bd989 100644 --- a/docs/dev/architecture.md +++ b/docs/dev/architecture.md @@ -306,10 +306,13 @@ browser tab being open.** The fleet runs server-side in the venice edge function on two kinds of trigger: - **Chat-turn tail** - `getStreamingResponse` registers a - sequential curation -> samskara -> reflection chain under + sequential curation -> samskara chain under `EdgeRuntime.waitUntil` after each completed turn. The tail is the low-latency driver for work the user notices - in-session: thread titles, samskara mints, fresh memories. + in-session: thread titles, samskara mints. Reflection is + deliberately NOT in the chain - memory formation is + sweep-only so it keeps a fixed, predictable cadence (see + `memory.md`). - **pg_cron sweeps** - scheduled jobs that pg_net-POST a function route, as the catch-up and maintenance drivers. The minute ladder: embed backfill `*/5`, bias `:03`, wiki diff --git a/docs/dev/chat.md b/docs/dev/chat.md index 6e457e2b..c760b7b0 100644 --- a/docs/dev/chat.md +++ b/docs/dev/chat.md @@ -412,15 +412,13 @@ A chat turn goes: `last_summarised_msg_id`. The chat loop creates that assistant message; the background workers pick it up on their next poll. See `./summaries.md`. -- **Reflection** — driven directly from the completed-turn - tail. `getStreamingResponse` (the streaming orchestrator) - fires `reflectOneThread` via `EdgeRuntime.waitUntil` after - the chat response ships, draining one day-gate-eligible - thread from the reflection queue as background work. The - chat loop creates the terminal assistant message that makes - a thread eligible; reflection acts on it on the same turn's - tail (after at least a calendar day has elapsed). See - `./memory.md`. +- **Reflection** — no direct call. The chat loop creates the + terminal assistant message that makes a thread eligible; + the hourly reflection sweep (reflection's only driver) + picks the thread up once its newest message is at least a + calendar day old. Deliberately not tail-driven, so edits + and retries any time before that sweep are what reflection + sees. See `./memory.md`. - **Topics** — `Chat.svelte` owns the `selectedTopics` / `topicsVocabulary` state for the drawer's topic-filter dropdown and threads `selectedTopics` through the three diff --git a/docs/dev/memory.md b/docs/dev/memory.md index 85c1c2d0..ee37f88c 100644 --- a/docs/dev/memory.md +++ b/docs/dev/memory.md @@ -41,11 +41,11 @@ ride the system prompt independently. ## Role in the app Every turn the main chat model can call `memory_recall` to pull in -relevant memories; at the tail of each completed streaming chat turn, -the venice edge function fires a reflection pass that reads a settled -thread and decides whether to write, update, or invalidate memories -based on what it saw. The store lives in each user's own Supabase; -the writes happen through the same tool harness the main chat uses. +relevant memories; on an hourly cron cadence, the venice edge +function's reflection sweep reads settled threads and decides whether +to write, update, or invalidate memories based on what it saw. The +store lives in each user's own Supabase; the writes happen through +the same tool harness the main chat uses. From the user's perspective this is the Memory feature documented in `docs/user/memory.md`. The dev side has five moving parts: @@ -53,8 +53,8 @@ in `docs/user/memory.md`. The dev side has five moving parts: 1. **The store** — `memories` table, RLS-scoped, with a pgvector `embedding` column populated asynchronously. 2. **The writer** — the reflection agent runs in the venice edge - function, reads settled threads end-to-end, and uses a - write-scoped subset of the memory tools. + function on the hourly sweep, reads settled threads end-to-end, + and uses a write-scoped subset of the memory tools. 3. **The reader** — the recall agent runs in the venice edge function, inline in the tool dispatch when the main model invokes the `memory_recall` tool. Read-only. @@ -124,9 +124,9 @@ in `docs/user/memory.md`. The dev side has five moving parts: `memory_conversation` row per surfaced memory - the co-occurrence hint queue rem drains. - `supabase/functions/venice/agents/reflection.ts` — the reflection - agent. Exports `reflectOneThread(adminClient, userId)`; runs - write-scoped with no return value (side effects = memory tool - calls). Its prompt instructs TIMELESS memories - no "this session", + agent. Exports `runReflectionSweepTick(adminClient)`; runs + write-scoped (side effects = memory tool calls) and returns a + per-tick drain summary. Its prompt instructs TIMELESS memories - no "this session", no write-date narration, no first-person AI self-logging - because a body that stamps when it was written reads back later as a current-chat event (the row's `created_at` already records when it @@ -327,26 +327,25 @@ in `docs/user/memory.md`. The dev side has five moving parts: headless read-only tool loop on the fast tier. Returns a structured JSON output the tool encodes as the `role='tool'` message payload for the next round. -- **Reflection (edge function tail)** — at the end of each - successfully completed streaming chat turn, `getStreamingResponse` - fires `reflectOneThread(adminClient, userId)` via - `EdgeRuntime.waitUntil` as background work after the chat response - ships. Each invocation opportunistically drains one day-gate- - eligible thread from the existing reflection queue - NOT - necessarily the thread that just finished. Claim mutual exclusion - is the per-thread claim RPC (each call uses a fresh random holder - id); no `worker_leases` row is involved. -- **Reflection catch-up sweep** — pg_cron job +- **Reflection sweep (the only driver)** — pg_cron job `nak-reflection-sweep` (hourly, minute 27) pg_net-POSTs - `/reflection-sweep` -> `runReflectionSweepTick`, which claims the - most-overdue eligible thread across ALL users + `/reflection-sweep` -> `runReflectionSweepTick`, which drains the + day-gated queue one thread at a time: claim the most-overdue + eligible thread across ALL users (`claim_next_thread_for_reflection_sweep`, SECURITY DEFINER, - per-owner timezone off the profile) and runs the same shared - reflect body. Exists because the tail only fires when its owner - converses - without it a dormant account's queue never moves. The - per-thread claim makes tail + sweep double-driving safe. The dev - shim ticks this route too. -- **Reflection attempt cap** — both reflection claims count + per-owner timezone off the profile), reflect it, claim the next - + until the queue is empty, the per-tick cap (5 threads) is hit, or + the tick's time budget (180s of new claims) is spent; the next + hourly tick resumes. Reflection deliberately does NOT ride the + chat turn's waitUntil tail (unlike curation and samskara): the + fixed cadence gives the user a predictable window to edit or + retry a settled conversation before reflection reads it, and the + strictly sequential drain keeps at most one reflection agent + writing to the store per tick. Claim mutual exclusion is the + per-thread claim RPC (each cycle uses a fresh random holder id); + no `worker_leases` row is involved, and overlapping ticks simply + claim different threads. The dev shim ticks this route too. +- **Reflection attempt cap** — the sweep claim counts ATTEMPTS at claim time (`threads.reflection_attempt_msg_id` + `reflection_attempt_count`): three claims against the same terminal message and the thread stops being offered, until a new @@ -503,7 +502,7 @@ in `docs/user/memory.md`. The dev side has five moving parts: - **Reflection claim columns** — `threads.reflection_holder_id` and `threads.reflection_claim_expires_at` are the mutual- exclusion primitive for the server-side reflection path. - The claim-RPC pair (`claim_next_thread_for_reflection` / + The claim-RPC pair (`claim_next_thread_for_reflection_sweep` / `mark_thread_reflected_if_claimed`) uses a fresh random holder id per call; there is no `worker_leases` row for reflection. - **`memories.last_librarian_visit_at timestamptz`** — per-row @@ -651,14 +650,16 @@ from the same ports); the browser carries only the wire schemas. `done`); attaching it also injects the `activity` narration parameter into the tools' wire schemas, so the sweep paths (no listener) stay narration-free. -- `reflectOneThread(adminClient, userId)` — the edge function - entry point. Claims one day-gate-eligible thread (newest message - on a prior calendar day in the user's timezone, with >= 2 user - messages), runs the reflection agent's headless tool loop, and - stamps `last_reflected_msg_id` via a claim-guarded RPC. The - agent's "answer" is whatever `memory_*` tool calls it made; - the final text is discarded. Returns without error when the - queue is empty or the claim is lost. +- `runReflectionSweepTick(adminClient)` — the edge function + entry point. Drains day-gate-eligible threads (newest message + on a prior calendar day in the owner's timezone, with >= 2 user + messages) one at a time up to the per-tick cap/budget. Each + cycle runs the reflection agent's headless tool loop and stamps + `last_reflected_msg_id` via a claim-guarded RPC. The agent's + "answer" is whatever `memory_*` tool calls it made; the final + text is discarded. Non-throwing; returns a per-tick summary + (`reflected` / `claimLost` / `emptySlice` counters plus what + stopped the drain). ## The body-length budget (non-growth rule) @@ -826,11 +827,11 @@ renders the delta as a chip (`memorySizeDelta` in `./topics.md` under "Memory topics" for the unit shape, the schema deltas, and the trigger / claim discipline. - **Summaries / conversation recall** — separate store (thread - rows), separate agents. Summary and reflection both run in the - venice edge function, fired from the completed-chat-turn tail - with an hourly sweep as catch-up. Both use the same per-row - claim-RPC pattern on `threads` but have independent claim - columns and no shared lease. + rows), separate agents. Both run in the venice edge function: + summary from the completed-chat-turn tail with an hourly sweep + as catch-up, reflection from its hourly sweep only. Both use + the same per-row claim-RPC pattern on `threads` but have + independent claim columns and no shared lease. - **Logging** - the reflection agent and both librarian passes emit breadcrumbs through `createEdgeLogger` (sources `reflection`, `rem`, `deep-sleep`), which both writes to the diff --git a/docs/dev/samskara.md b/docs/dev/samskara.md index c1e76f88..5f8e0fc0 100644 --- a/docs/dev/samskara.md +++ b/docs/dev/samskara.md @@ -254,7 +254,7 @@ toast is just a glance cue that the bias model is forming. LLM work. - **`samskaraOnTurnTail(admin, userId)`** - fired from `getStreamingResponse`'s `EdgeRuntime.waitUntil` tail on - completed turns, sequenced curation -> samskara -> reflection. + completed turns, sequenced curation -> samskara. Runs the session-responsive phases: a capped assimilate drain, then one pair-relate probe, then one mint-tier1 probe. (Reaction scoring is no longer a tail phase - it moved to the next-day @@ -566,8 +566,8 @@ embeddings backfill's claim -> process -> save shape. Two drivers run the phases, split by timing sensitivity: - **Turn tail** (`samskaraOnTurnTail`) - an assimilate drain - capped at `TAIL_ASSIMILATE_CAP` (3) so the - chain never delays reflection behind it, then one pair-relate + capped at `TAIL_ASSIMILATE_CAP` (3) so one tail invocation + never monopolises the background budget, then one pair-relate probe, then one mint-tier1 probe (the in-session toast surface). - **Hourly sweep** (`runSamskaraSweepTick`) - a cross-user @@ -1175,7 +1175,7 @@ summarizer reads samskaras to feed the agent. options struct, and samskara is currently the only caller. Function side, `getStreamingResponse`'s waitUntil tail drives `samskaraOnTurnTail` on every completed turn, sequenced - curation -> samskara -> reflection. `Chat.svelte` mounts the + curation -> samskara. `Chat.svelte` mounts the single `` component and owns the `subscribeToSamskaraInserts` realtime subscription that turns `samskara-mint` Broadcast events into `SAMSKARA_MINT_EVENT`. See @@ -1192,9 +1192,9 @@ summarizer reads samskaras to feed the agent. predictive bias the model formed on its own. No data flows between them. The reflection agent reads thread transcripts and writes memories; the samskara assimilator reads - individual exchanges and writes substrate. Both ride the same - waitUntil tail, samskara first (reflection can span minutes - and samskara carries the fleet's only hard timing window). + individual exchanges and writes substrate. Samskara rides the + waitUntil tail (it carries the fleet's only hard timing + window); reflection is sweep-only on an hourly cadence. See `./memory.md`. - **Bias profile** - sibling server-side pipeline, no data flow. Bias profile aggregates cognitive-bias observations across diff --git a/docs/dev/second-thoughts.md b/docs/dev/second-thoughts.md index 8e9298db..281f3b03 100644 --- a/docs/dev/second-thoughts.md +++ b/docs/dev/second-thoughts.md @@ -63,7 +63,7 @@ user live. Runs as a turn-tail unit (`secondThoughtsOnTurnTail`), FIRST in the `EdgeRuntime.waitUntil` tail of `getStreamingResponse.ts` where -`curateOnTurnTail` / `samskaraOnTurnTail` / `reflectOneThread` also run, +`curateOnTurnTail` / `samskaraOnTurnTail` also run, and only on `terminalKind === 'completed'`. Detached, so it adds zero latency to the user-visible turn. Best-effort and non-throwing: a failure leaves the row without a verdict, never breaks the turn. @@ -331,7 +331,7 @@ composition + wiring. ## Interactions - **Chat ([`chat.md`](./chat.md))** - the reviewer is a turn-tail unit - next to curation/samskara/reflection. The refinement reuses the + next to curation/samskara. The refinement reuses the browser send/regenerate flow for one extra APPEND turn anchored to the original user message; it touches the send path, `commit_assistant_message`'s anchor handling (append does not conflict diff --git a/docs/qa/use-cases/context-recall-priming.md b/docs/qa/use-cases/context-recall-priming.md index 89a7c7da..f914cc2e 100644 --- a/docs/qa/use-cases/context-recall-priming.md +++ b/docs/qa/use-cases/context-recall-priming.md @@ -170,7 +170,7 @@ the recall-quality properties those subsystems now owe. 8. **Timeless writer (reflection).** In a SEPARATE existing thread (>= 2 user messages, last activity on a prior calendar day so the drain is eligible), hold a short exchange that teaches one clear new fact about - the user. Let reflection run (chat-turn tail, or the hourly sweep; see + the user. Let reflection run (the hourly sweep; see [reflection-drain](./reflection-drain.md) to force it). Inspect the newest memory: diff --git a/docs/qa/use-cases/reflection-drain.md b/docs/qa/use-cases/reflection-drain.md index 7aab68b9..be85e6ed 100644 --- a/docs/qa/use-cases/reflection-drain.md +++ b/docs/qa/use-cases/reflection-drain.md @@ -1,11 +1,14 @@ -# Reflection: tail drain, catch-up sweep, attempt cap +# Reflection: sweep drain loop, attempt cap ## Covers -The reflection agent's two drivers - the chat-turn waitUntil tail -and the hourly `/reflection-sweep` cron route - plus the per-thread -claim mutual exclusion and the attempt cap +The reflection agent's single driver - the hourly `/reflection-sweep` +cron route's per-tick drain loop (claim one thread, reflect it, claim +the next, up to the cap/time budget) - plus the per-thread claim +mutual exclusion and the attempt cap ([dev: memory](../../dev/memory.md), "Reflection" entries). +Reflection deliberately does NOT run on the chat-turn tail: a +completed turn must produce no reflection activity. ## Preconditions @@ -22,13 +25,17 @@ claim mutual exclusion and the attempt cap where id = ''; ``` +- For the multi-thread drain check (step 2), make two or more + threads eligible with the same statement. - `SR` = the service-role key from `supabase status -o json`. ## Steps -1. Tail drain: send a chat message in ANY thread and let the turn - complete. Watch the Logs drawer's `reflection` source. -2. Sweep: tick the route directly and watch the same source: +1. No tail drive: with an eligible thread queued, send a chat + message in ANY thread and let the turn complete. Watch the Logs + drawer's `reflection` source. +2. Sweep drain: tick the route directly with 2+ eligible threads + queued and watch the same source: ```sh curl -s -X POST \ @@ -56,15 +63,17 @@ claim mutual exclusion and the attempt cap ## Expected -- (1) On a turn whose user has an eligible OLDER thread queued, the - drawer shows `[reflection] picked up thread ...` then - `finished thread ... (N tool calls over M messages)`; new - memories appear for content-bearing threads. With an empty queue - the tail is silent at default levels (trace line only). +- (1) The completed turn produces NO `reflection` lines in the + drawer and the eligible thread's `last_reflected_msg_id` does not + advance - the tail no longer drives reflection. - (2) Immediate `{"accepted":true}` (the tick runs detached); the - drawer shows the same picked-up/finished pair when a thread was - eligible, `last_reflected_msg_id` advances, and - `reflection_attempt_count` resets to 0 on the mark. + drawer shows a `picked up thread ...` / `finished thread ... (N + tool calls over M messages)` pair PER eligible thread, one after + another (sequential, not interleaved), until the queue empties or + the cap (5 threads) / time budget (180s of new claims) stops the + loop. `last_reflected_msg_id` advances and + `reflection_attempt_count` resets to 0 on each mark; new memories + appear for content-bearing threads. - (3) Gateway 401 without a JWT; route-level `{"error":"forbidden"}` 403 with a non-service JWT. - (4) The count reaches 3 and the fourth claim returns no row for @@ -73,7 +82,8 @@ claim mutual exclusion and the attempt cap hosted invocation wall clock, OR caps out at 3 attempts and stops burning Venice calls. Local measurement: ~9 minutes end-to-end on a 14-message/69KB thread - likely over the hosted window; the cap - is the backstop. + is the backstop, and the 180s claim cutoff keeps one slow thread + from dragging later claims past the wall clock with it. ## Cleanup @@ -90,7 +100,7 @@ update threads set reflection_attempt_count = 0, | Date | Env | Commit | Result | Notes | | ---- | --- | ------ | ------ | ----- | -| 2026-06-09 | local | 4e33cc3 | pass (1) | four queued reflections drained in order across turns, drawer lines live | -| 2026-06-10 | local | d37dbcd | pass (2,3) | detached tick accepted in 215ms; sweep claimed cross-user; 401/403 posture held | +| 2026-06-09 | local | 4e33cc3 | pass (1) | pre-rework baseline: four queued reflections drained in order across turns via the then-extant tail driver, drawer lines live | +| 2026-06-10 | local | d37dbcd | pass (2,3) | pre-rework baseline: detached tick accepted in 215ms; sweep claimed cross-user; 401/403 posture held | | 2026-06-10 | local | 2e37c8b | pass (4) | counter hit 3, fourth claim skipped the thread | | 2026-06-10 | local | d37dbcd | note | full reflect+mark on the 69KB thread took ~9 min detached; completed only after the TTL fix (600s) | diff --git a/docs/user/memory.md b/docs/user/memory.md index e4c6a2f1..27e50ec0 100644 --- a/docs/user/memory.md +++ b/docs/user/memory.md @@ -34,9 +34,12 @@ or overly behavioral. That's intentional. ## How it grows -When a conversation settles (no new messages for a while), a -background agent reads the whole thread and decides whether anything -it saw is worth remembering. For each candidate memory, the agent: +When a conversation settles (its newest message is at least a +calendar day old), an hourly background pass reads the whole thread +and decides whether anything it saw is worth remembering. The lag is +deliberate: until that pass runs, you can still edit or retry any +part of the conversation and memory formation only ever sees the +corrected version. For each candidate memory, the agent: 1. **Searches existing memories** for anything close. Duplicates are worse than nothing — they dilute search. diff --git a/scripts/dev-backfill-cron.mjs b/scripts/dev-backfill-cron.mjs index ea87a2d5..e6a55ee4 100644 --- a/scripts/dev-backfill-cron.mjs +++ b/scripts/dev-backfill-cron.mjs @@ -18,8 +18,8 @@ // (the two memory librarians; same most-overdue-user claim shape // with their own 12h cadences); // - `nak_trigger_reflection_sweep()` hourly -> POST /reflection-sweep -// (reflection's catch-up drain - the chat-turn tail is the primary -// driver, this reaches queues whose owners stopped conversing); +// (reflection's only driver - each tick drains eligible threads +// one at a time until its cap or time budget stops it); // - `nak_trigger_curation_sweep()` hourly -> POST /curation-sweep, // `nak_trigger_bias_sweep()` hourly -> POST /bias-sweep, and // `nak_trigger_samskara_sweep()` hourly -> POST /samskara-sweep diff --git a/src/lib/supabase/realtime.ts b/src/lib/supabase/realtime.ts index b22bc13e..9513b42c 100644 --- a/src/lib/supabase/realtime.ts +++ b/src/lib/supabase/realtime.ts @@ -488,7 +488,7 @@ export function subscribeToGroceryChanges( /** * Subscribe to any change on the signed-in user's memories. The * wiki-articles twin above, for the memory writers that all live - * server-side now (reflection on the chat-turn tail, the rem and + * server-side now (the hourly reflection sweep, the rem and * deep-sleep librarian sweeps): the caller (Chat.svelte) routes the * notification into emitMemoryChange so an open Memories panel * refetches through the path it already had. Same coarse contract - diff --git a/src/screens/Chat.svelte b/src/screens/Chat.svelte index 43820299..cf281c2a 100644 --- a/src/screens/Chat.svelte +++ b/src/screens/Chat.svelte @@ -2002,7 +2002,7 @@ }); // Realtime: the memories twin of the wiki relay above. Every memory - // writer is server-side now (reflection on the chat-turn tail, the + // writer is server-side now (the hourly reflection sweep, the // rem / deep-sleep librarian sweeps), so this subscription is how an // open Memories panel learns a background write landed. $effect(() => { diff --git a/supabase/functions/venice/agents/reflection.ts b/supabase/functions/venice/agents/reflection.ts index c0e84ced..82a3ba44 100644 --- a/supabase/functions/venice/agents/reflection.ts +++ b/supabase/functions/venice/agents/reflection.ts @@ -6,28 +6,29 @@ // user. The model's final text is discarded - the memory_* side effects // ARE the output. // -// Drive shape - it drains OLDER threads, not "this" one. This module is -// fired from getStreamingResponse's terminal tail (via edgeWaitUntil) -// once per completed chat turn, but it does NOT reflect the thread that -// just finished. claim_next_thread_for_reflection only claims a thread -// whose newest message lands on a PRIOR calendar day in the user's -// timezone and that carries >= 2 user messages. So each turn-completion -// opportunistically drains ONE reflection-eligible thread from the -// existing day-gated queue. The day-gate exists because memory_recall -// has no per-conversation source attribution: a memory derived from a +// Drive shape - the hourly cron sweep is the ONLY driver. Each tick +// drains the day-gated queue one thread at a time (claim -> reflect -> +// claim the next) until the queue is empty, the per-tick cap is hit, +// or the tick's time budget runs out; the next hourly tick resumes. +// The sweep claim only offers a thread whose newest message lands on a +// PRIOR calendar day in the owner's timezone and that carries >= 2 +// user messages. The day-gate exists because memory_recall has no +// per-conversation source attribution: a memory derived from a // half-finished thought must not ride straight back into the same -// conversation that produced it. Faithful to the browser supervisor's -// behaviour (same queue, same gate); only the driver changed from a -// supervisor poll to a chat-activity piggyback. +// conversation that produced it. Reflection deliberately does NOT ride +// the chat turn's waitUntil tail the way curation and samskara do: +// memory formation gets a fixed, predictable cadence (edit or retry a +// conversation any time before the next hourly tick after midnight and +// reflection only ever sees the corrected thread), and the queue +// drains for dormant users at the same rate as active ones. // -// No lease coordinator. The browser ran reflection under a -// LeaseCoordinator so that only one of several open tabs/devices drove -// the supervisor at a time. Server-side that coordination is moot: the -// claim RPC's atomic per-thread claim+TTL IS the mutual exclusion. Two -// concurrent edge invocations that both call claim simply get two -// different threads (or one gets none) - more reflection throughput, not -// a correctness problem. So this module claims with a fresh per-call -// holder id and skips the lease machinery entirely. +// No lease coordinator, and no global singleton guard. The claim RPC's +// atomic per-thread claim+TTL IS the mutual exclusion: two overlapping +// ticks that both call claim simply get two different threads (or one +// gets none). Within a tick the drain loop is strictly sequential, so +// at most one reflection agent per tick is writing to the memory store +// at a time - which is what keeps near-simultaneous duplicate writes +// unlikely without any cross-invocation coordination. import type { SupabaseClient } from '@supabase/supabase-js'; import { createEdgeLogger } from '../../_shared/edge-log.ts'; @@ -450,106 +451,17 @@ function buildReflectionToolbox(): Toolbox { }; } -/** - * Resolve the user's display timezone for the day-gate. Stored in - * profiles.settings.displayTimezone (Settings -> AI -> About you). - * Falls back to UTC, matching the claim RPC's own p_timezone default. - */ -async function loadDisplayTimezone( - adminClient: SupabaseClient, - userId: string, -): Promise { - const { data, error } = await adminClient - .from('profiles') - .select('settings') - .eq('user_id', userId) - .maybeSingle<{ settings: Record | null }>(); - if (error || !data?.settings) return 'UTC'; - const tz = data.settings.displayTimezone; - return typeof tz === 'string' && tz.length > 0 ? tz : 'UTC'; -} - -/** Outcome of one reflectOneThread cycle, for the caller's diagnostic log. */ -export interface ReflectionCycleResult { +/** Outcome of one claim+reflect cycle inside a sweep tick's drain loop. */ +interface ReflectionCycleResult { outcome: 'no-thread' | 'empty-slice' | 'reflected' | 'claim-lost' | 'error'; threadId?: string; toolCalls?: number; } /** - * Run one reflection cycle for `userId`: claim the oldest day-gate- - * eligible thread, reflect on it, mark it done. A no-op when the queue - * is empty. Best-effort and NON-throwing by contract - the caller fires - * this from a chat turn's background tail and must not let a reflection - * failure touch the turn's recorded outcome, so every failure path is - * caught here, logged, and folded into an `error` result. Progress is - * logged through an edge logger so the browser Logs drawer sees the - * cycle even though it runs server-side; flush() at the end guarantees - * the final line lands before the waitUntil tail tears down. - */ -export async function reflectOneThread( - adminClient: SupabaseClient, - userId: string, -): Promise { - const log = createEdgeLogger(userId, 'reflection'); - try { - const timezone = await loadDisplayTimezone(adminClient, userId); - // Fresh holder per call - see the no-lease rationale in the file - // preamble. The claim+mark pair share this one holder; nothing else - // needs to recognise it. - const holderId = crypto.randomUUID(); - - // Claim atomically. p_user_id is the b-strict escape hatch: the - // service-role admin client has no auth.uid(), so the RPC scopes to - // the thread owner via coalesce(p_user_id, auth.uid()). - const { data: claimRows, error: claimErr } = await adminClient.rpc( - 'claim_next_thread_for_reflection', - { - p_holder_id: holderId, - p_ttl_seconds: REFLECTION_CLAIM_TTL_SECONDS, - p_timezone: timezone, - p_user_id: userId, - }, - ); - if (claimErr) { - throw new Error(`claim_next_thread_for_reflection failed: ${claimErr.message}`); - } - const claim = Array.isArray(claimRows) ? claimRows[0] : claimRows; - if (!claim || typeof claim.thread_id !== 'string') { - // Routine: the day-gated queue is empty on most turns. trace tier - // so it's available when actively watching but stays out of the - // default drawer view. - log.trace('no reflection-eligible thread to drain this turn'); - return { outcome: 'no-thread' }; - } - const threadId = claim.thread_id as string; - const terminalMsgId = claim.terminal_msg_id as string; - return await reflectClaimedThread( - adminClient, - userId, - log, - threadId, - terminalMsgId, - holderId, - ); - } catch (err) { - log.error( - 'reflection cycle failed', - err instanceof Error ? err : new Error(String(err)), - ); - return { outcome: 'error' }; - } finally { - // Flush before the waitUntil tail settles so the outcome line (the - // one worth seeing) isn't dropped as an un-awaited broadcast. - await log.flush(); - } -} - -/** - * The run half shared by both reflection drivers (the chat-turn tail - * and the cron catch-up sweep): the caller already holds the + * The run half of one drain cycle: the caller already holds the * per-thread claim; this reflects the thread and marks it done. - * Throws on infrastructure failure - each driver owns its own + * Throws on infrastructure failure - the cycle wrapper owns the * catch/log/flush posture. */ async function reflectClaimedThread( @@ -590,10 +502,9 @@ async function reflectClaimedThread( toolbox: buildReflectionToolbox(), baseCtx, apiKey, - // No outer turn to cancel - reflection runs in a background - // tail or a cron tick, after any user-visible work already - // shipped. A never-aborting signal lets runHeadlessAgent run - // to its own maxRounds backstop. + // No outer turn to cancel - reflection runs in a cron tick, + // never on a user-visible path. A never-aborting signal lets + // runHeadlessAgent run to its own maxRounds backstop. signal: new AbortController().signal, }, // parentDepth 0: reflection is a top-level agent (depth 1), same @@ -629,18 +540,94 @@ async function reflectClaimedThread( } /** - * One cron catch-up tick: claim the most-overdue reflection-eligible - * thread across ALL users and reflect on it. The chat-turn tail is - * reflection's primary driver but only fires when its owner - * converses; this sweep drains queues the tail can't reach. One - * thread per tick - the hourly schedule resumes the drain, matching - * the other fleets' pacing. Double-driving with the tail is safe: - * the per-thread claim columns are the mutual exclusion, so - * whichever driver claims first wins. Non-throwing, same contract - * as reflectOneThread. + * Per-tick thread cap for the sweep's drain loop. Bounds one tick's + * worst-case Venice spend; a backlog deeper than the cap drains + * across successive hourly ticks. Sequential on purpose - one + * reflection agent writing to the memory store at a time keeps + * near-simultaneous duplicate writes unlikely. + */ +const REFLECTION_SWEEP_MAX_THREADS = 5; + +/** + * Wall-clock cutoff for claiming ANOTHER thread within one tick. The + * hosted edge runtime kills an isolate at roughly 400s (see + * DEEP_SLEEP_BUDGET_MS in deep_sleep.ts for the fuller story), so a + * loop that kept claiming until empty would compound several long + * reflections into a guaranteed mid-flight death. Stopping new claims + * at 180s leaves a reflection claimed at the cutoff over 200s to + * finish - most do. One that doesn't dies exactly as it would have + * under a single-claim tick, and the attempt cap (3 claims per + * terminal message) keeps such a thread from wedging the queue. + */ +const REFLECTION_SWEEP_BUDGET_MS = 180_000; + +/** What one sweep tick did, and why its drain loop stopped. */ +export interface ReflectionSweepSummary { + reflected: number; + claimLost: number; + emptySlice: number; + stoppedBy: 'empty-queue' | 'cap' | 'budget' | 'error'; +} + +/** + * One cron tick: drain the reflection queue across ALL users, one + * thread at a time - claim the most-overdue eligible thread, reflect + * it, claim the next - until the queue is empty, the per-tick cap is + * hit, or the time budget is spent. The hourly schedule resumes the + * drain. This is reflection's ONLY driver (see the file preamble for + * why it does not ride the chat turn's tail). Non-throwing: cycle + * failures are folded into the summary, and an `error` cycle stops + * the loop rather than hot-looping a broken dependency. */ export async function runReflectionSweepTick( adminClient: SupabaseClient, +): Promise { + const startedAt = Date.now(); + const summary: ReflectionSweepSummary = { + reflected: 0, + claimLost: 0, + emptySlice: 0, + stoppedBy: 'cap', + }; + for (let i = 0; i < REFLECTION_SWEEP_MAX_THREADS; i++) { + // The first cycle always runs regardless of the budget - a tick + // must make at least one unit of progress, same guarantee the + // old single-claim tick gave. + if (i > 0 && Date.now() - startedAt > REFLECTION_SWEEP_BUDGET_MS) { + summary.stoppedBy = 'budget'; + return summary; + } + const cycle = await sweepClaimAndReflectOnce(adminClient); + switch (cycle.outcome) { + case 'no-thread': + summary.stoppedBy = 'empty-queue'; + return summary; + case 'error': + summary.stoppedBy = 'error'; + return summary; + case 'reflected': + summary.reflected += 1; + break; + case 'claim-lost': + // Another tick's claim outlived ours mid-run; the queue may + // still hold more work, so keep draining. + summary.claimLost += 1; + break; + case 'empty-slice': + summary.emptySlice += 1; + break; + } + } + return summary; +} + +/** + * One drain cycle: claim the most-overdue reflection-eligible thread + * across all users and reflect it. Non-throwing; every failure path + * folds into an `error` result for the loop to act on. + */ +async function sweepClaimAndReflectOnce( + adminClient: SupabaseClient, ): Promise { const holderId = crypto.randomUUID(); let claim: { thread_id?: unknown; terminal_msg_id?: unknown; user_id?: unknown } | null; diff --git a/supabase/functions/venice/agents/second_thoughts.ts b/supabase/functions/venice/agents/second_thoughts.ts index fb700bd8..6fa65a7b 100644 --- a/supabase/functions/venice/agents/second_thoughts.ts +++ b/supabase/functions/venice/agents/second_thoughts.ts @@ -2,7 +2,7 @@ // // Runs from the streaming function's completed-turn waitUntil tail // (getStreamingResponse.ts), a sibling to curateOnTurnTail / -// samskaraOnTurnTail / reflectOneThread. After a turn commits, it +// samskaraOnTurnTail. After a turn commits, it // re-reads what the model just said and reports a FELT CONFIDENCE - // stands behind it, or something feels off - onto the terminal // assistant row's `second_thoughts` jsonb column. diff --git a/supabase/functions/venice/getStreamingResponse.ts b/supabase/functions/venice/getStreamingResponse.ts index baab63e6..67a52650 100644 --- a/supabase/functions/venice/getStreamingResponse.ts +++ b/supabase/functions/venice/getStreamingResponse.ts @@ -75,7 +75,6 @@ import { stripGeneratedImage, type GeneratedImagePayload, } from './tools/_generated_image.ts'; -import { reflectOneThread } from './agents/reflection.ts'; import { curateOnTurnTail } from './agents/curation.ts'; import { samskaraOnTurnTail } from './agents/samskara.ts'; import { secondThoughtsOnTurnTail } from './agents/second_thoughts.ts'; @@ -1169,26 +1168,20 @@ export async function getStreamingResponse( `${runId} end terminalKind=${terminalKind} persistedId=${persistedId || 'none'}`, ); - // Reflection piggyback. A completed chat turn is the trigger that - // drains ONE day-gate-eligible OLDER thread from the reflection - // queue (not this thread - see agents/reflection.ts for why). The - // hourly /reflection-sweep cron route is the catch-up sibling for - // users who stopped conversing; the per-thread claim makes the two - // drivers safe together. Runs here in the already-detached waitUntil tail, - // after the response shipped and the channels tore down, so it never - // delays the user-visible turn. reflectOneThread is non-throwing and - // logs its own outcome (to the function log + the user's Logs - // drawer); the catch is a defensive backstop so a reflection bug - // still can't disturb this turn's already-committed row. + // Reflection deliberately does NOT run here. Memory formation is + // sweep-only (the hourly /reflection-sweep cron drain) so it keeps + // a fixed, predictable cadence: the user can edit or retry a + // conversation any time before the first hourly tick after the + // day-gate opens and reflection only ever sees the corrected + // thread. See the drive-shape preamble in agents/reflection.ts. if (terminalKind === 'completed') { // Second-thoughts reflex, FIRST in the tail: re-read the turn we // just committed and write a self-doubt verdict onto the terminal // assistant row. The browser hydrates it via the messages UPDATE // echo (subscribeToMessages listens for UPDATE), so the // per-message slide-down lands a beat after the reply settles. - // Ordered ahead of curation/samskara/reflection so the - // user-visible verdict isn't starved behind reflection, which can - // span minutes of tool rounds. Guarded on persistedId - a turn + // Ordered ahead of curation/samskara so the user-visible verdict + // ships before the housekeeping work. Guarded on persistedId - a turn // that committed no assistant row (should not happen on the // 'completed' path, but cheap to check) has nothing to review. if (persistedId) { @@ -1204,34 +1197,26 @@ export async function getStreamingResponse( log.error(`${runId} second-thoughts tail failed:`, err); } } - // Curation piggyback, BEFORE reflection on purpose: the chain is - // sequential and reflection can span minutes of tool rounds, - // while curation is a handful of quick completions whose first + // Curation piggyback: a handful of quick completions whose first // unit (auto-title) is the user-visible one - a brand-new // conversation sits on the 'New conversation' placeholder until - // it runs. Same non-throwing contract and hourly catch-up - // sibling (/curation-sweep) as reflection below. + // it runs. Non-throwing, with an hourly catch-up sibling + // (/curation-sweep) for users who stop conversing. try { await curateOnTurnTail(opts.adminClient, opts.userId); } catch (err) { log.error(`${runId} curation tail failed:`, err); } - // Samskara before reflection: reflection can span minutes of - // tool rounds, and the samskara rotation carries the fleet's - // only hard timing window (reaction-classify must catch a fired - // cohort 1-10 minutes after the fire - this turn's user message - // is what resolves the PREVIOUS turn's cohort). The hourly + // Samskara last: the rotation carries the fleet's only hard + // timing window (reaction-classify must catch a fired cohort + // 1-10 minutes after the fire - this turn's user message is what + // resolves the PREVIOUS turn's cohort). The hourly // /samskara-sweep cron is the catch-up sibling. try { await samskaraOnTurnTail(opts.adminClient, opts.userId); } catch (err) { log.error(`${runId} samskara tail failed:`, err); } - try { - await reflectOneThread(opts.adminClient, opts.userId); - } catch (err) { - log.error(`${runId} reflection tail failed:`, err); - } } // Drain pending drawer broadcasts before the waitUntil promise diff --git a/supabase/functions/venice/index.ts b/supabase/functions/venice/index.ts index 9de59cbb..ecccf7c6 100644 --- a/supabase/functions/venice/index.ts +++ b/supabase/functions/venice/index.ts @@ -861,16 +861,15 @@ const handleWikiRecordsSweep = sweepHandler(runWikiRecordsSweepTick); const handleWikiLibrarianSweep = sweepHandler(runWikiLibrarianSweepTick); const handleRemSweep = sweepHandler(runRemSweepTick); const handleDeepSleepSweep = sweepHandler(runDeepSleepSweepTick); -// Reflection's catch-up drain. The primary driver stays the chat -// turn's waitUntil tail in getStreamingResponse; this route exists so -// a user who stops conversing still gets their queue drained, and so -// reflection's trigger surface is visible in this routing table like -// every other fleet's. +// Reflection's ONLY driver: an hourly drain loop that claims and +// reflects eligible threads one at a time until the queue is empty or +// the tick's cap/budget stops it. Deliberately not tail-driven - see +// the drive-shape preamble in agents/reflection.ts. const handleReflectionSweep = sweepHandler(runReflectionSweepTick); // Curation catch-up drain (auto-title, thread topics, summaries, // memory topics, recipe topics). The primary driver is the chat -// turn's waitUntil tail in getStreamingResponse, same dual-driver -// shape as reflection; this route is what drains work created +// turn's waitUntil tail in getStreamingResponse; this route is +// what drains work created // server-side (rem / deep-sleep consolidations re-queue memory tags) // or left behind by a failed tail attempt. const handleCurationSweep = sweepHandler(runCurationSweepTick); diff --git a/supabase/schema.sql b/supabase/schema.sql index b1228308..ba62d9b6 100644 --- a/supabase/schema.sql +++ b/supabase/schema.sql @@ -992,8 +992,8 @@ alter table public.threads add column if not exists last_reflected_msg_id uuid references public.messages(id) on delete set null, add column if not exists reflection_holder_id text, add column if not exists reflection_claim_expires_at timestamptz, - -- Attempt accounting, stamped AT CLAIM TIME by both reflection - -- claims. Counting attempts (not failures) is deliberate: a run + -- Attempt accounting, stamped AT CLAIM TIME by the sweep claim. + -- Counting attempts (not failures) is deliberate: a run -- that dies to the invocation wall clock never reaches an error -- handler, so a failure counter would miss exactly the deaths that -- need bounding. Three attempts at the same terminal message and @@ -4541,8 +4541,8 @@ grant execute on function public.nak_sweep_stale_streams() to service_role; -- Reflection pipeline RPCs ----------------------------------------------- -- --- The reflection agent's worker runs on the same claim/lease pattern as --- the embeddings worker, but against `threads` instead of `memories` +-- The reflection agent runs on the same claim/lease pattern as the +-- embeddings worker, but against `threads` instead of `memories` -- and with a different "what does 'needs work' mean?" predicate. -- -- "Needs reflection" = there exists a terminal assistant message in the @@ -4554,121 +4554,25 @@ grant execute on function public.nak_sweep_stale_streams() to service_role; -- user prompt + assistant reply, no follow-up) doesn't burn Venice -- calls reflecting on a conversation that hadn't actually started yet. -- --- The function returns `(thread_id, terminal_msg_id)` atomically. The --- worker fetches messages up to `terminal_msg_id` (so a race where the +-- The claim returns `(thread_id, terminal_msg_id)` atomically. The +-- agent fetches messages up to `terminal_msg_id` (so a race where the -- user adds more turns mid-reflection just queues the thread for the -- next cycle), runs its tool-call loop, and calls -- `mark_thread_reflected_if_claimed` with the same msg_id it got here. --- If the claim was lost (device B took over mid-reflection) the mark --- returns false and the whole run is discarded — device B will redo it. - --- Claim the oldest thread in need of reflection and return its id plus --- the terminal assistant message we should reflect up to. `for update --- skip locked` is belt-and-suspenders under the lease invariant (only --- one device should be claiming at a time); it costs nothing and --- removes an entire class of wrong answer from the corner where two --- devices briefly both think they hold the lease. +-- If the claim was lost (another run took over mid-reflection) the mark +-- returns false and the whole run is discarded — the winner redoes it. + +-- The per-user claim variant (claim_next_thread_for_reflection) is +-- retired: reflection is sweep-only, and the cross-user sweep claim +-- below is the single claiming path. The drops cover every signature +-- the variant ever shipped with. drop function if exists public.claim_next_thread_for_reflection(text, int); drop function if exists public.claim_next_thread_for_reflection(text, int, text); -create or replace function public.claim_next_thread_for_reflection( - p_holder_id text, - p_ttl_seconds int, - -- User's display timezone from Settings -> AI -> About you; - -- determines the calendar day the eligibility gate buckets on. - -- Same shape as claim_next_thread_for_wiki - we want the - -- reflection pass to leave in-flight conversations alone so a - -- memory derived from a half-finished thought doesn't land - -- before the user has a chance to correct or extend it. The - -- memory_recall tool has no per-conversation source attribution - -- on memories, so a same-day write could ride straight back into - -- the conversation that produced it. - p_timezone text default 'UTC', - -- b-strict escape hatch (see search_memories_by_embedding): the - -- browser supervisor calls with auth.uid() in scope and leaves this - -- null; the venice edge function fires reflection from a chat turn's - -- waitUntil tail with a service-role client that has no uid, so it - -- passes the thread owner's id explicitly. security invoker stays - -- correct because service_role bypasses RLS and the coalesce scopes - -- the claim to one user either way. - p_user_id uuid default null -) returns table (thread_id uuid, terminal_msg_id uuid) -language sql security invoker as $$ - with candidate as ( - -- Oldest thread (by updated_at ascending) that has a terminal - -- assistant message newer than what we've reflected on, passes the - -- token-volume guard, lands on a calendar day strictly before - -- today in the user's timezone, and isn't currently claimed. The - -- terminal-message lookup is a lateral join so we get both the - -- thread row AND the specific msg id to mark up to, in one - -- round trip. The newest-message lookup is a second lateral so - -- the day-gate buckets on messages.created_at - same source - -- the wiki claim uses, stable against unrelated bumps to - -- threads.updated_at. - select t.id as thread_id, term.msg_id as terminal_msg_id - from public.threads t - cross join lateral ( - select m.id as msg_id - from public.messages m - where m.thread_id = t.id - and m.role = 'assistant' - and (m.tool_calls is null - or jsonb_typeof(m.tool_calls) <> 'array' - or jsonb_array_length(m.tool_calls) = 0) - and m.content is not null - and length(m.content) > 0 - order by m.created_at desc - limit 1 - ) term - cross join lateral ( - select m2.created_at - from public.messages m2 - where m2.thread_id = t.id - order by m2.created_at desc - limit 1 - ) newest - where t.user_id = coalesce(p_user_id, auth.uid()) - and term.msg_id is distinct from t.last_reflected_msg_id - -- Attempt cap: stop offering a terminal message that has - -- already burned three claims (see the column comment on - -- reflection_attempt_count). A different terminal message - -- means new conversation turns landed - fresh budget. - and (term.msg_id is distinct from t.reflection_attempt_msg_id - or t.reflection_attempt_count < 3) - and (t.reflection_claim_expires_at is null - or t.reflection_claim_expires_at < now()) - and (newest.created_at at time zone p_timezone)::date - < (now() at time zone p_timezone)::date - and ( - -- At least two user messages on the thread. A single user - -- prompt + assistant reply is a one-shot Q&A; we only want - -- to reflect once the user came back with a follow-up, which - -- is the cheapest signal that the conversation has substance - -- worth turning into memories. - select count(*) - from public.messages m2 - where m2.thread_id = t.id - and m2.role = 'user' - ) >= 2 - order by t.updated_at asc - limit 1 - for update of t skip locked - ) - update public.threads t - set reflection_holder_id = p_holder_id, - reflection_claim_expires_at = now() + make_interval(secs => p_ttl_seconds), - reflection_attempt_count = case - when t.reflection_attempt_msg_id is distinct from c.terminal_msg_id then 1 - else t.reflection_attempt_count + 1 - end, - reflection_attempt_msg_id = c.terminal_msg_id - from candidate c - where t.id = c.thread_id - returning t.id as thread_id, c.terminal_msg_id; -$$; +drop function if exists public.claim_next_thread_for_reflection(text, int, text, uuid); -- Record a completed reflection IF the claim is still ours. Returns -- true on success, false when the claim expired or was stolen (another --- device took over). The worker treats false the same way +-- run took over). The agent treats false the same way -- save_memory_embedding_if_claimed does: drop the work, loop to the -- next row. Any memory writes the agent already made during the run -- stay — they're owned by the user, not the claim, and re-reflection @@ -4679,9 +4583,9 @@ create or replace function public.mark_thread_reflected_if_claimed( p_thread_id uuid, p_holder_id text, p_msg_id uuid, - -- b-strict escape hatch, same as the claim RPC above: null from the - -- browser (auth.uid() in scope), the thread owner's id from the - -- service-role edge-function caller. + -- b-strict escape hatch: the service-role edge-function caller has + -- no auth.uid(), so it passes the thread owner's id explicitly + -- (coalesced below). p_user_id uuid default null ) returns boolean language plpgsql security invoker as $$ @@ -4701,12 +4605,9 @@ begin return updated > 0; end $$; --- service_role grants for the edge-function reflection driver (the --- venice function fires reflection from a chat turn's waitUntil tail). --- The browser keeps calling these as the authenticated user; these --- grants just let the service-role client reach them too. -grant execute on function - public.claim_next_thread_for_reflection(text, int, text, uuid) to service_role; +-- service_role grant for the edge-function reflection driver (the +-- venice function's sweep tick marks threads with a service-role +-- client, passing the owner's id through the b-strict escape hatch). grant execute on function public.mark_thread_reflected_if_claimed(uuid, text, uuid, uuid) to service_role; @@ -4736,15 +4637,14 @@ exception when others then return 'UTC'; end $$; --- Global reflection sweep claim: the cron catch-up drain's variant of --- claim_next_thread_for_reflection. Same candidate predicate, but --- across ALL users - the timezone comes off each owner's profile --- (nak_safe_timezone, UTC fallback) instead of a parameter. The --- per-turn waitUntil tail only drains when its owner converses, so --- without this sweep a dormant account's reflection queue never --- moves. Tail + sweep double-driving is safe by construction: the --- per-thread claim columns are the mutual exclusion, so whichever --- driver claims first wins and the other sees no candidate. +-- Global reflection sweep claim: reflection's only claiming path. +-- Picks the most-overdue eligible thread across ALL users - the +-- timezone comes off each owner's profile (nak_safe_timezone, UTC +-- fallback). The sweep tick calls this in a drain loop (claim -> +-- reflect -> claim the next), and overlapping ticks are safe by +-- construction: the per-thread claim columns are the mutual +-- exclusion, so whichever caller claims first wins and the other +-- sees a different candidate or none. drop function if exists public.claim_next_thread_for_reflection_sweep(text, int); create or replace function public.claim_next_thread_for_reflection_sweep( p_holder_id text, @@ -13063,17 +12963,17 @@ end $cron$; -- --------------------------------------------------------------------------- --- Scheduled reflection catch-up sweep (pg_cron -> pg_net -> venice function) --- --- Reflection's primary driver is the chat turn's waitUntil tail in --- getStreamingResponse - one eligible older thread drains per --- completed turn. This hourly sweep is the catch-up path for queues --- the tail can't reach: a user who stops conversing leaves eligible --- threads stranded (no turns -> no draining). One thread per tick, --- claimed across all users by claim_next_thread_for_reflection_sweep; --- the per-thread claim columns make tail + sweep double-driving safe. --- Same Vault-secret custody and no-op-until-seeded behavior as the --- other sweep triggers above. +-- Scheduled reflection sweep (pg_cron -> pg_net -> venice function) +-- +-- Reflection's ONLY driver. Each hourly tick drains the day-gated +-- queue one thread at a time (claimed across all users by +-- claim_next_thread_for_reflection_sweep) until it runs empty or the +-- tick's cap/time budget stops it; the next tick resumes. Memory +-- formation deliberately does not ride the chat turn's tail - the +-- fixed cadence gives the user a predictable window to edit or retry +-- a settled conversation before reflection reads it. Same +-- Vault-secret custody and no-op-until-seeded behavior as the other +-- sweep triggers above. -- --------------------------------------------------------------------------- create or replace function public.nak_trigger_reflection_sweep()