""" builtin_actions.py Registry of built-in automation actions that can be executed by the task scheduler without needing an LLM call. """ import logging import os import json from datetime import datetime from typing import Tuple from src.auth_helpers import owner_filter from core.platform_compat import IS_WINDOWS, find_bash from core.constants import internal_api_base from src.constants import DATA_DIR, DEEP_RESEARCH_DIR, TIDY_CALENDAR_STATE_FILE, EMAIL_URGENCY_CACHE_DIR, COOKBOOK_STATE_FILE from src.interactive_gate import wait_for_interactive_quiet logger = logging.getLogger(__name__) def _read_email_urgency_state(state_path): """Read one atomic urgency checkpoint, tolerating the legacy shape.""" from pathlib import Path state_path = Path(state_path) try: state = ( json.loads(state_path.read_text(encoding="utf-8")) if state_path.exists() else {} ) except Exception: return {} return state if isinstance(state, dict) else {} def _email_urgency_account_generations(state): """Return normalized per-account checkpoint/complete generations. Checkpoint generations fence every accepted state mutation. Complete generations advance only for a non-stale complete scan. Missing metadata is the legacy generation zero. """ raw = state.get("account_generations", {}) if isinstance(state, dict) else {} if not isinstance(raw, dict): return {} generations = {} for account_id, value in raw.items(): if isinstance(value, dict): checkpoint = value.get("checkpoint", 0) complete = value.get("complete", 0) else: # Tolerate an intermediate scalar representation as one completed # checkpoint generation instead of discarding its fence. checkpoint = value complete = value try: checkpoint = max(0, int(checkpoint)) except (TypeError, ValueError): checkpoint = 0 try: complete = max(0, int(complete)) except (TypeError, ValueError): complete = 0 generations[str(account_id)] = { "checkpoint": checkpoint, "complete": complete, } return generations def _email_urgency_string_set(value): if not isinstance(value, (list, tuple, set, frozenset)): return set() return {str(item) for item in value if isinstance(item, (str, int))} def _acquire_email_urgency_state_lock( state_path, lock_db_path, cancel_event, timeout_seconds=120, ): """Acquire the cross-process urgency lock without blocking the app loop.""" import sqlite3 import time from pathlib import Path state_path = Path(state_path) state_path.parent.mkdir(parents=True, exist_ok=True) deadline = time.monotonic() + timeout_seconds while not cancel_event.is_set(): remaining = deadline - time.monotonic() if remaining <= 0: raise sqlite3.OperationalError("timed out waiting for urgency state lock") conn = sqlite3.connect( str(lock_db_path), timeout=min(0.25, max(0.01, remaining)), check_same_thread=False, ) try: conn.execute("BEGIN IMMEDIATE") except sqlite3.OperationalError as exc: conn.close() if "locked" not in str(exc).lower(): raise cancel_event.wait(min(0.05, max(0.0, remaining))) continue except BaseException: conn.close() raise if cancel_event.is_set(): conn.rollback() conn.close() return None, None return conn, _read_email_urgency_state(state_path) return None, None def _close_email_urgency_state_lock(conn): if conn is None: return try: try: conn.rollback() except Exception: pass finally: conn.close() def _commit_email_urgency_state(conn, state_path, next_state): """Atomically publish JSON before releasing the SQLite write lock.""" import uuid from pathlib import Path state_path = Path(state_path) temp_path = state_path.with_name( f".{state_path.name}.{uuid.uuid4().hex}.tmp" ) try: temp_path.write_text(json.dumps(next_state), encoding="utf-8") temp_path.replace(state_path) conn.commit() except BaseException: conn.rollback() raise finally: temp_path.unlink(missing_ok=True) conn.close() async def _run_email_urgency_state_transaction( state_path, lock_db_path, operation, ): """Serialize one urgency decision while keeping async work on this loop. Only lock acquisition waits in a worker thread. ``operation`` is awaited on the caller's long-lived event loop, where shared async clients, locks, and the browser-notification queue belong. Cancellation rolls back the SQLite transaction and never publishes a checkpoint. """ import asyncio import threading loop = asyncio.get_running_loop() cancel_event = threading.Event() acquire_future = loop.run_in_executor( None, _acquire_email_urgency_state_lock, state_path, lock_db_path, cancel_event, ) try: conn, prior = await asyncio.shield(acquire_future) except asyncio.CancelledError as cancelled: cancel_event.set() # The acquisition worker owns any connection until it returns. Wait # for its short busy-poll to observe cancellation, then close a lock it # may have won concurrently with the cancellation request. while True: try: conn, _prior = await asyncio.shield(acquire_future) break except asyncio.CancelledError: continue except Exception: conn = None break _close_email_urgency_state_lock(conn) raise cancelled if conn is None: raise asyncio.CancelledError try: result, next_state = await operation(prior) # Keep this small atomic publish synchronous. There is no await between # the successful operation and commit, so cancellation cannot be # observed and then followed by a checkpoint. try: _commit_email_urgency_state(conn, state_path, next_state) finally: conn = None return result except BaseException: _close_email_urgency_state_lock(conn) raise def _email_urgency_account_key(message_key): return str(message_key).split(":", 1)[0] def _email_urgency_payload_account_ids(state): """Return account IDs that still own user-visible urgency payload.""" if not isinstance(state, dict): return set() per_uid = state.get("per_uid", {}) per_uid_keys = per_uid if isinstance(per_uid, dict) else {} return { _email_urgency_account_key(key) for key in per_uid_keys } | { _email_urgency_account_key(key) for key in _email_urgency_string_set(state.get("notified_uids", [])) } def _email_urgency_known_account_ids(state): """Return payload owners plus generation-only active/retired markers.""" return _email_urgency_payload_account_ids(state) | set( _email_urgency_account_generations(state) ) def _email_urgency_stale_accounts( prior, base_account_generations, account_ids, ): prior_generations = _email_urgency_account_generations(prior) base_generations = _email_urgency_account_generations( {"account_generations": base_account_generations} ) return { str(account_id) for account_id in account_ids if prior_generations.get(str(account_id), {}).get("checkpoint", 0) != base_generations.get(str(account_id), {}).get("checkpoint", 0) } def _merge_email_urgency_state( prior, *, owner, per_uid_scores, notified_uids, all_unread_keys, fully_scanned_account_ids, base_account_generations, timestamp, retired_account_ids=(), base_payload_account_ids=(), known_account_ids=(), ): """Merge a scan without letting an older snapshot erase newer facts.""" prior_per_uid = prior.get("per_uid", {}) if not isinstance(prior_per_uid, dict): prior_per_uid = {} complete = {str(account_id) for account_id in fully_scanned_account_ids} prior_generations = _email_urgency_account_generations(prior) retire_requested = {str(account_id) for account_id in retired_account_ids} observed_accounts = { _email_urgency_account_key(key) for key in per_uid_scores } | complete | retire_requested stale_accounts = _email_urgency_stale_accounts( prior, base_account_generations, observed_accounts, ) prior_payload_accounts = _email_urgency_payload_account_ids(prior) base_payload_accounts = { str(account_id) for account_id in base_payload_account_ids } # A selected account can be absent from the base snapshot. If another # worker creates its first payload before this transaction wins the lock, # membership itself is a fence even when both snapshots normalize to the # legacy generation zero. retired_accounts = { account_id for account_id in retire_requested - stale_accounts if not ( account_id in prior_payload_accounts and account_id not in base_payload_accounts ) } fresh_complete = complete - stale_accounts - retired_accounts changed_accounts = set(fresh_complete) merged_per_uid = { key: value for key, value in prior_per_uid.items() if _email_urgency_account_key(key) not in retired_accounts } for key in list(merged_per_uid): account_id = _email_urgency_account_key(key) if account_id in fresh_complete: merged_per_uid.pop(key, None) changed_accounts.add(account_id) # Partial scans may add or refresh facts, but absence from a partial scan # is not evidence that another checkpoint or UI row is stale. When another # worker committed after this scan captured its base generation, discard # this account's whole stale snapshot. A key absent from the newer state # may have been removed/read, so even a stale-only key is not safely # additive without another fresh scan. for key, value in per_uid_scores.items(): account_id = _email_urgency_account_key(key) if account_id in stale_accounts or account_id in retired_accounts: continue if merged_per_uid.get(key) != value: changed_accounts.add(account_id) merged_per_uid[key] = value prior_notified = _email_urgency_string_set(prior.get("notified_uids", [])) merged_notified = { key for key in prior_notified if _email_urgency_account_key(key) not in retired_accounts } for key in _email_urgency_string_set(notified_uids) - prior_notified: account_id = _email_urgency_account_key(key) if account_id in stale_accounts or account_id in retired_accounts: continue merged_notified.add(key) changed_accounts.add(account_id) for key in list(merged_notified): if ( _email_urgency_account_key(key) in fresh_complete and key not in all_unread_keys ): merged_notified.discard(key) changed_accounts.add(_email_urgency_account_key(key)) next_generations = { account_id: dict(value) for account_id, value in prior_generations.items() } for account_id in changed_accounts: generation = next_generations.setdefault( account_id, {"checkpoint": 0, "complete": 0}, ) generation["checkpoint"] += 1 if account_id in fresh_complete: generation["complete"] += 1 for account_id in {str(value) for value in known_account_ids}: next_generations.setdefault( account_id, {"checkpoint": 0, "complete": 0}, ) for account_id in retired_accounts: # Every authoritative absence advances its generation, even when the # prior state is already a payload-empty tombstone. A re-enabled scan # may have captured that previous tombstone immediately before the # account was disabled/deleted again; monotonic advancement is what # makes that in-flight scan stale. generation = next_generations.setdefault( account_id, {"checkpoint": 0, "complete": 0}, ) generation["checkpoint"] += 1 total_unread = 0 total_urgent = 0 max_score = 0 for value in merged_per_uid.values(): if not isinstance(value, dict): continue try: score = max(0, min(3, int(value.get("score", 0)))) except (TypeError, ValueError): score = 0 max_score = max(max_score, score) if value.get("unread"): total_unread += 1 if score >= 2: total_urgent += 1 return { "ts": timestamp, "owner": owner or "", "total_unread": total_unread, "total_urgent": total_urgent, "max_score": max_score, "per_uid": merged_per_uid, "notified_uids": sorted(merged_notified), "account_generations": next_generations, } class TaskNoop(BaseException): """Raised by an action when it determined there's nothing to do. Inherits from BaseException (not Exception) so the standard `except Exception` wrappers each action uses for real error handling don't accidentally catch it. The scheduler explicitly catches TaskNoop, drops the queued TaskRun row, advances last_run / next_run, and exits silently. Nothing appears in the Activity log; the message is logged server-side only. """ class TaskDeferred(BaseException): """Raised when a task should run later without recording a skipped run.""" def __init__(self, reason: str, delay_seconds: int = 20 * 60): super().__init__(reason) self.reason = reason self.delay_seconds = delay_seconds async def action_tidy_sessions(owner: str, **kwargs) -> Tuple[str, bool]: """Delete empty sessions for the owner. Pure heuristic — the LLM folder-sort phase is skipped (user opted to keep this task LLM-free; sorting can be triggered manually via the Chats UI).""" try: import asyncio from src.session_actions import run_auto_sort result = await asyncio.wait_for( run_auto_sort(owner, skip_llm=True, delete_throwaway=False), timeout=60, ) return result, True except asyncio.TimeoutError: logger.error("tidy_sessions action timed out") return "Chat session tidy timed out", False except Exception as e: logger.error(f"tidy_sessions action failed: {e}") return str(e), False async def action_tidy_documents(owner: str, **kwargs) -> Tuple[str, bool]: """Run tidy on documents for the owner.""" try: from src.document_actions import run_document_tidy result = await run_document_tidy(owner) return result, True except Exception as e: logger.error(f"tidy_documents action failed: {e}") return str(e), False async def action_consolidate_memory(owner: str, **kwargs) -> Tuple[str, bool]: """Consolidate/deduplicate memories for the owner.""" try: import json import re from difflib import SequenceMatcher from src.constants import DATA_DIR from src.llm_core import llm_call_async_with_fallback from src.memory import MemoryManager manager = MemoryManager(DATA_DIR) all_memories = manager.load_all() _owner_clean = (owner or "").strip() text_limit = 2000 def _memory_owner(mem: dict) -> str: return (mem.get("owner") or "").strip() # Built-in housekeeping can run without an owner. In that case scan all # memories, but keep every AI prompt/apply step owner-local. if _owner_clean: memory_groups = { _owner_clean: [m for m in all_memories if _memory_owner(m) == _owner_clean] } else: memory_groups = {} for mem in all_memories: memory_groups.setdefault(_memory_owner(mem), []).append(mem) memory_groups = {group_owner: group for group_owner, group in memory_groups.items() if group} if not memory_groups: raise TaskNoop("no memories to consolidate") total_removed = 0 total_cleaned = 0 total_scanned = 0 removed_examples = [] ai_reasons = [] ai_used = False def _normalized_memory_text(mem: dict) -> str: text = (mem.get("text") or "").lower() text = re.sub(r"[^a-z0-9@._+-]+", " ", text) return " ".join(text.split()) def _memory_rank(mem: dict) -> tuple: text = (mem.get("text") or "").strip() return ( 1 if mem.get("pinned") else 0, 1 if (mem.get("source") or "") == "user" else 0, int(mem.get("uses") or 0), -len(text), int(mem.get("timestamp") or 0), ) def _same_memory_fact(a: dict, b: dict) -> bool: a_cat = (a.get("category") or "fact").strip().lower() b_cat = (b.get("category") or "fact").strip().lower() if a_cat != b_cat: return False a_text = _normalized_memory_text(a) b_text = _normalized_memory_text(b) if not a_text or not b_text: return False if a_text == b_text: return True shorter, longer = sorted((a_text, b_text), key=len) if len(shorter) >= 24 and shorter in longer: return True return SequenceMatcher(None, a_text, b_text).ratio() >= 0.88 def _dedupe_group(group_memories: list) -> tuple[list, int]: kept = [] removed = 0 for mem in group_memories: text = (mem.get("text") or "").strip() if not text: removed += 1 if len(removed_examples) < 3: removed_examples.append("(empty)") continue duplicate_idx = next( (idx for idx, kept_mem in enumerate(kept) if _same_memory_fact(mem, kept_mem)), None, ) if duplicate_idx is None: kept.append(mem) continue removed += 1 if _memory_rank(mem) > _memory_rank(kept[duplicate_idx]): if len(removed_examples) < 3: old_text = (kept[duplicate_idx].get("text") or "").strip() removed_examples.append(old_text[:60] + ("..." if len(old_text) > 60 else "")) kept[duplicate_idx] = mem elif len(removed_examples) < 3: removed_examples.append(text[:60] + ("..." if len(text) > 60 else "")) return kept, removed async def _try_ai_tidy_group(group_owner: str, group_memories: list) -> bool: nonlocal all_memories, total_removed, total_cleaned, total_scanned, ai_used if len(group_memories) < 2: return False from src.task_endpoint import resolve_task_candidates candidates = resolve_task_candidates(owner=group_owner or None) if not candidates: return False try: items = [ { "id": m.get("id"), "category": m.get("category", "fact"), "text": (m.get("text") or "").strip()[:text_limit], "truncated": len((m.get("text") or "").strip()) > text_limit, } for m in group_memories if m.get("id") and (m.get("text") or "").strip() ] if len(items) < 2: return False truncated_ids = {item["id"] for item in items if item.get("truncated")} prompt = ( "You are tidying a user's saved personal memories. Return ONLY raw JSON, no markdown.\n" "Remove memories that are empty, broken, trivial conversation filler, duplicates, or obsolete " "because a clearer newer memory replaces them. Preserve useful personal facts, preferences, " "contacts, project context, and instructions. If memories conflict, keep the clearest/latest " "one and drop the obsolete one.\n\n" "JSON shape:\n" "{\"keep\":[{\"id\":\"existing id\",\"text\":\"cleaned text\",\"category\":\"fact|preference|identity|event|contact|project|instruction\"}]," "\"drop\":[{\"id\":\"existing id\",\"reason\":\"short reason\"}]}\n\n" f"MEMORIES:\n{json.dumps(items, ensure_ascii=False)}" ) await wait_for_interactive_quiet("memory consolidation action") raw = await llm_call_async_with_fallback( candidates, messages=[{"role": "user", "content": prompt}], temperature=0.0, max_tokens=4096, timeout=120, ) from src.text_helpers import strip_think raw = strip_think(raw or "", prose=False, prompt_echo=False).strip() raw = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw, flags=re.MULTILINE).strip() start = raw.find("{") end = raw.rfind("}") if start != -1 and end != -1 and end > start: decision = json.loads(raw[start:end + 1]) keep_items = decision.get("keep") if isinstance(decision, dict) else None drop_items = decision.get("drop") if isinstance(decision, dict) else None if isinstance(keep_items, list) and isinstance(drop_items, list): by_id = {m.get("id"): m for m in group_memories if m.get("id")} cleaned_by_id = {} for item in keep_items: if not isinstance(item, dict): continue mid = item.get("id") if mid not in by_id: continue text = (item.get("text") or "").strip() if not text: continue cleaned = { "category": (item.get("category") or by_id[mid].get("category") or "fact").strip(), } original_text = (by_id[mid].get("text") or "").strip() if len(original_text) <= text_limit: cleaned["text"] = text cleaned_by_id[mid] = cleaned # Delete only memories the model EXPLICITLY dropped, never # ones it merely omitted from `keep`. Treating the # complement of `keep` as deletions meant a model that # forgot to re-list an id (common) silently destroyed that # memory. Honor the explicit `drop` set instead. drop_ids = { d.get("id") for d in drop_items if isinstance(d, dict) and d.get("id") in by_id } # Never delete a memory the model only saw truncated. drop_ids -= truncated_ids if drop_ids or cleaned_by_id: changed_text = 0 group_ref_ids = {id(m) for m in group_memories} kept_all = [] for mem in all_memories: if id(mem) not in group_ref_ids: kept_all.append(mem) continue mid = mem.get("id") if mid in drop_ids: continue cleaned = cleaned_by_id.get(mid) or {} if mid in truncated_ids: cleaned.pop("text", None) if cleaned.get("text") and cleaned["text"] != mem.get("text"): mem["text"] = cleaned["text"] changed_text += 1 if cleaned.get("category"): mem["category"] = cleaned["category"] kept_all.append(mem) removed = sum(1 for m in group_memories if m.get("id") in drop_ids) if removed or changed_text: all_memories = kept_all total_removed += removed total_cleaned += changed_text ai_used = True ai_reasons.extend([ (d.get("reason") or "").strip() for d in drop_items if isinstance(d, dict) and (d.get("reason") or "").strip() ]) return True except Exception as ai_err: logger.warning("AI memory tidy failed; falling back to duplicate cleanup: %s", ai_err) return False for group_owner, group_memories in memory_groups.items(): total_scanned += len(group_memories) deduped_group, group_removed = _dedupe_group(group_memories) if group_removed: group_ref_ids = {id(m) for m in group_memories} keep_ref_ids = {id(m) for m in deduped_group} all_memories = [ m for m in all_memories if id(m) not in group_ref_ids or id(m) in keep_ref_ids ] total_removed += group_removed group_memories = deduped_group if await _try_ai_tidy_group(group_owner, group_memories): continue seen = {} keep_refs = set() for mem in group_memories: text = (mem.get("text") or "").strip() key = " ".join(text.lower().split()) if not key: if len(removed_examples) < 3: removed_examples.append("(empty)") continue if key in seen: if len(removed_examples) < 3: removed_examples.append(text[:60] + ("..." if len(text) > 60 else "")) continue seen[key] = mem keep_refs.add(id(mem)) group_removed = len(group_memories) - len(keep_refs) if group_removed == 0: continue group_ref_ids = {id(m) for m in group_memories} all_memories = [ m for m in all_memories if id(m) not in group_ref_ids or id(m) in keep_refs ] total_removed += group_removed if total_removed or total_cleaned: manager.save(all_memories) if ai_used: reasons = ai_reasons[:3] reason_text = f": {'; '.join(reasons)}" if reasons else "" return ( f"AI tidied {total_scanned} memories: " f"removed {total_removed}, cleaned {total_cleaned}{reason_text}", True, ) preview = "; ".join(removed_examples) extra = f" (+{total_removed - len(removed_examples)} more)" if total_removed > len(removed_examples) else "" return f"Removed {total_removed} duplicate(s) of {total_scanned}: {preview}{extra}", True raise TaskNoop(f"scanned {total_scanned} memories, no duplicates") except Exception as e: logger.error(f"consolidate_memory action failed: {e}") return str(e), False # Registry: action name -> async function(owner, **kwargs) -> (result_str, success_bool) async def _run_subprocess(argv, *, shell: bool = False, timeout: int = 120, label: str = "Command") -> Tuple[str, bool]: """Shared subprocess runner. Wraps the blocking subprocess.run in asyncio.to_thread so the event loop stays responsive.""" import asyncio import subprocess try: result = await asyncio.to_thread( subprocess.run, argv, shell=shell, capture_output=True, text=True, timeout=timeout, ) output = (result.stdout or "").strip() if result.returncode != 0 and result.stderr: output += "\nSTDERR: " + result.stderr.strip() return output or "(no output)", result.returncode == 0 except subprocess.TimeoutExpired: return f"{label} timed out ({timeout}s)", False except Exception as e: return str(e), False async def action_ssh_command(owner: str, command: str = "", host: str = "localhost", **kwargs) -> Tuple[str, bool]: """Run a shell command locally or on a remote host via SSH.""" if not command: return "No command specified", False if host in ("localhost", "127.0.0.1", "local"): if IS_WINDOWS: bash = find_bash() if bash: return await _run_subprocess([bash, "-c", command], timeout=120, label="Command") return await _run_subprocess(command, shell=True, timeout=120, label="Command") return await _run_subprocess(["bash", "-c", command], timeout=120, label="Command") return await _run_subprocess( ["ssh", "-o", "ConnectTimeout=10", host, command], timeout=120, label="Command", ) async def action_run_script(owner: str, script: str = "", host: str = "", **kwargs) -> Tuple[str, bool]: """Run a script locally, or via SSH when a host is configured.""" if not script: return "No script specified", False target_host = (host or os.getenv("ODYSSEUS_SCRIPT_HOST", "localhost")).strip() if target_host in ("", "localhost", "127.0.0.1", "local"): if IS_WINDOWS and find_bash(): return await _run_subprocess([find_bash(), "-c", script], timeout=300, label="Script") return await _run_subprocess(script, shell=True, timeout=300, label="Script") return await _run_subprocess(["ssh", target_host, script], timeout=300, label="Script") async def action_run_local(owner: str, script: str = "", **kwargs) -> Tuple[str, bool]: """Run a script locally (no SSH).""" if not script: return "No script specified", False if IS_WINDOWS and find_bash(): return await _run_subprocess([find_bash(), "-c", script], timeout=300, label="Script") return await _run_subprocess(script, shell=True, timeout=300, label="Script") async def action_tidy_research(owner: str, **kwargs) -> Tuple[str, bool]: """Remove only broken research files (empty or unparseable JSON). Research history lives entirely in data/deep_research/.json and is NOT backed by chat-session rows — so a file must never be deleted just because no chat session matches its id. Only prune files that fail to load. A broken file has no readable owner stamp, so it cannot be matched against `owner`. Clearing one is privileged: admins and the single-user operator (AUTH_ENABLED=false) may, a regular user may not, and neither may anyone during the pre-setup window before an admin exists. """ try: from pathlib import Path import json as _json from src.tool_security import owner_is_admin_or_single_user research_dir = Path(DEEP_RESEARCH_DIR) if not research_dir.exists(): raise TaskNoop("no research directory") if not owner_is_admin_or_single_user(owner): # Return before the glob rather than filtering inside the loop: the # loop reports "none broken" off an empty `removed`, which reaches # Activity as a false report to a user whose files it skipped, and a # regular user need not read every owner's file to learn it may # delete none of them. raise TaskNoop("not permitted to remove unattributable research files") files = list(research_dir.glob("*.json")) removed = [] for p in files: try: txt = p.read_text(encoding="utf-8").strip() if not txt: raise ValueError("empty file") _json.loads(txt) # valid JSON → keep except Exception: p.unlink(missing_ok=True) removed.append(p.stem[:8]) if not removed: raise TaskNoop(f"scanned {len(files)} research file(s), none broken") return f"Removed {len(removed)} broken research file(s) of {len(files)}", True except Exception as e: logger.error(f"tidy_research action failed: {e}") return str(e), False async def action_tidy_calendar(owner: str, **kwargs) -> Tuple[str, bool]: """Find duplicate calendar events (same title + start time) and DELETE the dups, keeping the oldest (first-seen) instance. Incremental: remembers the newest `created_at` already scanned in data/tidy_calendar_state.json. If no events have been added since then, short-circuits. Otherwise only events newer than the watermark are candidates for deletion, but they're checked against the FULL existing set so a new duplicate of an old event still gets caught. """ try: import json from pathlib import Path from core.database import SessionLocal, CalendarEvent from sqlalchemy import func STATE_FILE = Path(TIDY_CALENDAR_STATE_FILE) last_watermark = None try: if STATE_FILE.exists(): saved = json.loads(STATE_FILE.read_text(encoding="utf-8")) if saved.get("last_created_at"): last_watermark = datetime.fromisoformat(saved["last_created_at"]) except Exception: last_watermark = None db = SessionLocal() try: newest = db.query(func.max(CalendarEvent.created_at)).scalar() db.query(CalendarEvent).count() # Short-circuit: nothing new since last run if last_watermark is not None and newest is not None and newest <= last_watermark: raise TaskNoop(f"no new events since watermark {last_watermark.strftime('%Y-%m-%d %H:%M')}") events = db.query(CalendarEvent).order_by(CalendarEvent.dtstart).all() # Build full seen-set from events at or before the watermark (known-clean). # Events after the watermark are candidates for deletion. seen = {} candidates = [] no_title = 0 for e in events: title = (e.summary or "").strip() if not title: no_title += 1 continue if last_watermark is None or (e.created_at and e.created_at <= last_watermark): # Known-clean region: first occurrence wins key = (title.lower(), e.dtstart) if key not in seen: seen[key] = e # If a dup exists in the known-clean region (first run, or imported later # with the same created_at), still remove it — fall through to candidate check. else: candidates.append(e) else: candidates.append(e) removed = [] for e in candidates: title = (e.summary or "").strip() key = (title.lower(), e.dtstart) if key in seen: when = e.dtstart.strftime('%Y-%m-%d %H:%M') if e.dtstart else '?' removed.append(f"{title} @ {when}") db.delete(e) else: seen[key] = e if removed: db.commit() # Persist the new watermark (newest created_at among events that survive) try: STATE_FILE.parent.mkdir(parents=True, exist_ok=True) if newest is not None: STATE_FILE.write_text(json.dumps({ "last_created_at": newest.isoformat(), "last_run_at": datetime.utcnow().isoformat(), "scanned": len(events), "removed": len(removed), }, indent=2), encoding="utf-8") except Exception as se: logger.warning(f"tidy_calendar watermark save failed: {se}") new_since = len(candidates) parts = [f"Scanned {len(events)} event(s), {new_since} new since last run"] if removed: preview = "; ".join(removed[:5]) if len(removed) > 5: preview += f" (+{len(removed) - 5} more)" parts.append(f"removed {len(removed)} duplicate(s): {preview}") if no_title: parts.append(f"{no_title} untitled (kept)") if not removed and not no_title: parts.append("no duplicates") return " · ".join(parts), True finally: db.close() except Exception as e: logger.error(f"tidy_calendar action failed: {e}") return str(e), False def _result_has_work(result: str | None) -> bool: """Heuristic: did the email pass actually process anything? `_run_auto_summarize_once` returns strings like 'Processed 0 emails', 'No new emails to summarize', 'Tagged 0 / Moved 0', etc. when nothing was done. Used to decide whether to record the run or noop it. """ if not isinstance(result, str) or not result: return False low = result.lower() if "processed 0" in low or "no new" in low or "nothing to" in low: return False # "Tagged 0 / Moved 0" or similar zero-count summaries if low.count(" 0") >= 2 and ("tagged" in low or "moved" in low or "drafted" in low): return False return True def _result_is_config_error(result: str | None) -> bool: if not isinstance(result, str): return False low = result.lower() return ( "no model configured" in low or "no model endpoint configured" in low or "no llm endpoint available" in low ) def _email_task_account_id(kwargs) -> str | None: prompt = (kwargs.get("prompt") or "").strip() if not prompt: return None try: data = json.loads(prompt) if isinstance(data, dict): val = data.get("account_id") or data.get("email_account_id") return str(val).strip() or None except Exception: pass for line in prompt.splitlines(): if "=" not in line: continue key, val = line.split("=", 1) if key.strip().lower() in {"account_id", "email_account_id"}: return val.strip() or None return None async def action_summarize_emails(owner: str, **kwargs) -> Tuple[str, bool]: """Run one pass of email summary background processing.""" try: from routes.email_pollers import _run_auto_summarize_once result = await _run_auto_summarize_once( do_summary=True, do_reply=False, account_id=_email_task_account_id(kwargs), ) if _result_is_config_error(result): return result, False if not _result_has_work(result): raise TaskNoop(f"summarize: {result or 'no new emails'}") return result, True except Exception as e: logger.error(f"summarize_emails action failed: {e}") return str(e), False async def action_draft_email_replies(owner: str, **kwargs) -> Tuple[str, bool]: """Run one pass of AI reply drafting.""" try: from routes.email_pollers import _run_auto_summarize_once result = await _run_auto_summarize_once( do_summary=False, do_reply=True, account_id=_email_task_account_id(kwargs), days_back=7, progress_cb=kwargs.get("progress_cb"), ) if _result_is_config_error(result): return result, False if not _result_has_work(result): raise TaskNoop(f"draft replies: {result or 'no new emails'}") return result, True except Exception as e: logger.error(f"draft_email_replies action failed: {e}") return str(e), False async def action_email_auto_translate(owner: str, **kwargs) -> Tuple[str, bool]: """Detect recent foreign-language emails and cache translated text. The reader still shows the original body; it simply checks this cache before calling the LLM on demand. Keep the scheduled pass deliberately small so translation never turns into a mailbox-wide background crawl. """ try: import email as _email_mod import json as _json import re as _re import sqlite3 as _sql3 from datetime import datetime as _dt, timedelta as _td from core.database import EmailAccount as _EA, SessionLocal as _SL from routes.email_helpers import ( SCHEDULED_DB, _decode_header, _email_cache_owner_clause, _extract_reply, _extract_text, _imap_connect, email_translation_body_hash, ) from src.settings import load_settings from src.task_endpoint import task_llm_call_async settings = load_settings() if not settings.get("email_auto_translate", False): raise TaskNoop("email auto-translate is disabled") target_language = (settings.get("email_translate_language") or "English").strip() or "English" account_id = _email_task_account_id(kwargs) days_back = 7 max_process = 5 try: data = _json.loads((kwargs.get("prompt") or "").strip() or "{}") if isinstance(data, dict): days_back = max(1, min(30, int(data.get("days_back") or days_back))) max_process = max(1, min(20, int(data.get("max_process") or max_process))) except Exception: pass db = _SL() try: from sqlalchemy import and_ as _and, or_ as _or q = db.query(_EA).filter(_EA.enabled == True) # noqa: E712 if owner: unowned = _or(_EA.owner == None, _EA.owner == "") # noqa: E711 same_mailbox = _or(_EA.imap_user == owner, _EA.from_address == owner) q = q.filter(_or(_EA.owner == owner, _and(unowned, same_mailbox))) if account_id: q = q.filter(_EA.id == account_id) accounts = q.all() finally: db.close() if not accounts: raise TaskNoop("no email accounts configured") def _cached(body_hash: str) -> bool: c = _sql3.connect(SCHEDULED_DB) try: owner_clause, owner_params = _email_cache_owner_clause(owner) row = c.execute( f"SELECT 1 FROM email_translations " f"WHERE body_hash = ? AND target_language = ? AND {owner_clause} LIMIT 1", (body_hash, target_language, *owner_params), ).fetchone() return bool(row) finally: c.close() def _store( body_hash: str, *, uid: str, folder: str, subject: str, sender: str, translation: str, same_language: bool, model_used: str, ) -> None: c = _sql3.connect(SCHEDULED_DB) try: c.execute(""" INSERT OR REPLACE INTO email_translations (body_hash, owner, target_language, uid, folder, subject, sender, translation, same_language, model_used, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( body_hash, owner, target_language, uid, folder, subject, sender, translation, 1 if same_language else 0, model_used, _dt.utcnow().isoformat(), )) c.commit() finally: c.close() async def _translate(body: str, subject: str, sender: str) -> tuple[str, bool]: content = await task_llm_call_async( [ { "role": "system", "content": ( "You translate emails faithfully. Preserve meaning, names, dates, money, addresses, " "bullet structure, and tone. Do not summarize or answer the email. " "Output only the translation between <<>> and <<>>. " "If the email is already primarily in the target language, output exactly " "<<>>." ), }, { "role": "user", "content": ( f"Target language: {target_language}\n\n" f"From: {sender}\nSubject: {subject}\n\n{body[:16000]}\n\n" "Translate the email unless it is already primarily in the target language.\n" "Return only:\n<<>>\ntranslated text\n<<>>" ), }, ], owner=owner, temperature=0.2, max_tokens=8192, timeout=180, ) content = (content or "").strip() content = _extract_reply(content) if "<<>>" in content: return "", True marker = _re.search(r"<<>>\s*(.*?)\s*<<>>", content, _re.S | _re.I) if marker: content = marker.group(1).strip() else: content = _re.sub(r"^\s*<<>>\s*", "", content, flags=_re.I).strip() content = _re.sub(r"\s*<<>>\s*$", "", content, flags=_re.I).strip() return content, False since = (_dt.utcnow() - _td(days=days_back)).strftime("%d-%b-%Y") examined = 0 cached = 0 translated = 0 same_language = 0 skipped = 0 failures = 0 processed = 0 for acct in accounts: if processed >= max_process: break imap = None try: imap = _imap_connect(acct.id, owner=owner) imap.select("INBOX", readonly=True) status, data = imap.uid("SEARCH", None, f'(SINCE {since})') if status != "OK" or not data or not data[0]: continue uids = list(reversed(data[0].split()))[:50] for uid_b in uids: if processed >= max_process: break uid = uid_b.decode("utf-8", errors="ignore") if isinstance(uid_b, bytes) else str(uid_b) status, msg_data = imap.uid("FETCH", uid, "(RFC822)") if status != "OK" or not msg_data: continue raw = None for part in msg_data: if isinstance(part, tuple) and len(part) > 1: raw = part[1] break if not raw: continue msg = _email_mod.message_from_bytes(raw) subject = _decode_header(msg.get("Subject", "")) sender = _decode_header(msg.get("From", "")) body = (_extract_text(msg) or "").strip() examined += 1 if len(body) < 80: skipped += 1 continue body_hash = email_translation_body_hash(body) if _cached(body_hash): cached += 1 continue translation, is_same_language = await _translate(body, subject, sender) if is_same_language: _store( body_hash, uid=uid, folder="INBOX", subject=subject, sender=sender, translation="", same_language=True, model_used="background-task", ) same_language += 1 processed += 1 continue if not translation: failures += 1 continue _store( body_hash, uid=uid, folder="INBOX", subject=subject, sender=sender, translation=translation, same_language=False, model_used="background-task", ) translated += 1 processed += 1 except Exception as acct_e: failures += 1 logger.warning(f"email_auto_translate account scan failed for {getattr(acct, 'id', '?')}: {acct_e}") finally: if imap: try: imap.logout() except Exception: pass if translated == 0 and same_language == 0: result = ( f"no uncached foreign-language emails found " f"(examined {examined}, cached {cached}, skipped {skipped}, failures {failures})" ) if failures: return f"Email Auto Translate failed: {result}", False raise TaskNoop(result) return ( f"Email Auto Translate cached {translated} translation(s), marked {same_language} same-language " f"(examined {examined}, already cached {cached}, skipped {skipped}, failures {failures})", True, ) except TaskNoop: raise except Exception as e: logger.error(f"email_auto_translate action failed: {e}") return str(e), False _TYPE_COLORS = { "work": "#5b8abf", # blue "personal": "#a07ae0", # purple "health": "#e06c75", # red "travel": "#e5a33a", # orange "meal": "#d8b974", # tan "social": "#82c882", # green "admin": "#888888", # gray "other": "#6b9cb5", # default } _HEURISTIC_TYPES = { "health": ["doctor", "dentist", "clinic", "hospital", "appointment", "checkup", "therapy", "physio", "chiropract", "vaccine", "blood test", "xray", "scan", "surgery"], "travel": ["flight", "airport", "train", "shinkansen", "boarding", "uber", "taxi", "trip", "hotel", "airbnb", "depart", "arrival", "check-in", "checkout"], "meal": ["lunch", "dinner", "breakfast", "brunch", "coffee", "drinks", "restaurant", "reservation", "bar", "cafe"], "social": ["birthday", "party", "hangout", "wedding", "date with", "drinks with", "anniversary", "baby shower", "graduation", "picnic", "bbq"], "admin": ["bill", "renewal", "tax", "deadline", "filing", "submit", "due date", "registration", "license", "passport", "visa", "form"], "work": ["meeting", "standup", "sync", "1:1", "1on1", "review", "interview", "demo", "presentation", "kickoff", "retro", "all-hands", "town hall", "call with", "client", "deck"], } _HEURISTIC_HIGH = ["flight", "interview", "wedding", "surgery", "exam", "deadline", "court", "presentation", "demo", "kickoff", "launch"] _HEURISTIC_CRITICAL = ["surgery", "court", "wedding day", "funeral", "delivery date"] def _classify_event_heuristic(summary: str) -> tuple: """Quick heuristic classification — returns (event_type, importance) or (None, None) if unclear.""" s = (summary if isinstance(summary, str) else "").lower() etype = None for t, kws in _HEURISTIC_TYPES.items(): if any(k in s for k in kws): etype = t break if any(k in s for k in _HEURISTIC_CRITICAL): return etype, "critical" if any(k in s for k in _HEURISTIC_HIGH): return etype, "high" return etype, None def _memory_context_lines(mems, limit: int = 40) -> list: """Render Memory rows into short personal-context bullets for event classify. Reads the Memory ORM `text` column. The previous inline code read a non-existent `content` attribute, so it raised AttributeError on the first row, the surrounding except swallowed it, and the classifier ran with no personal context at all. getattr keeps it robust to future schema drift. """ lines: list = [] for m in mems: c = (getattr(m, "text", "") or "").strip() if c: lines.append(f"- {c[:200]}") if len(lines) >= limit: break return lines async def action_classify_events(owner: str, **kwargs) -> Tuple[str, bool]: """Hybrid classification of upcoming calendar events: fast heuristic for obvious cases, LLM fallback for ambiguous ones. Assigns event_type + importance + color. Re-classifies anything not already set.""" try: from datetime import timedelta from core.database import SessionLocal, CalendarEvent from src.llm_core import llm_call_async_with_fallback import re as _re, json as _json db = SessionLocal() try: now = datetime.utcnow() horizon = now + timedelta(days=30) events = db.query(CalendarEvent).filter( CalendarEvent.dtstart >= now, CalendarEvent.dtstart <= horizon, CalendarEvent.status != "cancelled", ).all() if not events: return "No upcoming events to classify", True from src.task_endpoint import resolve_task_candidates llm_candidates = resolve_task_candidates(owner=owner) llm_available = bool(llm_candidates) # Pull user memories so the LLM has personal context (relationships, # job, hobbies). Helps it know e.g. " is your spouse" so their # events are personal/social, not work. _memory_context = "" try: from core.database import Memory as _Mem _mems = db.query(_Mem).filter(_Mem.owner == owner).limit(60).all() if owner else [] _lines = _memory_context_lines(_mems) if _lines: _memory_context = "USER CONTEXT (relationships, work, life):\n" + "\n".join(_lines) + "\n\n" except Exception as _me: logger.warning(f"Could not load memory for classify: {_me}") classified_h = 0 classified_llm = 0 failed = 0 unchanged = 0 # Pass 1: heuristic for obvious cases, collect ambiguous for LLM batch llm_queue = [] # list of CalendarEvent objects needing LLM for ev in events: if ev.event_type and ev.importance and ev.importance != "normal": unchanged += 1 continue etype, importance = _classify_event_heuristic(ev.summary or "") if etype and importance: ev.event_type = etype ev.color = _TYPE_COLORS.get(etype) ev.importance = importance classified_h += 1 continue # Apply partial heuristic; queue for LLM to fill missing if etype: ev.event_type = etype ev.color = _TYPE_COLORS.get(etype) if llm_available: llm_queue.append(ev) elif etype: classified_h += 1 # Persist heuristic results before LLM pass (in case LLM is slow/unavailable) try: db.commit() except Exception: pass # Pass 2: batch LLM classification (10 events per call) BATCH = 10 for i in range(0, len(llm_queue), BATCH): batch = llm_queue[i:i+BATCH] items = [ {"i": idx, "title": (ev.summary or "")[:120], "when": ev.dtstart.isoformat() if ev.dtstart else "", "loc": (ev.location or "")[:80]} for idx, ev in enumerate(batch) ] prompt = ( _memory_context + "Classify these calendar events using the USER CONTEXT above (people they know, " "their job, hobbies). Return ONLY a raw JSON array, no prose, no markdown.\n" "Each item: {\"i\": , \"type\": \"work|personal|health|travel|meal|social|admin|other\", " "\"importance\": \"low|normal|high|critical\"}\n\n" "Type guidance:\n" "- personal = family, partner, kids, pets, errands, home stuff\n" "- social = friends, parties, birthdays, hangouts\n" "- work = the user's own job/career commitments only (not their partner's)\n" "- health = doctor, gym, therapy\n" "- travel = flights, trips, hotels\n" "- meal = lunch/dinner/coffee specifically\n" "- admin = bills, taxes, paperwork\n" "- other = anything else\n\n" "Importance guide: critical = surgery/court/wedding day; high = flight/interview/big presentation/exam; " "normal = regular meetings/appointments; low = recurring routine.\n\n" f"EVENTS: {_json.dumps(items)}" ) try: await wait_for_interactive_quiet("calendar classification action") raw = await llm_call_async_with_fallback( llm_candidates, messages=[{"role": "user", "content": prompt}], temperature=0.1, max_tokens=16384, timeout=180, ) from src.text_helpers import strip_think as _st raw = _st(raw or "", prose=False, prompt_echo=False) raw = _re.sub(r"^```(?:json)?\s*|\s*```$", "", raw, flags=_re.MULTILINE).strip() m = _re.search(r"\[.*\]", raw, _re.DOTALL) if not m: logger.warning(f"[classify-llm] no JSON array in response: {raw[:300]!r}") failed += len(batch) continue arr = _json.loads(m.group()) by_idx = {x.get("i"): x for x in arr if isinstance(x, dict)} for idx, ev in enumerate(batch): x = by_idx.get(idx) if not x: failed += 1 continue t = (x.get("type") or "other").lower() imp = (x.get("importance") or "normal").lower() if t in _TYPE_COLORS: ev.event_type = t ev.color = _TYPE_COLORS[t] if imp in ("low", "normal", "high", "critical"): ev.importance = imp classified_llm += 1 logger.info(f"[classify-llm] '{ev.summary}' → type={t} importance={imp}") except Exception as e: logger.warning(f"[classify-llm] batch failed: {e}") failed += len(batch) # Commit after each batch so partial progress persists try: db.commit() except Exception as ce: logger.warning(f"[classify-llm] commit failed: {ce}") # Final commit covers heuristic-only updates from pass 1 db.commit() parts = [f"Scanned {len(events)} upcoming event(s)"] if classified_h: parts.append(f"{classified_h} via heuristic") if classified_llm: parts.append(f"{classified_llm} via LLM") if unchanged: parts.append(f"{unchanged} already set (skipped)") if failed: parts.append(f"{failed} LLM failed") return " · ".join(parts), True finally: db.close() except Exception as e: logger.error(f"classify_events action failed: {e}") return str(e), False async def action_ping_events(owner: str, **kwargs) -> Tuple[str, bool]: """Calendar event reminders are now dispatched by Notes.""" raise TaskNoop("calendar event reminders are handled by Notes") async def action_extract_email_events(owner: str, **kwargs) -> Tuple[str, bool]: """Scan recent emails for booking confirmations / meetings / events and auto-add them to the calendar.""" import asyncio as _aio try: from routes.email_pollers import _run_auto_summarize_once account_id = _email_task_account_id(kwargs) attempts = [ ("3d window, 3 emails", 3, 3, 240), ("3d window, 2 emails", 3, 2, 150), ("1d window, 1 email", 1, 1, 90), ] timed_out = [] last_result = "" for label, days_back, max_process, timeout in attempts: try: result = await _aio.wait_for( _run_auto_summarize_once( do_summary=False, do_reply=False, do_calendar=True, days_back=days_back, account_id=account_id, max_process=max_process, ), timeout=timeout, ) last_result = result or "" if _result_is_config_error(result): return f"{result} ({label})", False if _result_has_work(result): suffix = f"{label}" if not timed_out else f"{label}; retried after timeout" return f"{result} ({suffix})", True raise TaskNoop(f"email→calendar: {result or 'no new emails'} ({label})") except _aio.TimeoutError: timed_out.append(label) logger.warning(f"email calendar extraction timed out for {label}; retrying smaller batch") continue if timed_out: raise TaskNoop( "email→calendar: calendar extraction timed out on smaller batches; " "will retry on the next scheduled run" ) raise TaskNoop(f"email→calendar: {last_result or 'no new emails'}") except Exception as e: logger.error(f"extract_email_events action failed: {e}") return str(e), False # Sender local-parts (matched exactly or by prefix) whose mail never carries a # personal signature worth learning. These compare against the local-part # (before "@"), so role names must NOT include a trailing "@" — "support@" etc. # could never match a local-part of "support" and were silently dead. _SIG_SKIP_PREFIXES = ( "noreply", "no-reply", "donotreply", "do-not-reply", "mailer-daemon", "notifications", "notification", "bounce", "newsletter", "support", "info", "admin", ) async def action_learn_sender_signatures(owner: str, **kwargs) -> Tuple[str, bool]: """For each sender with ≥3 recent inbox emails, ask the LLM to extract the common signature block across their messages. The cached sig is served on the `/read` endpoint so the renderer can fold signatures consistently from that address (no more heuristic regex juggling). Caps at 20 senders per pass; re-runs after 30 days per sender.""" try: import sqlite3 as _sql3 import re as _re import email as _email_mod import asyncio as _aio from datetime import datetime as _dt, timedelta as _td from routes.email_helpers import _email_cache_owner_clause, _imap_connect, SCHEDULED_DB from src.llm_core import llm_call_async_with_fallback # 1. Pull recent UIDs + From headers cheaply (header-only fetch). def _pull_headers(): results = [] conn = _imap_connect(None, owner=owner) try: conn.select("INBOX", readonly=True) status, data = conn.uid("SEARCH", None, "ALL") if status != "OK" or not data or not data[0]: return results uids = data[0].split()[-300:][::-1] # newest 300 for uid in uids: try: st, msg_data = conn.uid( "FETCH", uid, "(BODY.PEEK[HEADER.FIELDS (FROM)])" ) if st != "OK" or not msg_data or not msg_data[0]: continue raw = msg_data[0][1] if isinstance(msg_data[0], tuple) else None if not raw: continue msg = _email_mod.message_from_bytes(raw) from_raw = msg.get("From", "") from_addr = _email_mod.utils.parseaddr(from_raw)[1].lower().strip() if not from_addr or "@" not in from_addr: continue results.append({ "uid": uid.decode() if isinstance(uid, bytes) else str(uid), "from_address": from_addr, }) except Exception: continue finally: try: conn.logout() except Exception: pass return results mails = await _aio.to_thread(_pull_headers) if not mails: return "No emails to scan", True # 2. Group by sender; drop addresses that don't carry useful sigs. by_sender: dict[str, list[dict]] = {} for m in mails: addr = m["from_address"] local = addr.split("@", 1)[0] if any(local == p or local.startswith(p) for p in _SIG_SKIP_PREFIXES): continue # Skip plus-aliases / list-style addresses too. if "+" in local or "-noreply" in addr or "-bounces" in addr: continue by_sender.setdefault(addr, []).append(m) # 3. Eligibility: ≥3 emails AND (no cache OR cache > 30 days old). try: conn = _sql3.connect(SCHEDULED_DB) owner_clause, owner_params = _email_cache_owner_clause(owner) cached = { r[0]: r[1] for r in conn.execute( f"SELECT from_address, last_built_at FROM sender_signatures WHERE {owner_clause}", owner_params, ).fetchall() } conn.close() except Exception: cached = {} cutoff_iso = (_dt.utcnow() - _td(days=30)).isoformat() eligible: list[tuple[str, list[dict]]] = [] for addr, msgs in by_sender.items(): if len(msgs) < 3: continue if cached.get(addr, "") > cutoff_iso: continue eligible.append((addr, msgs[:5])) # use up to last 5 emails if not eligible: return "All sender sigs already cached (or no eligible senders)", True from src.task_endpoint import resolve_task_candidates candidates = resolve_task_candidates(owner=owner) if not candidates: return "No LLM endpoint available", False model = candidates[0][1] analyzed = 0 no_sig = 0 for addr, msgs in eligible[:20]: # cost cap per run def _fetch_bodies(_msgs): bodies = [] conn2 = _imap_connect(None, owner=owner) try: conn2.select("INBOX", readonly=True) for mm in _msgs: try: st, data = conn2.uid("FETCH", mm["uid"], "(BODY.PEEK[TEXT])") if st != "OK" or not data or not data[0]: continue raw = data[0][1] if isinstance(data[0], tuple) else None if not raw: continue text = raw.decode("utf-8", errors="replace") bodies.append(text[:4000]) except Exception: continue finally: try: conn2.logout() except Exception: pass return bodies try: bodies = await _aio.to_thread(_fetch_bodies, msgs) except Exception as e: logger.warning(f"sig learner: fetch bodies failed for {addr}: {e}") continue if len(bodies) < 2: continue joined = "\n\n---NEXT EMAIL---\n\n".join(bodies[:5]) prompt = ( "You are extracting the literal common SIGNATURE block that " "appears at the END of multiple emails from the same sender.\n\n" "Return ONLY the exact signature text, verbatim, with original " "line breaks preserved. If there is no clear common signature " "block across these emails, respond with the single token: " "NONE\n\n" "INCLUDE: title, company, address, phone, email/url lines, " "legal disclaimer block.\n" "EXCLUDE: greetings ('Hi', 'Dear'), closing phrases on their " "own ('Best regards'), the sender's name on its own line, the " "body content, quoted/forwarded threads (lines starting with " "'>' or 'On ... wrote:' or 'From: ... Sent:').\n\n" f"EMAILS FROM {addr}:\n{joined}" ) try: await wait_for_interactive_quiet("sender signature action") raw = await llm_call_async_with_fallback( candidates, messages=[{"role": "user", "content": prompt}], temperature=0.0, max_tokens=600, timeout=60, ) from src.text_helpers import strip_think as _st sig = _st(raw or "", prose=False, prompt_echo=False).strip() # Strip surrounding code fences if the LLM added them. sig = _re.sub(r"^```[\w]*\n?", "", sig) sig = _re.sub(r"\n?```\s*$", "", sig) sig = sig.strip() except Exception as e: logger.warning(f"sig LLM call failed for {addr}: {e}") continue # NONE sentinel or out-of-bounds → cache a NULL row so we don't # re-try for 30 days, then move on. if ( not sig or sig.upper().strip().strip(".") == "NONE" or len(sig) < 15 or len(sig) > 3000 ): cached_sig: str | None = None no_sig += 1 else: cached_sig = sig try: conn = _sql3.connect(SCHEDULED_DB) owner_value = (owner or "").strip() conn.execute( "INSERT OR REPLACE INTO sender_signatures " "(from_address, owner, signature_text, sample_count, last_built_at, model_used, source) " "VALUES (?, ?, ?, ?, ?, ?, ?)", (addr, owner_value, cached_sig, len(bodies), _dt.utcnow().isoformat(), model, "llm"), ) conn.commit() conn.close() analyzed += 1 except Exception as e: logger.warning(f"sig cache write failed for {addr}: {e}") return f"Learned sigs: {analyzed - no_sig} found, {no_sig} no-sig, of {len(eligible)} eligible", True except Exception as e: logger.error(f"learn_sender_signatures failed: {e}") return str(e), False async def action_daily_brief(owner: str, **kwargs) -> Tuple[str, bool]: """Build a short morning digest: today's calendar events, unread email count + top-N senders/subjects, active todos.""" try: from datetime import datetime as _dt, timedelta as _td import json as _json from core.database import SessionLocal, CalendarEvent, CalendarCal, Note from routes.email_helpers import _imap_connect, _decode_header # ----- Calendar: today's events ----- today = _dt.now().replace(hour=0, minute=0, second=0, microsecond=0) tomorrow = today + _td(days=1) # v2 review HIGH-12: gate the OR-null branch on single-user # (unconfigured) deploys only. In a multi-user deploy, one # user's daily brief must not include another user's notes or # events that happen to be stored with owner=None. try: from core.auth import AuthManager _allow_null = not AuthManager().is_configured except Exception: _allow_null = False db = SessionLocal() try: ev_q = db.query(CalendarEvent).join(CalendarCal).filter( CalendarEvent.dtstart < tomorrow, CalendarEvent.dtend > today, CalendarEvent.status != "cancelled", ) if owner: ev_q = owner_filter(ev_q, CalendarCal, owner, include_shared=_allow_null) events = ev_q.order_by(CalendarEvent.dtstart).all() # ----- Notes: pinned + non-archived todos with at least one undone item ----- n_q = db.query(Note).filter(Note.archived == False) # noqa: E712 if owner: n_q = owner_filter(n_q, Note, owner, include_shared=_allow_null) notes = n_q.all() finally: db.close() # ----- Email: unread count + top 5 inbox subjects (best-effort) ----- # Direct IMAP: cheaper than the full _list_emails_sync helper and # avoids the module/import coupling that broke this once already. unread_count = 0 recent_subjects: list[tuple[str, str]] = [] try: import email as _email conn = _imap_connect(None) try: conn.select("INBOX", readonly=True) status, data = conn.uid("SEARCH", None, "UNSEEN") uids = (data[0].split() if status == "OK" and data and data[0] else []) unread_count = len(uids) # Grab headers for the most recent 5 unread (UIDs increase with arrival) for uid in uids[-5:][::-1]: try: _, msg_data = conn.uid("FETCH", uid, "(BODY.PEEK[HEADER.FIELDS (FROM SUBJECT)])") if not msg_data or not msg_data[0]: continue hdr = msg_data[0][1] if isinstance(msg_data[0], tuple) else msg_data[0] parsed = _email.message_from_bytes(hdr) subject = _decode_header(parsed.get("Subject") or "") or "(no subject)" from_raw = _decode_header(parsed.get("From") or "") or "?" # Extract just the display name if "Name " form if "<" in from_raw: name = from_raw.split("<", 1)[0].strip().strip('"') or from_raw else: name = from_raw recent_subjects.append((name, subject)) except Exception as fe: logger.debug(f"daily_brief: header fetch for uid {uid} failed: {fe}") finally: try: conn.logout() except Exception: pass except Exception as ee: logger.debug(f"daily_brief: email fetch failed: {ee}") # Pull active todo items from notes todo_lines: list[str] = [] for n in notes: if n.note_type == "checklist" and n.items: try: items = _json.loads(n.items) pending = [it.get("text", "") for it in items if not it.get("done")] for t in pending[:3]: if t: todo_lines.append(f"{n.title or 'Checklist'}: {t}") except Exception: continue elif n.pinned and n.title: todo_lines.append(n.title) # ----- Compose ----- # %-d is GNU-only; format the day with str() so the brief works on # Windows / non-glibc Python builds too. date_label = today.strftime(f"%A, %B {today.day}, %Y") plain = [f"Daily brief — {date_label}", ""] if events: plain.append("Calendar:") for e in events: t = e.dtstart.strftime("%H:%M") if not e.all_day else "all day" loc = f" @ {e.location}" if e.location else "" plain.append(f" {t} {e.summary}{loc}") plain.append("") else: plain.append("Calendar: nothing scheduled.") plain.append("") plain.append(f"Email: {unread_count} unread") for sender, subj in recent_subjects: plain.append(f" · {sender} — {subj}") plain.append("") if todo_lines: plain.append("Todos:") for t in todo_lines[:10]: plain.append(f" · {t}") else: plain.append("Todos: none active.") plain_body = "\n".join(plain) return plain_body, True except Exception as e: logger.error(f"daily_brief action failed: {e}") return str(e), False async def action_test_skills(owner: str, **kwargs) -> Tuple[str, bool]: """Run the per-skill Test on every skill: agent runs the procedure in a sandbox, LLM judges the transcript, verdict is recorded on the skill. ADVISORY ONLY — only writes set_audit (never rewrites SKILL.md, never demotes status, never overrides confidence).""" try: from services.memory.skills import SkillsManager from src.constants import DATA_DIR from routes.skills_routes import _run_skill_test_once, _skill_test_task # #3 SCOPE GUARD: refuse to run on a None/empty owner — otherwise # `sm.load(owner=None)` returns every user's skills and we'd cross- # test (and write audit verdicts to) other users' data in a # multi-user deployment. if not owner: return "test_skills requires an owner on the task — refusing to run without scope.", False sm = SkillsManager(DATA_DIR) skills = sm.load(owner=owner) names = [s.get("name") for s in skills if s.get("name")] if not names: raise TaskNoop("no skills to test") from src.task_endpoint import resolve_task_candidates candidates = resolve_task_candidates(owner=owner) if not candidates: return "No Default/Utility model configured — set one in Settings.", False # #2 NO SILENT MODEL SWAP: if the configured model isn't served by the # endpoint, try a basename match — but fail loudly instead of grabbing # `avail[0]` which could be an embedding-only model and produce 36 # garbage transcripts → 36 'unknown' verdicts with no hint why. url, model, headers = candidates[0] try: from src.llm_core import list_model_ids import os as _os selected = None mismatch_notes = [] for cand_url, cand_model, cand_headers in candidates: avail = list_model_ids(cand_url, headers=cand_headers) if not avail or cand_model in avail: selected = (cand_url, cand_model, cand_headers) break base = _os.path.basename((cand_model or "").rstrip("/")) matched = next((a for a in avail if _os.path.basename(a.rstrip("/")) == base), None) if matched: selected = (cand_url, matched, cand_headers) break mismatch_notes.append( f"{cand_model} not served by {cand_url}; available: " f"{', '.join(avail[:8])}{'...' if len(avail) > 8 else ''}" ) if selected: url, model, headers = selected elif mismatch_notes: return "No configured task fallback model is served. " + " | ".join(mismatch_notes[:3]), False except Exception as _e: logger.warning(f"test_skills model resolve check failed (continuing): {_e}") logger.info(f"test_skills: starting on {len(names)} skills, model={model}, owner={owner!r}") from collections import Counter tally = Counter() per_skill_log = [] for skill in skills: name = skill.get("name") if not name: continue md = sm.read_skill_md(name, owner=owner) or "" if not md: tally["skipped"] += 1 per_skill_log.append(f"{name}: skipped (no SKILL.md)") continue task = _skill_test_task(skill) try: transcript, verdict = await _run_skill_test_once(md, task, url, model, headers, owner) v = (verdict or {}).get("verdict") or "unknown" tally[v] += 1 summary = (verdict or {}).get("summary") or "" tlen = len(transcript or "") detail = "" if v in ("unknown", "inconclusive", "fail", "needs_work"): bits = [] if summary: bits.append(summary[:160]) if tlen < 200: bits.append(f"transcript {tlen}b") if bits: detail = " — " + "; ".join(bits) per_skill_log.append(f"{name}: {v}{detail}") # #4 + #8 + #12: ONLY persist a real verdict (pass / needs_work / # fail / inconclusive). Skip 'unknown' — that's the judge's # "couldn't parse" sentinel, not a real result, and persisting # it pollutes the verified-badge UI. Also skip the confidence # rewrite entirely — update_skill() re-serialises SKILL.md # (contradicts "advisory only" docstring) and overwriting a # user-set value (e.g. 1.0 → 0.95) is destructive. if v in ("pass", "needs_work", "fail", "inconclusive"): try: sm.set_audit(name, v, by_teacher=False, worker_model=model, owner=owner) except Exception as _e: logger.warning(f"test_skills set_audit({name}) failed: {_e}") if v == "unknown": logger.warning(f"test_skills: {name} → unknown — {summary[:200]}; transcript_len={tlen}") except Exception as e: logger.exception(f"test_skills: {name} errored") tally["error"] += 1 per_skill_log.append(f"{name}: error — {str(e)[:200]}") parts = [] for k in ("pass", "needs_work", "fail", "inconclusive", "unknown", "skipped", "error"): if tally.get(k): parts.append(f"{tally[k]} {k}") header = f"Tested {len(names)} skill(s): " + (" · ".join(parts) or "0") # Multi-line result: summary first, then per-skill detail. The Tasks # Activity feed renders this verbatim, so the user can see per-skill # outcomes + the judge's "why" without checking uvicorn stdout. body = "\n".join(per_skill_log) return f"{header}\nmodel={model}\n\n{body}", True except TaskNoop: raise except Exception as e: logger.error(f"test_skills action failed: {e}") return str(e), False async def action_audit_skills(owner: str, **kwargs) -> Tuple[str, bool]: """Run the real skills audit pipeline for skills that have not been audited. Unlike test_skills, this uses the same audit logic as the UI Audit all flow: metadata narrowing, self-edit/retry, optional teacher rewrite, necessity tagging, and publish/draft finalization from the user's confidence threshold. """ try: from services.memory.skills import SkillsManager from src.constants import DATA_DIR from routes.skills_routes import ( _resolve_audit_models, _run_audit_all_job, _skill_audit_jobs, ) if not owner: return "audit_skills requires an owner — refusing to run without scope.", False key = (owner or "",) existing = _skill_audit_jobs.get(key) if existing and existing.get("status") == "running": raise TaskNoop("skill audit already running") sm = SkillsManager(DATA_DIR) skills = sm.load(owner=owner) names = [ s.get("name") for s in skills if s.get("name") and not s.get("audit_verdict") ] if not names: raise TaskNoop("no unaudited skills") url, model, headers, teacher = _resolve_audit_models() try: from src.llm_core import seconds_since_model_activity recent = seconds_since_model_activity(url, model) except Exception: recent = None if recent is not None and recent < (20 * 60): raise TaskDeferred( f"audit model {model} was used {int(recent)}s ago; waiting for quiet window", delay_seconds=20 * 60, ) import time as _time _skill_audit_jobs[key] = { "status": "running", "scope": "scheduled-unchecked", "model": model, "teacher": teacher[1] if teacher else None, "total": len(names), "done": 0, "current": None, "results": [], "log": [ f"Scheduled audit of {len(names)} unaudited skill(s) with {model}" + (f"; teacher {teacher[1]}" if teacher else "") ], "started": _time.time(), "cancel": False, } await _run_audit_all_job(key, sm, names, url, model, headers, teacher, owner) job = _skill_audit_jobs.get(key, {}) counts = {} for r in job.get("results", []): k = r.get("result") or "unknown" counts[k] = counts.get(k, 0) + 1 summary = " · ".join(f"{v} {k}" for k, v in sorted(counts.items())) or "0 results" return f"Audited {job.get('done', 0)}/{len(names)} unaudited skill(s): {summary}", True except TaskNoop: raise except Exception as e: logger.error(f"audit_skills action failed: {e}") return str(e), False async def action_ping_notes(owner: str, **kwargs) -> Tuple[str, bool]: """Background note-due scanner. Fires a reminder for any note whose `due_date` falls in the current ±5-minute window and hasn't been pinged within the last 25 minutes. Mirrors `action_ping_events` for calendar. State (`data/note_pings.json`): {note_id: iso_ts_of_last_ping}. Pruned on each run by dropping entries for notes that are gone/archived/replied. """ try: import json as _json import time as _time from datetime import datetime as _dt, timezone as _tz, timedelta as _td from pathlib import Path as _P from core.database import SessionLocal as _SL, Note as _N # Per-owner state file so cache-pruning doesn't cross-delete other # users' entries (review C4). Legacy path kept as fallback so a # single-user install (empty owner) doesn't lose its history. _owner_slug = "".join(c if (c.isalnum() or c in "-_.@") else "_" for c in (owner or "default")) STATE = _P(DATA_DIR) / f"note_pings_{_owner_slug}.json" STATE.parent.mkdir(parents=True, exist_ok=True) # One-time migration: if legacy global file exists and per-owner file # doesn't, seed from global (entries for OTHER owners still get pruned # on their first run — acceptable, prevents silent loss). _legacy = _P(DATA_DIR) / "note_pings.json" if _legacy.exists() and not STATE.exists(): try: STATE.write_text(_legacy.read_text(encoding="utf-8"), encoding="utf-8") except Exception: pass # Scanner ticks every 60s in _note_pings_loop. 90s window guarantees # every note's due time lands inside at least one tick's window. WINDOW_SEC = 90 REPING_MIN = 25 # don't re-ping same note more often than this def _parse_due(s: str): """Accept '2026-05-29T16:31' (local) or '...Z' (UTC). Returns UTC datetime.""" if not s: return None try: # Handle the JS-style 'Z' suffix. if s.endswith("Z"): return _dt.fromisoformat(s[:-1]).replace(tzinfo=_tz.utc) # Naive → assume local server time. d = _dt.fromisoformat(s) if d.tzinfo is None: d = d.astimezone().astimezone(_tz.utc) return d.astimezone(_tz.utc) except Exception: return None try: cache = _json.loads(STATE.read_text(encoding="utf-8")) if STATE.exists() else {} except Exception: cache = {} db = _SL() try: q = db.query(_N).filter(_N.archived == False) # noqa: E712 q = q.filter(_N.due_date.isnot(None), _N.due_date != "") if owner: # Match owner OR legacy null-owner notes (single-user installs). q = owner_filter(q, _N, owner) notes = q.all() if not notes: raise TaskNoop("no notes with due dates") now = _dt.now(_tz.utc) window = _td(seconds=WINDOW_SEC) reping_cutoff = now - _td(minutes=REPING_MIN) seen_ids = set() sent = [] for n in notes: seen_ids.add(n.id) due = _parse_due(n.due_date) if not due: continue # Inside the ±5min window? if abs((due - now).total_seconds()) > window.total_seconds(): continue # Recently pinged? Skip. last = cache.get(n.id) if last: try: if isinstance(last, dict): last = last.get("at") last_dt = _dt.fromisoformat(str(last)) if last_dt.tzinfo is None: last_dt = last_dt.replace(tzinfo=_tz.utc) if last_dt >= reping_cutoff: continue except Exception: pass # Compose + dispatch. title = (n.title or "Reminder").strip() or "Reminder" body_parts = [] if n.content: body_parts.append(n.content[:400]) # Items: list pending checklist entries inline. if n.items: try: items = _json.loads(n.items) pending = [ it.get("text", "") for it in items if not it.get("done") and not it.get("checked") ] if pending: body_parts.append("Pending:\n" + "\n".join(f"- {t}" for t in pending[:8])) except Exception: pass body = "\n\n".join(p for p in body_parts if p) or title try: from routes.note_routes import dispatch_reminder await dispatch_reminder( title=title, note_body=body, note_id=n.id, owner=n.owner or owner or "", ) cache[n.id] = now.isoformat() sent.append(title) except Exception as e: logger.warning(f"ping_notes: dispatch failed for {n.id}: {e}") # Prune cache entries for notes that no longer exist. for stale in [k for k in cache if k not in seen_ids]: cache.pop(stale, None) try: STATE.write_text(_json.dumps(cache), encoding="utf-8") except Exception as e: logger.warning(f"ping_notes: cache write failed: {e}") if not sent: raise TaskNoop(f"scanned {len(notes)} note(s), none due in ±{WINDOW_SEC}s") preview = "; ".join(sent[:3]) extra = f" (+{len(sent) - 3} more)" if len(sent) > 3 else "" return f"Pinged {len(sent)} note(s): {preview}{extra}", True finally: db.close() except TaskNoop: raise except Exception as e: logger.exception("ping_notes action failed") return str(e), False async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]: """Scan unread emails across all accounts, LLM-triage new ones, cache per-UID verdicts, tag the inbox, and fire a reminder when a previously unseen UID scores reply-soon/urgent (>=2). State persists under data/email_urgency_state_* so the UI can color the unread dot by tier. Design notes: - Only classifies emails newer than 7 days (first-run scale guard). - Cache key = `:` so the same UID across accounts doesn't collide. - Re-notify gate: only when at least one UID NEW to `notified_uids` scores ≥2. Repeat scans where the set is unchanged stay silent. """ from src.settings import load_settings try: settings = load_settings() import json as _json import email as _email_mod import asyncio as _aio import os as _os import re as _re import time as _time import httpx from datetime import datetime as _dt, timedelta as _td from pathlib import Path as _P from core.database import SessionLocal as _SL, EmailAccount as _EA from routes.email_helpers import _imap_connect, _decode_header from src.llm_core import llm_call_async_with_fallback # Per-owner state file so multi-user runs don't clobber each other's # notified_uids / urgency counts. Empty owner falls back to a generic # filename for single-user installs (matches prior behaviour). _owner_slug = "".join(c if (c.isalnum() or c in "-_.@") else "_" for c in (owner or "default")) STATE_PATH = _P(DATA_DIR) / f"email_urgency_state_{_owner_slug}.json" STATE_LOCK_DB = STATE_PATH.with_suffix(".lock.sqlite3") CACHE_DIR = _P(EMAIL_URGENCY_CACHE_DIR) CACHE_DIR.mkdir(parents=True, exist_ok=True) STATE_PATH.parent.mkdir(parents=True, exist_ok=True) AGE_CUTOFF = _dt.utcnow() - _td(days=7) TRIAGE_VERSION = 10 CATEGORY_TAGS = { "bills", "receipt", "travel", "calendar", "action-needed", } VISIBLE_EMAIL_TAGS = CATEGORY_TAGS | {"urgent", "reply-soon"} MANAGED_TAGS = VISIBLE_EMAIL_TAGS | { "newsletter", "marketing", "notification", "finance", "security", "shopping", "social", "work", "personal", "legal", "support", "promo", } # Resolve with the task owner as before, but defer the availability # gate until after authoritative account cleanup. State retirement must # still run when no model is configured. from src.task_endpoint import resolve_task_candidates candidates = resolve_task_candidates(owner=owner) target_account_id = _email_task_account_id(kwargs) # ── 1. Enumerate enabled accounts. Match this task's owner AND fall # back to the legacy "unowned account whose imap_user / from_address # == this owner" pattern — same rule `_get_email_config` uses, so a # pre-multi-user account row still gets picked up for the seeded task. def _enumerate_enabled_accounts(): db = _SL() try: from sqlalchemy import and_ as _and, or_ as _or q = db.query(_EA).filter(_EA.enabled == True) # noqa: E712 if owner: unowned = _or(_EA.owner == None, _EA.owner == "") # noqa: E711 same_mailbox = _or( _EA.imap_user == owner, _EA.from_address == owner, ) q = q.filter( _or(_EA.owner == owner, _and(unowned, same_mailbox)) ) if target_account_id: q = q.filter(_EA.id == target_account_id) return q.all() finally: db.close() initial_accounts = _enumerate_enabled_accounts() initial_account_ids = { str(account.id) for account in initial_accounts } # Register every account before IMAP work, including its first-ever # scan. A concurrent zero-account cleanup can then advance this marker # and fence delivery even before the scan has produced payload. registered_state = None if initial_account_ids: async def _register_accounts(prior): next_state = _merge_email_urgency_state( prior, owner=owner, per_uid_scores={}, notified_uids=prior.get("notified_uids", []), all_unread_keys=set(), fully_scanned_account_ids=set(), base_account_generations=( _email_urgency_account_generations(prior) ), timestamp=_time.time(), known_account_ids=initial_account_ids, ) # Return the exact state committed by registration. This is # the scan's generation token: adopting a later checkpoint # after account cleanup would let the stale scan appear fresh. return next_state, next_state registered_state = await _run_email_urgency_state_transaction( STATE_PATH, STATE_LOCK_DB, _register_accounts, ) # Revalidate after registration. If deletion/disable and its cleanup # completed before the marker was published, this second enumeration # observes the absence and this action retires its own marker instead # of starting IMAP. Accounts newly appearing between the two reads are # left for the next pass rather than scanned without prior registration. verified_accounts = _enumerate_enabled_accounts() enabled_account_ids = { str(account.id) for account in verified_accounts } accounts = [ account for account in verified_accounts if str(account.id) in initial_account_ids ] # Capture the checkpoint basis before cleanup or IMAP. A full # owner-wide enumeration authoritatively retires all known state IDs # absent from the current enabled/visible set. A scoped task may retire # only its selected missing/disabled account. Existing accounts remain # present even if their later network scan fails, so transient IMAP # failure never erases their last known state. base_state = ( registered_state if registered_state is not None else _read_email_urgency_state(STATE_PATH) ) base_account_generations = _email_urgency_account_generations( base_state ) base_payload_account_ids = _email_urgency_payload_account_ids(base_state) known_state_account_ids = _email_urgency_known_account_ids(base_state) if target_account_id: retired_account_ids = ( {str(target_account_id)} if str(target_account_id) not in enabled_account_ids else set() ) else: retired_account_ids = ( known_state_account_ids - enabled_account_ids ) if retired_account_ids: async def _retire_accounts(prior): next_state = _merge_email_urgency_state( prior, owner=owner, per_uid_scores={}, notified_uids=prior.get("notified_uids", []), all_unread_keys=set(), fully_scanned_account_ids=set(), base_account_generations=base_account_generations, timestamp=_time.time(), retired_account_ids=retired_account_ids, base_payload_account_ids=base_payload_account_ids, ) return None, next_state await _run_email_urgency_state_transaction( STATE_PATH, STATE_LOCK_DB, _retire_accounts, ) if not accounts: raise TaskNoop("no email accounts configured") # ── 2. Account retirement above is state maintenance and does not # depend on model availability. Scanning still requires the utility # primary/fallback candidates resolved for this task owner. if not candidates: return "No LLM endpoint available", False urgency_prompt = settings.get("urgent_email_prompt", "") per_uid_scores = {} # key = ":" → {"score": 0-3, "reason": "..."} all_unread_keys = set() llm_attempts = 0 saved_classifications = 0 failed_classifications = [] tag_write_details = [] scanned = 0 fully_scanned_account_ids = set() def _heuristic_email_verdict(item: dict) -> dict: blob = ( f"{item.get('headers','')}\n{item.get('from','')}\n" f"{item.get('subject','')}\n{item.get('body','')}" ).lower() response_tags = [] type_candidates = [] def add_response(tag: str): if tag in CATEGORY_TAGS and tag not in response_tags: response_tags.append(tag) def add_type(tag: str): if tag in CATEGORY_TAGS and tag not in type_candidates: type_candidates.append(tag) bulkish = bool(_re.search( r"\b(list-unsubscribe|list-id|mailchimp|mailchimpapp|view this email in your browser|unsubscribe|newsletter|digest|precedence:\s*bulk)\b", blob, )) marketingish = bool(_re.search( r"\b(advertisement|sponsored|promo|promotion|sale|discount|offer|limited time|deal|coupon|shop now|buy now|membership|rewards?)\b", blob, )) if bulkish or marketingish: add_type("newsletter") if _re.search(r"\b(receipt|order|注文|payment confirmation|delivery|shipment|tracking|お届け|購入)\b", blob): add_type("receipt") if _re.search(r"\b(bill|billing|amount due|overdue|pay by|payment due|subscription could not be renewed)\b", blob): add_type("bills") if _re.search(r"\b(court|charge|legal|lawyer|solicitor|claim|judgment|registration fee|debt)\b", blob): add_type("legal") if _re.search(r"\b(flight|hotel|booking|reservation|itinerary|train|ticket|trip|旅|予約)\b", blob): add_type("travel") if _re.search(r"\b(ticket|case|support|helpdesk|request)\b", blob): add_type("support") if _re.search(r"\b(meeting|appointment|calendar|invite|event|schedule|予定|保育園|連絡帳)\b", blob): add_response("calendar") if _re.search( r"\b(action required|required action|please reply|please respond|deadline|by \d{1,2} |pay within|submit|sign|confirm|approval|waiting outside|locked out|can't get in|cannot get in|invoice|bill|billing|payment|balance|debt|subscription|renewal|overdue|amount due|court|charge|legal|lawyer|solicitor|claim|judgment)\b", blob, ): add_response("action-needed") type_priority = ("bills", "receipt", "travel") tags = [*response_tags] for type_tag in type_priority: if type_tag in type_candidates and type_tag not in tags: tags.append(type_tag) if len(tags) >= len(response_tags) + 2: break score = 0 reason = "categorized by email metadata" if "action-needed" in response_tags: score = 2 reason = "action likely needed" if _re.search(r"\b(urgent|immediately|final notice|locked out|waiting outside|can't get in|cannot get in)\b", blob): score = 3 reason = "urgent wording" if (bulkish or marketingish) and score < 2: score = 0 reason = "bulk marketing/newsletter" _from_raw = item.get("from", "") or "" if "<" in _from_raw: _from_short = _from_raw.split("<", 1)[0].strip().strip('"') or _from_raw else: _from_short = _from_raw return { "score": max(0, min(3, score)), "tags": tags[:4], "spam": False, "reason": reason, "subject": (item.get("subject") or "")[:200], "from": _from_short[:120], "triage_version": TRIAGE_VERSION, "message_id": (item.get("message_id") or "").strip(), "unread": bool(item.get("unread")), "ts": _time.time(), } # ── 3. Per-account scan: pull headers + lightweight body for new UIDs # since 7 days ago, score via LLM, cache the verdict. for acc in accounts: cache_file = CACHE_DIR / f"{acc.id}.json" try: cache = _json.loads(cache_file.read_text(encoding="utf-8")) if cache_file.exists() else {"uids": {}} except Exception: cache = {"uids": {}} def _scan_one(account=acc, cache_uids=cache.get("uids", {})): """Sync IMAP work runs in a thread.""" results = [] scan_complete = True conn = _imap_connect(account.id) try: select_status, _select_data = conn.select("INBOX", readonly=True) if select_status != "OK": return results, False # Tag recent inbox mail, not only unread mail. Urgency # reminders below still only notify for unread messages. since_str = AGE_CUTOFF.strftime("%d-%b-%Y") status, data = conn.uid("SEARCH", None, f'(SINCE {since_str})') if status != "OK": return results, False if not data or not data[0]: return results, True matching_uids = data[0].split() if len(matching_uids) > 30: # The scale guard deliberately processes only the most # recent 30. That is a partial account snapshot, so it # cannot justify pruning older checkpoint facts. scan_complete = False uids = matching_uids[-30:] for uid_b in uids: uid = uid_b.decode() if isinstance(uid_b, bytes) else str(uid_b) key = f"{account.id}:{uid}" cached = cache_uids.get(uid) cached_ok = isinstance(cached, dict) and cached.get("triage_version") == TRIAGE_VERSION results.append({"key": key, "uid": uid, "cached": cached if cached_ok else None}) if cached_ok: # Cached verdicts still need a lightweight FLAGS # refresh. Without it a cached unread message looks # read and its successful notification checkpoint # is pruned on the next pass. try: st, flag_data = conn.uid("FETCH", uid_b, "(UID FLAGS)") if st != "OK" or not flag_data: scan_complete = False results.pop() continue flag_parts = [] for part in flag_data: if isinstance(part, (bytes, bytearray)): flag_parts.append(bytes(part)) elif ( isinstance(part, tuple) and part and isinstance(part[0], (bytes, bytearray)) ): flag_parts.append(bytes(part[0])) flags_blob = b" ".join(flag_parts) results[-1]["unread"] = b"\\Seen" not in flags_blob except Exception as _fe: scan_complete = False results.pop() logger.debug( f"urgency: flag fetch for uid {uid} failed: {_fe}" ) continue # Pull headers + first ~800 chars of plaintext body. try: st, msg_data = conn.uid("FETCH", uid_b, "(UID FLAGS RFC822.HEADER BODY.PEEK[TEXT]<0.800>)") if st != "OK" or not msg_data: scan_complete = False results.pop() continue flags_blob = b" ".join( part[0] for part in msg_data if isinstance(part, tuple) and part and isinstance(part[0], (bytes, bytearray)) ) is_unread = b"\\Seen" not in flags_blob # Headers + body land in different tuples in the # response — concatenate the bytes for parsing. raw = b"" for part in msg_data: if isinstance(part, tuple) and part[1]: raw += part[1] + b"\n\n" if not raw: scan_complete = False results.pop() continue msg = _email_mod.message_from_bytes(raw) # Skip Odysseus-generated reminders so the scanner # doesn't classify its own emails as urgent and # trigger a feedback loop. Match on either the # stamped headers OR the subject prefix. _ody_origin = (msg.get("X-Odysseus-Origin") or "").strip().lower() _ody_kind = (msg.get("X-Odysseus-Kind") or "").strip().lower() _raw_subj = (msg.get("Subject") or "").lower() # MCP path drops custom headers (email_server's # schema doesn't accept them), so we ALSO match the # `[Task]` subject prefix that `_deliver_via_mcp` # always stamps. Anything that looks self-generated # is dropped before classification to prevent the # scanner from labelling its own emails "urgent". if (_ody_origin == "odysseus-ui" or _ody_kind == "reminder" or _raw_subj.startswith("reminder (odysseus):") or _raw_subj.startswith("reminder:") or _raw_subj.startswith("[task]")): # Drop this candidate entirely — don't list it # in results so its UID never enters the cache # nor counts toward `scanned`. results.pop() continue subject = _decode_header(msg.get("Subject") or "") from_raw = _decode_header(msg.get("From") or "") header_blob = "\n".join( f"{name}: {msg.get(name, '')}" for name in ( "From", "Subject", "List-Unsubscribe", "List-ID", "Precedence", "X-Mailchimp-Campaign-Id", "X-Campaign", "X-MC-User", ) if msg.get(name) ) body_snippet = "" try: if msg.is_multipart(): for part in msg.walk(): if part.get_content_type() == "text/plain": body_snippet = part.get_payload(decode=True).decode("utf-8", errors="ignore")[:1600] break else: body_snippet = (msg.get_payload(decode=True) or b"").decode("utf-8", errors="ignore")[:1600] except Exception: body_snippet = "" results[-1].update({ "subject": subject, "from": from_raw, "headers": header_blob, "body": body_snippet.strip(), "message_id": (msg.get("Message-ID") or "").strip(), "unread": is_unread, }) except Exception as _fe: scan_complete = False results.pop() logger.debug(f"urgency: header fetch for uid {uid} failed: {_fe}") finally: try: conn.logout() except Exception: pass return results, scan_complete try: items, scan_complete = await _aio.to_thread(_scan_one) except Exception as e: logger.warning(f"urgency: IMAP scan failed for account {acc.id}: {e}") continue if scan_complete: fully_scanned_account_ids.add(str(acc.id)) for item in items: scanned += 1 key = item["key"] if item.get("unread"): all_unread_keys.add(key) if item.get("cached"): cached_v = dict(item["cached"]) cached_v["unread"] = bool(item.get("unread")) per_uid_scores[key] = cached_v continue # Skip uids we couldn't fetch (no subject/from/body). if not item.get("subject") and not item.get("from"): continue verdict = _heuristic_email_verdict(item) cache.setdefault("uids", {})[item["uid"]] = verdict per_uid_scores[key] = verdict saved_classifications += 1 continue # ── LLM-classify. JSON-only response; bullet-proof parse. llm_attempts += 1 prompt = ( "You are triaging ONE email. Return ONLY JSON: " "{\"score\":0|1|2|3,\"tags\":[\"...\"],\"spam\":false," "\"reason\":\"one short phrase\"}.\n" "0 = trivial / promotional · 1 = informational, no reply needed · " "2 = should reply within a day · 3 = urgent, reply now (deadline, blocker).\n\n" "Allowed visible tags: urgent, reply-soon, action-needed, calendar, bills, receipt, travel.\n" "Use action-needed when the user likely needs to reply, pay, sign, book, or decide. " "Use bills for bills or debts, receipt for purchases/deliveries, travel for reservations/trips, " "and calendar only when a calendar event/reminder is involved. spam=true for scams, phishing, " "junk, cold sales, generic ads, or no-personal-action bulk mail.\n" "Important: 'I'm outside', 'I am outside', 'waiting outside', 'at the door', " "'locked out', or 'can't get in' means score 3 unless clearly historical.\n\n" f"User's rules:\n{urgency_prompt}\n\n" f"Email:\nFrom: {item.get('from','')}\nSubject: {item.get('subject','')}\n" f"Snippet:\n{item.get('body','')}\n" ) try: await wait_for_interactive_quiet("email urgency action") raw = await llm_call_async_with_fallback( candidates, [{"role": "user", "content": prompt}], temperature=0.1, max_tokens=220, timeout=30, ) # Tolerant JSON-parse: strip code fences if present. txt = (raw or "").strip() if txt.startswith("```"): txt = txt.strip("`") # Drop a leading "json\n" or any tag. nl = txt.find("\n") if nl >= 0: txt = txt[nl + 1:] # Find first { ... } in the response. s = txt.find("{") e = txt.rfind("}") if s < 0 or e <= s: failed_classifications.append({ "subject": item.get("subject") or "(no subject)", "from": item.get("from") or "", "reason": "model returned no JSON", }) continue obj = _json.loads(txt[s:e + 1]) score = int(obj.get("score", 0)) reason = str(obj.get("reason", ""))[:200] raw_tags = obj.get("tags") or [] if isinstance(raw_tags, str): raw_tags = [raw_tags] tags = [] for t in raw_tags: if not isinstance(t, str): continue tag = t.strip().lower().replace("_", "-") if tag == "promo": tag = "marketing" if tag in CATEGORY_TAGS and tag not in tags: tags.append(tag) _spam_raw = obj.get("spam") if isinstance(_spam_raw, bool): spam = _spam_raw elif isinstance(_spam_raw, (int, float)): spam = bool(_spam_raw) else: spam = str(_spam_raw or "").strip().lower() in {"1", "true", "yes", "y"} _blob = f"{item.get('headers','')}\n{item.get('subject','')}\n{item.get('body','')}".lower() if _re.search(r"\b(i'?m|i am|im|we'?re|we are)\s+outside\b", _blob) or _re.search( r"\b(waiting outside|at the door|locked out|can'?t get in|cannot get in)\b", _blob ): if score < 3: reason = "person is waiting outside" score = max(score, 3) bulkish = bool(_re.search( r"\b(list-unsubscribe|list-id|mailchimp|mailchimpapp|view this email in your browser|unsubscribe|newsletter|digest|precedence:\s*bulk)\b", _blob, )) marketingish = bool(_re.search( r"\b(advertisement|sponsored|promo|promotion|sale|discount|offer|limited time|deal|tickets?|tour|merch|stream|purchase|sold out|low tickets|coupon|shop now|buy now)\b", _blob, )) if (bulkish or marketingish) and score < 2: score = 0 if not reason or "urgent" in reason.lower(): reason = "bulk mail; no personal reply needed" # Strip "Name " to bare display name for compact summary. _from_raw = item.get("from", "") or "" if "<" in _from_raw: _from_short = _from_raw.split("<", 1)[0].strip().strip('"') or _from_raw else: _from_short = _from_raw verdict = { "score": max(0, min(3, score)), "tags": tags[:4], "spam": spam, "reason": reason, "subject": (item.get("subject") or "")[:200], "from": _from_short[:120], "triage_version": TRIAGE_VERSION, # Cache the message_id too so re-scans of already-cached # UIDs can still write the inbox tag without re-LLM'ing. "message_id": (item.get("message_id") or "").strip(), "unread": bool(item.get("unread")), "ts": _time.time(), } cache.setdefault("uids", {})[item["uid"]] = verdict per_uid_scores[key] = verdict saved_classifications += 1 except Exception as e: failed_classifications.append({ "subject": item.get("subject") or "(no subject)", "from": item.get("from") or "", "reason": str(e)[:120] or "classification failed", }) logger.debug(f"urgency: LLM classify failed for {key}: {e}") continue if scan_complete: # Only a complete account scan proves a cached UID left the # recent window. Partial/failing scans preserve prior facts. seen_uids = {it["uid"] for it in items} cache_uids = cache.get("uids", {}) for stale in [u for u in cache_uids if u not in seen_uids]: cache_uids.pop(stale, None) try: cache_file.write_text(_json.dumps(cache), encoding="utf-8") except Exception as e: logger.warning(f"urgency: cache write failed for {acc.id}: {e}") # ── 3.5 Mirror triage verdicts into email_tags so inbox filters and # pills show urgency + category tags. Runs for BOTH cached and freshly # classified items; message_id lives on the cached verdict so this is cheap. try: import sqlite3 as _sql3 from routes.email_helpers import SCHEDULED_DB, _init_scheduled_db from datetime import datetime as _dt2 _init_scheduled_db() _conn = _sql3.connect(SCHEDULED_DB) try: for _key, _v in per_uid_scores.items(): _msg_id = (_v.get("message_id") or "").strip() _score = _v.get("score", 0) if not _msg_id: continue _new_tags = [] if _score >= 3: _new_tags.append("urgent") elif _score >= 2: _new_tags.append("reply-soon") for _tag in (_v.get("tags") or []): _tag = str(_tag).strip().lower().replace("_", "-") if _tag == "promo": _tag = "marketing" if _tag == "action-needed" and any(t in _new_tags for t in ("urgent", "reply-soon")): continue if _tag in VISIBLE_EMAIL_TAGS and _tag not in _new_tags: _new_tags.append(_tag) _spam = 1 if _v.get("spam") else 0 # _key is ":" — extract uid for the row. _acc_id, _uid_only = (_key.split(":", 1) + [""])[:2] _owner_key = owner or "" _row = _conn.execute( "SELECT tags FROM email_tags WHERE message_id=? AND owner=? AND account_id=?", (_msg_id, _owner_key, _acc_id), ).fetchone() if _row: try: _existing = _json.loads(_row[0] or "[]") if not isinstance(_existing, list): _existing = [] except Exception: _existing = [] # Drop previous triage-owned tags so re-classification # can upgrade/downgrade/clear without touching manual tags. _existing = [ str(t).strip().lower().replace("_", "-") for t in _existing if str(t).strip().lower().replace("_", "-") not in MANAGED_TAGS ] for _tag in _new_tags: if _tag not in _existing: _existing.append(_tag) if _new_tags or _spam: tag_write_details.append({ "uid": _uid_only, "subject": _v.get("subject", ""), "from": _v.get("from", ""), "tags": list(_new_tags), "spam": _spam, "reason": _v.get("reason", ""), "updated": True, }) _conn.execute( "UPDATE email_tags SET tags=?, spam_verdict=?, spam_reason=?, uid=?, folder=?, subject=?, sender=? " "WHERE message_id=? AND owner=? AND account_id=?", (_json.dumps(_existing), _spam, _v.get("reason", ""), _uid_only, "INBOX", _v.get("subject", ""), _v.get("from", ""), _msg_id, _owner_key, _acc_id), ) else: if not _new_tags and not _spam: continue _conn.execute( "INSERT INTO email_tags " "(message_id, owner, account_id, uid, folder, subject, sender, tags, spam_verdict, spam_reason, created_at) " "VALUES (?, ?, ?, ?, 'INBOX', ?, ?, ?, ?, ?, ?)", (_msg_id, _owner_key, _acc_id, _uid_only, _v.get("subject", ""), _v.get("from", ""), _json.dumps(_new_tags), _spam, _v.get("reason", ""), _dt2.utcnow().isoformat()), ) tag_write_details.append({ "uid": _uid_only, "subject": _v.get("subject", ""), "from": _v.get("from", ""), "tags": list(_new_tags), "spam": _spam, "reason": _v.get("reason", ""), "updated": False, }) _conn.commit() finally: _conn.close() except Exception as _te: logger.warning(f"urgency: bulk tag write failed: {_te}") # ── 4. Aggregate state. urgent = score ≥ 2. urgent_keys = [k for k, v in per_uid_scores.items() if v.get("score", 0) >= 2 and v.get("unread")] # ── 5. Fire a reminder only when a previously-unnotified UID scores # urgent. The read, decision, delivery, and checkpoint are serialized # below so two scheduler workers cannot both act on the same stale # state or overwrite each other's successful checkpoint. newly_notified = set() notify_failed = set() def _urgency_reminder_payload(reminder_keys): total = len(reminder_keys) title = "Urgent email" if total == 1 else f"{total} urgent emails" sorted_urgent = sorted( ((key, per_uid_scores[key]) for key in reminder_keys), key=lambda item: item[1].get("score", 0), reverse=True, )[:10] _pub = (settings.get("app_public_url") or "").strip().rstrip("/") from urllib.parse import quote as _quote lines = [ f"{total} email" + ("" if total == 1 else "s") + " need an urgent reply:", "", ] for i, (key, value) in enumerate(sorted_urgent, 1): subj = (value.get("subject") or "(no subject)")[:160] frm = value.get("from") or "" why = value.get("reason") or "" uid_for_link = str(key).split(":", 1)[-1] hash_link = f"#email={_quote('INBOX', safe='')}:{uid_for_link}" open_link = f"{_pub}/{hash_link}" if _pub else hash_link line = f"{i}. {subj}" if frm: line += f" — {frm}" if why: line += f" · {why}" lines.append(line) lines.append(f" Open email: {open_link}") if total > len(sorted_urgent): lines.append("") lines.append(f"…and {total - len(sorted_urgent)} more.") return title, "\n".join(lines) async def _dispatch_urgency_reminder(reminder_keys): # Call dispatch_reminder directly: a scheduler has no browser # session cookie with which to call the HTTP endpoint. from routes.note_routes import dispatch_reminder title, body = _urgency_reminder_payload(reminder_keys) return await dispatch_reminder( title=title, note_body=body, note_id="urgent-email", owner=owner or "", ) async def _dispatch_and_checkpoint(prior): notified_uids = _email_urgency_string_set( prior.get("notified_uids", []) ) observed_accounts = { _email_urgency_account_key(key) for key in per_uid_scores } | fully_scanned_account_ids stale_accounts = _email_urgency_stale_accounts( prior, base_account_generations, observed_accounts, ) # Generation fencing must happen before delivery, not only during # merge. A stale-only unread UID may have been removed, read, or # downgraded by the newer completed scan. deliverable_urgent = [ key for key in urgent_keys if _email_urgency_account_key(key) not in stale_accounts ] new_urgent = [ key for key in deliverable_urgent if key not in notified_uids ] if new_urgent: try: dispatch_result = await _dispatch_urgency_reminder( deliverable_urgent ) channel = (settings.get("reminder_channel") or "browser").strip().lower() delivered = bool(dispatch_result.get("browser_sent")) if channel == "email": delivered = bool(dispatch_result.get("email_sent")) elif channel == "ntfy": delivered = bool(dispatch_result.get("ntfy_sent")) elif channel == "webhook": delivered = bool(dispatch_result.get("webhook_sent")) if delivered: newly_notified.update(new_urgent) notified_uids.update(new_urgent) else: notify_failed.update(new_urgent) logger.warning( "urgency: reminder dispatch returned no successful " f"delivery path: {dispatch_result}" ) except Exception as e: logger.warning(f"urgency: reminder dispatch failed: {e}") notify_failed.update(new_urgent) next_state = _merge_email_urgency_state( prior, owner=owner, per_uid_scores=per_uid_scores, notified_uids=notified_uids, all_unread_keys=all_unread_keys, fully_scanned_account_ids=fully_scanned_account_ids, base_account_generations=base_account_generations, timestamp=_time.time(), ) return notified_uids, next_state try: await _run_email_urgency_state_transaction( STATE_PATH, STATE_LOCK_DB, _dispatch_and_checkpoint, ) except Exception as e: logger.warning(f"urgency: state transaction failed: {e}") # ── 6. Activity-log summary — counts line on top, then per-tier # bulleted breakdown so the user can see WHICH emails ranked where # (subject · sender · reason) and which ones triggered notifications. tier_counts = {0: 0, 1: 0, 2: 0, 3: 0} for v in per_uid_scores.values(): tier_counts[v.get("score", 0)] = tier_counts.get(v.get("score", 0), 0) + 1 if scanned == 0: raise TaskNoop("no unread emails in last 7 days") head = ( f"scanned {scanned} · urgent {tier_counts[3]} · " f"reply-soon {tier_counts[2]} · info {tier_counts[1]} · trivial {tier_counts[0]} · " f"{saved_classifications} saved classifications" ) if failed_classifications: head += f" · {len(failed_classifications)} failed" if newly_notified: head += f" · notified {len(newly_notified)}" if notify_failed: head += f" · notify failed {len(notify_failed)}" def _fmt_tag_write(v): subj = (v.get("subject") or "(no subject)")[:80] frm = v.get("from") or "" tags = list(v.get("tags") or []) if v.get("spam"): tags.append("spam") tag_txt = ", ".join(tags) if tags else "cleared managed tags" why = v.get("reason") or "" op = "updated" if v.get("updated") else "created" line = f"- **{subj}**" + (f" — _{frm}_" if frm else "") line += f" — `{tag_txt}` ({op})" if why: line += f" · {why}" return line def _fmt_one(v, newly_notified_set, failed_set, key): subj = (v.get("subject") or "(no subject)")[:80] frm = v.get("from") or "" why = v.get("reason") or "" tag = " · *notified now*" if key in newly_notified_set else (" · *notify failed*" if key in failed_set else "") line = f"- **{subj}**" + (f" — _{frm}_" if frm else "") if why: line += f" — {why}" return line + tag # Sort each tier by reason length (longest reason first → most info). by_tier = {3: [], 2: [], 1: [], 0: []} for k, v in per_uid_scores.items(): by_tier.setdefault(v.get("score", 0), []).append((k, v)) lines = [head] if tag_write_details: lines.append("") lines.append(f"**Applied tags ({len(tag_write_details)}):**") for v in tag_write_details[:16]: lines.append(_fmt_tag_write(v)) if len(tag_write_details) > 16: lines.append(f"…and {len(tag_write_details) - 16} more") tier_labels = {3: "Urgent", 2: "Reply soon", 1: "Informational", 0: "Trivial"} for tier in (3, 2, 1, 0): items_t = by_tier.get(tier, []) if not items_t: continue lines.append("") lines.append(f"**{tier_labels[tier]} ({len(items_t)}):**") # Cap each tier at 8 rows to keep the activity entry readable. for k, v in items_t[:8]: lines.append(_fmt_one(v, newly_notified, notify_failed, k)) if len(items_t) > 8: lines.append(f"…and {len(items_t) - 8} more") if failed_classifications: lines.append("") lines.append(f"**Unclassified ({len(failed_classifications)}):**") for v in failed_classifications[:8]: subj = (v.get("subject") or "(no subject)")[:80] frm = v.get("from") or "" why = v.get("reason") or "" line = f"- **{subj}**" + (f" — _{frm}_" if frm else "") if why: line += f" — {why}" lines.append(line) if len(failed_classifications) > 8: lines.append(f"…and {len(failed_classifications) - 8} more") return "\n".join(lines), True except TaskNoop: raise except Exception as e: logger.exception("check_email_urgency action failed") return str(e), False async def action_cookbook_serve( owner: str, task_name: str = "", progress_cb=None, command: str = "", **kwargs, ) -> Tuple[str, bool]: """Launch a Cookbook model serve as a scheduled task. `command` is the JSON config string the task carries in `prompt`, of shape: {"preset": "name"} OR {"repo_id": "...", "cmd": "...", "host": "..."}. Optional `end_after_min: N` schedules a hard-stop N minutes after launch (handled by cookbook_serve_lifecycle_loop in src/cookbook_serve_lifecycle.py). """ import json import time as _time import httpx from pathlib import Path from core.middleware import INTERNAL_TOOL_HEADER, INTERNAL_TOOL_TOKEN from core.atomic_io import atomic_write_json headers = {INTERNAL_TOOL_HEADER: INTERNAL_TOOL_TOKEN} try: cfg = json.loads(command or "{}") except Exception: return f"Invalid JSON config: {command!r}", False if not isinstance(cfg, dict): return "Config must be a JSON object", False # Resolve the preset (if named) OR fall through with explicit fields. preset_name = (cfg.get("preset") or "").strip() repo_id = (cfg.get("repo_id") or "").strip() cmd = (cfg.get("cmd") or "").strip() host = (cfg.get("host") or cfg.get("remote_host") or "").strip() try: end_after_min = int(cfg.get("end_after_min") or 0) except Exception: end_after_min = 0 set_default = bool(cfg.get("set_default", True)) state_path = Path(COOKBOOK_STATE_FILE) try: state = json.loads(state_path.read_text(encoding="utf-8")) if state_path.exists() else {} except Exception: state = {} # Preset lookup. Try three matching strategies in order so the # schedule still works even when the user's preset is named # differently from the model's short name: # # 1. Exact preset.name == preset_name (case-insensitive) # 2. preset.model / preset.modelId == repo_id (caller knows the repo) # 3. preset.model's short name (after final /) == preset_name # # Without #2 and #3, scheduling "Qwen3.5-397B-A17B-AWQ" failed when # the saved preset was named "vllm-qwen-397b" or had the model field # populated with the full HF repo path. Either should resolve. def _short(name: str) -> str: return (name or "").rsplit("/", 1)[-1].lower() if not cmd or not repo_id: presets = state.get("presets") or [] chosen = None # Strategy 1: exact name match. if preset_name: chosen = next( (p for p in presets if isinstance(p, dict) and (p.get("name") or "").lower() == preset_name.lower()), None, ) # Strategy 2: repo_id matches the preset's model field. if chosen is None and repo_id: chosen = next( (p for p in presets if isinstance(p, dict) and (p.get("model") or p.get("modelId") or "").lower() == repo_id.lower()), None, ) # Strategy 3: model's short name matches the preset_name. if chosen is None and preset_name: chosen = next( (p for p in presets if isinstance(p, dict) and _short(p.get("model") or p.get("modelId") or "") == preset_name.lower()), None, ) if chosen is not None: repo_id = repo_id or chosen.get("model") or chosen.get("modelId") or "" cmd = cmd or (chosen.get("cmd") or "").strip() host = host or chosen.get("host") or chosen.get("remoteHost") or "" if not repo_id or not cmd or cmd.startswith("(adopted"): # Surface what we tried so the user can name their preset to match. preset_names = [(p.get("name") or "") for p in (state.get("presets") or []) if isinstance(p, dict)] hint = f" Saved presets: {preset_names!r}" if preset_names else "" return (f"No launchable config for {preset_name!r} (repo_id={repo_id!r}). " f"Check Cookbook → Presets has a real cmd, not 'adopted'.{hint}", False) # Resolve env_prefix etc. from the host's saved cookbook server entry, # matching the chat agent's serve_model path. body = {"repo_id": repo_id, "cmd": cmd} if host: body["remote_host"] = host env = (state.get("env") or {}) srv = next( (s for s in (env.get("servers") or []) if isinstance(s, dict) and (s.get("host") == host or s.get("name") == host)), {}, ) if srv.get("env") == "venv" and srv.get("envPath"): body["env_prefix"] = f"source {srv['envPath']}/bin/activate" elif srv.get("env") == "conda" and srv.get("envPath"): body["env_prefix"] = f"conda activate {srv['envPath']}" if srv.get("hfToken"): body["hf_token"] = srv["hfToken"] if srv.get("port"): body["ssh_port"] = str(srv["port"]) if srv.get("platform"): body["platform"] = srv["platform"] try: async with httpx.AsyncClient(timeout=30) as client: r = await client.post(f"{internal_api_base()}/api/model/serve", json=body, headers=headers) data = r.json() if r.content else {} except Exception as e: return f"Launch HTTP failed: {e}", False if not data.get("ok"): return f"Launch rejected: {data.get('error') or data.get('detail') or 'unknown'}", False sid = data.get("session_id") or "" endpoint_id = data.get("endpoint_id") or "" # Scheduled serves are usually meant to become the active local model for # chat/tools while their time window is open. Persist both endpoint and # model so task/utility/default resolution does not keep routing to a stale # API fallback. Allow explicit opt-out with {"set_default": false}. if endpoint_id and set_default: try: selected_model = repo_id try: from core.database import SessionLocal as _SL, ModelEndpoint as _ME _db = _SL() try: _ep = _db.query(_ME).filter(_ME.id == endpoint_id).first() if _ep and _ep.cached_models: _models = json.loads(_ep.cached_models or "[]") if isinstance(_models, list) and _models: selected_model = str(_models[0]) finally: _db.close() except Exception: pass from src.settings import load_settings as _load_settings, save_settings as _save_settings _settings = _load_settings() _settings["default_endpoint_id"] = endpoint_id _settings["default_model"] = selected_model # Keep background tasks aligned unless the user explicitly chose a # separate task model. if not (_settings.get("task_endpoint_id") or "").strip(): _settings["task_endpoint_id"] = endpoint_id _settings["task_model"] = selected_model if not (_settings.get("utility_endpoint_id") or "").strip(): _settings["utility_endpoint_id"] = endpoint_id _settings["utility_model"] = selected_model _save_settings(_settings) if owner: from routes.prefs_routes import _load_for_user, _save_for_user _prefs = _load_for_user(owner) _prefs["default_endpoint_id"] = endpoint_id _prefs["default_model"] = selected_model if not (_prefs.get("utility_endpoint_id") or "").strip(): _prefs["utility_endpoint_id"] = endpoint_id _prefs["utility_model"] = selected_model _save_for_user(owner, _prefs) except Exception as e: logger.warning(f"cookbook_serve: default endpoint update failed: {e}") # Register the new task in cookbook_state.json + stamp it with our # scheduler-owner markers. /api/model/serve spawns the tmux session # but leaves the state-write to the UI — when a scheduled action # launches a serve from server-side, NOBODY writes the task into # state, so the Cookbook tab never shows it. We do the write here. if sid: try: # Re-read fresh (the route may have updated state already). try: fresh = json.loads(state_path.read_text(encoding="utf-8")) except Exception: fresh = {} if not isinstance(fresh, dict): fresh = {} tasks = fresh.get("tasks") if isinstance(fresh.get("tasks"), list) else [] existing = next( (t for t in tasks if isinstance(t, dict) and t.get("sessionId") == sid), None, ) if existing is None: display_name = repo_id.split("/")[-1] if "/" in repo_id else repo_id ssh_port = str(srv.get("port") or cfg.get("ssh_port") or "") platform = str(srv.get("platform") or cfg.get("platform") or "linux") placeholder = ( f"Launched by scheduled task {task_name!r} — waiting for tmux output…\n" f" session: {sid}\n" f" target: {host or 'local'}\n" f" cmd: {cmd[:200]}{'…' if len(cmd) > 200 else ''}" ) existing = { "id": sid, "sessionId": sid, "name": display_name, "modelId": repo_id, "type": "serve", "status": "running", "output": placeholder, "ts": int(_time.time() * 1000), "payload": {"repo_id": repo_id, "remote_host": host or "", "_cmd": cmd}, "remoteHost": host or "", "sshPort": ssh_port or "", "platform": platform or "linux", "_serveReady": False, "_endpointAdded": bool(endpoint_id), } tasks.append(existing) # Stamp ownership + end-at on the task entry. existing["_scheduledByTask"] = task_name or "" existing["_scheduledByOwner"] = owner or "" if endpoint_id: existing["_endpointId"] = endpoint_id existing["endpointId"] = endpoint_id existing["_endpointAdded"] = True if end_after_min > 0: existing["_scheduledStopAtMs"] = int(_time.time() * 1000) + end_after_min * 60 * 1000 fresh["tasks"] = tasks atomic_write_json(state_path, fresh) except Exception as e: logger.warning(f"cookbook_serve: state register/stamp failed: {e}") # Don't try to render absolute clock time in the message — the # server runs in UTC (Docker default), the user reads it as local, # and the offset depends on the user's TZ which the action doesn't # have a reliable handle on. The Tasks UI already shows the RUN # timestamp in the user's local time right above this message, so # "stops 8 min after that" gives the user everything they need. if end_after_min: return ( f"Launched {repo_id} (session {sid}); stops {end_after_min} min after this ran", True, ) return f"Launched {repo_id} (session {sid})", True BUILTIN_ACTIONS = { "tidy_sessions": action_tidy_sessions, "tidy_documents": action_tidy_documents, "consolidate_memory": action_consolidate_memory, "tidy_research": action_tidy_research, "summarize_emails": action_summarize_emails, "draft_email_replies": action_draft_email_replies, "email_auto_translate": action_email_auto_translate, "extract_email_events": action_extract_email_events, "classify_events": action_classify_events, # ping_events removed from the user-facing registry. Calendar reminders # are represented as Notes, so note pings are the single dispatch path. "daily_brief": action_daily_brief, "learn_sender_signatures": action_learn_sender_signatures, "ssh_command": action_ssh_command, "run_script": action_run_script, "run_local": action_run_local, "test_skills": action_test_skills, "audit_skills": action_audit_skills, "check_email_urgency": action_check_email_urgency, "cookbook_serve": action_cookbook_serve, # ping_notes removed from the registry — runs only inside `_note_pings_loop`. } # Descriptions for the UI/API BUILTIN_ACTION_INFO = { "tidy_sessions": "Clean up empty chat sessions and auto-sort into folders", "tidy_documents": "Remove junk/empty documents", "consolidate_memory": "Remove duplicate memories", "tidy_research": "Remove orphaned research files (sessions that were deleted)", "summarize_emails": "Pre-generate AI summaries for new inbox emails", "draft_email_replies": "Pre-draft AI reply suggestions for new inbox emails", "email_auto_translate": "Detect foreign-language emails and cache translated text for the email reader", "extract_email_events": "Scan emails for booking/meeting confirmations and auto-add to calendar", "classify_events": "Tag upcoming events with importance (low/normal/high/critical) and type (work/health/travel/etc.); colors them too", "daily_brief": "Build a morning digest: today's calendar, unread email count + top senders, active todos", "learn_sender_signatures": "LLM learns each sender's signature from 3+ of their recent emails; cached per address so future renders fold sigs reliably without heuristics", "ssh_command": "Run a shell command on a local or remote host", "run_script": "Run a script locally or on ODYSSEUS_SCRIPT_HOST", "test_skills": "Run the per-skill Test on every skill: agent run + LLM judge → records verdict on the skill (pass/needs_work/fail/inconclusive). Advisory only — never rewrites or demotes anything.", "audit_skills": "Audit unaudited skills after enough new skills are added: test, narrow metadata, self-edit/retry, optional teacher rewrite, tag duplicates/trivial skills, and publish/draft using the auto-approve threshold.", "check_email_urgency": "Scan unread emails hourly, tag urgent/reply-soon/newsletter/marketing/spam, and send a reminder when a new email needs a fast reply.", }