diff --git a/tools/histcapture.py b/tools/histcapture.py
new file mode 100644
index 0000000..72b8f6c
--- /dev/null
+++ b/tools/histcapture.py
@@ -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())
diff --git a/tools/histcapture.service b/tools/histcapture.service
new file mode 100644
index 0000000..a25c7ef
--- /dev/null
+++ b/tools/histcapture.service
@@ -0,0 +1,11 @@
+[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
new file mode 100644
index 0000000..54224df
--- /dev/null
+++ b/tools/histcapture.state
@@ -0,0 +1 @@
+12606
\ No newline at end of file
diff --git a/tools/histcapture.timer b/tools/histcapture.timer
new file mode 100644
index 0000000..7a14286
--- /dev/null
+++ b/tools/histcapture.timer
@@ -0,0 +1,11 @@
+[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