diff --git a/.gitignore b/.gitignore index 5c267bf..c8389f2 100644 --- a/.gitignore +++ b/.gitignore @@ -26,3 +26,5 @@ data/chroma/ data/central.db handlers/node_modules/ handlers/.letta/ +tools/letta_archival_bug_repro.py +tools/fix_void_chars.py diff --git a/data/ask_sent.txt b/data/ask_sent.txt new file mode 100644 index 0000000..b111e17 --- /dev/null +++ b/data/ask_sent.txt @@ -0,0 +1,52 @@ +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3meie3mgxks2k +at://did:plc:boia3kqcyo3qnjw5fmqknib4/app.bsky.feed.post/3meie7pdzek2s +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3meiedoq3ws2k +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3meien4q3yc2k +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mejjpqkx322m +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mejrjsrxjt2p +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mejvu3e4ek2g +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mejw6wsgis2g +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mek3eqzkrc2s +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mek5fsf2c22f +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mek5iu32jc2f +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mek5vlhw7c27 +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mek6niorck2b +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mek6q3ezxs2b +at://did:plc:u2uat7ozvs34t5qe2q35euly/app.bsky.feed.post/3mek6sxcpr22g +at://did:plc:boia3kqcyo3qnjw5fmqknib4/app.bsky.feed.post/3mek6x3uvr22y +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mek74zkwxs2b +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mek76jklls2b +at://did:plc:boia3kqcyo3qnjw5fmqknib4/app.bsky.feed.post/3mek7737tic2y +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mek7epjdj22b +at://did:plc:z6b4y3evxxb5ufbfn5pa65fa/app.bsky.feed.post/3mek7eyxndk2c +at://did:plc:boia3kqcyo3qnjw5fmqknib4/app.bsky.feed.post/3mek7yuhyz22y +at://did:plc:27ivzcszryxp6mehutodmcxo/app.bsky.feed.post/3meka7md4zyfg +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekahdfn5c2e +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekarl54qs2c +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekast2grk24 +at://did:plc:6jrk46j6jdnumfecn6yfs27n/app.bsky.feed.post/3mekay6s34s2a +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekba6dkss24 +at://did:plc:boia3kqcyo3qnjw5fmqknib4/app.bsky.feed.post/3mekba2qxt22h +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekbbtf53c24 +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekbibcmec2u +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekbibcmec2u +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekbmrdsis2u +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekbp2pjhc2u +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekbp2pjhc2u +at://did:plc:boia3kqcyo3qnjw5fmqknib4/app.bsky.feed.post/3mekbtrqwfk2h +at://did:plc:z6b4y3evxxb5ufbfn5pa65fa/app.bsky.feed.post/3mekbw2alvc2h +at://did:plc:qcnyz4einvy2snq6rjykmays/app.bsky.feed.post/3mekcmz3fjk2w +at://did:plc:l46arqe6yfgh36h3o554iyvr/app.bsky.feed.post/3mekdfnzuak2r +at://did:plc:l46arqe6yfgh36h3o554iyvr/app.bsky.feed.post/3mekdlpqers2r +at://did:plc:l46arqe6yfgh36h3o554iyvr/app.bsky.feed.post/3meke2s4bys2r +at://did:plc:nbfjoeficjzf3pejpontvril/app.bsky.feed.post/3mekehdejp224 +at://did:plc:nbfjoeficjzf3pejpontvril/app.bsky.feed.post/3mekelmol3224 +at://did:plc:vtdq3y23vehztw33eohawmpz/app.bsky.feed.post/3mekemlqhr227 +at://did:plc:vtdq3y23vehztw33eohawmpz/app.bsky.feed.post/3mekemlqhr227 +at://did:plc:nbfjoeficjzf3pejpontvril/app.bsky.feed.post/3mekelmol3224 +at://did:plc:gfrmhdmjvxn2sjedzboeudef/app.bsky.feed.post/3mekerkqj5c2u +at://did:plc:mxzuau6m53jtdsbqe6f4laov/app.bsky.feed.post/3mekexz5tb22r +at://did:plc:azy5dtqi56fxhgohal7ytb2a/app.bsky.feed.post/3mekfdyasek2f +at://did:plc:l46arqe6yfgh36h3o554iyvr/app.bsky.feed.post/3mekltrl6l22r +at://did:plc:boia3kqcyo3qnjw5fmqknib4/app.bsky.feed.post/3meknmqwazc2y +at://did:plc:boia3kqcyo3qnjw5fmqknib4/app.bsky.feed.post/3meko2oalr22y diff --git a/handlers/src/notification-handler.ts b/handlers/src/notification-handler.ts index 7b74c7f..2c6f493 100644 --- a/handlers/src/notification-handler.ts +++ b/handlers/src/notification-handler.ts @@ -39,6 +39,16 @@ interface QueueItem { /** * Check if we've already processed this notification */ +function yamlEscape(text: string): string { + // Replace newlines with spaces, escape quotes, and wrap in quotes + const clean = text + .replace(/\n/g, " ") + .replace(/"/g, '\\"') + .replace(/\\/g, "\\\\") + .slice(0, 500); // Truncate long texts + return `"${clean}"`; +} + function alreadyProcessed(id: string): boolean { const patterns = [ path.join(BLUESKY_DRAFTS, `reply-${id}.txt`), @@ -228,7 +238,7 @@ type: escalate reason: "brief reason this needs Central" uri: {notification uri} author: {author} -original_text: "{text}" +original_text: ${yamlEscape(text)} --- \\\`\\\`\\\` Only write simple, short replies for straightforward mentions (acknowledgments, simple factual responses, thanks). @@ -259,7 +269,7 @@ reply_root: {root uri} reply_root_cid: {root cid} author: {author} priority: {priority} -original_text: "{text}" +original_text: ${yamlEscape(text)} drafted_at: {ISO timestamp} --- Your actual response here (under 280 chars) diff --git a/handlers/src/publisher.ts b/handlers/src/publisher.ts index 850bbbd..57443e4 100644 --- a/handlers/src/publisher.ts +++ b/handlers/src/publisher.ts @@ -297,8 +297,10 @@ async function publishDrafts(options: { all?: boolean; auto?: boolean }) { console.log(` Reason: ${(draft.frontmatter as any).reason || "unknown"}`); console.log(` Moving to review queue for Central to handle`); // Move to review dir with escalate prefix so handler picks it up next cycle - const filename = path.basename(filePath); - const reviewPath = path.join(REVIEW_DRAFTS, `escalated-${filename}`); + let filename = path.basename(filePath); + // Strip any existing escalate/escalated prefixes to prevent name growth + filename = filename.replace(/^(escalated?-)+/, ""); + const reviewPath = path.join(REVIEW_DRAFTS, `escalate-${filename}`); fs.renameSync(filePath, reviewPath); continue; } diff --git a/handlers/src/x-handler.ts b/handlers/src/x-handler.ts index b85d3bd..90a7e10 100644 --- a/handlers/src/x-handler.ts +++ b/handlers/src/x-handler.ts @@ -35,6 +35,15 @@ interface XQueueItem { /** * Check if we've already processed this notification */ +function yamlEscape(text: string): string { + const clean = text + .replace(/\n/g, " ") + .replace(/"/g, '\\"') + .replace(/\\/g, "\\\\") + .slice(0, 500); + return `"${clean}"`; +} + function alreadyProcessed(id: string): boolean { const patterns = [ path.join(X_DRAFTS, `reply-${id}.txt`), @@ -149,7 +158,7 @@ type: reply reply_to: {tweet_id} author: {author} priority: {priority} -original_text: "{text}" +original_text: ${yamlEscape(text)} drafted_at: {ISO timestamp} --- Your actual response here (under 280 chars) diff --git a/indexer/backfill.py b/indexer/backfill.py new file mode 100644 index 0000000..f777652 --- /dev/null +++ b/indexer/backfill.py @@ -0,0 +1,229 @@ +"""Backfill historical records from ATProto into the indexer. + +Fetches all records for each agent's relevant collections from their PDS, +embeds them, and stores in the database. Skips records already indexed. + +Usage: + uv run python -m indexer.backfill # All agents + uv run python -m indexer.backfill --did did:plc:... # Specific agent + uv run python -m indexer.backfill --dry-run # Count only +""" + +import argparse +import logging +import time +from datetime import datetime +from typing import Optional + +import httpx + +from indexer import db, embeddings +from indexer.worker import BASE_COLLECTIONS, SEED_DIDS + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", +) +logger = logging.getLogger(__name__) + + +def resolve_pds(did: str) -> Optional[str]: + """Resolve DID to PDS endpoint.""" + try: + resp = httpx.get(f"https://plc.directory/{did}", timeout=10) + resp.raise_for_status() + doc = resp.json() + return doc["service"][0]["serviceEndpoint"] + except Exception as e: + logger.error(f"Failed to resolve PDS for {did}: {e}") + return None + + +def resolve_handle(did: str) -> Optional[str]: + """Resolve DID to handle.""" + try: + resp = httpx.get(f"https://plc.directory/{did}", timeout=10) + resp.raise_for_status() + doc = resp.json() + for alias in doc.get("alsoKnownAs", []): + if alias.startswith("at://"): + return alias[5:] + except Exception: + pass + return None + + +def list_records(pds: str, did: str, collection: str): + """List all records in a collection with pagination.""" + cursor = None + while True: + params = { + "repo": did, + "collection": collection, + "limit": 100, + } + if cursor: + params["cursor"] = cursor + + resp = httpx.get( + f"{pds}/xrpc/com.atproto.repo.listRecords", + params=params, + timeout=30, + ) + + if resp.status_code == 400: + # Collection doesn't exist for this repo + return + resp.raise_for_status() + data = resp.json() + + records = data.get("records", []) + if not records: + return + + for record in records: + yield record + + cursor = data.get("cursor") + if not cursor: + break + + +def get_existing_uris(engine) -> set: + """Get set of all URIs already in the database.""" + session = db.get_session(engine) + try: + from sqlalchemy import text + result = session.execute(text("SELECT uri FROM cognition_records")) + return {row[0] for row in result} + finally: + session.close() + + +def backfill_agent(engine, did: str, handle: str, pds: str, collections: list[str], dry_run: bool = False): + """Backfill all collections for a single agent.""" + logger.info(f"Backfilling @{handle} ({did}) from {pds}") + + existing = get_existing_uris(engine) + total_new = 0 + total_skipped = 0 + + for collection in collections: + new = 0 + skipped = 0 + batch_texts = [] + batch_records = [] + + for record in list_records(pds, did, collection): + uri = record["uri"] + if uri in existing: + skipped += 1 + continue + + value = record.get("value", {}) + content = embeddings.extract_content(value) + if not content: + skipped += 1 + continue + + rkey = uri.split("/")[-1] + created_at = None + if created_str := value.get("createdAt"): + try: + created_at = datetime.fromisoformat( + created_str.replace("Z", "+00:00") + ) + except ValueError: + pass + + batch_texts.append(content) + batch_records.append({ + "uri": uri, + "did": did, + "collection": collection, + "rkey": rkey, + "content": content, + "created_at": created_at, + "handle": handle, + }) + + # Process in batches of 100 + if len(batch_texts) >= 100: + if not dry_run: + _store_batch(engine, batch_texts, batch_records) + new += len(batch_texts) + batch_texts = [] + batch_records = [] + # Rate limit + time.sleep(0.5) + + # Process remaining + if batch_texts: + if not dry_run: + _store_batch(engine, batch_texts, batch_records) + new += len(batch_texts) + + if new > 0 or skipped > 0: + logger.info(f" {collection}: {new} new, {skipped} skipped") + total_new += new + total_skipped += skipped + + logger.info(f" Total: {total_new} new, {total_skipped} skipped") + return total_new + + +def _store_batch(engine, texts: list[str], records: list[dict]): + """Embed and store a batch of records.""" + try: + embs = embeddings.embed_batch(texts) + except Exception as e: + logger.error(f"Embedding failed: {e}") + return + + session = db.get_session(engine) + try: + for rec, emb in zip(records, embs): + db.upsert_record( + session, + uri=rec["uri"], + did=rec["did"], + collection=rec["collection"], + rkey=rec["rkey"], + content=rec["content"], + embedding=emb, + created_at=rec["created_at"], + handle=rec["handle"], + ) + except Exception as e: + logger.error(f"Store failed: {e}") + session.rollback() + finally: + session.close() + + +def main(): + parser = argparse.ArgumentParser(description="Backfill ATProto records into indexer") + parser.add_argument("--did", help="Backfill specific DID only") + parser.add_argument("--dry-run", action="store_true", help="Count records without indexing") + args = parser.parse_args() + + engine = db.get_engine() + db.init_db(engine) + + dids = [args.did] if args.did else sorted(SEED_DIDS) + collections = sorted(BASE_COLLECTIONS) + + total = 0 + for did in dids: + pds = resolve_pds(did) + if not pds: + continue + handle = resolve_handle(did) or did[:30] + count = backfill_agent(engine, did, handle, pds, collections, dry_run=args.dry_run) + total += count + + logger.info(f"Backfill complete. {total} new records indexed.") + + +if __name__ == "__main__": + main() diff --git a/indexer/indexer/db.py b/indexer/indexer/db.py index 16eb7ac..9ff1e02 100644 --- a/indexer/indexer/db.py +++ b/indexer/indexer/db.py @@ -89,27 +89,60 @@ def init_db(engine): ) conn.commit() - # Create IVFFlat index for vector similarity (after table exists) + # Migrate embedding dimension if needed (384 -> 1536) with engine.connect() as conn: - # Check if index exists result = conn.execute( text( - "SELECT 1 FROM pg_indexes WHERE indexname = 'idx_embedding_ivfflat'" + "SELECT atttypmod FROM pg_attribute " + "WHERE attrelid = 'cognition_records'::regclass " + "AND attname = 'embedding'" ) ) - if not result.fetchone(): - # Create IVFFlat index with cosine similarity + row = result.fetchone() + if row and row[0] != EMBEDDING_DIM: + old_dim = row[0] + print(f"Migrating embedding column: {old_dim} -> {EMBEDDING_DIM} dimensions") + # Drop old IVFFlat index first + conn.execute(text("DROP INDEX IF EXISTS idx_embedding_ivfflat")) + # Clear old embeddings first (can't alter type with mismatched dimensions) + conn.execute( + text("UPDATE cognition_records SET embedding = NULL") + ) + # Alter column dimension conn.execute( text( - """ - CREATE INDEX idx_embedding_ivfflat - ON cognition_records - USING ivfflat (embedding vector_cosine_ops) - WITH (lists = 100) - """ + f"ALTER TABLE cognition_records " + f"ALTER COLUMN embedding TYPE vector({EMBEDDING_DIM})" ) ) conn.commit() + print(f"Migration complete. All embeddings cleared for re-generation.") + + # Create IVFFlat index for vector similarity (after table exists) + with engine.connect() as conn: + # Check if index exists + result = conn.execute( + text( + "SELECT 1 FROM pg_indexes WHERE indexname = 'idx_embedding_ivfflat'" + ) + ) + if not result.fetchone(): + # Need enough records for IVFFlat (lists * 39 minimum) + count = conn.execute( + text("SELECT count(*) FROM cognition_records WHERE embedding IS NOT NULL") + ).scalar() + if count and count >= 100: + conn.execute( + text( + """ + CREATE INDEX idx_embedding_ivfflat + ON cognition_records + USING ivfflat (embedding vector_cosine_ops) + WITH (lists = 50) + """ + ) + ) + conn.commit() def upsert_record( diff --git a/indexer/indexer/embeddings.py b/indexer/indexer/embeddings.py index b8521a8..fcdc324 100644 --- a/indexer/indexer/embeddings.py +++ b/indexer/indexer/embeddings.py @@ -1,23 +1,34 @@ -"""Embedding generation using local models via fastembed (ONNX runtime).""" +"""Embedding generation using OpenAI text-embedding-3-small API.""" +import os from typing import Optional -from fastembed import TextEmbedding +import httpx -# all-MiniLM-L6-v2: 384 dimensions, ~22MB, fast on CPU -DEFAULT_MODEL = "sentence-transformers/all-MiniLM-L6-v2" -EMBEDDING_DIM = 384 +# OpenAI text-embedding-3-small: 1536 dimensions, $0.02/1M tokens +EMBEDDING_MODEL = "text-embedding-3-small" +EMBEDDING_DIM = 1536 +OPENAI_API_URL = "https://api.openai.com/v1/embeddings" -# Lazy-loaded singleton -_model: Optional[TextEmbedding] = None +# Reusable client +_client: Optional[httpx.Client] = None -def get_model() -> TextEmbedding: - """Get or create the embedding model (lazy singleton).""" - global _model - if _model is None: - _model = TextEmbedding(model_name=DEFAULT_MODEL) - return _model +def _get_client() -> httpx.Client: + """Get or create the HTTP client.""" + global _client + if _client is None: + api_key = os.environ.get("OPENAI_API_KEY") + if not api_key: + raise ValueError("OPENAI_API_KEY not set") + _client = httpx.Client( + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + timeout=30, + ) + return _client def embed_text(text: str) -> list[float]: @@ -25,11 +36,15 @@ def embed_text(text: str) -> list[float]: Generate embedding for a single text. Returns: - List of floats (384-dim for all-MiniLM-L6-v2) + List of floats (1536-dim for text-embedding-3-small) """ - model = get_model() - embeddings = list(model.embed([text])) - return embeddings[0].tolist() + client = _get_client() + resp = client.post( + OPENAI_API_URL, + json={"input": text, "model": EMBEDDING_MODEL}, + ) + resp.raise_for_status() + return resp.json()["data"][0]["embedding"] def embed_batch(texts: list[str]) -> list[list[float]]: @@ -37,7 +52,7 @@ def embed_batch(texts: list[str]) -> list[list[float]]: Generate embeddings for multiple texts. Args: - texts: List of texts to embed + texts: List of texts to embed (max 2048 per request) Returns: List of embeddings in same order as input @@ -45,9 +60,23 @@ def embed_batch(texts: list[str]) -> list[list[float]]: if not texts: return [] - model = get_model() - embeddings = list(model.embed(texts)) - return [e.tolist() for e in embeddings] + client = _get_client() + all_embeddings = [] + + # OpenAI supports up to 2048 inputs per request + for i in range(0, len(texts), 2048): + batch = texts[i : i + 2048] + resp = client.post( + OPENAI_API_URL, + json={"input": batch, "model": EMBEDDING_MODEL}, + ) + resp.raise_for_status() + data = resp.json()["data"] + # Sort by index to maintain order + data.sort(key=lambda x: x["index"]) + all_embeddings.extend([d["embedding"] for d in data]) + + return all_embeddings def extract_content(record: dict) -> Optional[str]: diff --git a/indexer/indexer/worker.py b/indexer/indexer/worker.py index 7258092..3b52ffe 100644 --- a/indexer/indexer/worker.py +++ b/indexer/indexer/worker.py @@ -10,8 +10,10 @@ import logging import os import signal import sys +import threading import time from datetime import datetime +from pathlib import Path from typing import Optional import httpx @@ -58,6 +60,8 @@ BASE_COLLECTIONS = [ "systems.witchcraft.announcement", # site.standard.* - astral's essays "site.standard.document", + # social.astral.* - astral's agent catalog + "social.astral.catalog.agent", ] # Seed DIDs - always indexed, even without a profile record @@ -77,6 +81,72 @@ SEED_DIDS = { # Profile collection used for self-registration PROFILE_COLLECTION = "network.comind.agent.profile" +# Ask agent configuration +ASK_DID = "did:plc:i2kylvv6t74i7ikrudlzowms" +ASK_HANDLE = "ask.comind.network" +ASK_SENT_FILE = Path(__file__).parent.parent.parent / "data" / "ask_sent.txt" + + +def _ask_respond_thread(uri: str, cid: str, did: str, text: str): + """Handle an @ask.comind.network mention in a background thread.""" + # Import here to avoid circular imports and keep worker standalone + sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + from tools.ask_responder import ( + build_source_text_and_facets, + extract_sources_from_response, + get_bsky_session, + get_thread_context, + get_thread_root, + post_reply, + save_sent, + send_to_agent, + ) + + try: + # Strip handle from question + question = text.replace(f"@{ASK_HANDLE}", "").strip() + if not question: + save_sent(uri) + return + + logger.info(f"[ask] Processing mention: {question[:80]}") + + # Small delay for public API propagation (firehose is faster than appview) + time.sleep(2) + + session = get_bsky_session() + context = get_thread_context(session, uri) + reply_text, raw_results = send_to_agent(question, context) + + if not reply_text: + logger.warning(f"[ask] No response from agent for {uri}") + save_sent(uri) + return + + clean_answer, source_uris = extract_sources_from_response(reply_text, raw_results) + logger.info(f"[ask] Answer ({len(clean_answer)} chars): {clean_answer[:100]}...") + logger.info(f"[ask] Sources: {len(source_uris)} URIs, raw_results: {len(raw_results)}") + if source_uris: + logger.info(f"[ask] Source URIs: {source_uris[:2]}") + source_text, source_facets = build_source_text_and_facets(source_uris) + + root_uri, root_cid = get_thread_root(session, uri) + result = post_reply( + session, clean_answer, uri, cid, + root_uri=root_uri, root_cid=root_cid, + source_text=source_text if source_text else None, + source_facets=source_facets if source_facets else None, + ) + logger.info(f"[ask] Replied: {result.get('uri', 'ok')}") + save_sent(uri) + + except Exception as e: + logger.error(f"[ask] Error responding to {uri}: {e}") + try: + save_sent(uri) + except Exception: + pass + class IndexerWorker: """Worker that consumes Jetstream and indexes cognition records. @@ -200,6 +270,31 @@ class IndexerWorker: collection = commit.get("collection") did = message.get("did") + # Check for @ask.comind.network mentions in any post + if ( + collection == "app.bsky.feed.post" + and operation == "create" + and did != ASK_DID + ): + record = commit.get("record", {}) + post_text = record.get("text", "") + if f"@{ASK_HANDLE}" in post_text: + rkey = commit.get("rkey", "") + uri = f"at://{did}/app.bsky.feed.post/{rkey}" + # Check dedup + sent = set() + if ASK_SENT_FILE.exists(): + sent = set(ASK_SENT_FILE.read_text().strip().split("\n")) + if uri not in sent: + cid = commit.get("cid", "") + t = threading.Thread( + target=_ask_respond_thread, + args=(uri, cid, did, post_text), + daemon=True, + ) + t.start() + logger.info(f"[ask] Spawned response thread for {uri}") + # Self-registration: accept profile records from ANY DID if ( collection == PROFILE_COLLECTION diff --git a/indexer/pyproject.toml b/indexer/pyproject.toml index 2d77d80..6c660a1 100644 --- a/indexer/pyproject.toml +++ b/indexer/pyproject.toml @@ -7,7 +7,7 @@ dependencies = [ "flask>=3.0", "lexrpc[flask]>=2.0", "atproto>=0.0.55", - "fastembed>=0.4", + "psycopg2-binary>=2.9", "pgvector>=0.3", "sqlalchemy>=2.0", diff --git a/tools/ask_responder.py b/tools/ask_responder.py new file mode 100644 index 0000000..00f4ac1 --- /dev/null +++ b/tools/ask_responder.py @@ -0,0 +1,569 @@ +""" +Responder for @ask.comind.network + +Checks notifications, sends questions to the ask agent via Letta API, +posts replies to Bluesky. + +Usage: + uv run python -m tools.ask_responder # Check and respond + uv run python -m tools.ask_responder --dry-run # Show what would happen + +Env vars: + ASK_BSKY_HANDLE, ASK_APP_PASSWORD, ASK_ATPROTO_PDS + CAMERON_LETTA_API_KEY (or LETTA_API_KEY) +""" + +import argparse +import json +import os +import sys +import time +from pathlib import Path + +import httpx + +ASK_AGENT_ID = "agent-5f18498b-9656-4a28-aabd-fa3d9c43c2a0" +LETTA_BASE = "https://api.letta.com/v1" +SENT_FILE = Path(__file__).parent.parent / "data" / "ask_sent.txt" + + +def get_bsky_session(): + """Authenticate with Bluesky.""" + pds = os.environ["ASK_ATPROTO_PDS"] + handle = os.environ["ASK_BSKY_HANDLE"] + password = os.environ["ASK_APP_PASSWORD"] + + resp = httpx.post( + f"{pds}/xrpc/com.atproto.server.createSession", + json={"identifier": handle, "password": password}, + timeout=10, + ) + resp.raise_for_status() + return resp.json() + + +def get_notifications(session): + """Fetch recent notifications.""" + pds = os.environ["ASK_ATPROTO_PDS"] + resp = httpx.get( + f"{pds}/xrpc/app.bsky.notification.listNotifications", + headers={"Authorization": f"Bearer {session['accessJwt']}"}, + params={"limit": 20}, + timeout=10, + ) + resp.raise_for_status() + return resp.json().get("notifications", []) + + +def get_post(session, uri): + """Fetch a post by AT URI.""" + pds = os.environ["ASK_ATPROTO_PDS"] + resp = httpx.get( + f"{pds}/xrpc/com.atproto.repo.getRecord", + headers={"Authorization": f"Bearer {session['accessJwt']}"}, + params={ + "repo": uri.split("/")[2], + "collection": uri.split("/")[3], + "rkey": uri.split("/")[4], + }, + timeout=10, + ) + if resp.status_code == 200: + return resp.json() + return None + + +def get_thread_context(session, uri, depth=3): + """Get thread context for a mention.""" + pds = os.environ["ASK_ATPROTO_PDS"] + # Use public API for thread resolution + resp = httpx.get( + "https://public.api.bsky.app/xrpc/app.bsky.feed.getPostThread", + params={"uri": uri, "depth": 0, "parentHeight": depth}, + timeout=10, + ) + if resp.status_code != 200: + return "" + + thread = resp.json().get("thread", {}) + parts = [] + + # Walk up parent chain + current = thread + while current.get("parent"): + current = current["parent"] + post = current.get("post", {}) + author = post.get("author", {}).get("handle", "unknown") + text = post.get("record", {}).get("text", "") + if text: + parts.append(f"@{author}: {text}") + + parts.reverse() + return "\n".join(parts[-3:]) # Last 3 parents max + + +INDEXER_BASE = "https://comind-indexer.fly.dev/xrpc" + + +def search_index(query, limit=10): + """Search the comind XRPC indexer, filtering out low-value infrastructure records.""" + resp = httpx.get( + f"{INDEXER_BASE}/network.comind.search.query", + params={"q": query, "limit": limit}, + timeout=15, + ) + if resp.status_code != 200: + return [] + + results = resp.json().get("results", []) + + # Deprioritize infrastructure noise (reasoning traces, activity records) + noise_collections = {"network.comind.reasoning", "network.comind.activity", "network.comind.response"} + good = [r for r in results if r.get("collection") not in noise_collections] + rest = [r for r in results if r.get("collection") in noise_collections] + + return (good + rest)[:limit] + + +def list_agents(): + """Get the agent directory from the indexer.""" + resp = httpx.get( + f"{INDEXER_BASE}/network.comind.agents.list", + timeout=15, + ) + if resp.status_code != 200: + return [] + return resp.json().get("agents", []) + + +def format_search_results(results): + """Format search results for the agent prompt.""" + if not results: + return "No results found in the index." + + lines = [] + for i, r in enumerate(results, 1): + handle = r.get("handle", r.get("did", "unknown")[:30]) + collection = r.get("collection", "") + content = (r.get("content") or "")[:300] + uri = r.get("uri", "") + score = r.get("score", 0) + lines.append( + f"Result {i} (score: {score:.2f}, @{handle}, {collection}):\n" + f" \"{content}\"\n" + f" URI: {uri}" + ) + return "\n\n".join(lines) + + +def format_agent_directory(agents): + """Format agent directory for the agent prompt.""" + lines = [] + for a in agents: + handle = a.get("handle", a.get("did", "unknown")) + count = a.get("recordCount", 0) + collections = ", ".join(a.get("collections", [])) + profile = a.get("profile", "") + line = f"@{handle}: {count} records. Collections: {collections}" + if profile: + line += f". Profile: {profile[:150]}" + lines.append(line) + return "\n".join(lines) + + +def is_directory_question(question): + """Check if the question is about what agents are indexed.""" + q = question.lower() + return any(kw in q for kw in [ + "what agents", "who is indexed", "which agents", "list agents", + "who's in", "who is in", "agent directory", "how many agents", + "what's indexed", "what is indexed", + ]) + + +def build_agent_prompt(question, thread_context="", search_results="", agent_directory=""): + """Build the full prompt with pre-fetched context.""" + parts = [] + + if thread_context: + parts.append(f"Thread context:\n{thread_context}") + + if agent_directory: + parts.append(f"Agent directory:\n{agent_directory}") + + if search_results: + parts.append(f"Search results:\n{search_results}") + + parts.append(f'Question: "{question}"') + parts.append("Synthesize the search results into a reply. Under 280 chars. Just the reply text.") + + return "\n\n".join(parts) + + +def send_to_agent(question, thread_context=""): + """Send question to the ask agent and let it search with its tools. + + Returns (response_text, source_results). + """ + key = os.environ.get("LETTA_API_KEY") or os.environ.get("CAMERON_LETTA_API_KEY") + headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"} + + parts = [] + if thread_context: + parts.append(f"Thread context:\n{thread_context}") + + parts.append(f'Question from a Bluesky user: "{question}"') + parts.append("Search the index, then write a reply under 280 chars. Just the reply text.") + + prompt = "\n\n".join(parts) + + resp = httpx.post( + f"{LETTA_BASE}/agents/{ASK_AGENT_ID}/messages", + headers=headers, + json={ + "messages": [{"role": "user", "content": prompt}], + "stream_tokens": False, + }, + timeout=120, + ) + resp.raise_for_status() + messages = resp.json().get("messages", []) + + # Extract source URIs from tool returns + source_results = [] + seen_uris = set() + for msg in messages: + if msg.get("message_type") in ("tool_return", "tool_return_message"): + content = msg.get("content", "") or msg.get("tool_return", "") + current_handle = "" + for line in content.split("\n"): + line = line.strip() + if line.startswith("Result ") and "@" in line: + try: + current_handle = line.split("@")[1].split(",")[0] + except (IndexError, ValueError): + current_handle = "" + if line.startswith("URI: at://"): + uri = line.replace("URI: ", "") + if uri not in seen_uris: + seen_uris.add(uri) + source_results.append({"uri": uri, "handle": current_handle}) + + # Find the assistant's text response + for msg in messages: + if msg.get("message_type") in ("assistant_message", "assistant"): + content = msg.get("content", "") + if content: + return content, source_results[:3] + + return None, source_results[:3] + + +def at_uri_to_web_url(uri, handle=None): + """Convert an AT URI to a web URL where possible.""" + # at://did:plc:xxx/app.bsky.feed.post/rkey -> https://bsky.app/profile/handle/post/rkey + parts = uri.replace("at://", "").split("/") + if len(parts) < 3: + return None + did, collection, rkey = parts[0], parts[1], parts[2] + + if collection == "app.bsky.feed.post": + profile = handle or did + return f"https://bsky.app/profile/{profile}/post/{rkey}" + + # For non-post records, link to the indexer + return f"https://comind-indexer.fly.dev/xrpc/network.comind.search.query?q={rkey}" + + +def build_source_text_and_facets(source_uris): + """Build the 'Sources: [1] [2] [3]' text with link facets. + + Returns (source_text, facets) where facets have byte offsets relative to source_text. + """ + if not source_uris: + return "", [] + + source_text = "Sources: " + labels = [] + for i, (uri, handle) in enumerate(source_uris, 1): + labels.append(f"[{i}]") + source_text += " ".join(labels) + + facets = [] + for i, (uri, handle) in enumerate(source_uris, 1): + label = f"[{i}]" + label_bytes = label.encode("utf-8") + search_start = source_text.encode("utf-8").find(label_bytes) + if search_start == -1: + continue + web_url = at_uri_to_web_url(uri, handle) + if not web_url: + continue + facets.append({ + "index": { + "byteStart": search_start, + "byteEnd": search_start + len(label_bytes), + }, + "features": [{ + "$type": "app.bsky.richtext.facet#link", + "uri": web_url, + }], + }) + + return source_text, facets + + +def extract_sources_from_response(response_text, search_results): + """Extract the answer text and map sources back to URIs. + + Returns (clean_answer, [(uri, handle), ...]) + """ + # Split on "Sources:" or "Source:" line + answer = response_text + for separator in ["\nSources:", "\nSource:", "\n\nSources:", "\n\nSource:"]: + if separator in response_text: + answer = response_text[:response_text.index(separator)].strip() + break + + # Map results to (uri, handle) pairs, deduped + seen = set() + source_uris = [] + for r in search_results: + uri = r.get("uri", "") + handle = r.get("handle", "") + if uri and uri not in seen: + seen.add(uri) + source_uris.append((uri, handle)) + + # Limit to 3 sources to keep compact + return answer, source_uris[:3] + + +def _create_post(session, text, reply_ref, facets=None): + """Create a single post record. Returns {uri, cid}.""" + pds = os.environ["ASK_ATPROTO_PDS"] + + record = { + "$type": "app.bsky.feed.post", + "text": text, + "reply": reply_ref, + "createdAt": time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime()), + } + + if facets: + record["facets"] = facets + + resp = httpx.post( + f"{pds}/xrpc/com.atproto.repo.createRecord", + headers={"Authorization": f"Bearer {session['accessJwt']}"}, + json={ + "repo": session["did"], + "collection": "app.bsky.feed.post", + "record": record, + }, + timeout=10, + ) + resp.raise_for_status() + return resp.json() + + +def _split_text(text, limit=300): + """Split text into chunks of at most `limit` graphemes, breaking at sentence boundaries.""" + if len(text) <= limit: + return [text] + + chunks = [] + remaining = text + while len(remaining) > limit: + # Try to break at a sentence boundary + cut = remaining[:limit] + # Look for last sentence-ending punctuation + best = -1 + for sep in [". ", "! ", "? ", ".\n", "!\n", "?\n"]: + idx = cut.rfind(sep) + if idx > best: + best = idx + + if best > limit // 3: + # Break after the punctuation + chunks.append(remaining[:best + 1].rstrip()) + remaining = remaining[best + 1:].lstrip() + else: + # No good sentence break, try comma or space + space = cut.rfind(" ") + if space > limit // 3: + chunks.append(remaining[:space].rstrip()) + remaining = remaining[space + 1:].lstrip() + else: + # Hard break + chunks.append(remaining[:limit]) + remaining = remaining[limit:] + + if remaining.strip(): + chunks.append(remaining.strip()) + + return chunks + + +def post_reply(session, answer_text, reply_to_uri, reply_to_cid, root_uri=None, root_cid=None, source_text=None, source_facets=None): + """Post a reply to Bluesky. Threads long answers, then appends sources.""" + root_ref = { + "uri": root_uri or reply_to_uri, + "cid": root_cid or reply_to_cid, + } + + # Split answer into 300-char chunks + chunks = _split_text(answer_text) + + # Post first chunk as reply to the mention + reply_ref = { + "root": root_ref, + "parent": {"uri": reply_to_uri, "cid": reply_to_cid}, + } + result = _create_post(session, chunks[0], reply_ref) + last_result = result + + # Post remaining chunks as thread + for chunk in chunks[1:]: + reply_ref = { + "root": root_ref, + "parent": {"uri": last_result["uri"], "cid": last_result["cid"]}, + } + last_result = _create_post(session, chunk, reply_ref) + + # Post sources as final thread reply + if source_text: + source_reply_ref = { + "root": root_ref, + "parent": {"uri": last_result["uri"], "cid": last_result["cid"]}, + } + _create_post(session, source_text, source_reply_ref, source_facets) + + return result + + +def load_sent(): + """Load set of already-responded URIs.""" + if SENT_FILE.exists(): + return set(SENT_FILE.read_text().strip().split("\n")) + return set() + + +def save_sent(uri): + """Record a URI as responded to.""" + SENT_FILE.parent.mkdir(parents=True, exist_ok=True) + with open(SENT_FILE, "a") as f: + f.write(uri + "\n") + + +def get_thread_root(session, uri): + """Get the root post of a thread.""" + resp = httpx.get( + "https://public.api.bsky.app/xrpc/app.bsky.feed.getPostThread", + params={"uri": uri, "depth": 0, "parentHeight": 10}, + timeout=10, + ) + if resp.status_code != 200: + return None, None + + thread = resp.json().get("thread", {}) + # Walk to root + current = thread + while current.get("parent"): + current = current["parent"] + + post = current.get("post", {}) + return post.get("uri"), post.get("cid") + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + + session = get_bsky_session() + print(f"Authenticated as @{session['handle']}") + + notifications = get_notifications(session) + sent = load_sent() + + # Filter to direct mentions only (post must contain our handle) + our_handle = session["handle"] + mentions = [ + n for n in notifications + if n.get("reason") == "mention" + and n.get("uri") not in sent + and n.get("author", {}).get("handle") != our_handle # skip self + and f"@{our_handle}" in (n.get("record", {}).get("text", "")) # must directly mention us + ] + + print(f"Found {len(mentions)} new mentions") + + for mention in mentions: + author = mention.get("author", {}).get("handle", "unknown") + uri = mention["uri"] + cid = mention["cid"] + text = mention.get("record", {}).get("text", "") + + # Strip the @ask.comind.network from the question + question = text.replace("@ask.comind.network", "").strip() + + print(f"\n[@{author}] {question[:100]}") + + if not question: + print(" (empty question, skipping)") + save_sent(uri) + continue + + if args.dry_run: + print(" [DRY RUN] Would send to agent and reply") + continue + + # Get thread context + context = get_thread_context(session, uri) + if context: + print(f" Thread context: {context[:100]}...") + + # Send to agent + print(" Sending to ask agent...") + try: + reply_text, raw_results = send_to_agent(question, context) + except Exception as e: + print(f" ERROR from agent: {e}") + save_sent(uri) + continue + + if not reply_text: + print(" (no response from agent)") + save_sent(uri) + continue + + # Strip agent's source lines and build source facets + clean_answer, source_uris = extract_sources_from_response(reply_text, raw_results) + source_text, source_facets = build_source_text_and_facets(source_uris) + + print(f" Answer: {clean_answer[:120]}...") + print(f" Sources: {len(source_uris)} linked") + + # Get thread root for proper threading + root_uri, root_cid = get_thread_root(session, uri) + + # Post reply (answer + threaded sources) + try: + result = post_reply( + session, clean_answer, uri, cid, + root_uri=root_uri, root_cid=root_cid, + source_text=source_text if source_text else None, + source_facets=source_facets if source_facets else None, + ) + print(f" Posted: {result.get('uri', 'ok')}") + except Exception as e: + print(f" POST ERROR: {e}") + + save_sent(uri) + + print("\nDone.") + + +if __name__ == "__main__": + main() diff --git a/tools/ask_search_tool.py b/tools/ask_search_tool.py new file mode 100644 index 0000000..747ac0b --- /dev/null +++ b/tools/ask_search_tool.py @@ -0,0 +1,49 @@ +import httpx + +INDEXER_BASE = "https://comind-indexer.fly.dev/xrpc" + + +def search_comind_index(query: str, limit: int = 5) -> str: + """Search the comind collective intelligence index. + + Searches cognition records (thoughts, memories, concepts, posts) from AI agents on ATProtocol. + Returns results ranked by semantic similarity. + + Args: + query: The search query. Be specific for better results. + limit: Number of results to return (1-20). Default 5. + + Returns: + Formatted search results with content, author handle, collection type, and AT URI. + """ + if limit < 1: + limit = 1 + if limit > 20: + limit = 20 + + resp = httpx.get( + INDEXER_BASE + "/network.comind.search.query", + params={"q": query, "limit": limit}, + timeout=15, + ) + if resp.status_code != 200: + return "Search failed: HTTP " + str(resp.status_code) + + results = resp.json().get("results", []) + if not results: + return "No results found." + + lines = [] + for i, r in enumerate(results, 1): + handle = r.get("handle", r.get("did", "unknown")) + collection = r.get("collection", "") + content = (r.get("content") or "")[:300] + uri = r.get("uri", "") + score = r.get("score", 0) + line = "Result " + str(i) + line += " (score: " + "{:.2f}".format(score) + line += ", @" + handle + ", " + collection + "):\n" + line += " " + content + "\n" + line += " URI: " + uri + lines.append(line) + return "\n\n".join(lines) diff --git a/tools/ask_tools.py b/tools/ask_tools.py new file mode 100644 index 0000000..798557a --- /dev/null +++ b/tools/ask_tools.py @@ -0,0 +1,81 @@ +"""Custom tools for the @ask.comind.network agent.""" + +import httpx + +INDEXER_BASE = "https://comind-indexer.fly.dev/xrpc" + + +def search_comind_index(query: str, limit: int = 5) -> str: + """Search the comind collective intelligence index. + + Searches cognition records (thoughts, memories, concepts, posts) from AI agents on ATProtocol. + Returns results ranked by semantic similarity. + + Args: + query: The search query. Be specific for better results. + limit: Number of results to return (1-20). Default 5. + + Returns: + Formatted search results with content, author handle, collection type, and AT URI. + """ + if limit < 1: + limit = 1 + if limit > 20: + limit = 20 + + resp = httpx.get( + INDEXER_BASE + "/network.comind.search.query", + params={"q": query, "limit": limit}, + timeout=15, + ) + if resp.status_code != 200: + return "Search failed: HTTP " + str(resp.status_code) + + results = resp.json().get("results", []) + if not results: + return "No results found." + + lines = [] + for i, r in enumerate(results, 1): + handle = r.get("handle", r.get("did", "unknown")) + collection = r.get("collection", "") + content = (r.get("content") or "")[:300] + uri = r.get("uri", "") + score = r.get("score", 0) + lines.append( + "Result " + str(i) + " (score: " + "{:.2f}".format(score) + + ", @" + handle + ", " + collection + "):\n" + + " " + content + "\n" + + " URI: " + uri + ) + return "\n\n".join(lines) + + +def list_indexed_agents() -> str: + """List all agents in the comind index with their record counts and collections. + + Returns: + Agent directory showing handle, record count, collections, and profile info. + """ + resp = httpx.get( + INDEXER_BASE + "/network.comind.agents.list", + timeout=15, + ) + if resp.status_code != 200: + return "Failed: HTTP " + str(resp.status_code) + + agents = resp.json().get("agents", []) + if not agents: + return "No agents indexed." + + lines = [] + for a in agents: + handle = a.get("handle", a.get("did", "unknown")) + count = a.get("recordCount", 0) + collections = ", ".join(a.get("collections", [])) + profile = a.get("profile", "") + line = "@" + handle + ": " + str(count) + " records. Collections: " + collections + if profile: + line += ". Profile: " + profile[:150] + lines.append(line) + return "\n".join(lines)