add aborist activity — agent-readable recent timeline
New CLI: aborist [--shards-dir DIR] activity [--limit N] [--since-seconds S]
[--preview-chars N]
Aggregates across all shards into a single JSON document with:
shards the SQLite files queried
corpus counts: documents_total/surface/core,
providence_total/live/stale/failed,
audit_events_total
recent_qa providence_cache rows (cache_key, question, answer
preview, hit_count, state, last_hit_at) newest first
recent_ingests audit events of type=ingest with body's document_uri,
source_type, chunks, supersedes link
recent_derives audit events of type=derive (distillation results,
compression_ratio, compression_depth)
recent_falsifications records marked failed/stale/quarantined,
with reason and by_actor
--limit caps each list independently. --since-seconds filters to recent
events only. --preview-chars trims long answers.
Designed for an agent to read before deciding the next action — what
was just asked, what just got integrated, what's the current state.
Makefile target: make activity ACTIVITY_LIMIT=N
66 tests still passing.
This commit is contained in:
parent
641a4c8bde
commit
76446d4349
2 changed files with 198 additions and 0 deletions
4
Makefile
4
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
|
||||
|
|
|
|||
194
aborist/cli.py
194
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)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue