#![allow(dead_code)]
use anyhow::Result;
use chrono::Local;
use serde_json::json;
use std::collections::{HashMap, VecDeque};
use std::path::Path;
use std::sync::mpsc::{Receiver, RecvTimeoutError};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use crate::worker::claude::ClaudeCli;
use crate::worker::config::WorkerConfig;
use crate::worker::db::{Db, Event};
use crate::worker::mem0::Mem0Client;
use crate::worker::prompt_constants::BUILTIN_RULES;
use crate::worker::supabase::SupabaseClient;
pub struct BurstReady {
pub project: String,
pub paths: Vec<String>,
}
#[derive(Debug, Default)]
pub struct ProjectBurst {
pub paths: Vec<String>,
pub last_event: Option<Instant>,
pub started: Option<Instant>,
}
impl ProjectBurst {
pub fn record(&mut self, path: String) {
if self.started.is_none() {
self.started = Some(Instant::now());
}
self.last_event = Some(Instant::now());
if !self.paths.contains(&path) {
self.paths.push(path);
}
}
pub fn is_idle(&self, idle: Duration) -> bool {
self.last_event
.map(|t| t.elapsed() >= idle)
.unwrap_or(false)
}
pub fn drain(&mut self) -> Vec<String> {
self.started = None;
self.last_event = None;
std::mem::take(&mut self.paths)
}
}
pub struct HourlyLimiter {
cap: u32,
hits: Mutex<VecDeque<Instant>>,
}
impl HourlyLimiter {
pub fn new(cap: u32) -> Self {
Self {
cap,
hits: Mutex::new(VecDeque::new()),
}
}
pub fn try_consume(&self) -> bool {
let mut hits = self.hits.lock().unwrap();
let now = Instant::now();
while let Some(&front) = hits.front() {
if now.duration_since(front) > Duration::from_secs(3600) {
hits.pop_front();
} else {
break;
}
}
if (hits.len() as u32) >= self.cap {
return false;
}
hits.push_back(now);
true
}
pub fn current(&self) -> u32 {
let hits = self.hits.lock().unwrap();
hits.len() as u32
}
}
pub struct AdviceWorker {
cfg: WorkerConfig,
db: Db,
claude: ClaudeCli,
mem0: Option<Mem0Client>,
supabase: Option<Arc<SupabaseClient>>,
advice_limit: HashMap<String, Arc<HourlyLimiter>>,
mem0_limit: Arc<HourlyLimiter>,
}
impl AdviceWorker {
pub fn new(cfg: WorkerConfig) -> Result<Self> {
let db = Db::open(&cfg.db_path)?;
let claude = ClaudeCli::new(&cfg.claude_bin);
let mem0 = match (&cfg.mem0_api_key, &cfg.mem0_user_id) {
(Some(k), Some(u)) if !k.is_empty() && !u.is_empty() => {
Some(Mem0Client::new(k.clone(), u.clone())?)
}
_ => None,
};
let supabase = make_heartbeat_client(&cfg).map(Arc::new);
let mem0_limit = Arc::new(HourlyLimiter::new(cfg.mem0_max_writes_per_hour));
Ok(Self {
cfg,
db,
claude,
mem0,
supabase,
advice_limit: HashMap::new(),
mem0_limit,
})
}
fn load_reso_constraints(&self, project: &str) -> Vec<crate::worker::supabase::MemoryRow> {
let Some(sb) = &self.supabase else {
return Vec::new();
};
let mut out = Vec::new();
match sb.list_memories(&["constraint", "strong"], Some("user"), None, &[]) {
Ok(rows) => out.extend(rows),
Err(e) => log_line(&format!("[reso-load-err user] {}", e)),
}
match sb.list_memories(
&["constraint", "strong"],
Some("project"),
Some(project),
&[],
) {
Ok(rows) => out.extend(rows),
Err(e) => log_line(&format!("[reso-load-err project] {}", e)),
}
out.truncate(50);
out
}
fn resolve_locale(&self) -> String {
if let Some(sb) = &self.supabase {
if let Some(user_lang) = sb.get_user_locale() {
return user_lang;
}
}
self.cfg.advice_locale.clone()
}
pub fn run(mut self, rx: Receiver<BurstReady>) {
log_line("[advice] thread up");
let heartbeat = self.supabase.clone();
let mut last_beat = Instant::now() - Duration::from_secs(60);
loop {
if last_beat.elapsed() >= Duration::from_secs(10) {
if let Some(c) = heartbeat.as_ref() {
let _ = c.heartbeat("advice");
}
last_beat = Instant::now();
}
let msg = match rx.recv_timeout(Duration::from_secs(5)) {
Ok(m) => m,
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => break,
};
if !self.cfg.advice_enabled {
log_line("[advice] disabled — skipping");
continue;
}
if msg.paths.len() < self.cfg.advice_min_batch {
continue;
}
let limiter = self
.advice_limit
.entry(msg.project.clone())
.or_insert_with(|| Arc::new(HourlyLimiter::new(self.cfg.advice_max_per_hour)))
.clone();
if !limiter.try_consume() {
log_line(&format!(
"[advice] rate-limited for project {} ({}/h)",
msg.project, self.cfg.advice_max_per_hour
));
continue;
}
if let Err(e) = self.process_burst(&msg) {
log_line(&format!("[advice-err] {}: {}", msg.project, e));
}
}
}
fn process_burst(&self, msg: &BurstReady) -> Result<()> {
let project = &msg.project;
let paths = &msg.paths;
log_line(&format!(
"[advice] processing {} ({} paths)",
project,
paths.len()
));
let mem0_query = build_mem0_query(project, paths);
let prior = match &self.mem0 {
Some(m) => match m.search(&mem0_query, 5) {
Ok(v) => v,
Err(e) => {
log_line(&format!("[mem0-search-err] {}", e));
Vec::new()
}
},
None => Vec::new(),
};
let monitor = self.cfg.monitor_dir.clone();
let snippets = collect_snippets(&monitor, project, paths, 8, 1500);
let reso_constraints = self.load_reso_constraints(project);
let project_dir = self.cfg.monitor_dir.join(project);
let tech_tags = crate::worker::domain::detect_tech(&project_dir);
let tech_cards = crate::worker::domain::cards_for(&tech_tags);
let mut combined_lines = Vec::new();
for c in &tech_cards {
combined_lines.push(format!("- ({} · {}) {}", c.priority, c.tag, c.text));
}
for m in &reso_constraints {
combined_lines.push(format!("- ({}) {}", m.priority, m.text));
}
let constraints_text: String = if combined_lines.is_empty() {
String::new()
} else {
format!(
"Memories (always apply — devist-curated tech wisdom + your past decisions):\n{}\n\n---\n\n",
combined_lines.join("\n")
)
};
let prior_text: String = prior
.iter()
.map(|m| format!("- {}", m.memory))
.collect::<Vec<_>>()
.join("\n");
let locale = self.resolve_locale();
log_line(&format!(
"[advice] locale={} reso_constraints={} tech_cards={} ({})",
locale,
reso_constraints.len(),
tech_cards.len(),
tech_tags.join(",")
));
let combined_rules = format!("{}\n\n{}", BUILTIN_RULES.trim_end(), constraints_text);
let prompt = build_prompt(project, &snippets, &prior_text, &combined_rules, &locale);
let result = self
.claude
.ask_json(&prompt, Duration::from_secs(60))
.map_err(|e| {
let _ = self.db.insert(&Event {
id: None,
project: project.clone(),
event_type: "advice_error".into(),
path: None,
payload: json!({"error": e.to_string()}).to_string(),
severity: "warn".into(),
created_at: Local::now().to_rfc3339(),
synced_at: None,
confirmed_at: None,
});
e
})?;
if let Some(advice) = result.get("advice").and_then(|v| v.as_array()) {
for item in advice {
let text = item
.get("text")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let severity = item
.get("severity")
.and_then(|v| v.as_str())
.unwrap_or("info")
.to_string();
let verifiable = item
.get("verifiable")
.and_then(|v| v.as_bool())
.unwrap_or(false);
let paths: Vec<String> = item
.get("paths")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|x| x.as_str().map(String::from))
.collect()
})
.unwrap_or_default();
if text.is_empty() {
continue;
}
let severity_for_mem0 = severity.clone();
let _ = self.db.insert(&Event {
id: None,
project: project.clone(),
event_type: "advice".into(),
path: None,
payload: json!({
"text": text.clone(),
"verifiable": verifiable,
"paths": paths,
})
.to_string(),
severity,
created_at: Local::now().to_rfc3339(),
synced_at: None,
confirmed_at: None,
});
if severity_for_mem0 == "info" {
self.persist_memory(project, &text, "project", "info");
}
}
}
if let Some(facts) = result.get("facts").and_then(|v| v.as_array()) {
for fact in facts {
let text = fact.get("text").and_then(|v| v.as_str()).unwrap_or("");
let conf = fact
.get("confidence")
.and_then(|v| v.as_f64())
.unwrap_or(0.0) as f32;
if text.is_empty() {
continue;
}
if conf < self.cfg.mem0_confidence_threshold {
continue;
}
self.persist_memory(project, text, "project", "preference");
}
}
Ok(())
}
fn persist_memory(&self, project: &str, text: &str, scope: &str, priority: &str) {
let memory_id = match &self.supabase {
Some(sb) => match sb.insert_memory(text, scope, priority, "claude", Some(project), &[])
{
Ok(id) => Some(id),
Err(e) => {
log_line(&format!("[memory-insert-err] {}", e));
None
}
},
None => None,
};
let Some(client) = &self.mem0 else { return };
if !self.mem0_limit.try_consume() {
log_line(&format!(
"[mem0] write cap hit ({}/h) — skipping mem0 add (Supabase row #{:?} kept)",
self.cfg.mem0_max_writes_per_hour, memory_id
));
return;
}
let meta = json!({
"project": project,
"scope": scope,
"priority": priority,
"supabase_id": memory_id,
"source": "devist-worker",
});
match client.add(text, Some(meta)) {
Ok(resp) => {
if let (Some(mem0_id), Some(mid), Some(sb)) =
(extract_mem0_id(&resp), memory_id, &self.supabase)
{
if let Err(e) = sb.set_memory_mem0_id(mid, &mem0_id) {
log_line(&format!("[memory-link-err] {}", e));
}
}
}
Err(e) => log_line(&format!("[mem0-add-err {}] {}", priority, e)),
}
}
}
fn extract_mem0_id(resp: &serde_json::Value) -> Option<String> {
if let Some(id) = resp.get("id").and_then(|v| v.as_str()) {
return Some(id.to_string());
}
if let Some(arr) = resp.get("results").and_then(|v| v.as_array()) {
if let Some(first) = arr.first() {
if let Some(id) = first.get("id").and_then(|v| v.as_str()) {
return Some(id.to_string());
}
}
}
if let Some(arr) = resp.as_array() {
if let Some(first) = arr.first() {
if let Some(id) = first.get("id").and_then(|v| v.as_str()) {
return Some(id.to_string());
}
}
}
None
}
fn build_prompt(project: &str, snippets: &str, prior: &str, rules: &str, locale: &str) -> String {
let rules_section = if rules.is_empty() {
String::new()
} else {
format!("Rules and memories (follow strictly):\n\n{rules}\n\n---\n\n")
};
let locale_section = format!(
"All natural-language strings in your output (every \"text\" field, \
every \"explain\" field) MUST be written in language code `{locale}`. \
Field names and JSON structure stay English.\n\n",
);
format!(
r#"{rules_section}{locale_section}You are a code observer for the project "{project}".
The user just made these changes (paths + first lines shown):
{snippets}
Existing long-term memory about this user/project:
{prior}
Output STRICT JSON (no markdown fences, no commentary). Schema:
{{
"facts": [
{{"text": "<...>", "confidence": <0..1>}}
],
"advice": [
{{"text": "<actionable suggestion>", "severity": "info|suggest|warn|block", "verifiable": true|false, "paths": ["<file paths the advice applies to>"]}}
]
}}
============================================================
FACTS — extraction rules (read carefully, then apply strictly)
============================================================
A fact is something a future maintainer or AI assistant would WANT TO
KNOW but CANNOT learn by simply reading the code. The bar is high.
When in doubt, output an empty `facts: []` array. Most bursts produce
zero facts — that is the expected outcome.
MUST extract (these are the only valid fact categories):
1. **Decision rationale** — WHY a non-obvious choice was made.
✅ "rustls-tls was chosen to avoid OpenSSL system dependency"
✅ "develop branch holds detailed history; main only has squashed release commits per user workflow"
2. **Hard constraints / invariants** — rules that future code MUST follow.
✅ "secrets must never be written to local SQLite (defense in depth even after path filter)"
✅ "release is only triggered on user's explicit '배포해줘' — never auto-tagged"
3. **Cross-file invariants & contracts** — relationships between modules a single file doesn't reveal.
✅ "worker_jobs.kind='apply_advice' is consumed by jobs.rs and writes confirmed_at via verify thread"
4. **Past incidents / scars** — bugs that motivated current shape.
✅ "channel name collision crashed History tab in v0.17 — channel suffix added to fix"
MUST NOT extract (zero exceptions — these have NEVER been useful):
❌ Restating dependency names or versions visible in Cargo.toml / package.json
("rusqlite 0.39 is used", "release profile uses strip+lto")
❌ Restating file paths, module structure, or public API surface visible in code
("daemon.rs has run_loop", "paths.rs centralizes ~/.devist locations")
❌ Restating CI / workflow config visible in .github/workflows
("CI runs on PR", "publish-dry-run was removed")
❌ Restating documentation that already exists in CLAUDE.md / README.md / module doc comments
❌ Snapshots of "current architecture" or "this version uses X" — these go stale on the next release
❌ Statements of the form "this project uses ~" / "X is implemented with Y" — observations, not facts
CONFIDENCE — be conservative:
- 0.9–1.0: explicit user intent stated in commit message, doc comment, or code comment
- 0.85–0.9: strong cross-file evidence of an invariant
- < 0.85: do not output
============================================================
ADVICE — extraction rules (read carefully, then apply strictly)
============================================================
A piece of advice is worth raising ONLY IF acting on it would meaningfully
improve the code. The user already sees a long list of changes they made;
they don't need them narrated back. Think: "would a thoughtful senior
reviewer bother to flag this in a PR comment?" If no, output nothing for
that observation.
When `prior` (existing memory above) shows a similar concern was already
surfaced or already covered by a 'strong'/'constraint' user memory,
DO NOT re-raise it. Trust the memory.
MUST raise:
1. **Concrete bugs / risks** with a specific file location.
✅ "src/worker/daemon.rs:217 — main loop blocks indefinitely on rx.recv() with no timeout; advice thread shutdown is wedged"
2. **Code-doc / code-code drift** with both sides cited.
✅ "verify.rs L12 doc comment says 60s tick but TICK_INTERVAL_SECS=20"
3. **Cross-cutting invariants violated** — security, RLS, secrets, etc.
✅ "auth.users.user_metadata.lang is read in advice.rs but no fallback when the column is missing"
4. **Concrete refactor opportunity** with a clear win (not "consider refactoring").
MUST NOT raise (zero exceptions — these have been the dominant noise):
❌ Anything about temp / swap / build artifacts (*.tmp.*, *~, 4913,
*.timestamp-*.mjs, tsbuildinfo, SMOKE_*.md). The watcher already
filters most of these; if one slips through, ignore it — the next
release tightens the filter.
❌ ".gitignore should probably include X" — these are noise unless X
is a secret-bearing pattern.
❌ "Consider adding tests" — the user has decided when tests are needed.
Only flag testing if a concrete invariant is going untested.
❌ "Documentation might be stale" — only raise drift with both sides cited
(see MUST #2).
❌ Restating what the change did ("you renamed X to Y; consider…") —
the user can read their own diff.
❌ Anything already addressed by a prior memory (see `prior` block above
and the user/project 'strong'/'constraint' rules at the top).
SEVERITY (be conservative — most advice is `info` or `suggest`):
- `block` : action would break prod / leak secrets / corrupt data.
Almost never used.
- `warn` : real bug / drift / security issue with a specific location.
- `suggest`: meaningful improvement opportunity.
- `info` : observation worth recording but not requesting action; goes
to the History tab, not the actionable inbox.
`verifiable`:
- true if the issue is OBJECTIVELY checkable in current file content (e.g., "missing input validation in src/auth.tsx", "secret literal in .env.example", "TODO left at line N")
- false if the advice is SUBJECTIVE / opinion (e.g., "consider refactoring", "naming could be clearer", "add more tests")
`paths`: list of repo-relative file paths the advice references. Empty array if not applicable.
If you have nothing valuable to say, output `"advice": []`. An empty
inbox is a healthy outcome — most bursts should add at most 1–2 items.
"#
)
}
fn collect_snippets(
monitor: &Path,
project: &str,
rel_paths: &[String],
max_files: usize,
max_bytes_per_file: usize,
) -> String {
let mut out = String::new();
for rel in rel_paths.iter().take(max_files) {
let abs = monitor.join(rel);
if crate::worker::secrets::is_secret_path(&abs) {
continue;
}
let display = rel
.strip_prefix(&format!("{}/", project))
.unwrap_or(rel)
.to_string();
out.push_str(&format!("--- {} ---\n", display));
match std::fs::read(&abs) {
Ok(bytes) => {
if looks_binary(&bytes) {
out.push_str("(binary, skipped)\n\n");
continue;
}
let n = bytes.len().min(max_bytes_per_file);
let snippet = String::from_utf8_lossy(&bytes[..n]);
out.push_str(&snippet);
if bytes.len() > n {
out.push_str("\n…(truncated)\n");
}
out.push_str("\n\n");
}
Err(_) => out.push_str("(unreadable / deleted)\n\n"),
}
}
if rel_paths.len() > max_files {
out.push_str(&format!(
"(+ {} more files not shown)\n",
rel_paths.len() - max_files
));
}
out
}
fn make_heartbeat_client(cfg: &WorkerConfig) -> Option<SupabaseClient> {
let (url, key) = match (&cfg.supabase_url, &cfg.supabase_key) {
(Some(u), Some(k)) if !u.is_empty() && !k.is_empty() => (u.as_str(), k.as_str()),
_ => return None,
};
let client_id = cfg
.client_id
.as_deref()
.filter(|s| !s.is_empty())
.unwrap_or("unknown");
SupabaseClient::new(url, key, client_id).ok()
}
fn looks_binary(bytes: &[u8]) -> bool {
bytes.iter().take(2048).any(|&b| b == 0)
}
fn build_mem0_query(project: &str, paths: &[String]) -> String {
use std::collections::BTreeSet;
let mut dirs: BTreeSet<String> = BTreeSet::new();
let mut exts: BTreeSet<String> = BTreeSet::new();
let mut filenames: Vec<String> = Vec::new();
for p in paths.iter().take(20) {
let pp = Path::new(p);
if let Some(parent) = pp.parent().and_then(|x| x.to_str()) {
if !parent.is_empty() && parent != project {
dirs.insert(parent.to_string());
}
}
if let Some(ext) = pp.extension().and_then(|e| e.to_str()) {
exts.insert(ext.to_lowercase());
}
if let Some(name) = pp.file_name().and_then(|n| n.to_str()) {
filenames.push(name.to_string());
}
}
let dir_list = dirs.iter().take(8).cloned().collect::<Vec<_>>().join(", ");
let ext_list = exts.iter().cloned().collect::<Vec<_>>().join(", ");
let file_list = filenames
.iter()
.take(6)
.cloned()
.collect::<Vec<_>>()
.join(", ");
format!(
"Project '{project}': recent edits in {dir_list} \
affecting files like {file_list} ({ext_list}). \
Find prior conventions, preferences, and architectural decisions \
relevant to these areas of the codebase."
)
}
fn log_line(msg: &str) {
let now = Local::now().format("%Y-%m-%d %H:%M:%S");
println!("{} {}", now, msg);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn mem0_query_includes_dirs_and_extensions() {
let q = build_mem0_query(
"myapp",
&[
"myapp/src/auth/login.tsx".into(),
"myapp/src/auth/session.ts".into(),
"myapp/package.json".into(),
],
);
assert!(q.contains("myapp"));
assert!(q.contains("login.tsx") || q.contains("session.ts"));
assert!(q.contains("tsx") || q.contains("ts") || q.contains("json"));
assert!(q.contains("myapp/src/auth"));
}
#[test]
fn mem0_query_handles_empty_paths() {
let q = build_mem0_query("x", &[]);
assert!(q.contains("x"));
}
#[test]
fn limiter_caps_to_window() {
let l = HourlyLimiter::new(3);
assert!(l.try_consume());
assert!(l.try_consume());
assert!(l.try_consume());
assert!(!l.try_consume());
assert_eq!(l.current(), 3);
}
}