Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
7.1 kB · 205 lines
Rust
at dev
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206//! The medium loop (docs/04): seconds behind the write path, never blocking it.//! Drains the durable gloss queue: interpret each new completion with the full//! ensemble, canonicalize, publish gloss records, rebuild the user's term//! graph, update their posterior.
use std::sync::Arc;use std::time::Duration;
use smol_str::SmolStr;
use crate::atproto::records::{Gloss, Interpreter, Reading};use crate::graph;use crate::infer;use crate::state::AppState;use jacquard::api::com_atproto::repo::strong_ref::StrongRef;use jacquard::types::string::{AtUri, Cid, Datetime};
use super::canon;use super::{GlossInput, PROMPT_VERSION};
pub async fn run(state: Arc<AppState>) { loop { match drain(&state).await { Ok(0) => { // Idle: wake on notify or poll (Jetstream rows arrive without notify). tokio::select! { _ = state.gloss_notify.notified() => {}, _ = tokio::time::sleep(Duration::from_secs(30)) => {}, } } Ok(_) => {} // keep draining Err(e) => { tracing::error!("gloss worker error: {e:#}"); tokio::time::sleep(Duration::from_secs(5)).await; } } }}
async fn drain(state: &AppState) -> anyhow::Result<usize> { let batch: Vec<(i64, String, i64)> = sqlx::query_as( "SELECT id, completion_uri, attempts FROM gloss_queue WHERE attempts < 5 ORDER BY id LIMIT 8", ) .fetch_all(&state.pool) .await?;
let mut processed = 0; for (qid, completion_uri, _attempts) in batch { match gloss_one(state, &completion_uri).await { Ok(()) => { sqlx::query("DELETE FROM gloss_queue WHERE id = ?") .bind(qid) .execute(&state.pool) .await?; processed += 1; } Err(e) => { tracing::warn!(%completion_uri, "gloss failed: {e:#}"); sqlx::query("UPDATE gloss_queue SET attempts = attempts + 1 WHERE id = ?") .bind(qid) .execute(&state.pool) .await?; } } } Ok(processed)}
async fn gloss_one(state: &AppState, completion_uri: &str) -> anyhow::Result<()> { let row: Option<(String, Option<String>, String, String, String, Option<String>)> = sqlx::query_as( "SELECT c.did, c.text, c.mode, c.slot, c.square_uri, c.cid FROM completions c WHERE c.uri = ?", ) .bind(completion_uri) .fetch_optional(&state.pool) .await?; let Some((did, text, mode, _slot, square_uri, completion_cid)) = row else { // Completion deleted before glossing: nothing to interpret. return Ok(()); };
let square: Option<(Option<String>, Option<String>, Option<String>, Option<String>, String)> = sqlx::query_as("SELECT a, b, c, d, open_slot FROM squares WHERE uri = ?") .bind(&square_uri) .fetch_optional(&state.pool) .await?; let Some((a, b, c, d, open_slot)) = square else { anyhow::bail!("square {square_uri} not indexed yet"); };
let input = GlossInput { a, b, c, d, open_slot, text: text.clone(), mode: mode.clone(), };
// Skip if this ensemble already glossed this completion (idempotent re-runs). let (existing,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM glosses WHERE completion_uri = ?") .bind(completion_uri) .fetch_one(&state.pool) .await?; if existing >= state.ensemble.len() as i64 { return Ok(()); }
let completion_ref = StrongRef { uri: AtUri::new(SmolStr::new(completion_uri)).map_err(|e| anyhow::anyhow!("{e}"))?, cid: Cid::new_owned( completion_cid .as_deref() .unwrap_or("bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") .as_bytes(), ) .map_err(|e| anyhow::anyhow!("{e}"))?, extra_data: None, };
for member in &state.ensemble { let member_id = member.id(); let (already,): (i64,) = sqlx::query_as( "SELECT COUNT(*) FROM glosses WHERE completion_uri = ? AND ensemble_member = ?", ) .bind(completion_uri) .bind(&member_id) .fetch_one(&state.pool) .await?; if already > 0 { continue; }
let output = member.gloss(&input).await?;
// Canonicalize each reading into the growing relation space. let mut readings: Vec<Reading> = Vec::new(); let mut readings_json: Vec<serde_json::Value> = Vec::new(); for r in &output.readings { let canonical = canon::canonicalize(&state.pool, &state.embedder, &r.relation).await?; readings.push(Reading { relation: r.relation.clone(), canonical_relation: Some(canonical), weight: SmolStr::new(format!("{:.3}", r.weight)), valence: Some(SmolStr::new(format!("{:.3}", r.valence))), }); readings_json.push(serde_json::json!({ "relation": r.relation, "canonicalRelation": canonical, "weight": r.weight, "valence": r.valence, })); }
let record = Gloss { completion: completion_ref.clone(), readings, interpreter: Interpreter { model: SmolStr::new(&output.model), prompt_version: SmolStr::new(PROMPT_VERSION), ensemble_member: SmolStr::new(&member_id), temperature: output.temperature.map(|t| SmolStr::new(format!("{t}"))), }, created_at: Datetime::now(), }; let published = state.publisher.publish_gloss(&record).await?;
sqlx::query( "INSERT INTO glosses (uri, completion_uri, did, readings, interpreter, prompt_version, ensemble_member, created_at) VALUES (?,?,?,?,?,?,?,?) ON CONFLICT (uri) DO NOTHING", ) .bind(&published.uri) .bind(completion_uri) .bind(&did) .bind(serde_json::to_string(&readings_json)?) .bind(&output.model) .bind(PROMPT_VERSION) .bind(&member_id) .bind(record.created_at.as_str()) .execute(&state.pool) .await?; }
// Medium-loop conclusions: term graph and posterior for this user. graph::rebuild_user_graph(&state.pool, &did).await?; infer::update_posterior(&state.pool, &did).await?;
Ok(())}
pub async fn enqueue(pool: &sqlx::SqlitePool, completion_uri: &str) -> anyhow::Result<()> { sqlx::query( "INSERT INTO gloss_queue (completion_uri, enqueued_at) VALUES (?, ?) ON CONFLICT (completion_uri) DO NOTHING", ) .bind(completion_uri) .bind(crate::db::now()) .execute(pool) .await?; Ok(())}