aaron
fda61ad622
Claude can now write docx or pdf files to Aaron's Nextcloud Drafts/ when he asks for a document (bio, cover letter, statement, CV section) rather than chat text. Pandoc handles markdown -> docx and markdown -> pdf with the xelatex engine. Upload is a WebDAV PUT against the same Nextcloud instance dream.py already uses; NEXTCLOUD_URL / NEXTCLOUD_USER / NEXTCLOUD_PASSWORD in .env are reused. MKCOL ensures Drafts/ exists; PROPFIND-based collision check appends _2, _3, ... until unique. Filename sanitization strips path components and unsafe characters. System prompt instructs Claude to call save_document when the user wants a file (not chat text) and not to duplicate the file contents in the chat response — just write the file and tell Aaron where it landed. ingest.py and watcher.py now skip files under Drafts/ at ingest time so generated drafts don't pollute future retrieval. Drafts can still be opened, edited, and shipped; they just don't become part of the searchable corpus unless Aaron explicitly moves them out of Drafts/.
1690 lines
65 KiB
Python
1690 lines
65 KiB
Python
import os
|
|
import re
|
|
import json
|
|
import sqlite3
|
|
import subprocess
|
|
import hashlib
|
|
import requests
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
from dotenv import load_dotenv
|
|
from sentence_transformers import SentenceTransformer, CrossEncoder
|
|
import anthropic
|
|
from fastapi import FastAPI, Request, Response, Depends, HTTPException, BackgroundTasks
|
|
import psycopg2
|
|
import psycopg2.extras
|
|
from fastapi import UploadFile, File, Form
|
|
import tempfile
|
|
import os
|
|
try:
|
|
from faster_whisper import WhisperModel
|
|
HAS_WHISPER = True
|
|
except ImportError:
|
|
HAS_WHISPER = False
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
import secrets
|
|
import hashlib
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import uvicorn
|
|
import asyncio
|
|
from fastapi.responses import StreamingResponse
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
|
|
from encoding import extract_text as encoding_extract_text
|
|
from ingest import ingest_directory
|
|
|
|
load_dotenv(Path.home() / "aaronai" / ".env")
|
|
|
|
MEMORY_PATH = Path.home() / "aaronai" / "memory.md"
|
|
CONVERSATIONS_DB = str(Path.home() / "aaronai" / "conversations.db")
|
|
|
|
def _connect(path):
|
|
conn = sqlite3.connect(path, timeout=5.0)
|
|
conn.execute("PRAGMA synchronous=NORMAL")
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
return conn
|
|
|
|
def _connect_conversations():
|
|
return _connect(CONVERSATIONS_DB)
|
|
|
|
def _connect_sessions():
|
|
return _connect(SESSIONS_DB)
|
|
|
|
SETTINGS_PATH = Path.home() / "aaronai" / "settings.json"
|
|
WATCHER_LOG = str(Path.home() / "aaronai" / "watcher.log")
|
|
WATCHER_STATE = str(Path.home() / "aaronai" / "watcher_state.json")
|
|
NEXTCLOUD_PATH = "/home/aaron/nextcloud/data/data/aaron/files"
|
|
PYTHON = str(Path.home() / "aaronai" / "venv" / "bin" / "python3")
|
|
|
|
DEFAULT_SETTINGS = {
|
|
"theme": "light",
|
|
"font_size": "medium",
|
|
"web_search": True,
|
|
"show_sources": True,
|
|
"dream_hour_utc": 8,
|
|
"dream_minute_utc": 0,
|
|
"dream_mode": "pipeline",
|
|
"ingest_hour_utc": 2,
|
|
"ingest_minute_utc": 30,
|
|
"share_time": True,
|
|
}
|
|
|
|
print("Loading Aaron AI...")
|
|
PG_DSN = os.getenv("PG_DSN")
|
|
|
|
def get_pg():
|
|
return psycopg2.connect(PG_DSN)
|
|
WHISPER_PROMPT = (
|
|
"Grasshopper, Rhino, PolyJet, SLA, FDM, DMLS, "
|
|
"HVAMC, FWN3D, Mossygear, Nextcloud, Gitea, "
|
|
"computational design, additive manufacturing, fabrication, "
|
|
"Graphiti, FalkorDB, pgvector, BirdAI, Active Inference, "
|
|
"dreamer, consolidator, Extended Mind, "
|
|
"Aaron Nelson, SUNY New Paltz, University of Utah, MDD"
|
|
)
|
|
whisper_model = None
|
|
if HAS_WHISPER:
|
|
try:
|
|
whisper_model = WhisperModel("distil-large-v3", device="cpu", compute_type="int8", cpu_threads=4)
|
|
print("Whisper model loaded")
|
|
except Exception as e:
|
|
print(f"Whisper not available: {e}")
|
|
embedder = SentenceTransformer("all-MiniLM-L6-v2")
|
|
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
|
|
# ChromaDB removed — using pgvector
|
|
anthropic_client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
|
|
|
|
SYSTEM_PROMPT = """You are the personal AI assistant of Aaron Nelson — computational
|
|
designer, fabrication researcher, program builder, and creative
|
|
practitioner based in the Hudson Valley.
|
|
|
|
Aaron's work sits at the intersection of computational geometry,
|
|
additive manufacturing, and physical making. He resolves complex
|
|
systems into physical reality — from Grasshopper definitions to
|
|
large-scale steel structures, from archival photographs to 3D-printed
|
|
architectural restorations, from product concepts to manufactured
|
|
goods. This throughline — computation resolving into physical form —
|
|
defines his practice across academic, consulting, and creative contexts.
|
|
|
|
He built the Hudson Valley Additive Manufacturing Center (HVAMC) from
|
|
nothing and has directed it since 2016, alongside the DDF academic
|
|
program at SUNY New Paltz. He has the skills of a founder — equipment
|
|
selection, policy creation, client development, grant writing,
|
|
curriculum design — operating within an academic structure. He
|
|
consults with IBM, Braskem, Selux and others through FWN3D. He runs
|
|
Mossygear as a product business. He makes large-scale fabricated art.
|
|
|
|
His communication style is direct, precise, and intolerant of padding
|
|
or overclaiming. He flags inaccuracies immediately and expects the
|
|
same standard from you. When helping him write, match his voice —
|
|
economical, specific, never performative. When answering questions,
|
|
cite sources and acknowledge uncertainty rather than filling gaps with
|
|
plausible-sounding content.
|
|
|
|
You have a persistent memory file (always present below) that carries
|
|
Aaron's current context — treat it as ground truth for his present
|
|
situation.
|
|
|
|
For anything beyond what's in memory, you have a retrieve_documents
|
|
tool that searches his full knowledge base: personal documents,
|
|
reading library, conversation transcripts, and journal entries. Call
|
|
it whenever you need concrete information — names, dates, project
|
|
specifics, prior thinking, exhibition records, syllabi, anything you
|
|
don't already know. For compound questions, call it multiple times
|
|
with different concrete queries; one call per distinct information
|
|
need. Prefer specific tokens (named entities, project names, course
|
|
codes) over abstract instructional phrasing — search "FWN3D
|
|
consulting" not "my work." Results are unfiltered and ranked by
|
|
semantic similarity; judge each chunk for relevance and ignore
|
|
irrelevant hits rather than forcing them into the answer.
|
|
|
|
When Aaron asks for a document file — bio, cover letter, statement,
|
|
CV section, anything he wants to send or edit outside chat — use the
|
|
save_document tool to render the content to his Nextcloud Drafts/
|
|
folder as docx (editable) or pdf (typeset). Don't duplicate the full
|
|
content in the chat reply; just write the file and tell him where it
|
|
landed. He can open it from any of his synced devices.
|
|
|
|
Use web search automatically when current external information is
|
|
needed. Never re-brief on context that's already in memory or
|
|
retrieved chunks.
|
|
|
|
When making factual claims about Aaron — his history, credentials, locations, dates, relationships, projects, or any specific event — you must ground the claim in a specific retrieved document or the memory file. Cite the source by name inline. If no source supports the claim, say so explicitly rather than filling the gap with plausible-sounding content. Do not confabulate. If you are inferring rather than citing, mark it as inference."""
|
|
|
|
# Auth configuration
|
|
import os
|
|
SESSION_PASSWORD = os.getenv("AARON_AI_PASSWORD", "changeme")
|
|
SESSION_MAX_AGE_SECONDS = 60 * 60 * 24 * 365
|
|
SESSIONS_DB = str(Path.home() / "aaronai" / "sessions.db")
|
|
|
|
def _init_sessions():
|
|
conn = _connect_sessions()
|
|
conn.execute("CREATE TABLE IF NOT EXISTS sessions (token TEXT PRIMARY KEY, created_at TEXT)")
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
_init_sessions()
|
|
|
|
def make_session_token() -> str:
|
|
return secrets.token_urlsafe(32)
|
|
|
|
def hash_password(password: str) -> str:
|
|
return hashlib.sha256(password.encode()).hexdigest()
|
|
|
|
def save_session(token: str):
|
|
conn = _connect_sessions()
|
|
conn.execute("INSERT OR REPLACE INTO sessions VALUES (?, ?)", (token, datetime.now().isoformat()))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def delete_session(token: str):
|
|
conn = _connect_sessions()
|
|
conn.execute("DELETE FROM sessions WHERE token = ?", (token,))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def session_exists(token: str) -> bool:
|
|
conn = _connect_sessions()
|
|
cutoff = (datetime.now() - timedelta(seconds=SESSION_MAX_AGE_SECONDS)).isoformat()
|
|
conn.execute("DELETE FROM sessions WHERE created_at < ?", (cutoff,))
|
|
conn.commit()
|
|
row = conn.execute("SELECT 1 FROM sessions WHERE token = ? AND created_at >= ?", (token, cutoff)).fetchone()
|
|
conn.close()
|
|
return row is not None
|
|
|
|
def get_session(request: Request) -> str | None:
|
|
return request.cookies.get("aaronai_session")
|
|
|
|
def require_auth(request: Request):
|
|
token = get_session(request)
|
|
if not token or not session_exists(token):
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
return token
|
|
|
|
def init_conversations_db():
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute('''CREATE TABLE IF NOT EXISTS conversations (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
model TEXT DEFAULT 'claude-sonnet-4-6',
|
|
message_count INTEGER DEFAULT 0
|
|
)''')
|
|
c.execute('''CREATE TABLE IF NOT EXISTS messages (
|
|
id TEXT PRIMARY KEY,
|
|
conversation_id TEXT NOT NULL,
|
|
role TEXT NOT NULL,
|
|
content TEXT NOT NULL,
|
|
sources TEXT DEFAULT '[]',
|
|
timestamp TEXT NOT NULL,
|
|
FOREIGN KEY (conversation_id) REFERENCES conversations(id)
|
|
)''')
|
|
c.execute("PRAGMA journal_mode=WAL")
|
|
c.execute("CREATE INDEX IF NOT EXISTS idx_messages_conv_ts ON messages(conversation_id, timestamp DESC)")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
init_conversations_db()
|
|
|
|
def load_settings():
|
|
if SETTINGS_PATH.exists():
|
|
try:
|
|
s = json.loads(SETTINGS_PATH.read_text())
|
|
return {**DEFAULT_SETTINGS, **s}
|
|
except:
|
|
pass
|
|
return DEFAULT_SETTINGS.copy()
|
|
|
|
def save_settings(settings):
|
|
SETTINGS_PATH.write_text(json.dumps(settings, indent=2))
|
|
|
|
def load_memory():
|
|
if MEMORY_PATH.exists():
|
|
return MEMORY_PATH.read_text(encoding="utf-8")
|
|
return ""
|
|
|
|
def save_memory(content):
|
|
MEMORY_PATH.write_text(content, encoding="utf-8")
|
|
|
|
def add_to_memory(item):
|
|
memory = load_memory()
|
|
timestamp = datetime.now().strftime("%Y-%m-%d")
|
|
note = f"\n- [{timestamp}] {item}"
|
|
if "## Notes" not in memory:
|
|
memory += "\n\n## Notes"
|
|
memory += note
|
|
save_memory(memory)
|
|
|
|
def remove_from_memory(item):
|
|
memory = load_memory()
|
|
lines = memory.split("\n")
|
|
filtered = [l for l in lines if item.lower() not in l.lower()]
|
|
save_memory("\n".join(filtered))
|
|
return len(lines) - len(filtered)
|
|
|
|
HYBRID_CANDIDATES = 30
|
|
RRF_K = 60
|
|
FINAL_LIMIT = 8
|
|
|
|
_TSQUERY_SANITIZE_RE = re.compile(r"[^\w\s\"'-]")
|
|
|
|
|
|
def _websearch_query(text: str) -> str:
|
|
"""Strip characters websearch_to_tsquery doesn't handle cleanly. Quoted
|
|
phrases and 'or' are preserved by the function itself."""
|
|
return _TSQUERY_SANITIZE_RE.sub(" ", text).strip()
|
|
|
|
|
|
def _rerank(query: str, candidates: list[tuple]) -> list[tuple]:
|
|
"""Cross-encoder rerank. Candidates are (id, document, source, folder, created_at)
|
|
tuples. Returns the same tuples reordered by reranker score with created_at as
|
|
secondary key — so when two chunks score similarly the newer one wins, which
|
|
keeps memory/journal files biased toward the latest snapshot."""
|
|
if not candidates:
|
|
return []
|
|
pairs = [(query, row[1]) for row in candidates]
|
|
scores = reranker.predict(pairs)
|
|
return [row for row, _ in sorted(
|
|
zip(candidates, scores),
|
|
key=lambda x: (float(x[1]), x[0][4] or ""),
|
|
reverse=True,
|
|
)]
|
|
|
|
|
|
def _format_source(source: str, folder: str) -> str:
|
|
"""Surface folder context to the LLM so it can disambiguate same-named files
|
|
(e.g., 21 different CV.docx files across job-application folders)."""
|
|
source = source or "unknown"
|
|
if folder and folder not in ("", "."):
|
|
return f"{folder}/{source}"
|
|
return source
|
|
|
|
|
|
def _dedup_key(doc: str) -> str:
|
|
"""Collapse near-duplicates by content. Files copied to multiple folders
|
|
produce byte-identical chunks; this catches those without affecting
|
|
legitimately-different chunks of the same source (e.g., separate sections
|
|
of a conversation)."""
|
|
return hashlib.md5(doc[:300].lower().encode("utf-8", "ignore")).hexdigest()
|
|
|
|
|
|
def retrieve_context(query, n_results=FINAL_LIMIT):
|
|
"""Hybrid retrieval (dense + lexical, RRF fused) followed by cross-encoder rerank.
|
|
|
|
- Dense (pgvector) handles paraphrase / semantic similarity.
|
|
- Lexical (tsvector) catches rare named tokens (FWN3D, Sono-Tek, course codes)
|
|
the embedding model has no signal for.
|
|
- RRF combines the two rankings without calibrating score scales.
|
|
- Cross-encoder rerank scores each (query, chunk) pair jointly.
|
|
- Near-duplicate collapse on output so top-N slots aren't burned by
|
|
multi-folder copies of the same file.
|
|
|
|
No type or folder filtering: imposing a taxonomy at retrieval time is a
|
|
heuristic we've explicitly rejected. The reranker ranks, the caller (LLM)
|
|
decides what's relevant to its task."""
|
|
query_embedding = embedder.encode([query]).tolist()[0]
|
|
ts_query = _websearch_query(query)
|
|
|
|
context_pieces = []
|
|
sources = []
|
|
|
|
try:
|
|
pg = get_pg()
|
|
cur = pg.cursor()
|
|
|
|
cur.execute("""
|
|
SELECT id, document, source, metadata->>'folder' AS folder, created_at
|
|
FROM embeddings
|
|
ORDER BY embedding <=> %s::vector
|
|
LIMIT %s
|
|
""", (query_embedding, HYBRID_CANDIDATES))
|
|
dense_hits = cur.fetchall()
|
|
|
|
lexical_hits = []
|
|
if ts_query:
|
|
cur.execute("""
|
|
SELECT id, document, source, metadata->>'folder' AS folder, created_at
|
|
FROM embeddings
|
|
WHERE to_tsvector('english', document)
|
|
@@ websearch_to_tsquery('english', %s)
|
|
ORDER BY ts_rank(to_tsvector('english', document),
|
|
websearch_to_tsquery('english', %s)) DESC
|
|
LIMIT %s
|
|
""", (ts_query, ts_query, HYBRID_CANDIDATES))
|
|
lexical_hits = cur.fetchall()
|
|
|
|
pg.close()
|
|
|
|
scores = {}
|
|
rows_by_id = {}
|
|
for rank, row in enumerate(dense_hits):
|
|
scores[row[0]] = scores.get(row[0], 0) + 1.0 / (RRF_K + rank + 1)
|
|
rows_by_id[row[0]] = row
|
|
for rank, row in enumerate(lexical_hits):
|
|
scores[row[0]] = scores.get(row[0], 0) + 1.0 / (RRF_K + rank + 1)
|
|
rows_by_id[row[0]] = row
|
|
|
|
rrf_ranked = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)
|
|
candidates = [rows_by_id[doc_id] for doc_id, _ in rrf_ranked]
|
|
|
|
seen = set()
|
|
for _id, doc, source, folder, _created_at in _rerank(query, candidates):
|
|
key = _dedup_key(doc)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
context_pieces.append(doc)
|
|
sources.append(_format_source(source, folder))
|
|
if len(context_pieces) >= n_results:
|
|
break
|
|
|
|
except Exception as e:
|
|
print(f"hybrid retrieval error: {e}")
|
|
|
|
return context_pieces, sources
|
|
|
|
def get_conversation_history(conversation_id, limit=20):
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute('''SELECT role, content FROM messages
|
|
WHERE conversation_id = ?
|
|
ORDER BY timestamp DESC LIMIT ?''', (conversation_id, limit))
|
|
rows = c.fetchall()
|
|
conn.close()
|
|
return [{"role": r[0], "content": r[1]} for r in reversed(rows)]
|
|
|
|
def save_message(conversation_id, role, content, sources=None):
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
msg_id = hashlib.md5(f"{conversation_id}{role}{datetime.now().isoformat()}".encode()).hexdigest()
|
|
timestamp = datetime.now().isoformat()
|
|
c.execute('''INSERT INTO messages (id, conversation_id, role, content, sources, timestamp)
|
|
VALUES (?, ?, ?, ?, ?, ?)''',
|
|
(msg_id, conversation_id, role, content,
|
|
json.dumps(sources or []), timestamp))
|
|
c.execute('''UPDATE conversations SET updated_at = ?, message_count = message_count + 1
|
|
WHERE id = ?''', (timestamp, conversation_id))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def create_conversation(title="New conversation"):
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
conv_id = hashlib.md5(f"{datetime.now().isoformat()}".encode()).hexdigest()[:16]
|
|
now = datetime.now().isoformat()
|
|
c.execute('''INSERT INTO conversations (id, title, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?)''', (conv_id, title, now, now))
|
|
conn.commit()
|
|
conn.close()
|
|
return conv_id
|
|
|
|
NEXTCLOUD_URL = os.getenv("NEXTCLOUD_URL", "https://nextcloud.aaronnelson.studio")
|
|
NEXTCLOUD_USER = os.getenv("NEXTCLOUD_USER", "aaron")
|
|
NEXTCLOUD_PASSWORD = os.getenv("NEXTCLOUD_PASSWORD", "")
|
|
DRAFTS_WEBDAV = f"{NEXTCLOUD_URL}/remote.php/dav/files/{NEXTCLOUD_USER}/Drafts"
|
|
|
|
_FILENAME_SAFE_RE = re.compile(r"[^A-Za-z0-9_\-\. ]")
|
|
|
|
|
|
SAVE_DOCUMENT_TOOL = {
|
|
"name": "save_document",
|
|
"description": (
|
|
"Render markdown content to docx or pdf and save it to Aaron's Nextcloud "
|
|
"Drafts/ folder (syncs to his other devices and web UI). Use this when "
|
|
"Aaron asks for a document file rather than chat text — bios, cover "
|
|
"letters, statements, CV sections, anything he'll edit or send. Returns "
|
|
"the saved filename. Pick a descriptive filename (no extension) like "
|
|
"'Aaron_Nelson_Bio_Utah_2026-05'. Format is 'docx' for editable drafts, "
|
|
"'pdf' for typeset/print-ready output. Content should be well-formed "
|
|
"markdown — # headings, **bold**, *italic*, - bulleted lists. Don't "
|
|
"embed file content in the chat response too; just call this tool and "
|
|
"tell Aaron where it landed."
|
|
),
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"content": {
|
|
"type": "string",
|
|
"description": "Document content in markdown.",
|
|
},
|
|
"filename": {
|
|
"type": "string",
|
|
"description": "Descriptive filename without extension.",
|
|
},
|
|
"format": {
|
|
"type": "string",
|
|
"enum": ["docx", "pdf"],
|
|
"description": "Output format.",
|
|
},
|
|
},
|
|
"required": ["content", "filename", "format"],
|
|
},
|
|
}
|
|
|
|
|
|
def _safe_filename(name: str, ext: str) -> str:
|
|
"""Strip path components and unsafe chars; force the requested extension."""
|
|
base = Path(name).name
|
|
base = _FILENAME_SAFE_RE.sub("_", base).strip().rstrip(".")
|
|
if not base:
|
|
base = "untitled"
|
|
base = Path(base).stem
|
|
return f"{base}.{ext}"
|
|
|
|
|
|
def _webdav_unique_url(base_url: str, filename: str, auth) -> tuple[str, str]:
|
|
"""Return a WebDAV URL that doesn't collide with an existing file. Appends
|
|
_2, _3, ... until PROPFIND returns 404. Matches the convention dream.py uses."""
|
|
stem = Path(filename).stem
|
|
suffix = Path(filename).suffix
|
|
name = filename
|
|
i = 2
|
|
while True:
|
|
url = f"{base_url}/{name}"
|
|
check = requests.request("PROPFIND", url, auth=auth, timeout=10)
|
|
if check.status_code == 404:
|
|
return url, name
|
|
name = f"{stem}_{i}{suffix}"
|
|
i += 1
|
|
if i > 50:
|
|
raise RuntimeError("could not find a free filename")
|
|
|
|
|
|
def _execute_save_document(tool_input):
|
|
"""Generate a document via pandoc and PUT it to Nextcloud Drafts/.
|
|
Returns a user-facing status string for Claude to relay."""
|
|
if not NEXTCLOUD_PASSWORD:
|
|
return "save_document: NEXTCLOUD_PASSWORD not configured."
|
|
|
|
payload = tool_input or {}
|
|
content = payload.get("content", "")
|
|
raw_filename = payload.get("filename", "untitled")
|
|
fmt = payload.get("format", "docx")
|
|
|
|
if not content.strip():
|
|
return "save_document: empty content, nothing saved."
|
|
if fmt not in ("docx", "pdf"):
|
|
return f"save_document: unsupported format {fmt!r}; use 'docx' or 'pdf'."
|
|
|
|
safe_name = _safe_filename(raw_filename, fmt)
|
|
auth = (NEXTCLOUD_USER, NEXTCLOUD_PASSWORD)
|
|
|
|
# Ensure Drafts/ exists. 201 = created, 405 = already there — both fine.
|
|
try:
|
|
requests.request("MKCOL", DRAFTS_WEBDAV, auth=auth, timeout=10)
|
|
except requests.RequestException as e:
|
|
return f"save_document: could not reach Nextcloud ({e})."
|
|
|
|
try:
|
|
url, final_name = _webdav_unique_url(DRAFTS_WEBDAV, safe_name, auth)
|
|
except (requests.RequestException, RuntimeError) as e:
|
|
return f"save_document: filename probe failed ({e})."
|
|
|
|
cmd = ["pandoc", "-f", "markdown", "-t", fmt, "-o", "-"]
|
|
if fmt == "pdf":
|
|
cmd.insert(-2, "--pdf-engine=xelatex")
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd, input=content.encode("utf-8"),
|
|
capture_output=True, timeout=120,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return "save_document: pandoc timed out (>120s)."
|
|
except FileNotFoundError:
|
|
return "save_document: pandoc not installed."
|
|
if proc.returncode != 0:
|
|
err = proc.stderr.decode("utf-8", errors="replace")[:400]
|
|
return f"save_document: pandoc failed: {err}"
|
|
|
|
try:
|
|
put = requests.put(url, data=proc.stdout, auth=auth, timeout=60)
|
|
except requests.RequestException as e:
|
|
return f"save_document: WebDAV upload failed ({e})."
|
|
if put.status_code not in (200, 201, 204):
|
|
return f"save_document: WebDAV upload returned {put.status_code}."
|
|
|
|
return f"Saved to Nextcloud: Drafts/{final_name}"
|
|
|
|
|
|
RETRIEVE_DOCUMENTS_TOOL = {
|
|
"name": "retrieve_documents",
|
|
"description": (
|
|
"Search Aaron's knowledge base — personal documents, reading library, "
|
|
"conversation transcripts, and journal entries — for content relevant "
|
|
"to a query. Call whenever you need concrete information you don't "
|
|
"already have from the persistent memory file. For compound questions "
|
|
"(e.g. 'bio emphasizing consulting work and recent research'), call "
|
|
"this tool multiple times with different concrete queries; one call "
|
|
"per distinct information need. Prefer specific named entities, "
|
|
"project names, course codes, or topic-specific terms over abstract "
|
|
"instructional phrasing — 'FWN3D consulting' retrieves better than "
|
|
"'my work'. Results are ranked by semantic + lexical hybrid retrieval "
|
|
"and a cross-encoder reranker; no taxonomy is applied, so judge each "
|
|
"returned chunk on its own merits and ignore irrelevant hits."
|
|
),
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {
|
|
"type": "string",
|
|
"description": "The search query. Use concrete terms.",
|
|
},
|
|
},
|
|
"required": ["query"],
|
|
},
|
|
}
|
|
|
|
|
|
def _execute_retrieve_documents(tool_input):
|
|
"""Run retrieve_context for a tool call. Returns (tool_result_text, sources)."""
|
|
query = (tool_input or {}).get("query", "").strip()
|
|
if not query:
|
|
return ("No query provided.", [])
|
|
pieces, sources = retrieve_context(query)
|
|
if not pieces:
|
|
return (f"No results for query={query!r}.", [])
|
|
parts = []
|
|
for i, (piece, src) in enumerate(zip(pieces, sources), 1):
|
|
parts.append(f"[{i}] Source: {src}\n{piece}")
|
|
return ("\n\n---\n\n".join(parts), sources)
|
|
|
|
|
|
def chat(user_message, conversation_id, settings, client_time=None):
|
|
memory = load_memory()
|
|
history = get_conversation_history(conversation_id)
|
|
|
|
# System prompt + persistent memory are stable across the tool_use round-trip
|
|
# and across turns within the 5-minute cache TTL. Putting cache_control on the
|
|
# last system block creates a cache breakpoint here — the second LLM call in a
|
|
# tool_use turn reads this prefix from cache (~10% of standard input cost)
|
|
# instead of re-billing it. Memory lives here (not in the user message) so its
|
|
# position stays stable for cache hits.
|
|
system_blocks = [{"type": "text", "text": SYSTEM_PROMPT}]
|
|
if memory:
|
|
system_blocks.append({
|
|
"type": "text",
|
|
"text": f"Aaron's persistent memory:\n\n{memory}",
|
|
})
|
|
system_blocks[-1]["cache_control"] = {"type": "ephemeral"}
|
|
|
|
# client_time is per-turn dynamic, so it stays out of the cached prefix.
|
|
if client_time:
|
|
full_message = (
|
|
f"Current time (user-supplied, not logged): {client_time}\n\n"
|
|
f"---\n\n{user_message}"
|
|
)
|
|
else:
|
|
full_message = user_message
|
|
|
|
messages = history + [{"role": "user", "content": full_message}]
|
|
|
|
tools = [RETRIEVE_DOCUMENTS_TOOL, SAVE_DOCUMENT_TOOL]
|
|
if settings.get("web_search", True):
|
|
tools.append({"type": "web_search_20250305", "name": "web_search"})
|
|
|
|
accumulated_sources = []
|
|
|
|
while True:
|
|
response = anthropic_client.messages.create(
|
|
model="claude-sonnet-4-6",
|
|
max_tokens=2048,
|
|
system=system_blocks,
|
|
messages=messages,
|
|
tools=tools,
|
|
)
|
|
|
|
if response.stop_reason == "tool_use":
|
|
messages.append({"role": "assistant", "content": response.content})
|
|
tool_results = []
|
|
for block in response.content:
|
|
if block.type != "tool_use":
|
|
continue
|
|
if block.name == "retrieve_documents":
|
|
result_text, result_sources = _execute_retrieve_documents(block.input)
|
|
accumulated_sources.extend(result_sources)
|
|
tool_results.append({
|
|
"type": "tool_result",
|
|
"tool_use_id": block.id,
|
|
"content": result_text,
|
|
})
|
|
elif block.name == "save_document":
|
|
result_text = _execute_save_document(block.input)
|
|
tool_results.append({
|
|
"type": "tool_result",
|
|
"tool_use_id": block.id,
|
|
"content": result_text,
|
|
})
|
|
else:
|
|
tool_results.append({
|
|
"type": "tool_result",
|
|
"tool_use_id": block.id,
|
|
"content": "Search completed",
|
|
})
|
|
messages.append({"role": "user", "content": tool_results})
|
|
else:
|
|
assistant_message = ""
|
|
for block in response.content:
|
|
if hasattr(block, "text"):
|
|
assistant_message += block.text
|
|
return assistant_message, list(dict.fromkeys(accumulated_sources))
|
|
|
|
from contextlib import asynccontextmanager
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
reschedule_jobs()
|
|
scheduler.start()
|
|
print("Scheduler started")
|
|
yield
|
|
scheduler.shutdown()
|
|
print("Scheduler stopped")
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
|
|
|
|
@app.post("/auth/login")
|
|
async def login(request: Request, response: Response):
|
|
data = await request.json()
|
|
password = data.get("password", "")
|
|
if hash_password(password) != hash_password(SESSION_PASSWORD):
|
|
raise HTTPException(status_code=401, detail="Invalid password")
|
|
token = make_session_token()
|
|
save_session(token)
|
|
response.set_cookie(
|
|
key="aaronai_session",
|
|
value=token,
|
|
httponly=True,
|
|
secure=True,
|
|
samesite="lax",
|
|
max_age=SESSION_MAX_AGE_SECONDS
|
|
)
|
|
response.body = b'{"ok": true}'
|
|
response.status_code = 200
|
|
response.media_type = "application/json"
|
|
return response
|
|
|
|
@app.post("/auth/logout")
|
|
async def logout(request: Request, response: Response):
|
|
token = get_session(request)
|
|
if token:
|
|
delete_session(token)
|
|
response.delete_cookie("aaronai_session")
|
|
return JSONResponse({"ok": True})
|
|
|
|
@app.get("/auth/check")
|
|
async def check_auth(request: Request):
|
|
token = get_session(request)
|
|
if not token or not session_exists(token):
|
|
return JSONResponse({"authenticated": False})
|
|
return JSONResponse({"authenticated": True})
|
|
|
|
@app.get("/", response_class=FileResponse)
|
|
async def index():
|
|
return FileResponse("/home/aaron/aaronai/static/index.html")
|
|
|
|
@app.get("/api/settings")
|
|
async def get_settings(auth: str = Depends(require_auth)):
|
|
return JSONResponse(load_settings())
|
|
|
|
@app.post("/api/settings")
|
|
async def update_settings(request: Request, auth: str = Depends(require_auth)):
|
|
data = await request.json()
|
|
settings = load_settings()
|
|
settings.update(data)
|
|
save_settings(settings)
|
|
# Reschedule if schedule settings changed
|
|
schedule_keys = {"dream_hour_utc","dream_minute_utc","dream_mode","ingest_hour_utc","ingest_minute_utc"}
|
|
if any(k in data for k in schedule_keys):
|
|
reschedule_jobs()
|
|
return JSONResponse(settings)
|
|
|
|
@app.get("/api/conversations")
|
|
async def list_conversations(auth: str = Depends(require_auth)):
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute('''SELECT id, title, created_at, updated_at, message_count
|
|
FROM conversations ORDER BY updated_at DESC LIMIT 100''')
|
|
rows = c.fetchall()
|
|
conn.close()
|
|
return JSONResponse([{
|
|
"id": r[0], "title": r[1], "created_at": r[2],
|
|
"updated_at": r[3], "message_count": r[4]
|
|
} for r in rows])
|
|
|
|
@app.post("/api/conversations")
|
|
async def new_conversation(request: Request, auth: str = Depends(require_auth)):
|
|
data = await request.json()
|
|
title = data.get("title", "New conversation")
|
|
conv_id = create_conversation(title)
|
|
return JSONResponse({"id": conv_id, "title": title})
|
|
|
|
@app.get("/api/conversations/{conv_id}/messages")
|
|
async def get_messages(conv_id: str, auth: str = Depends(require_auth)):
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute('''SELECT role, content, sources, timestamp FROM messages
|
|
WHERE conversation_id = ? ORDER BY timestamp ASC''', (conv_id,))
|
|
rows = c.fetchall()
|
|
conn.close()
|
|
return JSONResponse([{
|
|
"role": r[0], "content": r[1],
|
|
"sources": json.loads(r[2]), "timestamp": r[3]
|
|
} for r in rows])
|
|
|
|
@app.patch("/api/conversations/{conv_id}")
|
|
async def rename_conversation(conv_id: str, request: Request, auth: str = Depends(require_auth)):
|
|
data = await request.json()
|
|
title = data.get("title", "")
|
|
if not title:
|
|
return JSONResponse({"error": "Title required"}, status_code=400)
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute("UPDATE conversations SET title = ? WHERE id = ?", (title, conv_id))
|
|
conn.commit()
|
|
conn.close()
|
|
return JSONResponse({"id": conv_id, "title": title})
|
|
|
|
@app.delete("/api/conversations/{conv_id}")
|
|
async def delete_conversation(conv_id: str, auth: str = Depends(require_auth)):
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute("DELETE FROM messages WHERE conversation_id = ?", (conv_id,))
|
|
c.execute("DELETE FROM conversations WHERE id = ?", (conv_id,))
|
|
conn.commit()
|
|
conn.close()
|
|
return JSONResponse({"deleted": conv_id})
|
|
|
|
@app.post("/api/chat")
|
|
async def chat_endpoint(request: Request, auth: str = Depends(require_auth)):
|
|
data = await request.json()
|
|
user_message = data.get("message", "").strip()
|
|
conversation_id = data.get("conversation_id", "")
|
|
client_time = data.get("client_time", None)
|
|
settings = load_settings()
|
|
|
|
if not user_message:
|
|
return JSONResponse({"error": "Empty message"})
|
|
|
|
if not conversation_id:
|
|
conversation_id = create_conversation("New conversation")
|
|
|
|
stripped = user_message.strip().lower()
|
|
|
|
if stripped == "show memory":
|
|
return JSONResponse({"response": load_memory(), "sources": [], "conversation_id": conversation_id})
|
|
|
|
if stripped.startswith("remember:"):
|
|
item = user_message[9:].strip()
|
|
add_to_memory(item)
|
|
save_message(conversation_id, "user", user_message)
|
|
save_message(conversation_id, "assistant", f"Saved to memory: '{item}'")
|
|
return JSONResponse({"response": f"Saved to memory: '{item}'", "sources": [], "conversation_id": conversation_id})
|
|
|
|
if stripped.startswith("forget:"):
|
|
item = user_message[7:].strip()
|
|
removed = remove_from_memory(item)
|
|
msg = f"Removed {removed} line(s) containing '{item}'" if removed else f"Nothing found containing '{item}'"
|
|
save_message(conversation_id, "user", user_message)
|
|
save_message(conversation_id, "assistant", msg)
|
|
return JSONResponse({"response": msg, "sources": [], "conversation_id": conversation_id})
|
|
|
|
save_message(conversation_id, "user", user_message)
|
|
|
|
# Auto-title conversation from first message
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute("SELECT message_count, title FROM conversations WHERE id = ?", (conversation_id,))
|
|
row = c.fetchone()
|
|
conn.close()
|
|
if row and row[0] <= 1 and row[1] == "New conversation":
|
|
auto_title = user_message[:60] + ("..." if len(user_message) > 60 else "")
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute("UPDATE conversations SET title = ? WHERE id = ?", (auto_title, conversation_id))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
response, sources = chat(user_message, conversation_id, settings, client_time=client_time)
|
|
save_message(conversation_id, "assistant", response, sources if settings.get("show_sources") else [])
|
|
|
|
return JSONResponse({
|
|
"response": response,
|
|
"sources": sources if settings.get("show_sources") else [],
|
|
"conversation_id": conversation_id
|
|
})
|
|
|
|
@app.get("/api/memory")
|
|
async def get_memory(auth: str = Depends(require_auth)):
|
|
return JSONResponse({"content": load_memory()})
|
|
|
|
@app.post("/api/memory")
|
|
async def update_memory(request: Request, auth: str = Depends(require_auth)):
|
|
data = await request.json()
|
|
content = data.get("content", "")
|
|
save_memory(content)
|
|
return JSONResponse({"saved": True})
|
|
|
|
@app.get("/api/status")
|
|
async def get_status(auth: str = Depends(require_auth)):
|
|
try:
|
|
pg = get_pg()
|
|
cur = pg.cursor()
|
|
cur.execute("SELECT COUNT(*) FROM embeddings")
|
|
chunk_count = cur.fetchone()[0]
|
|
pg.close()
|
|
except:
|
|
chunk_count = 0
|
|
|
|
# Watcher status
|
|
watcher_running = False
|
|
watcher_ingestion = {"status": "idle", "message": "", "file_count": 0}
|
|
last_indexed = "Unknown"
|
|
try:
|
|
import time as _time, json as _json
|
|
_sp = Path("/home/aaron/aaronai/watcher_status.json")
|
|
if _sp.exists():
|
|
_s = _json.loads(_sp.read_text())
|
|
_age = _time.time() - _s.get("timestamp", 0)
|
|
watcher_running = _s.get("running", False) and _age < 30
|
|
watcher_ingestion = _s.get("ingestion", watcher_ingestion)
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
log_path = Path(WATCHER_LOG)
|
|
if log_path.exists():
|
|
lines = log_path.read_text().strip().split("\n")
|
|
for line in reversed(lines):
|
|
if "Ingestion complete" in line:
|
|
last_indexed = line.split(" - ")[0].strip()
|
|
break
|
|
except:
|
|
pass
|
|
|
|
# File count from watcher state
|
|
file_count = 0
|
|
try:
|
|
state_path = Path(WATCHER_STATE)
|
|
if state_path.exists():
|
|
state = json.loads(state_path.read_text())
|
|
file_count = len(state)
|
|
else:
|
|
# Count files in Nextcloud directly
|
|
nc_path = Path(NEXTCLOUD_PATH)
|
|
if nc_path.exists():
|
|
file_count = sum(1 for f in nc_path.rglob("*")
|
|
if f.is_file() and f.suffix.lower() in {'.pdf','.docx','.pptx','.txt','.md'})
|
|
except:
|
|
pass
|
|
|
|
# Conversation count
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute("SELECT COUNT(*) FROM conversations")
|
|
conv_count = c.fetchone()[0]
|
|
conn.close()
|
|
|
|
return JSONResponse({
|
|
"aaron_ai": "running",
|
|
"watcher": "running" if watcher_running else "stopped",
|
|
"watcher_ingestion": watcher_ingestion,
|
|
"chunk_count": chunk_count,
|
|
"file_count": file_count,
|
|
"last_indexed": last_indexed,
|
|
"conversation_count": conv_count,
|
|
"model": "claude-sonnet-4-6",
|
|
"nextcloud_path": NEXTCLOUD_PATH
|
|
})
|
|
|
|
@app.post("/api/transcribe")
|
|
async def transcribe_audio(request: Request, audio: UploadFile = File(...), auth: str = Depends(require_auth)):
|
|
if not whisper_model:
|
|
raise HTTPException(status_code=503, detail="Whisper not available")
|
|
try:
|
|
suffix = ".webm"
|
|
if audio.content_type and "mp4" in audio.content_type:
|
|
suffix = ".mp4"
|
|
elif audio.content_type and "ogg" in audio.content_type:
|
|
suffix = ".ogg"
|
|
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
|
|
content = await audio.read()
|
|
tmp.write(content)
|
|
tmp_path = tmp.name
|
|
segments, info = whisper_model.transcribe(
|
|
tmp_path,
|
|
language="en",
|
|
vad_filter=True,
|
|
beam_size=1,
|
|
initial_prompt=WHISPER_PROMPT
|
|
)
|
|
transcript = " ".join(s.text.strip() for s in segments)
|
|
os.unlink(tmp_path)
|
|
return JSONResponse({"text": transcript, "language": info.language})
|
|
except Exception as e:
|
|
if os.path.exists(tmp_path):
|
|
os.unlink(tmp_path)
|
|
return JSONResponse({"ok": False, "error": str(e), "error_type": "transcription_failed"}, status_code=500)
|
|
|
|
@app.get("/api/dreamer/status")
|
|
async def dreamer_status(auth: str = Depends(require_auth)):
|
|
try:
|
|
state_path = Path.home() / "aaronai" / "dreamer_state.json"
|
|
if state_path.exists():
|
|
state = json.loads(state_path.read_text())
|
|
else:
|
|
state = {}
|
|
last_ts = state.get("last_dream_timestamp", 0)
|
|
last_dt = datetime.fromtimestamp(last_ts).strftime("%Y-%m-%d %H:%M") if last_ts else "never"
|
|
raw_mode = state.get("last_dream_mode", "none")
|
|
display_mode = "full pipeline" if raw_mode == "pipeline" else raw_mode
|
|
return JSONResponse({
|
|
"last_dream": last_dt,
|
|
"last_mode": display_mode,
|
|
"last_file": state.get("last_dream_file", ""),
|
|
})
|
|
except Exception as e:
|
|
return JSONResponse({"last_dream": "unknown", "last_mode": "none", "last_file": "", "error": str(e)})
|
|
|
|
@app.post("/api/dreamer/run")
|
|
async def run_dreamer(request: Request, auth: str = Depends(require_auth)):
|
|
try:
|
|
body = await request.json()
|
|
mode = body.get("mode", "nrem")
|
|
task = body.get("task", None)
|
|
dream_script = str(Path.home() / "aaronai" / "scripts" / "dream.py")
|
|
cmd = [PYTHON, dream_script, "--mode", mode]
|
|
if task:
|
|
cmd += ["--task", task]
|
|
subprocess.Popen(cmd, cwd=str(Path.home() / "aaronai"))
|
|
return JSONResponse({"started": True, "mode": mode})
|
|
except Exception as e:
|
|
return JSONResponse({"started": False, "error": str(e)})
|
|
|
|
def transcribe_and_save(tmp_path, timestamp, nextcloud_url, nextcloud_user, nextcloud_password):
|
|
"""Background task — transcribes audio and saves to Nextcloud after endpoint returns.
|
|
Audio is preserved in Journal/Media/ on every terminal path; failed and empty-transcript
|
|
captures still produce a markdown record in Journal/Captures/ with a status field."""
|
|
import requests as req_lib
|
|
nc_auth = (nextcloud_user, nextcloud_password)
|
|
month_dir = timestamp[:7]
|
|
audio_ext = os.path.splitext(tmp_path)[1] or ".webm"
|
|
audio_filename = f"{timestamp}-voice{audio_ext}"
|
|
audio_relpath = f"Journal/Media/{month_dir}/{audio_filename}"
|
|
|
|
def archive_audio() -> bool:
|
|
try:
|
|
with open(tmp_path, "rb") as f:
|
|
audio_bytes = f.read()
|
|
media_parent = f"{nextcloud_url}/remote.php/dav/files/{nextcloud_user}/Journal/Media"
|
|
media_dir = f"{media_parent}/{month_dir}"
|
|
req_lib.request("MKCOL", media_parent, auth=nc_auth, timeout=10)
|
|
req_lib.request("MKCOL", media_dir, auth=nc_auth, timeout=10)
|
|
req_lib.put(f"{media_dir}/{audio_filename}", data=audio_bytes, auth=nc_auth, timeout=60)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Audio archival failed for {timestamp}: {e}")
|
|
return False
|
|
finally:
|
|
if os.path.exists(tmp_path):
|
|
os.unlink(tmp_path)
|
|
|
|
def write_capture(filename: str, content_md: str, status: str):
|
|
captures_dir = f"{nextcloud_url}/remote.php/dav/files/{nextcloud_user}/Journal/Captures"
|
|
try:
|
|
req_lib.request("MKCOL", captures_dir, auth=nc_auth, timeout=10)
|
|
req_lib.put(f"{captures_dir}/{filename}", data=content_md.encode("utf-8"), auth=nc_auth, timeout=30)
|
|
except Exception as e:
|
|
print(f"Capture markdown write failed for {timestamp}: {e}")
|
|
return
|
|
try:
|
|
payload = {"type": "capture_saved", "filename": filename, "timestamp": timestamp, "status": status}
|
|
req_lib.post("http://localhost:8000/api/events/notify", json=payload, timeout=3)
|
|
req_lib.post("http://localhost:8000/api/captures/events/notify", json=payload, timeout=3)
|
|
except Exception:
|
|
pass
|
|
|
|
transcript = ""
|
|
transcribe_error = None
|
|
try:
|
|
segments, _ = whisper_model.transcribe(
|
|
tmp_path, language="en", vad_filter=True, beam_size=1, initial_prompt=WHISPER_PROMPT
|
|
)
|
|
transcript = " ".join(s.text.strip() for s in segments).strip()
|
|
except Exception as e:
|
|
transcribe_error = str(e)
|
|
|
|
audio_archived = archive_audio()
|
|
audio_line = f"**audio_path:** {audio_relpath}\n" if audio_archived else "**audio_archive_failed:** true\n"
|
|
|
|
if transcribe_error is not None:
|
|
filename = f"{timestamp}-voice-failed.md"
|
|
content_md = (
|
|
f"# Capture — {timestamp}\n\n"
|
|
f"**type:** voice\n**modality:** audio\n**status:** failed_transcription\n"
|
|
f"{audio_line}"
|
|
f"**error:** {transcribe_error}\n"
|
|
)
|
|
write_capture(filename, content_md, "failed_transcription")
|
|
print(f"Async transcription failed for {timestamp}: {transcribe_error}")
|
|
return
|
|
|
|
if not transcript:
|
|
filename = f"{timestamp}-voice-empty.md"
|
|
content_md = (
|
|
f"# Capture — {timestamp}\n\n"
|
|
f"**type:** voice\n**modality:** audio\n**status:** empty_transcript\n"
|
|
f"{audio_line}"
|
|
)
|
|
write_capture(filename, content_md, "empty_transcript")
|
|
print(f"Async transcription empty for {timestamp}: audio archived")
|
|
return
|
|
|
|
filename = f"{timestamp}-voice.md"
|
|
content_md = (
|
|
f"# Capture — {timestamp}\n\n"
|
|
f"**type:** voice\n**modality:** audio\n**status:** saved\n"
|
|
f"{audio_line}\n---\n\n{transcript}\n"
|
|
)
|
|
write_capture(filename, content_md, "saved")
|
|
print(f"Async transcription saved: {filename}")
|
|
|
|
|
|
@app.post("/api/capture")
|
|
async def capture_endpoint(
|
|
background_tasks: BackgroundTasks,
|
|
audio: UploadFile = File(None),
|
|
image: UploadFile = File(None),
|
|
project: str = Form(None),
|
|
):
|
|
"""Auth-free capture endpoint — handles voice, image, or image+voice."""
|
|
import requests as req_lib
|
|
import base64
|
|
|
|
nextcloud_url = os.getenv("NEXTCLOUD_URL", "")
|
|
nextcloud_user = os.getenv("NEXTCLOUD_USER", "aaron")
|
|
nextcloud_password = os.getenv("NEXTCLOUD_PASSWORD", "")
|
|
nc_auth = (nextcloud_user, nextcloud_password)
|
|
timestamp = datetime.now().strftime("%Y-%m-%d-%H-%M")
|
|
month_dir = datetime.now().strftime("%Y-%m")
|
|
|
|
# ── Image + optional voice ───────────────────────────────────────────────
|
|
if image is not None:
|
|
tmp_audio_path = None
|
|
try:
|
|
# Read image bytes
|
|
image_bytes = await image.read()
|
|
image_content_type = image.content_type or "image/jpeg"
|
|
# Determine extension
|
|
ext_map = {"image/jpeg": "jpg", "image/png": "png", "image/webp": "webp", "image/heic": "jpg"}
|
|
img_ext = ext_map.get(image_content_type, "jpg")
|
|
img_filename = f"{timestamp}-image.{img_ext}"
|
|
|
|
# Save raw image to Media/YYYY-MM/ via WebDAV
|
|
media_dir = f"{nextcloud_url}/remote.php/dav/files/{nextcloud_user}/Journal/Media/{month_dir}"
|
|
req_lib.request("MKCOL", f"{nextcloud_url}/remote.php/dav/files/{nextcloud_user}/Journal/Media", auth=nc_auth, timeout=10)
|
|
req_lib.request("MKCOL", media_dir, auth=nc_auth, timeout=10)
|
|
media_url = f"{media_dir}/{img_filename}"
|
|
req_lib.put(media_url, data=image_bytes, auth=nc_auth,
|
|
headers={"Content-Type": image_content_type}, timeout=60)
|
|
|
|
# Transcribe voice annotation if present
|
|
voice_annotation = None
|
|
if audio is not None and whisper_model:
|
|
audio_bytes = await audio.read()
|
|
suffix = ".webm"
|
|
if audio.content_type and "mp4" in audio.content_type:
|
|
suffix = ".mp4"
|
|
elif audio.content_type and "ogg" in audio.content_type:
|
|
suffix = ".ogg"
|
|
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
|
|
tmp.write(audio_bytes)
|
|
tmp_audio_path = tmp.name
|
|
segments, _ = whisper_model.transcribe(
|
|
tmp_audio_path, language="en", vad_filter=True, beam_size=1, initial_prompt=WHISPER_PROMPT
|
|
)
|
|
voice_annotation = " ".join(s.text.strip() for s in segments).strip() or None
|
|
os.unlink(tmp_audio_path)
|
|
tmp_audio_path = None
|
|
|
|
# Generate Claude vision description
|
|
image_b64 = base64.standard_b64encode(image_bytes).decode("utf-8")
|
|
annotation_line = f"Aaron said about this image: \"{voice_annotation}\"" if voice_annotation else ""
|
|
vision_prompt = f"""You are generating a memory description for an AI corpus belonging to Aaron Nelson — computational designer, fabrication researcher, and visual artist working in the Hudson Valley.
|
|
|
|
Describe this image for long-term memory indexing.
|
|
|
|
PERCEPTUAL: Composition, materials, light, color, texture, scale, spatial relationships. Be specific enough that this image could be distinguished from visually similar images.
|
|
|
|
CONTENT: What is this? What domain does it belong to? What is it an instance of?
|
|
|
|
{annotation_line}
|
|
|
|
End your response with a single line in this exact format:
|
|
ENTITIES: [comma-separated list of key entities — people, objects, materials, places, projects, tools]
|
|
|
|
Keep the full description to 150-250 words. Do not speculate beyond what is visible or stated. Write as continuous prose followed by the ENTITIES line."""
|
|
|
|
vision_response = anthropic_client.messages.create(
|
|
model="claude-sonnet-4-6",
|
|
max_tokens=800,
|
|
messages=[{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": image_content_type,
|
|
"data": image_b64,
|
|
}
|
|
},
|
|
{"type": "text", "text": vision_prompt}
|
|
]
|
|
}]
|
|
)
|
|
description = vision_response.content[0].text.strip()
|
|
|
|
# Build rich Graphiti-ready episode markdown
|
|
capture_type = "image+voice" if voice_annotation else "image"
|
|
modality = "visual+audio" if voice_annotation else "visual"
|
|
media_path = f"Journal/Media/{month_dir}/{img_filename}"
|
|
|
|
content_md = f"""# Capture — Image — {timestamp}
|
|
|
|
**type:** {capture_type}
|
|
**modality:** {modality}
|
|
**status:** saved
|
|
**media:** {media_path}
|
|
{f"**project:** {project}" if project else ""}
|
|
|
|
---
|
|
|
|
**Visual description:**
|
|
{description}
|
|
|
|
**Voice annotation:**
|
|
{voice_annotation if voice_annotation else "none recorded"}
|
|
|
|
---
|
|
"""
|
|
# Save description to Journal/Captures/ via WebDAV
|
|
captures_dir = f"{nextcloud_url}/remote.php/dav/files/{nextcloud_user}/Journal/Captures"
|
|
req_lib.request("MKCOL", captures_dir, auth=nc_auth, timeout=10)
|
|
cap_filename = f"{timestamp}-image.md"
|
|
cap_url = f"{captures_dir}/{cap_filename}"
|
|
req_lib.put(cap_url, data=content_md.encode("utf-8"), auth=nc_auth, timeout=30)
|
|
|
|
return JSONResponse({
|
|
"ok": True,
|
|
"filename": cap_filename,
|
|
"media": media_path,
|
|
"has_voice": voice_annotation is not None,
|
|
})
|
|
|
|
except Exception as e:
|
|
if tmp_audio_path and os.path.exists(tmp_audio_path):
|
|
os.unlink(tmp_audio_path)
|
|
return JSONResponse({"ok": False, "error": str(e), "error_type": "transcription_failed"}, status_code=500)
|
|
|
|
# ── Voice only ───────────────────────────────────────────────────────────
|
|
elif audio is not None:
|
|
if not whisper_model:
|
|
raise HTTPException(status_code=503, detail="Whisper not available")
|
|
try:
|
|
suffix = ".webm"
|
|
if audio.content_type and "mp4" in audio.content_type:
|
|
suffix = ".mp4"
|
|
elif audio.content_type and "ogg" in audio.content_type:
|
|
suffix = ".ogg"
|
|
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
|
|
content_bytes = await audio.read()
|
|
tmp.write(content_bytes)
|
|
tmp_path = tmp.name
|
|
background_tasks.add_task(
|
|
transcribe_and_save,
|
|
tmp_path=tmp_path,
|
|
timestamp=timestamp,
|
|
nextcloud_url=nextcloud_url,
|
|
nextcloud_user=nextcloud_user,
|
|
nextcloud_password=nextcloud_password,
|
|
)
|
|
return JSONResponse({"ok": True, "filename": f"{timestamp}-voice.md", "async": True})
|
|
except Exception as e:
|
|
return JSONResponse({"ok": False, "error": str(e), "error_type": "capture_failed"})
|
|
|
|
else:
|
|
raise HTTPException(status_code=400, detail="No audio or image provided")
|
|
|
|
@app.get("/api/captures")
|
|
async def list_captures():
|
|
"""Returns recent captures from Nextcloud Journal/Captures/ — auth-free"""
|
|
try:
|
|
import requests as req_lib
|
|
nextcloud_url = os.getenv("NEXTCLOUD_URL", "")
|
|
nextcloud_user = os.getenv("NEXTCLOUD_USER", "aaron")
|
|
nextcloud_password = os.getenv("NEXTCLOUD_PASSWORD", "")
|
|
captures_dir = f"{nextcloud_url}/remote.php/dav/files/{nextcloud_user}/Journal/Captures"
|
|
auth = (nextcloud_user, nextcloud_password)
|
|
|
|
propfind = req_lib.request("PROPFIND", captures_dir, auth=auth, timeout=10,
|
|
headers={"Depth": "1"})
|
|
|
|
if propfind.status_code == 404:
|
|
return JSONResponse({"captures": []})
|
|
|
|
import xml.etree.ElementTree as ET
|
|
root = ET.fromstring(propfind.text)
|
|
ns = {"d": "DAV:"}
|
|
captures = []
|
|
for resp in root.findall("d:response", ns):
|
|
href = resp.findtext("d:href", namespaces=ns) or ""
|
|
if href.endswith("/"):
|
|
continue
|
|
name = href.split("/")[-1]
|
|
if not name.endswith(".md"):
|
|
continue
|
|
captures.append({"name": name.replace(".md", ""), "duration": ""})
|
|
|
|
captures.sort(key=lambda x: x["name"], reverse=True)
|
|
return JSONResponse({"captures": captures[:10]})
|
|
except Exception as e:
|
|
return JSONResponse({"captures": []})
|
|
|
|
REINDEX_STATUS_PATH = Path.home() / "aaronai" / "reindex_status.json"
|
|
|
|
|
|
def _read_reindex_status() -> dict:
|
|
if REINDEX_STATUS_PATH.exists():
|
|
try:
|
|
return json.loads(REINDEX_STATUS_PATH.read_text())
|
|
except Exception:
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def _write_reindex_status(state: dict):
|
|
REINDEX_STATUS_PATH.write_text(json.dumps(state, indent=2))
|
|
|
|
|
|
def _reindex_running() -> bool:
|
|
return _read_reindex_status().get("status") == "running"
|
|
|
|
|
|
def _run_reindex_background():
|
|
"""Background-thread entry: shares api.py's module-level embedder."""
|
|
started = datetime.now().isoformat()
|
|
_write_reindex_status({"status": "running", "started_at": started})
|
|
try:
|
|
result = ingest_directory(Path(NEXTCLOUD_PATH), embedder=embedder)
|
|
_write_reindex_status({
|
|
"status": "complete",
|
|
"started_at": started,
|
|
"finished_at": datetime.now().isoformat(),
|
|
**result,
|
|
})
|
|
except Exception as e:
|
|
_write_reindex_status({
|
|
"status": "error",
|
|
"started_at": started,
|
|
"finished_at": datetime.now().isoformat(),
|
|
"error": str(e),
|
|
})
|
|
|
|
|
|
@app.post("/api/reindex")
|
|
async def trigger_reindex(background_tasks: BackgroundTasks,
|
|
auth: str = Depends(require_auth)):
|
|
if _reindex_running():
|
|
return JSONResponse(
|
|
{"started": False, "message": "reindex already running"},
|
|
status_code=409,
|
|
)
|
|
background_tasks.add_task(_run_reindex_background)
|
|
return JSONResponse({"started": True, "message": "Re-indexing started in background"})
|
|
|
|
|
|
@app.get("/api/reindex/status")
|
|
async def reindex_status(auth: str = Depends(require_auth)):
|
|
return JSONResponse(_read_reindex_status())
|
|
|
|
@app.delete("/api/conversations")
|
|
async def clear_all_conversations(auth: str = Depends(require_auth)):
|
|
conn = _connect_conversations()
|
|
c = conn.cursor()
|
|
c.execute("DELETE FROM messages")
|
|
c.execute("DELETE FROM conversations")
|
|
conn.commit()
|
|
conn.close()
|
|
return JSONResponse({"cleared": True})
|
|
|
|
|
|
|
|
# ─── Corpus Integrity Endpoints ─────────────────────────────────────────────
|
|
|
|
CORPUS_INTEGRITY_SCRIPT = str(Path.home() / "aaronai" / "scripts" / "corpus_integrity.py")
|
|
CORPUS_REPORT_PATH = Path.home() / "aaronai" / "corpus_integrity_report.json"
|
|
SUPPORTED_EXTS = {".pdf", ".docx", ".pptx", ".txt", ".md"}
|
|
MIGRATION_STATE_PATH = Path.home() / "aaronai" / "experiments" / "tier1_migration_state.json"
|
|
|
|
|
|
def get_corpus_status_data():
|
|
fs_count = 0
|
|
try:
|
|
root = Path(NEXTCLOUD_PATH)
|
|
for path in root.rglob("*"):
|
|
if path.is_file() and path.suffix.lower() in SUPPORTED_EXTS:
|
|
if path.name.startswith((".", "~$")): continue
|
|
if "Admin/Backups" in str(path) or "Backups" in path.parts: continue
|
|
if "Journal/Media" in str(path): continue
|
|
fs_count += 1
|
|
except Exception:
|
|
pass
|
|
|
|
pv_count = 0
|
|
try:
|
|
pg = get_pg()
|
|
cur = pg.cursor()
|
|
cur.execute("SELECT COUNT(DISTINCT source) FROM embeddings WHERE source IS NOT NULL")
|
|
pv_count = cur.fetchone()[0]
|
|
pg.close()
|
|
except Exception:
|
|
pass
|
|
|
|
gr_sources = set()
|
|
try:
|
|
if MIGRATION_STATE_PATH.exists():
|
|
state = json.loads(MIGRATION_STATE_PATH.read_text())
|
|
for fp in state.get("ingested", []):
|
|
gr_sources.add(Path(fp).name)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
pg = get_pg()
|
|
cur = pg.cursor()
|
|
cur.execute("SELECT DISTINCT source FROM stage_3_queue WHERE completed_at IS NOT NULL")
|
|
for row in cur.fetchall(): gr_sources.add(row[0])
|
|
pg.close()
|
|
except Exception:
|
|
pass
|
|
|
|
failures = []
|
|
try:
|
|
pg = get_pg()
|
|
cur = pg.cursor()
|
|
cur.execute("""
|
|
SELECT source, filepath, error, retry_count, first_failed_at, last_failed_at
|
|
FROM ingest_failures WHERE resolved = FALSE
|
|
ORDER BY last_failed_at DESC LIMIT 50
|
|
""")
|
|
for row in cur.fetchall():
|
|
failures.append({
|
|
"source": row[0], "filepath": row[1], "error": row[2],
|
|
"retry_count": row[3], "first_failed_at": str(row[4]),
|
|
"last_failed_at": str(row[5]),
|
|
})
|
|
pg.close()
|
|
except Exception:
|
|
pass
|
|
|
|
last_report = None
|
|
try:
|
|
if CORPUS_REPORT_PATH.exists():
|
|
report = json.loads(CORPUS_REPORT_PATH.read_text())
|
|
last_report = {
|
|
"timestamp": report.get("timestamp"),
|
|
"gaps": report.get("summary", {}).get("neither", 0),
|
|
"auto_queued": len(report.get("auto_queued", [])),
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"filesystem": fs_count,
|
|
"pgvector": pv_count,
|
|
"graphiti": len(gr_sources),
|
|
"failures": failures,
|
|
"failure_count": len(failures),
|
|
"last_reconciliation": last_report,
|
|
}
|
|
|
|
|
|
@app.get("/api/corpus/status")
|
|
async def corpus_status(auth: str = Depends(require_auth)):
|
|
try:
|
|
return JSONResponse(get_corpus_status_data())
|
|
except Exception as e:
|
|
return JSONResponse({"error": str(e)}, status_code=500)
|
|
|
|
|
|
@app.post("/api/corpus/retry")
|
|
async def corpus_retry(request: Request, auth: str = Depends(require_auth)):
|
|
try:
|
|
body = await request.json()
|
|
source = body.get("source", "")
|
|
if not source:
|
|
return JSONResponse({"error": "source required"}, status_code=400)
|
|
pg = get_pg()
|
|
cur = pg.cursor()
|
|
cur.execute("SELECT filepath FROM ingest_failures WHERE source = %s", (source,))
|
|
row = cur.fetchone()
|
|
pg.close()
|
|
if not row:
|
|
return JSONResponse({"error": "source not found in failures"}, status_code=404)
|
|
filepath = Path(row[0])
|
|
if not filepath.exists():
|
|
return JSONResponse({"error": f"file not found: {filepath}"}, status_code=404)
|
|
try:
|
|
text = encoding_extract_text(filepath)
|
|
except Exception as e:
|
|
return JSONResponse({"error": f"extraction failed: {e}"}, status_code=500)
|
|
if not text.strip():
|
|
return JSONResponse({"error": "file produces empty text — may be corrupt"}, status_code=422)
|
|
pg = get_pg()
|
|
cur = pg.cursor()
|
|
cur.execute("""
|
|
INSERT INTO stage_2_queue (source, full_text, char_length)
|
|
VALUES (%s, %s, %s)
|
|
ON CONFLICT (source) DO UPDATE SET
|
|
full_text = EXCLUDED.full_text, char_length = EXCLUDED.char_length,
|
|
enqueued_at = NOW(), completed_at = NULL, failed_at = NULL, attempts = 0
|
|
""", (source, text, len(text)))
|
|
cur.execute("""
|
|
UPDATE ingest_failures SET retry_count = retry_count + 1, last_failed_at = NOW()
|
|
WHERE source = %s
|
|
""", (source,))
|
|
pg.commit()
|
|
pg.close()
|
|
return JSONResponse({"queued": True, "source": source})
|
|
except Exception as e:
|
|
return JSONResponse({"error": str(e)}, status_code=500)
|
|
|
|
|
|
@app.post("/api/corpus/reconcile")
|
|
async def corpus_reconcile(request: Request, background_tasks: BackgroundTasks, auth: str = Depends(require_auth)):
|
|
try:
|
|
body = await request.json()
|
|
fix = body.get("fix", True)
|
|
except Exception:
|
|
fix = True
|
|
def run_reconcile():
|
|
try:
|
|
cmd = [PYTHON, CORPUS_INTEGRITY_SCRIPT]
|
|
if fix:
|
|
cmd.append("--fix")
|
|
subprocess.run(cmd, cwd=str(Path.home() / "aaronai"), timeout=300)
|
|
except Exception as e:
|
|
print(f"Reconciliation failed: {e}")
|
|
background_tasks.add_task(run_reconcile)
|
|
return JSONResponse({"started": True, "fix": fix})
|
|
|
|
# ─── Scheduler ──────────────────────────────────────────────────────────────
|
|
scheduler = BackgroundScheduler()
|
|
|
|
def run_dream_job():
|
|
"""Runs nightly dreamer at the mode set in settings.json (default: pipeline)."""
|
|
try:
|
|
import subprocess
|
|
settings = load_settings()
|
|
mode = settings.get("dream_mode", "pipeline")
|
|
valid_modes = {"pipeline", "nrem", "early-rem", "late-rem"}
|
|
if mode not in valid_modes:
|
|
print(f"Dreamer: invalid dream_mode={mode!r}; falling back to pipeline")
|
|
mode = "pipeline"
|
|
dream_script = str(Path.home() / "aaronai" / "scripts" / "dream.py")
|
|
cmd = [PYTHON, dream_script]
|
|
if mode != "pipeline":
|
|
cmd += ["--mode", mode]
|
|
result = subprocess.run(
|
|
cmd,
|
|
cwd=str(Path.home() / "aaronai"),
|
|
capture_output=True, text=True, timeout=600
|
|
)
|
|
print(f"Dreamer completed (mode={mode}): {result.stdout[-200:] if result.stdout else 'no output'}")
|
|
if result.returncode != 0:
|
|
print(f"Dreamer error: {result.stderr[-200:] if result.stderr else 'unknown'}")
|
|
except Exception as e:
|
|
print(f"Dreamer job failed: {e}")
|
|
|
|
def run_ingest_job():
|
|
"""Runs nightly conversation indexing."""
|
|
try:
|
|
import subprocess
|
|
ingest_script = str(Path.home() / "aaronai" / "scripts" / "ingest_conversations.py")
|
|
result = subprocess.run(
|
|
[PYTHON, ingest_script],
|
|
cwd=str(Path.home() / "aaronai"),
|
|
capture_output=True, text=True, timeout=300
|
|
)
|
|
print(f"Ingest completed: {result.stdout[-200:] if result.stdout else 'no output'}")
|
|
except Exception as e:
|
|
print(f"Ingest job failed: {e}")
|
|
|
|
def reschedule_jobs():
|
|
"""Update scheduler from current settings."""
|
|
settings = load_settings()
|
|
# Remove existing jobs
|
|
for job_id in ("dream_job", "ingest_job"):
|
|
try:
|
|
scheduler.remove_job(job_id)
|
|
except:
|
|
pass
|
|
# Add dream job
|
|
scheduler.add_job(
|
|
run_dream_job,
|
|
CronTrigger(hour=settings.get("dream_hour_utc", 8),
|
|
minute=settings.get("dream_minute_utc", 0),
|
|
timezone="UTC"),
|
|
id="dream_job",
|
|
max_instances=1,
|
|
replace_existing=True
|
|
)
|
|
# Add ingest job
|
|
scheduler.add_job(
|
|
run_ingest_job,
|
|
CronTrigger(hour=settings.get("ingest_hour_utc", 2),
|
|
minute=settings.get("ingest_minute_utc", 30),
|
|
timezone="UTC"),
|
|
id="ingest_job",
|
|
max_instances=1,
|
|
replace_existing=True
|
|
)
|
|
print(f"Scheduled: dream at {settings.get('dream_hour_utc',8):02d}:{settings.get('dream_minute_utc',0):02d} UTC, ingest at {settings.get('ingest_hour_utc',2):02d}:{settings.get('ingest_minute_utc',30):02d} UTC")
|
|
|
|
# SSE client registry
|
|
sse_clients: list[asyncio.Queue] = []
|
|
capture_sse_clients: list[asyncio.Queue] = []
|
|
|
|
async def sse_generator(queue: asyncio.Queue):
|
|
try:
|
|
yield 'data: {"type": "connected"}\n\n'
|
|
while True:
|
|
try:
|
|
event = await asyncio.wait_for(queue.get(), timeout=30.0)
|
|
import json as _json
|
|
yield 'data: ' + _json.dumps(event) + '\n\n'
|
|
except asyncio.TimeoutError:
|
|
yield 'data: {"type": "heartbeat"}\n\n'
|
|
|
|
except asyncio.CancelledError:
|
|
pass
|
|
finally:
|
|
if queue in sse_clients:
|
|
sse_clients.remove(queue)
|
|
|
|
async def capture_sse_generator(queue: asyncio.Queue):
|
|
try:
|
|
yield 'data: {"type": "connected"}\n\n'
|
|
while True:
|
|
try:
|
|
event = await asyncio.wait_for(queue.get(), timeout=30.0)
|
|
import json as _json
|
|
yield 'data: ' + _json.dumps(event) + '\n\n'
|
|
except asyncio.TimeoutError:
|
|
yield 'data: {"type": "heartbeat"}\n\n'
|
|
except asyncio.CancelledError:
|
|
pass
|
|
finally:
|
|
if queue in capture_sse_clients:
|
|
capture_sse_clients.remove(queue)
|
|
|
|
@app.get("/api/captures/events")
|
|
async def capture_sse_endpoint(request: Request):
|
|
"""Public SSE endpoint for capture page — no auth required."""
|
|
queue: asyncio.Queue = asyncio.Queue()
|
|
capture_sse_clients.append(queue)
|
|
return StreamingResponse(
|
|
capture_sse_generator(queue),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"X-Accel-Buffering": "no",
|
|
"Connection": "keep-alive",
|
|
}
|
|
)
|
|
|
|
@app.post("/api/captures/events/notify")
|
|
async def notify_capture_clients(request: Request):
|
|
"""Internal endpoint — called when transcription completes."""
|
|
client_host = request.client.host if request.client else ""
|
|
if client_host not in ("127.0.0.1", "::1", "localhost"):
|
|
raise HTTPException(status_code=403, detail="Internal only")
|
|
data = await request.json()
|
|
for queue in capture_sse_clients:
|
|
await queue.put(data)
|
|
return JSONResponse({"notified": len(capture_sse_clients)})
|
|
|
|
@app.get("/api/events")
|
|
async def sse_endpoint(request: Request, auth: str = Depends(require_auth)):
|
|
queue: asyncio.Queue = asyncio.Queue()
|
|
sse_clients.append(queue)
|
|
return StreamingResponse(
|
|
sse_generator(queue),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"X-Accel-Buffering": "no",
|
|
"Connection": "keep-alive",
|
|
}
|
|
)
|
|
|
|
@app.post("/api/events/notify")
|
|
async def notify_clients(request: Request):
|
|
"""Internal endpoint — called by dream.py when a dream is delivered"""
|
|
# Only allow from localhost
|
|
client_host = request.client.host if request.client else ""
|
|
if client_host not in ("127.0.0.1", "::1", "localhost"):
|
|
raise HTTPException(status_code=403, detail="Internal only")
|
|
data = await request.json()
|
|
for queue in sse_clients:
|
|
await queue.put(data)
|
|
return JSONResponse({"notified": len(sse_clients)})
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|