diff --git a/tools/histcapture.py b/tools/histcapture.py deleted file mode 100644 index 0235677..0000000 --- a/tools/histcapture.py +++ /dev/null @@ -1,253 +0,0 @@ -#!/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::" with a "TYPE: HIST-QNA" header + JSON body. - -DEFAULT BEHAVIOR — FORWARD ONLY (no backfill, no deletion) - 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. - - Saved history is NEVER deleted by this script. The one-time cleanup of an - early over-backfill bug was a manual, one-off action and is NOT a recurring - retention limit — historical entries that land in the cube stay. - -BOUNDED SEED (explicit, user-approved, one-time) - `--seed-hours N` populates the last N hours ONCE. This is the only - backfill path and is opt-in (used to bootstrap the initial 24h window). - -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. - -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 - - -# --- 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 - - -# --- 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 = None - try: - con = sqlite3.connect(STATE_DB) - cur = con.cursor() - cur.execute("SELECT MAX(id) FROM messages") - return cur.fetchone()[0] or 0 - finally: - if con: - con.close() - - -# --- 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 - 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: - 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) 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 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("--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: - return 0 - - 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()) diff --git a/tools/histcapture.service b/tools/histcapture.service deleted file mode 100644 index a25c7ef..0000000 --- a/tools/histcapture.service +++ /dev/null @@ -1,11 +0,0 @@ -[Unit] -Description=CUBELinux-2 historical-record capture (Hermes transcript -> cube 'hermes' ns) -After=network.target - -[Service] -Type=oneshot -# Runs as root: root can read /root/.hermes/state.db and reach the luulu -# cubed socket via the cubelinux group (gid 1002). Forward-only by default -# (no backfill); state file tracks last pushed message id. -ExecStart=/usr/bin/python3 /home/CUBELinux/CUBELinux-2/tools/histcapture.py -Nice=10 diff --git a/tools/histcapture.state b/tools/histcapture.state deleted file mode 100644 index 579c923..0000000 --- a/tools/histcapture.state +++ /dev/null @@ -1 +0,0 @@ -12631 \ No newline at end of file diff --git a/tools/histcapture.timer b/tools/histcapture.timer deleted file mode 100644 index 7a14286..0000000 --- a/tools/histcapture.timer +++ /dev/null @@ -1,11 +0,0 @@ -[Unit] -Description=Periodically capture Hermes history into the cube 'hermes' namespace - -[Timer] -OnBootSec=30s -OnCalendar=*:0/5 -Persistent=true -Unit=histcapture.service - -[Install] -WantedBy=timers.target