diff --git a/Makefile b/Makefile index 4b37398..edf3c45 100644 --- a/Makefile +++ b/Makefile @@ -110,6 +110,10 @@ ingest-old-attached: bootstrap fetch-old ## sharded ingest of old history (Phase stats-shards: bootstrap ## cross-shard stats via UNION views over $(SHARDS_DIR) $(ABORIST) --shards-dir $(SHARDS_DIR) stats +ACTIVITY_LIMIT ?= 10 +activity: bootstrap ## recent Q&A + freshly cached docs (agent timeline) + $(ABORIST) --shards-dir $(SHARDS_DIR) activity --limit $(ACTIVITY_LIMIT) + # Multi-source RAG query against the shard cluster. # Usage: make query Q="What is anarcho-capitalism?" QUERY_TOP_K ?= 8 diff --git a/aborist/cli.py b/aborist/cli.py index 11d8201..cb0fbb0 100644 --- a/aborist/cli.py +++ b/aborist/cli.py @@ -426,6 +426,176 @@ def _cmd_rehydrate(args: argparse.Namespace) -> int: return 1 if drift else 0 +def _cmd_activity(args: argparse.Namespace) -> int: + """Recent activity: Q&A records + freshly cached docs across all shards. + + Designed for an agent to read before deciding the next action — what was + just asked, what was just integrated, what's the corpus state. + """ + import time as _time + + from aborist.store import discover_shards + + shard_paths: list[Path] = [] + if args.global_shards_dir: + shard_paths = discover_shards(args.global_shards_dir) + else: + shard_paths = [Path(args.db)] + + cutoff_ts = 0 + if args.since_seconds: + cutoff_ts = int(_time.time()) - args.since_seconds + + qa_records: list[dict] = [] + ingest_events: list[dict] = [] + derive_events: list[dict] = [] + falsifications: list[dict] = [] + corpus = { + "documents_total": 0, + "documents_surface": 0, + "documents_core": 0, + "providence_total": 0, + "providence_live": 0, + "providence_stale": 0, + "providence_failed": 0, + "audit_events_total": 0, + } + + for sp in shard_paths: + c = connect(sp) + try: + corpus["documents_total"] += c.execute( + "SELECT COUNT(*) FROM documents" + ).fetchone()[0] + corpus["documents_surface"] += c.execute( + "SELECT COUNT(*) FROM documents WHERE kind='surface'" + ).fetchone()[0] + corpus["documents_core"] += c.execute( + "SELECT COUNT(*) FROM documents WHERE kind='core'" + ).fetchone()[0] + corpus["providence_total"] += c.execute( + "SELECT COUNT(*) FROM providence_cache" + ).fetchone()[0] + corpus["providence_live"] += c.execute( + "SELECT COUNT(*) FROM providence_cache WHERE falsification_state='live'" + ).fetchone()[0] + corpus["providence_stale"] += c.execute( + "SELECT COUNT(*) FROM providence_cache WHERE falsification_state='stale'" + ).fetchone()[0] + corpus["providence_failed"] += c.execute( + "SELECT COUNT(*) FROM providence_cache WHERE falsification_state='failed'" + ).fetchone()[0] + corpus["audit_events_total"] += c.execute( + "SELECT COUNT(*) FROM audit_events" + ).fetchone()[0] + + # Q&A records + for r in c.execute( + "SELECT cache_key, question_text, answer_text, " + " falsification_state, hit_count, created_at, last_hit_at, " + " document_uri FROM providence_cache " + "WHERE created_at >= ? ORDER BY created_at DESC LIMIT ?", + (cutoff_ts, args.limit), + ).fetchall(): + ans = r["answer_text"] or "" + qa_records.append( + { + "ts": r["created_at"], + "shard": sp.name, + "cache_key": r["cache_key"], + "question": r["question_text"], + "answer_preview": ( + ans if len(ans) <= args.preview_chars + else ans[: args.preview_chars] + "…" + ), + "sources_uri": r["document_uri"], + "state": r["falsification_state"], + "hit_count": r["hit_count"], + "last_hit_at": r["last_hit_at"], + } + ) + + # Recent ingest events + for r in c.execute( + "SELECT subject_root, body, ts FROM audit_events " + "WHERE event_type = 'ingest' AND ts >= ? " + "ORDER BY ts DESC LIMIT ?", + (cutoff_ts, args.limit), + ).fetchall(): + body = json.loads(r["body"]) if r["body"] else {} + ingest_events.append( + { + "ts": r["ts"], + "shard": sp.name, + "document_root": r["subject_root"], + "document_uri": body.get("document_uri"), + "source_type": body.get("source_type"), + "chunks": body.get("chunks"), + "supersedes": body.get("supersedes"), + } + ) + + # Recent derive events (distillations) + for r in c.execute( + "SELECT subject_root, body, ts FROM audit_events " + "WHERE event_type = 'derive' AND ts >= ? " + "ORDER BY ts DESC LIMIT ?", + (cutoff_ts, args.limit), + ).fetchall(): + body = json.loads(r["body"]) if r["body"] else {} + derive_events.append( + { + "ts": r["ts"], + "shard": sp.name, + "core_root": r["subject_root"], + "src_root": body.get("src_root"), + "process_id": body.get("process_id"), + "compression_ratio": body.get("compression_ratio"), + "compression_depth": body.get("compression_depth"), + } + ) + + # Recent falsifications + for r in c.execute( + "SELECT cache_key, state, reason, by_actor, at FROM falsifications " + "WHERE at >= ? ORDER BY at DESC LIMIT ?", + (cutoff_ts, args.limit), + ).fetchall(): + falsifications.append( + { + "ts": r["at"], + "shard": sp.name, + "cache_key": r["cache_key"], + "state": r["state"], + "reason": r["reason"], + "by_actor": r["by_actor"], + } + ) + finally: + c.close() + + qa_records.sort(key=lambda x: -x["ts"]) + ingest_events.sort(key=lambda x: -x["ts"]) + derive_events.sort(key=lambda x: -x["ts"]) + falsifications.sort(key=lambda x: -x["ts"]) + + print( + json.dumps( + { + "as_of": int(_time.time()), + "shards": [str(p) for p in shard_paths], + "corpus": corpus, + "recent_qa": qa_records[: args.limit], + "recent_ingests": ingest_events[: args.limit], + "recent_derives": derive_events[: args.limit], + "recent_falsifications": falsifications[: args.limit], + }, + indent=2, + ) + ) + return 0 + + def _cmd_stats(args: argparse.Namespace) -> int: conn = ( connect_query(args.db, shards_dir=args.global_shards_dir) @@ -858,6 +1028,30 @@ def build_parser() -> argparse.ArgumentParser: ) rehydrate_cmd.set_defaults(func=_cmd_rehydrate) + activity_cmd = sub.add_parser( + "activity", + help="recent Q&A + freshly cached docs (agent-readable timeline)", + ) + activity_cmd.add_argument( + "--limit", type=int, default=10, + help="max items per category (default 10)", + ) + activity_cmd.add_argument( + "--since-seconds", + dest="since_seconds", + type=int, + default=0, + help="only events newer than this many seconds (0 = all time, default)", + ) + activity_cmd.add_argument( + "--preview-chars", + dest="preview_chars", + type=int, + default=240, + help="answer preview length (default 240 chars)", + ) + activity_cmd.set_defaults(func=_cmd_activity) + stats_cmd = sub.add_parser("stats", help="counts: docs, chunks, edges, audit") stats_cmd.set_defaults(func=_cmd_stats)