tooling: histcapture.py (auto HIST-QNA capture) + systemd timer
Captures user/assistant messages from /root/.hermes/state.db into the 'hermes' cube namespace as TYPE: HIST-QNA entries. Forward-only by default (no backfill); --seed-hours N for bounded one-time population; idempotent; --prune-older-than-hours to undo over-backfill. systemd timer runs every 5 min. Tested: 24h seed = 1171 msgs; prune cleared a 3164-entry backfill to 44, then re-seeded correctly.
This commit is contained in:
CUBELinux-2
@@ -0,0 +1,295 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
histcapture.py — automated historical-record capture for CUBELinux-2 work.
|
||||
|
||||
PURPOSE
|
||||
The user wanted the per-exchange historical record (TYPE: HIST-QNA in the
|
||||
'hermes' cube namespace) captured automatically from the Hermes session
|
||||
transcript, so the agent no longer has to log each exchange by hand.
|
||||
|
||||
SOURCE: /root/.hermes/state.db, table `messages` (id, session_id, role,
|
||||
content, timestamp). We capture role IN ('user','assistant') — not 'tool'
|
||||
(tool results are not conversational Q&A).
|
||||
|
||||
DESTINATION: CUBE namespace 'hermes', one entry per message, named
|
||||
"HIST:<ISO8601>:<seq>" with a "TYPE: HIST-QNA" header + JSON body.
|
||||
|
||||
DEFAULT BEHAVIOR — FORWARD ONLY (no backfill)
|
||||
On first run with no state file, we DO NOT start from message id 0. We
|
||||
start from the CURRENT maximum message id, so only future exchanges are
|
||||
captured. This honors the standing "no backfill" rule.
|
||||
|
||||
BOUNDED SEED (explicit, user-approved)
|
||||
`--seed-hours N` populates the last N hours ONCE. This is the only
|
||||
backfill path and is opt-in. The user approved a 24h seed.
|
||||
|
||||
IDEMPOTENCY
|
||||
- A state file records the highest message id pushed; reruns only push
|
||||
newer messages (safe under a timer).
|
||||
- Writes check existence first and skip duplicates (so re-running a seed
|
||||
or a crash mid-run does not create dupes).
|
||||
|
||||
PRUNE
|
||||
`--prune-older-than-hours N` removes entries this script would have written
|
||||
that are older than N hours (used to undo an over-eager backfill). Names
|
||||
are reconstructed deterministically from the DB, so we can target exactly
|
||||
what this script wrote.
|
||||
|
||||
FAILURE
|
||||
If the cubed socket is down, cube_write returns False; we do not advance
|
||||
last_id past the failure point and exit non-zero so a timer sees it.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import base64
|
||||
import socket
|
||||
import sqlite3
|
||||
import argparse
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from typing import Optional
|
||||
|
||||
STATE_DB = "/root/.hermes/state.db"
|
||||
CUBED_SOCKET = os.environ.get("CUBED_SOCKET") or "/run/user/1000/cubelinux/cubed.sock"
|
||||
NAMESPACE = "hermes"
|
||||
STATE_FILE = "/home/CUBELinux/CUBELinux-2/tools/histcapture.state"
|
||||
WINDOW_SECONDS = 3600
|
||||
WALL1_GROUP = 1002 # cubelinux gid; we run as root, group gives socket access
|
||||
|
||||
|
||||
# --- cube RPC --------------------------------------------------------------
|
||||
def _rpc(req: dict) -> dict:
|
||||
payload = (json.dumps(req) + "\n").encode("utf-8")
|
||||
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||||
sock.settimeout(5)
|
||||
try:
|
||||
sock.connect(CUBED_SOCKET)
|
||||
sock.sendall(payload)
|
||||
buf = b""
|
||||
while b"\n" not in buf:
|
||||
chunk = sock.recv(4096)
|
||||
if not chunk:
|
||||
break
|
||||
buf += chunk
|
||||
return json.loads(buf.split(b"\n", 1)[0].decode("utf-8", "replace"))
|
||||
except OSError:
|
||||
return {"ok": False, "error": "socket"}
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
|
||||
def cube_write(name: str, content: str) -> bool:
|
||||
return bool(_rpc({
|
||||
"cmd": "write", "namespace": NAMESPACE, "name": name,
|
||||
"value": base64.b64encode(content.encode("utf-8")).decode("ascii"),
|
||||
"kind": "memory", "visibility": "private",
|
||||
}).get("ok"))
|
||||
|
||||
|
||||
def cube_read(name: str):
|
||||
r = _rpc({"cmd": "read", "namespace": NAMESPACE, "name": name})
|
||||
if r.get("ok") and r.get("value") is not None:
|
||||
return base64.b64decode(r["value"]).decode("utf-8", "replace")
|
||||
return None
|
||||
|
||||
|
||||
def cube_delete(name: str) -> bool:
|
||||
return bool(_rpc({
|
||||
"cmd": "delete", "namespace": NAMESPACE, "name": name,
|
||||
}).get("ok"))
|
||||
|
||||
|
||||
# --- state -----------------------------------------------------------------
|
||||
def read_last_id() -> int:
|
||||
try:
|
||||
with open(STATE_FILE) as f:
|
||||
return int(f.read().strip())
|
||||
except (FileNotFoundError, ValueError):
|
||||
return -1 # sentinel: not initialized
|
||||
|
||||
|
||||
def write_last_id(i: int) -> None:
|
||||
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
|
||||
with open(STATE_FILE, "w") as f:
|
||||
f.write(str(i))
|
||||
|
||||
|
||||
# --- DB --------------------------------------------------------------------
|
||||
def fetch_messages(since_id: int, before_ts: Optional[float] = None):
|
||||
con = sqlite3.connect(STATE_DB)
|
||||
con.row_factory = sqlite3.Row
|
||||
cur = con.cursor()
|
||||
if before_ts is None:
|
||||
cur.execute(
|
||||
"SELECT id, session_id, role, content, timestamp FROM messages "
|
||||
"WHERE id > ? AND role IN ('user','assistant') ORDER BY id ASC",
|
||||
(since_id,),
|
||||
)
|
||||
else:
|
||||
cur.execute(
|
||||
"SELECT id, session_id, role, content, timestamp FROM messages "
|
||||
"WHERE id > ? AND timestamp >= ? AND role IN ('user','assistant') "
|
||||
"ORDER BY id ASC",
|
||||
(since_id, before_ts),
|
||||
)
|
||||
rows = cur.fetchall()
|
||||
con.close()
|
||||
return rows
|
||||
|
||||
|
||||
def max_message_id() -> int:
|
||||
con = sqlite3.connect(STATE_DB)
|
||||
cur = con.cursor()
|
||||
cur.execute("SELECT MAX(id) FROM messages")
|
||||
m = cur.fetchone()[0] or 0
|
||||
con.close()
|
||||
return m
|
||||
|
||||
|
||||
# --- naming / entry --------------------------------------------------------
|
||||
def iso_from_ts(ts: float) -> str:
|
||||
return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
||||
|
||||
|
||||
def build_entry(row, seq: int) -> tuple[str, str]:
|
||||
ts_iso = iso_from_ts(row["timestamp"])
|
||||
name = f"HIST:{ts_iso}:{seq}"
|
||||
content = row["content"] or ""
|
||||
try:
|
||||
obj = json.loads(content)
|
||||
if isinstance(obj, dict):
|
||||
content = obj.get("text") or obj.get("content") or content
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
pass
|
||||
body = {"ts": ts_iso, "role": row["role"], "session": row["session_id"]}
|
||||
if row["role"] == "user":
|
||||
body["query"] = content
|
||||
else:
|
||||
body["response"] = content
|
||||
return name, "TYPE: HIST-QNA\n" + json.dumps(body, indent=2)
|
||||
|
||||
|
||||
def capture(rows, dry_run: bool) -> int:
|
||||
seen: dict[str, int] = {}
|
||||
pushed_max = 0
|
||||
written = 0
|
||||
skipped = 0
|
||||
for r in rows:
|
||||
ts_iso = iso_from_ts(r["timestamp"])
|
||||
seen[ts_iso] = seen.get(ts_iso, 0) + 1
|
||||
name, entry = build_entry(r, seen[ts_iso])
|
||||
if dry_run:
|
||||
print(f"[dry-run] {name} ({r['role']})")
|
||||
continue
|
||||
if cube_read(name) is not None:
|
||||
skipped += 1
|
||||
pushed_max = max(pushed_max, r["id"])
|
||||
continue
|
||||
if cube_write(name, entry):
|
||||
written += 1
|
||||
pushed_max = max(pushed_max, r["id"])
|
||||
else:
|
||||
print(f"WRITE FAILED id {r['id']} ({name})", file=sys.stderr)
|
||||
return pushed_max # stop; caller records progress
|
||||
return pushed_max if (written or skipped) else 0
|
||||
|
||||
|
||||
# --- commands --------------------------------------------------------------
|
||||
def cmd_forward(dry_run: bool) -> int:
|
||||
last = read_last_id()
|
||||
if last == -1:
|
||||
# FORWARD-ONLY default: start from current max, capture nothing old.
|
||||
last = max_message_id()
|
||||
print(f"no state: initializing forward-only from id {last} "
|
||||
f"(no backfill)")
|
||||
if not dry_run:
|
||||
write_last_id(last)
|
||||
return 0
|
||||
rows = fetch_messages(last)
|
||||
if not rows:
|
||||
print(f"nothing new since id {last}")
|
||||
return 0
|
||||
if dry_run:
|
||||
capture(rows, True)
|
||||
return 0
|
||||
pushed = capture(rows, False)
|
||||
if pushed:
|
||||
write_last_id(pushed)
|
||||
print(f"captured; last_id -> {pushed}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_seed(hours: int, dry_run: bool) -> int:
|
||||
now = datetime.now(timezone.utc).timestamp()
|
||||
cutoff = now - hours * WINDOW_SECONDS
|
||||
rows = fetch_messages(-1, before_ts=cutoff)
|
||||
if not rows:
|
||||
print(f"no messages in last {hours}h")
|
||||
return 0
|
||||
if dry_run:
|
||||
capture(rows, True)
|
||||
return 0
|
||||
pushed = capture(rows, False)
|
||||
# seed also advances the forward cursor so we don't re-capture the seed
|
||||
cur_max = max_message_id()
|
||||
write_last_id(max(pushed, cur_max))
|
||||
print(f"seeded last {hours}h ({len(rows)} msgs); last_id -> "
|
||||
f"{max(pushed, cur_max)}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_prune(hours: int, dry_run: bool) -> int:
|
||||
now = datetime.now(timezone.utc).timestamp()
|
||||
cutoff = now - hours * WINDOW_SECONDS
|
||||
# messages OLDER than the cutoff that this script would have written
|
||||
con = sqlite3.connect(STATE_DB)
|
||||
con.row_factory = sqlite3.Row
|
||||
cur = con.cursor()
|
||||
cur.execute(
|
||||
"SELECT id, session_id, role, content, timestamp FROM messages "
|
||||
"WHERE timestamp < ? AND role IN ('user','assistant') ORDER BY id ASC",
|
||||
(cutoff,),
|
||||
)
|
||||
rows = cur.fetchall()
|
||||
con.close()
|
||||
seen: dict[str, int] = {}
|
||||
removed = 0
|
||||
for r in rows:
|
||||
ts_iso = iso_from_ts(r["timestamp"])
|
||||
seen[ts_iso] = seen.get(ts_iso, 0) + 1
|
||||
name, _ = build_entry(r, seen[ts_iso])
|
||||
if dry_run:
|
||||
print(f"[dry-run prune] would delete {name}")
|
||||
continue
|
||||
if cube_delete(name):
|
||||
removed += 1
|
||||
print(f"pruned {removed} entries older than {hours}h")
|
||||
return 0
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--dry-run", action="store_true")
|
||||
ap.add_argument("--seed-hours", type=int, metavar="N",
|
||||
help="one-time bounded population of last N hours")
|
||||
ap.add_argument("--prune-older-than-hours", type=int, metavar="N",
|
||||
help="remove entries this script wrote older than N hours")
|
||||
ap.add_argument("--reset", action="store_true",
|
||||
help="clear last_id state")
|
||||
args = ap.parse_args()
|
||||
|
||||
if args.reset:
|
||||
write_last_id(0)
|
||||
print("reset last_id -> 0")
|
||||
if not (args.seed_hours or args.prune_older_than_hours):
|
||||
return 0
|
||||
|
||||
if args.prune_older_than_hours:
|
||||
return cmd_prune(args.prune_older_than_hours, args.dry_run)
|
||||
if args.seed_hours:
|
||||
return cmd_seed(args.seed_hours, args.dry_run)
|
||||
return cmd_forward(args.dry_run)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user