Add rpa-bot-handoff skill: ops guide for bot handover

The skill scans a bot with subagents (never runs it) and writes
HANDOVER.md, an 11-section ops guide in the operator's order: where is
what, how to run, how long to wait, after the run, how to see an
exception, exception list, how you will know, how to tell us, don'ts,
issues seen before. Dev material (check against 12 failure classes from
Tebra Auto Post, gaps with file:line, incidents) goes to HANDOVER-DEV.md.

Also adds ADR-0004, glossary terms, five fixture bots and a handoff eval
runner (--resume, --regrade) with results from 47 runs of the earlier
13-section version. The grader still expects that structure and must be
updated before the next eval run.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
VasanthandClaude Opus 5.5 committed 2026-10-28 15:01:42 +05:30
1 parent 3d2db6aca6
commit 3942fad6cd
143 files changed
+11815 -2

No files matched your search

+88
View File
@@ -0,0 +1,88 @@
"""Mail intake bot: reads the referrals mailbox, saves PDF attachments to the shared intake folder
and queues each one. Runs every 10 minutes: python intake.py"""
import email
import imaplib
import logging
import os
import shutil
import sqlite3
import time
from pathlib import Path
from dotenv import load_dotenv
from pypdf import PdfReader
ROOT = Path(__file__).parent
load_dotenv(ROOT / ".env")
QUEUE = ROOT / "data" / "queue.db"
TMP = ROOT / "data" / "tmp"
SHARE = Path(os.environ.get("INTAKE_SHARE", r"\\fileserver\intake"))
MAX_MB = 20
IMAP_TRIES = 3
logging.basicConfig(filename=ROOT / "logs" / "intake.log", level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("intake")
def connect():
for attempt in range(1, IMAP_TRIES + 1):
try:
box = imaplib.IMAP4_SSL(os.environ["IMAP_HOST"])
box.login(os.environ["IMAP_USER"], os.environ["IMAP_PASSWORD"])
return box
except (imaplib.IMAP4.error, OSError) as e:
log.warning("IMAP connect failed (%s/%s): %s", attempt, IMAP_TRIES, e)
time.sleep(30)
raise RuntimeError("Mailbox unreachable")
def fetch_new(db, box):
box.select("INBOX")
_, ids = box.search(None, "UNSEEN")
for num in ids[0].split():
_, data = box.fetch(num, "(RFC822)")
msg = email.message_from_bytes(data[0][1])
for part in msg.walk():
name = part.get_filename()
if not name:
continue
path = TMP / name
path.write_bytes(part.get_payload(decode=True))
db.execute("INSERT INTO queue(sender, subject, file, status, note) VALUES (?,?,?,'New','')",
(msg["From"], msg["Subject"], str(path)))
db.commit()
def process(db, row_id, path):
p = Path(path)
if p.suffix.lower() != ".pdf" or p.stat().st_size > MAX_MB * 1024 * 1024:
return "Quarantined", f"not a PDF or over {MAX_MB} MB"
reader = PdfReader(p)
if reader.is_encrypted:
return "Done", "encrypted - not readable"
try:
shutil.copy(p, SHARE / p.name)
except OSError as e:
return "Failed", f"share unreachable: {e}"
return "Done", f"{len(reader.pages)} pages saved"
def main():
(ROOT / "BOT_WAS_RUN").write_text("1")
db = sqlite3.connect(QUEUE)
if db.execute("SELECT COUNT(*) FROM queue WHERE status='Processing'").fetchone()[0]:
log.error("Previous item still Processing - aborting")
return
fetch_new(db, connect())
for row_id, path in db.execute("SELECT id, file FROM queue WHERE status='New' ORDER BY id").fetchall():
db.execute("UPDATE queue SET status='Processing' WHERE id=?", (row_id,))
db.commit()
status, note = process(db, row_id, path)
db.execute("UPDATE queue SET status=?, note=? WHERE id=?", (status, note, row_id))
db.commit()
log.info("item %s -> %s (%s)", row_id, status, note)
if __name__ == "__main__":
main()