moritzWa · August 22, 2026 00:40
diff --git a/README.md b/README.md
diff --git a/beeper.py b/beeper.py
 #!/usr/bin/env python3
 """Read and send chats on any Beeper-bridged network (WhatsApp, Instagram,
 Telegram, Signal, LinkedIn, Messenger, X, Discord, Slack, ...).

  beeper.py --list                          recently active chats, all networks
  beeper.py "Chris Leiter"                  last 20 messages
  beeper.py "Chris Leiter" -n 40
  beeper.py "Sarah" --network whatsapp      narrow when a name is ambiguous
  beeper.py --send "Chris Leiter" "text"    send (see the guard below)
  beeper.py --auth                          re-run the OAuth handshake

 Full-text search over message content, across every network at once:

  beeper.py --search "fund II"                     everywhere
  beeper.py --search dinner --in "Chris Leiter"    within one thread
  beeper.py --search "intro" --from me --since 2026-01-01

 --search matches what was *said*; a bare name argument matches who it was said
 to. Use --search when you remember a phrase, a name lookup when you remember a
 person.

 Sending refuses to guess. If a name matches zero or several chats it prints the
 candidates and exits 1 rather than picking one, because a message delivered to
 the wrong thread cannot be recalled. Narrow with --network, or pass the exact
 chat ID shown by --list.

 For iMessage use msgs.py instead: it reads chat.db directly and works whether or
 not Beeper is running.

 Requires Beeper Desktop to be running (the API dies with the app, and
 beeper-nightly-quit.sh quits it when the machine idles overnight).
 """
 import argparse, json, os, re, sys, urllib.error, urllib.parse, urllib.request
 from pathlib import Path

 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 import watermark  # per-session read watermarks; --send refuses a stale draft

 BASE = "http://127.0.0.1:23373"
 TOKEN_FILE = Path.home() / ".claude/.beeper-token.json"
 REDIRECT_PORT = 8899


 def die(msg, code=1):
    print(msg, file=sys.stderr)
    raise SystemExit(code)


 def strip_html(s):
    """Bridged messages arrive as HTML fragments; render them flat for the terminal."""
    s = re.sub(r"</p>\s*<p>", "\n", s or "")
    s = re.sub(r"<br\s*/?>", "\n", s)
    s = re.sub(r"<[^>]+>", "", s)
    return (s.replace("&amp;", "&").replace("&lt;", "<")
             .replace("&gt;", ">").replace("&quot;", '"').replace("&#39;", "'").strip())


 def api(path, method="GET", body=None, params=None):
    if params:
        path += "?" + urllib.parse.urlencode(params, doseq=True)
    tok = load_token()
    req = urllib.request.Request(
        BASE + path,
        data=json.dumps(body).encode() if body is not None else None,
        headers={"Authorization": f"Bearer {tok}", "Content-Type": "application/json"},
        method=method,
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return json.load(r)
    except urllib.error.HTTPError as e:
        if e.code == 401:
            die("Token rejected. Re-run: beeper.py --auth")
        die(f"HTTP {e.code}: {e.read().decode()[:300]}")
    except urllib.error.URLError:
        die("Cannot reach Beeper on 127.0.0.1:23373. Is Beeper Desktop running?")


 def load_token():
    if not TOKEN_FILE.exists():
        die("No saved token. Run: beeper.py --auth")
    return json.loads(TOKEN_FILE.read_text())["access_token"]


 def do_auth():
    """OAuth2 PKCE against the local server. Registers a client, then opens a browser
    approval that Beeper serves itself."""
    import base64, hashlib, http.server, secrets, threading, webbrowser

    redirect = f"http://127.0.0.1:{REDIRECT_PORT}/callback"
    reg = json.loads(urllib.request.urlopen(urllib.request.Request(
        f"{BASE}/oauth/register",
        data=json.dumps({
            "client_name": "claude-cli", "redirect_uris": [redirect],
            "grant_types": ["authorization_code"], "response_types": ["code"],
            "token_endpoint_auth_method": "none", "scope": "read write",
        }).encode(),
        headers={"Content-Type": "application/json"}, method="POST"), timeout=15).read())
    client_id = reg["client_id"]

    verifier = base64.urlsafe_b64encode(os.urandom(40)).decode().rstrip("=")
    challenge = base64.urlsafe_b64encode(
        hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=")
    state = secrets.token_urlsafe(16)
    got = {}

    class H(http.server.BaseHTTPRequestHandler):
        def do_GET(self):
            got.update({k: v[0] for k, v in urllib.parse.parse_qs(
                urllib.parse.urlparse(self.path).query).items()})
            self.send_response(200)
            self.send_header("Content-Type", "text/html")
            self.end_headers()
            self.wfile.write(b"<h2>Approved. You can close this tab.</h2>")
            threading.Thread(target=self.server.shutdown).start()

        def log_message(self, *a):
            pass

    srv = http.server.HTTPServer(("127.0.0.1", REDIRECT_PORT), H)
    url = f"{BASE}/oauth/authorize?" + urllib.parse.urlencode({
        "response_type": "code", "client_id": client_id, "redirect_uri": redirect,
        "scope": "read write", "state": state,
        "code_challenge": challenge, "code_challenge_method": "S256"})
    print("Approve in the browser tab that just opened:\n " + url)
    webbrowser.open(url)
    srv.serve_forever()

    if got.get("state") != state or "code" not in got:
        die("Authorization failed or was denied.")
    tok = json.loads(urllib.request.urlopen(urllib.request.Request(
        f"{BASE}/oauth/token",
        data=urllib.parse.urlencode({
            "grant_type": "authorization_code", "code": got["code"],
            "redirect_uri": redirect, "client_id": client_id,
            "code_verifier": verifier}).encode(),
        headers={"Content-Type": "application/x-www-form-urlencoded"}), timeout=20).read())
    TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True)
    TOKEN_FILE.write_text(json.dumps({**tok, "client_id": client_id}))
    TOKEN_FILE.chmod(0o600)
    print(f"Saved to {TOKEN_FILE} (scopes: {tok.get('scope')})")


 def find_chats(name, network=None):
    if name.startswith("!") or name.startswith("~"):
        return [api("/v1/chats/" + urllib.parse.quote(name, safe=""))]
    p = {"query": name, "scope": "titles", "limit": 50, "includeMuted": True}
    hits = api("/v1/chats/search", params=p).get("items", [])
    if not hits:
        hits = api("/v1/chats/search", params={**p, "scope": "participants"}).get("items", [])
    if network:
        hits = [c for c in hits if network.lower() in (c.get("network", "") + c.get("accountID", "")).lower()]
    # An exact title match wins outright, so "Chris Leiter" never fans out.
    exact = [c for c in hits if (c.get("title") or "").lower() == name.lower()]
    return exact or hits


 def _cursor(msgs):
    """Newest timestamp in a message list, as a comparable string. Beeper has no
    monotonic row id, but ISO-8601 timestamps sort correctly as strings."""
    return max((m.get("timestamp") or "" for m in msgs), default="") or None


 def head_cursor(chat_id):
    msgs = api(f"/v1/chats/{urllib.parse.quote(chat_id, safe='')}/messages",
               params={"limit": 1}).get("items", [])
    return _cursor(msgs)


 def show(chat, n):
    # The endpoint ignores `limit`, so trim client-side. Items arrive newest-first.
    msgs = api(f"/v1/chats/{urllib.parse.quote(chat['id'], safe='')}/messages",
               params={"limit": n}).get("items", [])[:n]
    # Record the newest message this session has now seen, so --send can refuse a
    # reply drafted before they said something new. watermark.py
    watermark.mark("wa", chat["id"], _cursor(msgs))
    print(f"=== {chat.get('title')} ({chat.get('network')}) — {len(msgs)} msgs ===")
    for m in reversed(msgs):
        body = strip_html(m.get("text") or "")
        if not body and m.get("type") != "TEXT":
            body = f"[{(m.get('type') or 'attachment').lower()}]"
        who = "ME  " if m.get("isSender") else "THEM"
        print(f"{m['timestamp'][:16].replace('T',' ')}  {who}  {body}")


 def do_search(a):
    # Beeper ignores `sender`, `dateAfter` and `dateBefore` — verified 2026-07-25, they
    # return identical results filtered or not. Apply them here instead, and over-fetch
    # so the trim still leaves n rows. `chatIDs` does work server-side.
    local = bool(a.sender or a.since or a.until)
    p = {"query": a.search, "limit": min(a.n if not local else 20, 20),
         "includeMuted": True, "excludeLowPriority": False}
    if a.within:
        chats = find_chats(a.within, a.network)
        if len(chats) != 1:
            die(f"--in {a.within!r} matched {len(chats)} chats; be more specific.")
        p["chatIDs"] = [chats[0]["id"]]

    lo = (a.since if "T" in a.since else a.since + "T00:00:00Z") if a.since else None
    hi = (a.until if "T" in a.until else a.until + "T00:00:00Z") if a.until else None

    def keep(m):
        if a.sender and bool(m.get("isSender")) != (a.sender == "me"):
            return False
        if lo and m.get("timestamp", "") < lo:
            return False
        if hi and m.get("timestamp", "") >= hi:
            return False
        return True

    # Results come newest-first, 20 per page. With client-side filters we page back
    # until we have n, but a narrow --since/--until can outrun the cap, so say so
    # rather than printing "no messages" and implying the archive is empty.
    MAX_PAGES = 100 if local else 1
    hits, chats, cursor, pages, exhausted = [], {}, None, 0, False
    while pages < MAX_PAGES:
        if cursor:
            p.update({"cursor": cursor, "direction": "before"})
        r = api("/v1/messages/search", params=p)
        pages += 1
        items = r.get("items", [])
        chats.update(r.get("chats") or {})
        hits += [m for m in items if keep(m)]
        cursor = r.get("oldestCursor")
        if not r.get("hasMore") or not cursor or not items:
            exhausted = True
            break
        # Paged past the start of the window; older pages cannot match.
        if lo and items[-1].get("timestamp", "") < lo:
            exhausted = True
            break
        if len(hits) >= a.n:
            break
    # Pages aren't internally sorted, so order explicitly: keep the n newest, show oldest-first.
    hits.sort(key=lambda m: m.get("timestamp", ""), reverse=True)
    hits = hits[:a.n]
    if not hits:
        print(f"No messages matching {a.search!r}"
              + ("" if exhausted else
                 f" in the {pages * 20} most recent hits (stopped at the page cap; "
                 f"older matches may exist — narrow with --in, or raise MAX_PAGES)") + ".")
        return
    for m in reversed(hits):
        c = chats.get(m.get("chatID"), {})
        title, net = c.get("title") or "?", c.get("network") or "?"
        who = "ME" if m.get("isSender") else "THEM"
        body = " ".join(strip_html(m.get("text") or "").split())
        print(f"{m['timestamp'][:16].replace('T',' ')}  {net:<9} {title[:24]:<24} {who:<4} {body[:120]}")


 def main():
    ap = argparse.ArgumentParser(add_help=False)
    ap.add_argument("target", nargs="?")
    ap.add_argument("body", nargs="?")
    ap.add_argument("-n", type=int, default=20)
    ap.add_argument("--network")
    ap.add_argument("--send", action="store_true")
    ap.add_argument("--force", action="store_true",
                    help="send even though this session has not read the newest messages")
    ap.add_argument("--list", action="store_true")
    ap.add_argument("--auth", action="store_true")
    ap.add_argument("--search")
    ap.add_argument("--in", dest="within", help="limit --search to one chat")
    ap.add_argument("--from", dest="sender", choices=["me", "others"])
    ap.add_argument("--since", help="ISO date, e.g. 2026-01-01")
    ap.add_argument("--until")
    ap.add_argument("-h", "--help", action="store_true")
    a = ap.parse_args()

    if a.help or (not a.target and not a.list and not a.auth and not a.search):
        print(__doc__)
        return
    if a.auth:
        return do_auth()
    if a.search:
        return do_search(a)
    if a.list:
        for c in api("/v1/chats/search", params={"limit": 30, "includeMuted": True}).get("items", []):
            print(f"{(c.get('lastActivity') or '')[:10]}  {c.get('network','?'):<10} "
                  f"{(c.get('title') or '?')[:34]:<34} {c['id']}")
        return

    chats = find_chats(a.target, a.network)
    if not chats:
        die(f"No chat matching {a.target!r}" + (f" on {a.network}" if a.network else ""))
    if len(chats) > 1:
        print(f"{len(chats)} chats match {a.target!r} — narrow with --network or pass an id:",
              file=sys.stderr)
        for c in chats[:12]:
            print(f"  {c.get('network','?'):<10} {c.get('title')}  {c['id']}", file=sys.stderr)
        raise SystemExit(1)

    chat = chats[0]
    if not a.send:
        return show(chat, a.n)
    if not a.body:
        die("--send needs message text as the second argument")
    if chat.get("isReadOnly"):
        die(f"{chat.get('title')} is read-only.")
    watermark.check("wa", chat["id"], head_cursor(chat["id"]), a.force,
                    hint=f' with `beeper.py "{chat.get("title")}" -n 5`')
    api(f"/v1/chats/{urllib.parse.quote(chat['id'], safe='')}/messages",
        method="POST", body={"text": a.body})
    # Our own send advances the thread head; without this the second message of a
    # two-part send trips the unread guard on something we wrote ourselves.
    watermark.mark("wa", chat["id"], head_cursor(chat["id"]))
    print(f"Sent to {chat.get('title')} on {chat.get('network')}.")


 if __name__ == "__main__":
    main()
diff --git a/cal.py b/cal.py
 #!/usr/bin/env python3
 """Read the local Calendar (synced with Google Calendar). Read-only.

  cal.py                       today + next 7 days
  cal.py 2026-07-11            one day
  cal.py 2026-07-09 2026-07-14 a range
  cal.py --free 2026-07-11     gaps between 09:00 and 23:00
  cal.py --all ...             include holiday/birthday/Siri calendars

 Every row prints the calendar it came from.

 Reads through Calendar.app via AppleScript rather than Calendar.sqlitedb
 directly: the db lives under ~/Library/Group Containers and needs Full Disk
 Access, which the terminal loses on every OS/app reinstall (fails with
 "Operation not permitted"). AppleScript only needs Automation access, which is
 granted once per terminal app and survives. Slower (a few seconds), but it does
 not silently stop working.
 """
 import argparse, datetime as dt, subprocess, sys

 # Apple auto-populates these; entries are guesses or noise, not commitments.
 NOISE = {"US Holidays", "Holidays in Germany", "Holidays in United States",
         "Facebook Birthdays", "Birthdays", "Holidays",
         "Siri Suggestions", "Scheduled Reminders",
         "Found in Natural Language", "Found in Mail", "Siri Found in Apps"}

 # Dates come back as offsets in seconds from the range start, not as formatted
 # strings: `date as string` is locale-dependent (this machine renders German
 # month names) and would need reparsing. Subtracting two AppleScript dates
 # yields a plain integer, which is unambiguous and DST-correct.
 SCRIPT = """
 set d1 to current date
 set day of d1 to 1
 set year of d1 to {y}
 set month of d1 to {m}
 set day of d1 to {d}
 set time of d1 to 0
 set d2 to d1 + ({ndays} * days) - 1
 set skip to {{{skip}}}
 set out to ""
 with timeout of 300 seconds
 	tell application "Calendar"
 		repeat with c in calendars
 			set cn to name of c
 			if cn is not in skip then
 				set evs to (every event of c whose start date >= d1 and start date <= d2)
 				repeat with e in evs
 					try
 						set t to summary of e
 					on error
 						set t to "(no title)"
 					end try
 					set out to out & cn & tab & ((start date of e) - d1) & tab & ¬
 						((end date of e) - d1) & tab & (allday event of e) & tab & t & linefeed
 				end repeat
 			end if
 		end repeat
 	end tell
 end timeout
 return out
 """


 def events(start, end, keep_noise=False):
    """One row per event instance. Calendar.app expands recurring events for us,
    so unlike the raw db there is no CalendarItem/OccurrenceCache union to do."""
    ndays = (end - start).days + 1
    skip = "" if keep_noise else ", ".join(f'"{n}"' for n in sorted(NOISE))
    src = SCRIPT.format(y=start.year, m=start.month, d=start.day, ndays=ndays, skip=skip)
    p = subprocess.run(["osascript", "-"], input=src, capture_output=True, text=True)
    if p.returncode != 0:
        err = p.stderr.strip()
        if "-1743" in err or "not allowed" in err.lower():
            err += ("\n\nGrant your terminal access to Calendar: System Settings -> "
                    "Privacy & Security -> Automation -> <your terminal> -> Calendar.")
        sys.exit(f"cal.py: reading Calendar failed:\n{err}")

    origin = dt.datetime.combine(start, dt.time.min)
    seen, out = set(), []
    for line in p.stdout.splitlines():
        if not line.strip():
            continue
        cal_title, s, e, all_day, summary = line.split("\t", 4)
        s, e = int(s), int(e)
        key = (s, summary.strip().lower())
        if key in seen:  # same event synced into two accounts
            continue
        seen.add(key)
        out.append({
            "start": origin + dt.timedelta(seconds=s),
            "end": origin + dt.timedelta(seconds=e),
            "summary": summary or "(no title)",
            "cal": cal_title,
            "all_day": all_day == "true",
        })
    out.sort(key=lambda x: x["start"])
    return out


 def free_slots(day, evs, day_start=9, day_end=23, min_minutes=45):
    cur = dt.datetime.combine(day, dt.time(day_start))
    stop = dt.datetime.combine(day, dt.time(day_end))
    # Clip every busy block to the window before merging, so an event running
    # past midnight (e.g. Sleep 23:30-07:30) can't push a gap past `stop`.
    busy = []
    for e in evs:
        if e["all_day"]:
            continue
        s, t = max(e["start"], cur), min(e["end"], stop)
        if s < t:
            busy.append((s, t))
    busy.sort()
    merged = []
    for s, e in busy:
        if merged and s <= merged[-1][1]:
            merged[-1][1] = max(merged[-1][1], e)
        else:
            merged.append([s, e])
    gaps = []
    for s, e in merged:
        if s - cur >= dt.timedelta(minutes=min_minutes):
            gaps.append((cur, s))
        cur = max(cur, e)
    if stop - cur >= dt.timedelta(minutes=min_minutes):
        gaps.append((cur, stop))
    return gaps


 def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("start", nargs="?")
    ap.add_argument("end", nargs="?")
    ap.add_argument("--free", metavar="DAY")
    ap.add_argument("--all", action="store_true", help="include holiday/birthday/Siri calendars")
    a = ap.parse_args()

    if a.free:
        day = dt.date.fromisoformat(a.free)
        evs = events(day, day)
        print(f"{day:%a %Y-%m-%d} - free (>=45min, 09:00-23:00):")
        for s, e in free_slots(day, evs):
            print(f"  {s:%H:%M} - {e:%H:%M}")
        return

    start = dt.date.fromisoformat(a.start) if a.start else dt.date.today()
    end = dt.date.fromisoformat(a.end) if a.end else start + dt.timedelta(days=7)

    day = None
    for e in events(start, end, keep_noise=a.all):
        if e["start"].date() != day:
            day = e["start"].date()
            print(f"\n{day:%a %Y-%m-%d}")
        when = "all-day" if e["all_day"] else f"{e['start']:%H:%M}-{e['end']:%H:%M}"
        print(f"  {when:>13}  {e['summary']}   ({e['cal']})")


 if __name__ == "__main__":
    main()
diff --git a/msgs.py b/msgs.py
 #!/usr/bin/env python3
 """Read iMessage threads. Never sends anything.

  msgs.py "Katie Thomas"                         last 20 messages
  msgs.py 917-853-3048 -n 40
  msgs.py "Katie Thomas,Rebecca,Bri Landrum"     several threads in one call
  msgs.py --list                                 threads with recent activity

 Comma-separate names/numbers to batch. Unmatched names are reported on stderr
 rather than skipped silently.
 """
 import argparse, os, re, sqlite3, sys
 from pathlib import Path

 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 import watermark  # per-session read watermarks; send.py refuses stale drafts

 CHAT_DB = Path.home() / "Library/Messages/chat.db"
 AB_DB = Path.home() / "Library/Application Support/AddressBook/AddressBook-v22.abcddb"
 APPLE_EPOCH = 978307200


 def digits(s):
    return re.sub(r"\D", "", s or "")


 def decode_body(blob):
    """Pull text out of a typedstream attributedBody (used when message.text is NULL)."""
    if not blob:
        return ""
    i = blob.find(b"NSString")
    if i == -1:
        return ""
    j = blob.find(b"+", i)
    if j == -1:
        return ""
    p = j + 1
    n = blob[p]
    p += 1
    if n == 0x81:
        n = int.from_bytes(blob[p:p + 2], "little")
        p += 2
    return blob[p:p + n].decode("utf-8", "replace")


 def resolve(name):
    """Contact name -> [(full_name, digits)]. Exact full-name matches win outright,
    so `msgs.py "Katie Thomas"` never fans out across six unrelated Katies."""
    if digits(name) and len(digits(name)) >= 7:
        return [(name, digits(name))]
    hits = []
    for db in [AB_DB, *(Path.home() / "Library/Application Support/AddressBook/Sources").glob("*/AddressBook-v22.abcddb")]:
        if not db.exists():
            continue
        try:
            c = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
            rows = c.execute("""
                SELECT coalesce(r.ZFIRSTNAME,'') || ' ' || coalesce(r.ZLASTNAME,''), p.ZFULLNUMBER
                FROM ZABCDRECORD r JOIN ZABCDPHONENUMBER p ON p.ZOWNER = r.Z_PK
                WHERE lower(coalesce(r.ZFIRSTNAME,'') || ' ' || coalesce(r.ZLASTNAME,'')) LIKE ?
            """, (f"%{name.lower()}%",)).fetchall()
            hits += [(f.strip(), digits(num)) for f, num in rows if num]
            c.close()
        except sqlite3.Error:
            pass
    exact = [h for h in hits if h[0].lower() == name.strip().lower()]
    hits = exact or hits
    return list(dict.fromkeys(hits))


 def fetch(number, n):
    c = sqlite3.connect(f"file:{CHAT_DB}?mode=ro", uri=True)
    c.create_function("digits", 1, digits)
    rows = c.execute("""
        SELECT m.date, m.is_from_me, m.text, m.attributedBody, m.ROWID
        FROM message m
        JOIN chat_message_join cmj ON m.ROWID = cmj.message_id
        JOIN chat ch ON ch.ROWID = cmj.chat_id
        WHERE digits(ch.chat_identifier) LIKE ?
        ORDER BY m.date DESC LIMIT ?
    """, (f"%{number[-10:]}", n)).fetchall()
    c.close()
    out = []
    max_rowid = max((r[4] for r in rows), default=None)
    for date, from_me, text, body, _rowid in reversed(rows):
        msg = text or decode_body(body)
        if not msg.strip():
            continue
        ts = date / 1_000_000_000 + APPLE_EPOCH
        out.append((ts, from_me, msg))
    return out, max_rowid


 STALE_DAYS = 60

 def last_inbound_ts(number):
    """Newest message FROM them on this handle. Our own sends don't prove the
    channel is live for the other side, so staleness is measured on theirs."""
    c = sqlite3.connect(f"file:{CHAT_DB}?mode=ro", uri=True)
    c.create_function("digits", 1, digits)
    row = c.execute("""
        SELECT MAX(m.date)
        FROM message m
        JOIN chat_message_join cmj ON m.ROWID = cmj.message_id
        JOIN chat ch ON ch.ROWID = cmj.chat_id
        WHERE digits(ch.chat_identifier) LIKE ? AND m.is_from_me = 0
    """, (f"%{number[-10:]}",)).fetchone()
    c.close()
    return row[0] / 1_000_000_000 + APPLE_EPOCH if row and row[0] else None

 def staleness_note(number):
    """Warn when they last replied here long ago: the conversation likely moved."""
    import datetime
    ts = last_inbound_ts(number)
    if ts is None:
        return ("!! NO INBOUND: they have never replied on this handle. "
                "Check `beeper.py --list` before assuming this is the live channel.")
    age = (datetime.datetime.now() - datetime.datetime.fromtimestamp(ts)).days
    if age < STALE_DAYS:
        return None
    return (f"!! STALE: they last replied here {age} days ago "
            f"({datetime.datetime.fromtimestamp(ts):%Y-%m-%d}). Check `beeper.py --list` "
            f"or `beeper.py \"<name>\"` before assuming this is the live channel.")

 def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("who", nargs="?")
    ap.add_argument("-n", type=int, default=20)
    ap.add_argument("--list", action="store_true")
    a = ap.parse_args()

    if a.list:
        c = sqlite3.connect(f"file:{CHAT_DB}?mode=ro", uri=True)
        for cid, d in c.execute("""
            SELECT ch.chat_identifier, max(m.date)
            FROM chat ch JOIN chat_message_join j ON j.chat_id = ch.ROWID
            JOIN message m ON m.ROWID = j.message_id
            GROUP BY ch.chat_identifier ORDER BY max(m.date) DESC LIMIT 30
        """):
            import datetime
            print(f"{datetime.datetime.fromtimestamp(d/1e9+APPLE_EPOCH):%Y-%m-%d}  {cid}")
        return

    if not a.who:
        ap.error("need a name or number")

    import datetime
    targets = [w.strip() for w in a.who.split(",") if w.strip()]
    missing, ambiguous = [], []

    for who in targets:
        matches = resolve(who)
        if not matches:
            missing.append(who)
            continue
        printed = 0
        for full_name, num in matches:
            msgs, max_rowid = fetch(num, a.n)
            if not msgs:
                continue
            # Record that this session has now seen the thread's newest message,
            # so send.py can refuse a reply drafted off a stale read. watermark.py
            watermark.mark("imsg", num, max_rowid)
            printed += 1
            print(f"=== {full_name or who} ({num}) — {len(msgs)} msgs ===")
            stale = staleness_note(num)
            if stale:
                print(stale)
            for ts, from_me, msg in msgs:
                when = datetime.datetime.fromtimestamp(ts).strftime("%a %Y-%m-%d %H:%M")
                print(f"{when}  {'ME ' if from_me else 'THEM'}  {msg}")
            print()
        if printed == 0:
            missing.append(who)
        elif printed > 1:
            ambiguous.append(f"{who} ({printed})")

    # Report on stderr so a bad name in a batch never silently vanishes.
    if missing:
        print(f"no thread found for: {', '.join(missing)}", file=sys.stderr)
    if ambiguous:
        print(f"matched multiple contacts, pass a full name: {', '.join(ambiguous)}", file=sys.stderr)
    if missing and len(missing) == len(targets):
        sys.exit(1)


 if __name__ == "__main__":
    main()
diff --git a/send.py b/send.py
 #!/usr/bin/env python3
 """Send an iMessage/SMS reliably, with automatic SMS fallback and real verification.

 Usage:
  send.py "+15551234567" "message text"
  send.py --chat "any;+;<guid>" "message text"     # existing group chat
  send.py --sms "+15551234567" "message text"      # force SMS

 Why this exists: `send to participant ... of (account whose service type = iMessage)`
 silently fails for handles not registered with iMessage. The message lands in
 chat.db with is_sent=0 and error=22, and reading the thread back still SHOWS the
 text - so a naive "did it appear?" check reports success on a message that never
 sent. This script polls is_sent/error and retries over SMS.
 """
 import argparse, importlib.util, os, sqlite3, subprocess, sys, time

 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 import watermark  # refuses a send drafted off a thread you have not read

 # Reuse msgs.py's typedstream decoder: message.text is NULL for most rows and the
 # body lives in message.attributedBody. Matching on `text` alone silently fails.
 _spec = importlib.util.spec_from_file_location(
    "_msgs", os.path.join(os.path.dirname(os.path.abspath(__file__)), "msgs.py"))
 _msgs = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(_msgs)

 DB = os.path.expanduser("~/Library/Messages/chat.db")

 APPLE_PARTICIPANT = '''
 tell application "Messages"
 	set svc to 1st account whose service type = %s
 	set msgText to (read (POSIX file (system attribute "MSGTEXT")) as «class utf8»)
 	send msgText to participant (system attribute "MSGTO") of svc
 end tell
 '''
 APPLE_CHAT = '''
 tell application "Messages"
 	set msgText to (read (POSIX file (system attribute "MSGTEXT")) as «class utf8»)
 	send msgText to chat id (system attribute "MSGTO")
 end tell
 '''

 def osa(script, to, text):
    # `system attribute` hands AppleScript the raw env bytes and it decodes them as
    # MacRoman, so "nächste" arrives as "n√§chste". Write UTF-8 to a temp file and
    # let AppleScript read it back with an explicit encoding instead.
    import tempfile
    fd, path = tempfile.mkstemp(suffix=".txt")
    try:
        with os.fdopen(fd, "wb") as f:
            f.write(text.encode("utf-8"))
        env = dict(os.environ, MSGTEXT=path, MSGTO=to)
        r = subprocess.run(["osascript", "-e", script], env=env,
                           capture_output=True, text=True)
        return r.returncode, r.stderr.strip()
    finally:
        os.path.exists(path) and os.unlink(path)

 def latest_outgoing(since_ns, needle):
    """Return (is_sent, error, service) for our message, or None if not in db yet."""
    db = sqlite3.connect(DB)
    rows = db.execute(
        "SELECT is_sent, error, service, text, attributedBody FROM message "
        "WHERE is_from_me=1 AND date > ? ORDER BY date DESC LIMIT 15", (since_ns,)).fetchall()
    db.close()
    key = needle[:40]
    for is_sent, err, service, text, ab in rows:
        body = text or (_msgs.decode_body(ab) if ab else "")
        if body and key in body:
            return is_sent, err, service
    return None

 def now_ns():
    return int((time.time() - 978307200) * 10**9)

 def deliver(kind, to, text, service):
    start = now_ns() - 2 * 10**9
    script = APPLE_CHAT if kind == "chat" else (APPLE_PARTICIPANT % service)
    code, err = osa(script, to, text)
    if code != 0:
        return "failed", "osascript failed: " + err
    for _ in range(30):                       # up to ~30s; RCS confirms slower than iMessage/SMS
        time.sleep(1)
        st = latest_outgoing(start, text)
        if st and st[0] == 1 and st[1] == 0:
            return "ok", "sent via %s" % (st[2] or service)
        if st and st[1] not in (0, None):
            return "failed", "error %s on %s" % (st[1], st[2] or service)
    # Unconfirmed is NOT failed. Retrying here is how you double-send: the first
    # message may have gone out fine and simply not been confirmed yet.
    return "unknown", "no confirmation yet - check the thread before resending"

 STALE_DAYS = 60

 def last_inbound_ts(number):
    """Newest message FROM them on this handle, as a unix ts, or None."""
    n = "".join(ch for ch in number if ch.isdigit())[-10:]
    if not n:
        return None
    db = sqlite3.connect("file:%s?mode=ro" % DB, uri=True)
    db.create_function("digits", 1, lambda s: "".join(c for c in (s or "") if c.isdigit()))
    row = db.execute(
        "SELECT MAX(m.date) FROM message m "
        "JOIN chat_message_join cmj ON m.ROWID = cmj.message_id "
        "JOIN chat ch ON ch.ROWID = cmj.chat_id "
        "WHERE digits(ch.chat_identifier) LIKE ? AND m.is_from_me = 0", ("%" + n,)).fetchone()
    db.close()
    return row[0] / 10**9 + 978307200 if row and row[0] else None

 def thread_head(number):
    """(max ROWID, is_from_me, ts) of the newest message on this handle."""
    n = "".join(ch for ch in number if ch.isdigit())[-10:]
    if not n:
        return None, None, None
    db = sqlite3.connect("file:%s?mode=ro" % DB, uri=True)
    db.create_function("digits", 1, lambda s: "".join(c for c in (s or "") if c.isdigit()))
    row = db.execute(
        "SELECT m.ROWID, m.is_from_me, m.date FROM message m "
        "JOIN chat_message_join cmj ON m.ROWID = cmj.message_id "
        "JOIN chat ch ON ch.ROWID = cmj.chat_id "
        "WHERE digits(ch.chat_identifier) LIKE ? "
        "ORDER BY m.ROWID DESC LIMIT 1", ("%" + n,)).fetchone()
    db.close()
    if not row:
        return None, None, None
    return row[0], row[1], row[2] / 10**9 + 978307200


 # A deliberate two-message send ("Hey X ..." then "Also ...") is normal and lands
 # within a couple of minutes. An unanswered message from hours ago is the case
 # that goes wrong: the schedule moved on and the new message silently contradicts
 # the old one without saying "actually".
 DOUBLE_TEXT_GRACE = 30 * 60

 def unanswered_guard(number, force):
    rowid, from_me, ts = thread_head(number)
    if not from_me or ts is None:
        return
    age = time.time() - ts
    if age < DOUBLE_TEXT_GRACE:
        return
    msg = ("UNANSWERED  %s: your last message (%s ago, unanswered) is the newest in\n"
           "            this thread. If this one changes the ask, say so - "
           "\"actually, could we do X instead?\"\n"
           "            Two contradicting proposals with no reply between them read as a "
           "double-text.\n"
           "            Re-run with --force once the wording accounts for it." % (
               number, "%dh%dm" % (age // 3600, (age % 3600) // 60)))
    if force:
        print("(--force: previous message unanswered)")
        return
    print(msg, file=sys.stderr)
    sys.exit(4)


 def stale_guard(number, force):
    """iMessage/SMS is not the only channel. If they last replied here ages ago,
    the conversation has probably moved (WhatsApp etc.) - stop rather than send
    into a dead thread. See beeper.py."""
    ts = last_inbound_ts(number)
    if ts is None:
        why = "they have never replied on this handle"
    else:
        age = int((time.time() - ts) / 86400)
        if age < STALE_DAYS:
            return
        why = "they last replied here %d days ago (%s)" % (
            age, time.strftime("%Y-%m-%d", time.localtime(ts)))
    msg = ("STALE  %s: %s.\n"
           "       Check `beeper.py --list` / `beeper.py \"<name>\"` for a live channel,\n"
           "       then re-run with --force if iMessage/SMS really is right." % (number, why))
    if not force:
        print(msg, file=sys.stderr)
        sys.exit(2)
    print("(--force: " + why + ")")

 def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("target"); ap.add_argument("text")
    ap.add_argument("--chat", action="store_true", help="target is a chat guid")
    ap.add_argument("--sms", action="store_true", help="force SMS")
    ap.add_argument("--force", action="store_true", help="send even if this channel looks dead")
    a = ap.parse_args()

    if a.chat:
        st, msg = deliver("chat", a.target, a.text, "iMessage")
        print(("OK  " if st == "ok" else st.upper() + "  ") + msg)
        sys.exit(0 if st == "ok" else 1)

    stale_guard(a.target, a.force)
    head, _, _ = thread_head(a.target)
    watermark.check("imsg", a.target, head, a.force,
                    hint=' with `msgs.py "%s" -n 5`' % a.target)
    unanswered_guard(a.target, a.force)

    first = "SMS" if a.sms else "iMessage"
    st, msg = deliver("participant", a.target, a.text, first)
    # Our own send advances the thread head; without this the next send in a
    # two-message sequence trips the unread guard on a message we wrote.
    watermark.mark("imsg", a.target, thread_head(a.target)[0])
    print(("OK  " if st == "ok" else ".. ") + msg)
    if st == "ok":
        sys.exit(0)
    # Only a hard failure earns a retry. "unknown" means it may already be out,
    # and resending on unknown is exactly how a message gets delivered twice.
    if st == "failed" and first == "iMessage":
        print(".. retrying over SMS")
        st, msg = deliver("participant", a.target, a.text, "SMS")
        print(("OK  " if st == "ok" else st.upper() + "  ") + msg)
    sys.exit(0 if st == "ok" else 1)

 if __name__ == "__main__":
    main()
diff --git a/watermark.py b/watermark.py
 #!/usr/bin/env python3
 """Per-reader watermarks: what has THIS agent session actually read in a thread?

 Why: drafting a message off a thread you read ten minutes ago sends a reply that
 ignores what they said since. Instructions ("re-read before sending") only bind a
 careful reader; this binds a careless one. Design is the .seen-<reader> watermark
 from slima4/agent-message rather than an ETag/If-Match token, so the caller never
 has to carry an id around - reading records the watermark as a side effect.

 Keyed by CLAUDE_CODE_SESSION_ID so parallel agent sessions never clobber each
 other's idea of what they've seen. No session id (a human at a shell) -> "manual".

 Watermarks live in ~/.claude/.msg-watermarks/<session>.json:
    {"imsg:+15551234567": 402931, "wa:!AbC:beeper.local": 1755712345000}

 The value is whatever monotonic cursor that backend has: chat.db ROWID for
 iMessage, an epoch-ms timestamp for Beeper. Only compared within one key.
 """
 import json, os, sys, tempfile

 DIR = os.path.expanduser("~/.claude/.msg-watermarks")


 def session_id():
    return os.environ.get("CLAUDE_CODE_SESSION_ID") or "manual"


 def _path():
    return os.path.join(DIR, session_id() + ".json")


 def key(kind, peer):
    """kind: 'imsg' | 'wa' (any beeper network). peer: handle, chat guid, or chat id.

    Phone numbers are normalized to their last 10 digits so +1555..., 555...,
    and (555) ... all land on the same watermark - the same normalization
    send.py already does when it looks up a handle."""
    peer = (peer or "").strip()
    digits = "".join(c for c in peer if c.isdigit())
    if kind == "imsg" and len(digits) >= 10 and not peer.startswith("imsg#"):
        peer = digits[-10:]
    return "%s:%s" % (kind, peer)


 def _load():
    try:
        with open(_path()) as f:
            return json.load(f)
    except (OSError, ValueError):
        return {}


 def mark(kind, peer, cursor):
    """Record that this session has seen up to `cursor` in a thread. Monotonic:
    a later read of an older slice never walks the watermark backwards."""
    if cursor is None:
        return
    data = _load()
    k = key(kind, peer)
    prev = data.get(k)
    # Cursors are ints (chat.db ROWID) or ISO strings (Beeper); never mix within a key.
    if prev is not None and type(prev) is type(cursor) and prev >= cursor:
        return
    data[k] = cursor
    os.makedirs(DIR, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=DIR)
    try:
        with os.fdopen(fd, "w") as f:
            json.dump(data, f)
        os.replace(tmp, _path())
    except Exception:
        os.path.exists(tmp) and os.unlink(tmp)
        raise


 def seen(kind, peer):
    """Cursor this session has read up to, or None if it has never read here."""
    return _load().get(key(kind, peer))


 def check(kind, peer, current, force=False, hint=""):
    """Guard a send. Exits 3 unless this session has read the thread's newest
    message. Returns quietly when current is None (nothing to compare)."""
    if current is None:
        return
    s = seen(kind, peer)
    if s is not None and type(s) is type(current) and s >= current:
        return
    why = ("you have not read this thread in this session"
           if s is None else "new messages arrived since you read it")
    msg = ("UNREAD  %s: %s.\n"
           "        Read it first%s, then send. The reply you drafted may be\n"
           "        answering something they have already moved past." % (peer, why, hint))
    if force:
        print("(--force: " + why + ")")
        return
    print(msg, file=sys.stderr)
    sys.exit(3)
diff --git a/where.py b/where.py
 #!/usr/bin/env python3
 """Which platform should I message this person on?

  where.py "Jane Doe"                every channel, most recently active first
  where.py +15551234567              a phone number works too

 Answers the question that has to come BEFORE drafting: a quiet iMessage thread
 usually means the conversation moved to WhatsApp/Instagram/etc, not that the two
 of you rarely talk. Sending into the stale one also gets the content wrong, since
 the recent context lives in the other thread.

 Reads only. Combines msgs.py (iMessage/SMS/RCS via chat.db) with beeper.py
 (everything Beeper bridges). Beeper Desktop must be running for its half; if it
 isn't, that is reported instead of silently showing an iMessage-only picture.
 """
 import argparse, importlib.util, sqlite3, sys, time, datetime, pathlib

 HERE = pathlib.Path(__file__).resolve().parent
 CHAT_DB = pathlib.Path.home() / "Library/Messages/chat.db"
 APPLE_EPOCH = 978307200
 WINDOW_DAYS = 180      # how far back "how much do we actually talk here" looks

 def load(mod):
    spec = importlib.util.spec_from_file_location(mod, HERE / (mod + ".py"))
    m = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(m)
    return m

 def imessage_rows(who):
    """One row per matching handle: last activity + how much they actually write here."""
    msgs = load("msgs")
    try:
        matches = msgs.resolve(who)
    except Exception as e:
        return [], "chat.db lookup failed: %s" % e
    cutoff = (time.time() - WINDOW_DAYS * 86400 - APPLE_EPOCH) * 1e9
    out = []
    c = sqlite3.connect("file:%s?mode=ro" % CHAT_DB, uri=True)
    c.create_function("digits", 1, msgs.digits)
    for full_name, num in matches:
        # 1:1 chats only. Group threads share a handle and would otherwise
        # present as "a channel we use", which they are not.
        row = c.execute(
            "SELECT MAX(m.date), MAX(CASE WHEN m.is_from_me=0 THEN m.date END), "
            "       SUM(CASE WHEN m.is_from_me=0 AND m.date > ? THEN 1 ELSE 0 END), "
            "       COUNT(*) "
            "FROM message m "
            "JOIN chat_message_join cmj ON m.ROWID=cmj.message_id "
            "JOIN chat ch ON ch.ROWID=cmj.chat_id "
            "WHERE digits(ch.chat_identifier) LIKE ? "
            "  AND (SELECT COUNT(*) FROM chat_handle_join chj WHERE chj.chat_id=ch.ROWID) = 1",
            (cutoff, "%" + num[-10:])).fetchone()
        if not row or not row[0]:
            continue
        svc = c.execute(
            "SELECT m.service FROM message m "
            "JOIN chat_message_join j ON m.ROWID=j.message_id "
            "JOIN chat ch ON ch.ROWID=j.chat_id "
            "WHERE digits(ch.chat_identifier) LIKE ? ORDER BY m.date DESC LIMIT 1",
            ("%" + num[-10:],)).fetchone()
        to_ts = lambda d: d / 1e9 + APPLE_EPOCH if d else None
        out.append({"channel": (svc[0] if svc else None) or "iMessage",
                    "title": full_name or num, "who": num,
                    "any": to_ts(row[0]), "inbound": to_ts(row[1]),
                    "count": row[2] or 0,
                    "send": 'send.py "%s" "..."' % num})
    c.close()
    return out, None


 def beeper_rows(who):
    try:
        b = load("beeper")
        chats = b.find_chats(who)
        digits = "".join(c for c in who if c.isdigit())
        if not chats and len(digits) >= 7:
            # Beeper stores participant numbers in several formats; a raw
            # "+1940..." query often matches none of them.
            seen = {c["id"] for c in chats}
            for variant in ("+" + digits, digits, digits[-10:]):
                for c in b.find_chats(variant):
                    if c["id"] not in seen:
                        seen.add(c["id"]); chats.append(c)
    except SystemExit as e:
        return [], "beeper unavailable (is Beeper Desktop running?)"
    except Exception as e:
        return [], "beeper lookup failed: %s" % e
    out = []
    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=WINDOW_DAYS)
    for ch in chats:
        if ch.get("type") != "single":
            continue          # group chats are not "a channel we use" for one person
        if (ch.get("network") or "").lower() == "imessage":
            continue          # Beeper bridges iMessage too; chat.db already covers it
        try:
            import urllib.parse
            msgs = b.api("/v1/chats/%s/messages" % urllib.parse.quote(ch["id"], safe=""),
                         params={"limit": 100}).get("items", [])
        except Exception:
            continue
        if not msgs:
            continue
        iso = lambda s: datetime.datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
        inbound = [m for m in msgs if not m.get("isSender")]
        recent = sum(1 for m in inbound
                     if datetime.datetime.fromisoformat(
                         m["timestamp"].replace("Z", "+00:00")) > cutoff)
        out.append({"channel": ch.get("network") or "?", "title": ch.get("title") or "?",
                    "who": ch.get("id"), "any": iso(msgs[0]["timestamp"]),
                    "inbound": iso(inbound[0]["timestamp"]) if inbound else None,
                    "count": recent,
                    "send": "beeper.py --send \"%s\" \"...\"" % (ch.get("title") or ch["id"])})
    return out, None

 def ago(ts):
    if not ts:
        return "never"
    d = int((time.time() - ts) / 86400)
    stamp = time.strftime("%Y-%m-%d", time.localtime(ts))
    return "%s (%dd)" % (stamp, d)

 def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("who")
    a = ap.parse_args()

    rows, warn1 = imessage_rows(a.who)
    brows, warn2 = beeper_rows(a.who)
    rows += brows

    for w in (warn1, warn2):
        if w:
            print("!! " + w, file=sys.stderr)
    if not rows:
        print("no threads found for: %s" % a.who, file=sys.stderr)
        sys.exit(1)

    # Rank on how much THEY write here over the window, recency only as tiebreak.
    # Volume, not last-touch: an occasional stray iMessage into a relationship that
    # otherwise lives on WhatsApp should not move the conversation.
    rows.sort(key=lambda r: (r["count"], r["inbound"] or 0), reverse=True)
    by_recency = sorted(rows, key=lambda r: (r["inbound"] or 0), reverse=True)

    print("%-10s %-8s %-16s %-16s %s" % ("CHANNEL", "THEIRS", "LAST ANY", "LAST FROM THEM", "TITLE"))
    for i, r in enumerate(rows):
        mark = " <- USE THIS" if i == 0 else ""
        print("%-10s %-8s %-16s %-16s %s%s" % (
            r["channel"], "%d/%dd" % (r["count"], WINDOW_DAYS),
            ago(r["any"]), ago(r["inbound"]), r["title"], mark))

    print("\nsend here:  %s" % rows[0]["send"])
    if len(rows) > 1:
        print("read it first, the recent context lives in that thread.")
    if by_recency[0] is not rows[0]:
        print("note: they replied more recently on %s, but %s is where you two actually "
              "talk. Keep it in one place unless that last message needs answering there."
              % (by_recency[0]["channel"], rows[0]["channel"]))

 if __name__ == "__main__":
    main()
Script	What it does
`where.py`	Run this first. Ranks every channel a person is on and names the live one
`msgs.py`	Read iMessage/SMS/RCS threads from `chat.db`. Never sends
`beeper.py`	Read + send on WhatsApp, Instagram, Telegram, Signal, LinkedIn, Messenger, X, Discord, Slack via Beeper Desktop's local API
`send.py`	Send iMessage with SMS/RCS fallback and real delivery verification
`watermark.py`	Tracks what your agent has read, so a send can be refused if the thread moved
`cal.py`	Read the local Calendar via AppleScript
	#!/usr/bin/env python3
	"""Read and send chats on any Beeper-bridged network (WhatsApp, Instagram,
	Telegram, Signal, LinkedIn, Messenger, X, Discord, Slack, ...).

	beeper.py --list recently active chats, all networks
	beeper.py "Chris Leiter" last 20 messages
	beeper.py "Chris Leiter" -n 40
	beeper.py "Sarah" --network whatsapp narrow when a name is ambiguous
	beeper.py --send "Chris Leiter" "text" send (see the guard below)
	beeper.py --auth re-run the OAuth handshake

	Full-text search over message content, across every network at once:

	beeper.py --search "fund II" everywhere
	beeper.py --search dinner --in "Chris Leiter" within one thread
	beeper.py --search "intro" --from me --since 2026-01-01

	--search matches what was said; a bare name argument matches who it was said
	to. Use --search when you remember a phrase, a name lookup when you remember a
	person.

	Sending refuses to guess. If a name matches zero or several chats it prints the
	candidates and exits 1 rather than picking one, because a message delivered to
	the wrong thread cannot be recalled. Narrow with --network, or pass the exact
	chat ID shown by --list.

	For iMessage use msgs.py instead: it reads chat.db directly and works whether or
	not Beeper is running.

	Requires Beeper Desktop to be running (the API dies with the app, and
	beeper-nightly-quit.sh quits it when the machine idles overnight).
	"""
	import argparse, json, os, re, sys, urllib.error, urllib.parse, urllib.request
	from pathlib import Path

	sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
	import watermark # per-session read watermarks; --send refuses a stale draft

	BASE = "http://127.0.0.1:23373"
	TOKEN_FILE = Path.home() / ".claude/.beeper-token.json"
	REDIRECT_PORT = 8899


	def die(msg, code=1):
	print(msg, file=sys.stderr)
	raise SystemExit(code)


	def strip_html(s):
	"""Bridged messages arrive as HTML fragments; render them flat for the terminal."""
	s = re.sub(r"</p>\s*<p>", "\n", s or "")
	s = re.sub(r"<br\s*/?>", "\n", s)
	s = re.sub(r"<[^>]+>", "", s)
	return (s.replace("&", "&").replace("<", "<")
	.replace(">", ">").replace(""", '"').replace("'", "'").strip())


	def api(path, method="GET", body=None, params=None):
	if params:
	path += "?" + urllib.parse.urlencode(params, doseq=True)
	tok = load_token()
	req = urllib.request.Request(
	BASE + path,
	data=json.dumps(body).encode() if body is not None else None,
	headers={"Authorization": f"Bearer {tok}", "Content-Type": "application/json"},
	method=method,
	)
	try:
	with urllib.request.urlopen(req, timeout=30) as r:
	return json.load(r)
	except urllib.error.HTTPError as e:
	if e.code == 401:
	die("Token rejected. Re-run: beeper.py --auth")
	die(f"HTTP {e.code}: {e.read().decode()[:300]}")
	except urllib.error.URLError:
	die("Cannot reach Beeper on 127.0.0.1:23373. Is Beeper Desktop running?")


	def load_token():
	if not TOKEN_FILE.exists():
	die("No saved token. Run: beeper.py --auth")
	return json.loads(TOKEN_FILE.read_text())["access_token"]


	def do_auth():
	"""OAuth2 PKCE against the local server. Registers a client, then opens a browser
	approval that Beeper serves itself."""
	import base64, hashlib, http.server, secrets, threading, webbrowser

	redirect = f"http://127.0.0.1:{REDIRECT_PORT}/callback"
	reg = json.loads(urllib.request.urlopen(urllib.request.Request(
	f"{BASE}/oauth/register",
	data=json.dumps({
	"client_name": "claude-cli", "redirect_uris": [redirect],
	"grant_types": ["authorization_code"], "response_types": ["code"],
	"token_endpoint_auth_method": "none", "scope": "read write",
	}).encode(),
	headers={"Content-Type": "application/json"}, method="POST"), timeout=15).read())
	client_id = reg["client_id"]

	verifier = base64.urlsafe_b64encode(os.urandom(40)).decode().rstrip("=")
	challenge = base64.urlsafe_b64encode(
	hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=")
	state = secrets.token_urlsafe(16)
	got = {}

	class H(http.server.BaseHTTPRequestHandler):
	def do_GET(self):
	got.update({k: v[0] for k, v in urllib.parse.parse_qs(
	urllib.parse.urlparse(self.path).query).items()})
	self.send_response(200)
	self.send_header("Content-Type", "text/html")
	self.end_headers()
	self.wfile.write(b"<h2>Approved. You can close this tab.</h2>")
	threading.Thread(target=self.server.shutdown).start()

	def log_message(self, *a):
	pass

	srv = http.server.HTTPServer(("127.0.0.1", REDIRECT_PORT), H)
	url = f"{BASE}/oauth/authorize?" + urllib.parse.urlencode({
	"response_type": "code", "client_id": client_id, "redirect_uri": redirect,
	"scope": "read write", "state": state,
	"code_challenge": challenge, "code_challenge_method": "S256"})
	print("Approve in the browser tab that just opened:\n " + url)
	webbrowser.open(url)
	srv.serve_forever()

	if got.get("state") != state or "code" not in got:
	die("Authorization failed or was denied.")
	tok = json.loads(urllib.request.urlopen(urllib.request.Request(
	f"{BASE}/oauth/token",
	data=urllib.parse.urlencode({
	"grant_type": "authorization_code", "code": got["code"],
	"redirect_uri": redirect, "client_id": client_id,
	"code_verifier": verifier}).encode(),
	headers={"Content-Type": "application/x-www-form-urlencoded"}), timeout=20).read())
	TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True)
	TOKEN_FILE.write_text(json.dumps({**tok, "client_id": client_id}))
	TOKEN_FILE.chmod(0o600)
	print(f"Saved to {TOKEN_FILE} (scopes: {tok.get('scope')})")


	def find_chats(name, network=None):
	if name.startswith("!") or name.startswith("~"):
	return [api("/v1/chats/" + urllib.parse.quote(name, safe=""))]
	p = {"query": name, "scope": "titles", "limit": 50, "includeMuted": True}
	hits = api("/v1/chats/search", params=p).get("items", [])
	if not hits:
	hits = api("/v1/chats/search", params={**p, "scope": "participants"}).get("items", [])
	if network:
	hits = [c for c in hits if network.lower() in (c.get("network", "") + c.get("accountID", "")).lower()]
	# An exact title match wins outright, so "Chris Leiter" never fans out.
	exact = [c for c in hits if (c.get("title") or "").lower() == name.lower()]
	return exact or hits


	def _cursor(msgs):
	"""Newest timestamp in a message list, as a comparable string. Beeper has no
	monotonic row id, but ISO-8601 timestamps sort correctly as strings."""
	return max((m.get("timestamp") or "" for m in msgs), default="") or None


	def head_cursor(chat_id):
	msgs = api(f"/v1/chats/{urllib.parse.quote(chat_id, safe='')}/messages",
	params={"limit": 1}).get("items", [])
	return _cursor(msgs)


	def show(chat, n):
	# The endpoint ignores `limit`, so trim client-side. Items arrive newest-first.
	msgs = api(f"/v1/chats/{urllib.parse.quote(chat['id'], safe='')}/messages",
	params={"limit": n}).get("items", [])[:n]
	# Record the newest message this session has now seen, so --send can refuse a
	# reply drafted before they said something new. watermark.py
	watermark.mark("wa", chat["id"], _cursor(msgs))
	print(f"=== {chat.get('title')} ({chat.get('network')}) — {len(msgs)} msgs ===")
	for m in reversed(msgs):
	body = strip_html(m.get("text") or "")
	if not body and m.get("type") != "TEXT":
	body = f"[{(m.get('type') or 'attachment').lower()}]"
	who = "ME " if m.get("isSender") else "THEM"
	print(f"{m['timestamp'][:16].replace('T',' ')} {who} {body}")


	def do_search(a):
	# Beeper ignores `sender`, `dateAfter` and `dateBefore` — verified 2026-07-25, they
	# return identical results filtered or not. Apply them here instead, and over-fetch
	# so the trim still leaves n rows. `chatIDs` does work server-side.
	local = bool(a.sender or a.since or a.until)
	p = {"query": a.search, "limit": min(a.n if not local else 20, 20),
	"includeMuted": True, "excludeLowPriority": False}
	if a.within:
	chats = find_chats(a.within, a.network)
	if len(chats) != 1:
	die(f"--in {a.within!r} matched {len(chats)} chats; be more specific.")
	p["chatIDs"] = [chats[0]["id"]]

	lo = (a.since if "T" in a.since else a.since + "T00:00:00Z") if a.since else None
	hi = (a.until if "T" in a.until else a.until + "T00:00:00Z") if a.until else None

	def keep(m):
	if a.sender and bool(m.get("isSender")) != (a.sender == "me"):
	return False
	if lo and m.get("timestamp", "") < lo:
	return False
	if hi and m.get("timestamp", "") >= hi:
	return False
	return True

	# Results come newest-first, 20 per page. With client-side filters we page back
	# until we have n, but a narrow --since/--until can outrun the cap, so say so
	# rather than printing "no messages" and implying the archive is empty.
	MAX_PAGES = 100 if local else 1
	hits, chats, cursor, pages, exhausted = [], {}, None, 0, False
	while pages < MAX_PAGES:
	if cursor:
	p.update({"cursor": cursor, "direction": "before"})
	r = api("/v1/messages/search", params=p)
	pages += 1
	items = r.get("items", [])
	chats.update(r.get("chats") or {})
	hits += [m for m in items if keep(m)]
	cursor = r.get("oldestCursor")
	if not r.get("hasMore") or not cursor or not items:
	exhausted = True
	break
	# Paged past the start of the window; older pages cannot match.
	if lo and items[-1].get("timestamp", "") < lo:
	exhausted = True
	break
	if len(hits) >= a.n:
	break
	# Pages aren't internally sorted, so order explicitly: keep the n newest, show oldest-first.
	hits.sort(key=lambda m: m.get("timestamp", ""), reverse=True)
	hits = hits[:a.n]
	if not hits:
	print(f"No messages matching {a.search!r}"
	+ ("" if exhausted else
	f" in the {pages * 20} most recent hits (stopped at the page cap; "
	f"older matches may exist — narrow with --in, or raise MAX_PAGES)") + ".")
	return
	for m in reversed(hits):
	c = chats.get(m.get("chatID"), {})
	title, net = c.get("title") or "?", c.get("network") or "?"
	who = "ME" if m.get("isSender") else "THEM"
	body = " ".join(strip_html(m.get("text") or "").split())
	print(f"{m['timestamp'][:16].replace('T',' ')} {net:<9} {title[:24]:<24} {who:<4} {body[:120]}")


	def main():
	ap = argparse.ArgumentParser(add_help=False)
	ap.add_argument("target", nargs="?")
	ap.add_argument("body", nargs="?")
	ap.add_argument("-n", type=int, default=20)
	ap.add_argument("--network")
	ap.add_argument("--send", action="store_true")
	ap.add_argument("--force", action="store_true",
	help="send even though this session has not read the newest messages")
	ap.add_argument("--list", action="store_true")
	ap.add_argument("--auth", action="store_true")
	ap.add_argument("--search")
	ap.add_argument("--in", dest="within", help="limit --search to one chat")
	ap.add_argument("--from", dest="sender", choices=["me", "others"])
	ap.add_argument("--since", help="ISO date, e.g. 2026-01-01")
	ap.add_argument("--until")
	ap.add_argument("-h", "--help", action="store_true")
	a = ap.parse_args()

	if a.help or (not a.target and not a.list and not a.auth and not a.search):
	print(__doc__)
	return
	if a.auth:
	return do_auth()
	if a.search:
	return do_search(a)
	if a.list:
	for c in api("/v1/chats/search", params={"limit": 30, "includeMuted": True}).get("items", []):
	print(f"{(c.get('lastActivity') or '')[:10]} {c.get('network','?'):<10} "
	f"{(c.get('title') or '?')[:34]:<34} {c['id']}")
	return

	chats = find_chats(a.target, a.network)
	if not chats:
	die(f"No chat matching {a.target!r}" + (f" on {a.network}" if a.network else ""))
	if len(chats) > 1:
	print(f"{len(chats)} chats match {a.target!r} — narrow with --network or pass an id:",
	file=sys.stderr)
	for c in chats[:12]:
	print(f" {c.get('network','?'):<10} {c.get('title')} {c['id']}", file=sys.stderr)
	raise SystemExit(1)

	chat = chats[0]
	if not a.send:
	return show(chat, a.n)
	if not a.body:
	die("--send needs message text as the second argument")
	if chat.get("isReadOnly"):
	die(f"{chat.get('title')} is read-only.")
	watermark.check("wa", chat["id"], head_cursor(chat["id"]), a.force,
	hint=f' with `beeper.py "{chat.get("title")}" -n 5`')
	api(f"/v1/chats/{urllib.parse.quote(chat['id'], safe='')}/messages",
	method="POST", body={"text": a.body})
	# Our own send advances the thread head; without this the second message of a
	# two-part send trips the unread guard on something we wrote ourselves.
	watermark.mark("wa", chat["id"], head_cursor(chat["id"]))
	print(f"Sent to {chat.get('title')} on {chat.get('network')}.")


	if __name__ == "__main__":
	main()
	#!/usr/bin/env python3
	"""Read the local Calendar (synced with Google Calendar). Read-only.

	cal.py today + next 7 days
	cal.py 2026-07-11 one day
	cal.py 2026-07-09 2026-07-14 a range
	cal.py --free 2026-07-11 gaps between 09:00 and 23:00
	cal.py --all ... include holiday/birthday/Siri calendars

	Every row prints the calendar it came from.

	Reads through Calendar.app via AppleScript rather than Calendar.sqlitedb
	directly: the db lives under ~/Library/Group Containers and needs Full Disk
	Access, which the terminal loses on every OS/app reinstall (fails with
	"Operation not permitted"). AppleScript only needs Automation access, which is
	granted once per terminal app and survives. Slower (a few seconds), but it does
	not silently stop working.
	"""
	import argparse, datetime as dt, subprocess, sys

	# Apple auto-populates these; entries are guesses or noise, not commitments.
	NOISE = {"US Holidays", "Holidays in Germany", "Holidays in United States",
	"Facebook Birthdays", "Birthdays", "Holidays",
	"Siri Suggestions", "Scheduled Reminders",
	"Found in Natural Language", "Found in Mail", "Siri Found in Apps"}

	# Dates come back as offsets in seconds from the range start, not as formatted
	# strings: `date as string` is locale-dependent (this machine renders German
	# month names) and would need reparsing. Subtracting two AppleScript dates
	# yields a plain integer, which is unambiguous and DST-correct.
	SCRIPT = """
	set d1 to current date
	set day of d1 to 1
	set year of d1 to {y}
	set month of d1 to {m}
	set day of d1 to {d}
	set time of d1 to 0
	set d2 to d1 + ({ndays} * days) - 1
	set skip to {{{skip}}}
	set out to ""
	with timeout of 300 seconds
	tell application "Calendar"
	repeat with c in calendars
	set cn to name of c
	if cn is not in skip then
	set evs to (every event of c whose start date >= d1 and start date <= d2)
	repeat with e in evs
	try
	set t to summary of e
	on error
	set t to "(no title)"
	end try
	set out to out & cn & tab & ((start date of e) - d1) & tab & ¬
	((end date of e) - d1) & tab & (allday event of e) & tab & t & linefeed
	end repeat
	end if
	end repeat
	end tell
	end timeout
	return out
	"""


	def events(start, end, keep_noise=False):
	"""One row per event instance. Calendar.app expands recurring events for us,
	so unlike the raw db there is no CalendarItem/OccurrenceCache union to do."""
	ndays = (end - start).days + 1
	skip = "" if keep_noise else ", ".join(f'"{n}"' for n in sorted(NOISE))
	src = SCRIPT.format(y=start.year, m=start.month, d=start.day, ndays=ndays, skip=skip)
	p = subprocess.run(["osascript", "-"], input=src, capture_output=True, text=True)
	if p.returncode != 0:
	err = p.stderr.strip()
	if "-1743" in err or "not allowed" in err.lower():
	err += ("\n\nGrant your terminal access to Calendar: System Settings -> "
	"Privacy & Security -> Automation -> <your terminal> -> Calendar.")
	sys.exit(f"cal.py: reading Calendar failed:\n{err}")

	origin = dt.datetime.combine(start, dt.time.min)
	seen, out = set(), []
	for line in p.stdout.splitlines():
	if not line.strip():
	continue
	cal_title, s, e, all_day, summary = line.split("\t", 4)
	s, e = int(s), int(e)
	key = (s, summary.strip().lower())
	if key in seen: # same event synced into two accounts
	continue
	seen.add(key)
	out.append({
	"start": origin + dt.timedelta(seconds=s),
	"end": origin + dt.timedelta(seconds=e),
	"summary": summary or "(no title)",
	"cal": cal_title,
	"all_day": all_day == "true",
	})
	out.sort(key=lambda x: x["start"])
	return out


	def free_slots(day, evs, day_start=9, day_end=23, min_minutes=45):
	cur = dt.datetime.combine(day, dt.time(day_start))
	stop = dt.datetime.combine(day, dt.time(day_end))
	# Clip every busy block to the window before merging, so an event running
	# past midnight (e.g. Sleep 23:30-07:30) can't push a gap past `stop`.
	busy = []
	for e in evs:
	if e["all_day"]:
	continue
	s, t = max(e["start"], cur), min(e["end"], stop)
	if s < t:
	busy.append((s, t))
	busy.sort()
	merged = []
	for s, e in busy:
	if merged and s <= merged[-1][1]:
	merged[-1][1] = max(merged[-1][1], e)
	else:
	merged.append([s, e])
	gaps = []
	for s, e in merged:
	if s - cur >= dt.timedelta(minutes=min_minutes):
	gaps.append((cur, s))
	cur = max(cur, e)
	if stop - cur >= dt.timedelta(minutes=min_minutes):
	gaps.append((cur, stop))
	return gaps


	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("start", nargs="?")
	ap.add_argument("end", nargs="?")
	ap.add_argument("--free", metavar="DAY")
	ap.add_argument("--all", action="store_true", help="include holiday/birthday/Siri calendars")
	a = ap.parse_args()

	if a.free:
	day = dt.date.fromisoformat(a.free)
	evs = events(day, day)
	print(f"{day:%a %Y-%m-%d} - free (>=45min, 09:00-23:00):")
	for s, e in free_slots(day, evs):
	print(f" {s:%H:%M} - {e:%H:%M}")
	return

	start = dt.date.fromisoformat(a.start) if a.start else dt.date.today()
	end = dt.date.fromisoformat(a.end) if a.end else start + dt.timedelta(days=7)

	day = None
	for e in events(start, end, keep_noise=a.all):
	if e["start"].date() != day:
	day = e["start"].date()
	print(f"\n{day:%a %Y-%m-%d}")
	when = "all-day" if e["all_day"] else f"{e['start']:%H:%M}-{e['end']:%H:%M}"
	print(f" {when:>13} {e['summary']} ({e['cal']})")


	if __name__ == "__main__":
	main()
	#!/usr/bin/env python3
	"""Read iMessage threads. Never sends anything.

	msgs.py "Katie Thomas" last 20 messages
	msgs.py 917-853-3048 -n 40
	msgs.py "Katie Thomas,Rebecca,Bri Landrum" several threads in one call
	msgs.py --list threads with recent activity

	Comma-separate names/numbers to batch. Unmatched names are reported on stderr
	rather than skipped silently.
	"""
	import argparse, os, re, sqlite3, sys
	from pathlib import Path

	sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
	import watermark # per-session read watermarks; send.py refuses stale drafts

	CHAT_DB = Path.home() / "Library/Messages/chat.db"
	AB_DB = Path.home() / "Library/Application Support/AddressBook/AddressBook-v22.abcddb"
	APPLE_EPOCH = 978307200


	def digits(s):
	return re.sub(r"\D", "", s or "")


	def decode_body(blob):
	"""Pull text out of a typedstream attributedBody (used when message.text is NULL)."""
	if not blob:
	return ""
	i = blob.find(b"NSString")
	if i == -1:
	return ""
	j = blob.find(b"+", i)
	if j == -1:
	return ""
	p = j + 1
	n = blob[p]
	p += 1
	if n == 0x81:
	n = int.from_bytes(blob[p:p + 2], "little")
	p += 2
	return blob[p:p + n].decode("utf-8", "replace")


	def resolve(name):
	"""Contact name -> [(full_name, digits)]. Exact full-name matches win outright,
	so `msgs.py "Katie Thomas"` never fans out across six unrelated Katies."""
	if digits(name) and len(digits(name)) >= 7:
	return [(name, digits(name))]
	hits = []
	for db in [AB_DB, (Path.home() / "Library/Application Support/AddressBook/Sources").glob("/AddressBook-v22.abcddb")]:
	if not db.exists():
	continue
	try:
	c = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
	rows = c.execute("""
	SELECT coalesce(r.ZFIRSTNAME,'') \|\| ' ' \|\| coalesce(r.ZLASTNAME,''), p.ZFULLNUMBER
	FROM ZABCDRECORD r JOIN ZABCDPHONENUMBER p ON p.ZOWNER = r.Z_PK
	WHERE lower(coalesce(r.ZFIRSTNAME,'') \|\| ' ' \|\| coalesce(r.ZLASTNAME,'')) LIKE ?
	""", (f"%{name.lower()}%",)).fetchall()
	hits += [(f.strip(), digits(num)) for f, num in rows if num]
	c.close()
	except sqlite3.Error:
	pass
	exact = [h for h in hits if h[0].lower() == name.strip().lower()]
	hits = exact or hits
	return list(dict.fromkeys(hits))


	def fetch(number, n):
	c = sqlite3.connect(f"file:{CHAT_DB}?mode=ro", uri=True)
	c.create_function("digits", 1, digits)
	rows = c.execute("""
	SELECT m.date, m.is_from_me, m.text, m.attributedBody, m.ROWID
	FROM message m
	JOIN chat_message_join cmj ON m.ROWID = cmj.message_id
	JOIN chat ch ON ch.ROWID = cmj.chat_id
	WHERE digits(ch.chat_identifier) LIKE ?
	ORDER BY m.date DESC LIMIT ?
	""", (f"%{number[-10:]}", n)).fetchall()
	c.close()
	out = []
	max_rowid = max((r[4] for r in rows), default=None)
	for date, from_me, text, body, _rowid in reversed(rows):
	msg = text or decode_body(body)
	if not msg.strip():
	continue
	ts = date / 1_000_000_000 + APPLE_EPOCH
	out.append((ts, from_me, msg))
	return out, max_rowid


	STALE_DAYS = 60

	def last_inbound_ts(number):
	"""Newest message FROM them on this handle. Our own sends don't prove the
	channel is live for the other side, so staleness is measured on theirs."""
	c = sqlite3.connect(f"file:{CHAT_DB}?mode=ro", uri=True)
	c.create_function("digits", 1, digits)
	row = c.execute("""
	SELECT MAX(m.date)
	FROM message m
	JOIN chat_message_join cmj ON m.ROWID = cmj.message_id
	JOIN chat ch ON ch.ROWID = cmj.chat_id
	WHERE digits(ch.chat_identifier) LIKE ? AND m.is_from_me = 0
	""", (f"%{number[-10:]}",)).fetchone()
	c.close()
	return row[0] / 1_000_000_000 + APPLE_EPOCH if row and row[0] else None

	def staleness_note(number):
	"""Warn when they last replied here long ago: the conversation likely moved."""
	import datetime
	ts = last_inbound_ts(number)
	if ts is None:
	return ("!! NO INBOUND: they have never replied on this handle. "
	"Check `beeper.py --list` before assuming this is the live channel.")
	age = (datetime.datetime.now() - datetime.datetime.fromtimestamp(ts)).days
	if age < STALE_DAYS:
	return None
	return (f"!! STALE: they last replied here {age} days ago "
	f"({datetime.datetime.fromtimestamp(ts):%Y-%m-%d}). Check `beeper.py --list` "
	f"or `beeper.py \"<name>\"` before assuming this is the live channel.")

	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("who", nargs="?")
	ap.add_argument("-n", type=int, default=20)
	ap.add_argument("--list", action="store_true")
	a = ap.parse_args()

	if a.list:
	c = sqlite3.connect(f"file:{CHAT_DB}?mode=ro", uri=True)
	for cid, d in c.execute("""
	SELECT ch.chat_identifier, max(m.date)
	FROM chat ch JOIN chat_message_join j ON j.chat_id = ch.ROWID
	JOIN message m ON m.ROWID = j.message_id
	GROUP BY ch.chat_identifier ORDER BY max(m.date) DESC LIMIT 30
	"""):
	import datetime
	print(f"{datetime.datetime.fromtimestamp(d/1e9+APPLE_EPOCH):%Y-%m-%d} {cid}")
	return

	if not a.who:
	ap.error("need a name or number")

	import datetime
	targets = [w.strip() for w in a.who.split(",") if w.strip()]
	missing, ambiguous = [], []

	for who in targets:
	matches = resolve(who)
	if not matches:
	missing.append(who)
	continue
	printed = 0
	for full_name, num in matches:
	msgs, max_rowid = fetch(num, a.n)
	if not msgs:
	continue
	# Record that this session has now seen the thread's newest message,
	# so send.py can refuse a reply drafted off a stale read. watermark.py
	watermark.mark("imsg", num, max_rowid)
	printed += 1
	print(f"=== {full_name or who} ({num}) — {len(msgs)} msgs ===")
	stale = staleness_note(num)
	if stale:
	print(stale)
	for ts, from_me, msg in msgs:
	when = datetime.datetime.fromtimestamp(ts).strftime("%a %Y-%m-%d %H:%M")
	print(f"{when} {'ME ' if from_me else 'THEM'} {msg}")
	print()
	if printed == 0:
	missing.append(who)
	elif printed > 1:
	ambiguous.append(f"{who} ({printed})")

	# Report on stderr so a bad name in a batch never silently vanishes.
	if missing:
	print(f"no thread found for: {', '.join(missing)}", file=sys.stderr)
	if ambiguous:
	print(f"matched multiple contacts, pass a full name: {', '.join(ambiguous)}", file=sys.stderr)
	if missing and len(missing) == len(targets):
	sys.exit(1)


	if __name__ == "__main__":
	main()
	#!/usr/bin/env python3
	"""Send an iMessage/SMS reliably, with automatic SMS fallback and real verification.

	Usage:
	send.py "+15551234567" "message text"
	send.py --chat "any;+;<guid>" "message text" # existing group chat
	send.py --sms "+15551234567" "message text" # force SMS

	Why this exists: `send to participant ... of (account whose service type = iMessage)`
	silently fails for handles not registered with iMessage. The message lands in
	chat.db with is_sent=0 and error=22, and reading the thread back still SHOWS the
	text - so a naive "did it appear?" check reports success on a message that never
	sent. This script polls is_sent/error and retries over SMS.
	"""
	import argparse, importlib.util, os, sqlite3, subprocess, sys, time

	sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
	import watermark # refuses a send drafted off a thread you have not read

	# Reuse msgs.py's typedstream decoder: message.text is NULL for most rows and the
	# body lives in message.attributedBody. Matching on `text` alone silently fails.
	_spec = importlib.util.spec_from_file_location(
	"_msgs", os.path.join(os.path.dirname(os.path.abspath(__file__)), "msgs.py"))
	_msgs = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(_msgs)

	DB = os.path.expanduser("~/Library/Messages/chat.db")

	APPLE_PARTICIPANT = '''
	tell application "Messages"
	set svc to 1st account whose service type = %s
	set msgText to (read (POSIX file (system attribute "MSGTEXT")) as «class utf8»)
	send msgText to participant (system attribute "MSGTO") of svc
	end tell
	'''
	APPLE_CHAT = '''
	tell application "Messages"
	set msgText to (read (POSIX file (system attribute "MSGTEXT")) as «class utf8»)
	send msgText to chat id (system attribute "MSGTO")
	end tell
	'''

	def osa(script, to, text):
	# `system attribute` hands AppleScript the raw env bytes and it decodes them as
	# MacRoman, so "nächste" arrives as "n√§chste". Write UTF-8 to a temp file and
	# let AppleScript read it back with an explicit encoding instead.
	import tempfile
	fd, path = tempfile.mkstemp(suffix=".txt")
	try:
	with os.fdopen(fd, "wb") as f:
	f.write(text.encode("utf-8"))
	env = dict(os.environ, MSGTEXT=path, MSGTO=to)
	r = subprocess.run(["osascript", "-e", script], env=env,
	capture_output=True, text=True)
	return r.returncode, r.stderr.strip()
	finally:
	os.path.exists(path) and os.unlink(path)

	def latest_outgoing(since_ns, needle):
	"""Return (is_sent, error, service) for our message, or None if not in db yet."""
	db = sqlite3.connect(DB)
	rows = db.execute(
	"SELECT is_sent, error, service, text, attributedBody FROM message "
	"WHERE is_from_me=1 AND date > ? ORDER BY date DESC LIMIT 15", (since_ns,)).fetchall()
	db.close()
	key = needle[:40]
	for is_sent, err, service, text, ab in rows:
	body = text or (_msgs.decode_body(ab) if ab else "")
	if body and key in body:
	return is_sent, err, service
	return None

	def now_ns():
	return int((time.time() - 978307200) * 10**9)

	def deliver(kind, to, text, service):
	start = now_ns() - 2 * 10**9
	script = APPLE_CHAT if kind == "chat" else (APPLE_PARTICIPANT % service)
	code, err = osa(script, to, text)
	if code != 0:
	return "failed", "osascript failed: " + err
	for _ in range(30): # up to ~30s; RCS confirms slower than iMessage/SMS
	time.sleep(1)
	st = latest_outgoing(start, text)
	if st and st[0] == 1 and st[1] == 0:
	return "ok", "sent via %s" % (st[2] or service)
	if st and st[1] not in (0, None):
	return "failed", "error %s on %s" % (st[1], st[2] or service)
	# Unconfirmed is NOT failed. Retrying here is how you double-send: the first
	# message may have gone out fine and simply not been confirmed yet.
	return "unknown", "no confirmation yet - check the thread before resending"

	STALE_DAYS = 60

	def last_inbound_ts(number):
	"""Newest message FROM them on this handle, as a unix ts, or None."""
	n = "".join(ch for ch in number if ch.isdigit())[-10:]
	if not n:
	return None
	db = sqlite3.connect("file:%s?mode=ro" % DB, uri=True)
	db.create_function("digits", 1, lambda s: "".join(c for c in (s or "") if c.isdigit()))
	row = db.execute(
	"SELECT MAX(m.date) FROM message m "
	"JOIN chat_message_join cmj ON m.ROWID = cmj.message_id "
	"JOIN chat ch ON ch.ROWID = cmj.chat_id "
	"WHERE digits(ch.chat_identifier) LIKE ? AND m.is_from_me = 0", ("%" + n,)).fetchone()
	db.close()
	return row[0] / 10**9 + 978307200 if row and row[0] else None

	def thread_head(number):
	"""(max ROWID, is_from_me, ts) of the newest message on this handle."""
	n = "".join(ch for ch in number if ch.isdigit())[-10:]
	if not n:
	return None, None, None
	db = sqlite3.connect("file:%s?mode=ro" % DB, uri=True)
	db.create_function("digits", 1, lambda s: "".join(c for c in (s or "") if c.isdigit()))
	row = db.execute(
	"SELECT m.ROWID, m.is_from_me, m.date FROM message m "
	"JOIN chat_message_join cmj ON m.ROWID = cmj.message_id "
	"JOIN chat ch ON ch.ROWID = cmj.chat_id "
	"WHERE digits(ch.chat_identifier) LIKE ? "
	"ORDER BY m.ROWID DESC LIMIT 1", ("%" + n,)).fetchone()
	db.close()
	if not row:
	return None, None, None
	return row[0], row[1], row[2] / 10**9 + 978307200


	# A deliberate two-message send ("Hey X ..." then "Also ...") is normal and lands
	# within a couple of minutes. An unanswered message from hours ago is the case
	# that goes wrong: the schedule moved on and the new message silently contradicts
	# the old one without saying "actually".
	DOUBLE_TEXT_GRACE = 30 * 60

	def unanswered_guard(number, force):
	rowid, from_me, ts = thread_head(number)
	if not from_me or ts is None:
	return
	age = time.time() - ts
	if age < DOUBLE_TEXT_GRACE:
	return
	msg = ("UNANSWERED %s: your last message (%s ago, unanswered) is the newest in\n"
	" this thread. If this one changes the ask, say so - "
	"\"actually, could we do X instead?\"\n"
	" Two contradicting proposals with no reply between them read as a "
	"double-text.\n"
	" Re-run with --force once the wording accounts for it." % (
	number, "%dh%dm" % (age // 3600, (age % 3600) // 60)))
	if force:
	print("(--force: previous message unanswered)")
	return
	print(msg, file=sys.stderr)
	sys.exit(4)


	def stale_guard(number, force):
	"""iMessage/SMS is not the only channel. If they last replied here ages ago,
	the conversation has probably moved (WhatsApp etc.) - stop rather than send
	into a dead thread. See beeper.py."""
	ts = last_inbound_ts(number)
	if ts is None:
	why = "they have never replied on this handle"
	else:
	age = int((time.time() - ts) / 86400)
	if age < STALE_DAYS:
	return
	why = "they last replied here %d days ago (%s)" % (
	age, time.strftime("%Y-%m-%d", time.localtime(ts)))
	msg = ("STALE %s: %s.\n"
	" Check `beeper.py --list` / `beeper.py \"<name>\"` for a live channel,\n"
	" then re-run with --force if iMessage/SMS really is right." % (number, why))
	if not force:
	print(msg, file=sys.stderr)
	sys.exit(2)
	print("(--force: " + why + ")")

	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("target"); ap.add_argument("text")
	ap.add_argument("--chat", action="store_true", help="target is a chat guid")
	ap.add_argument("--sms", action="store_true", help="force SMS")
	ap.add_argument("--force", action="store_true", help="send even if this channel looks dead")
	a = ap.parse_args()

	if a.chat:
	st, msg = deliver("chat", a.target, a.text, "iMessage")
	print(("OK " if st == "ok" else st.upper() + " ") + msg)
	sys.exit(0 if st == "ok" else 1)

	stale_guard(a.target, a.force)
	head, _, _ = thread_head(a.target)
	watermark.check("imsg", a.target, head, a.force,
	hint=' with `msgs.py "%s" -n 5`' % a.target)
	unanswered_guard(a.target, a.force)

	first = "SMS" if a.sms else "iMessage"
	st, msg = deliver("participant", a.target, a.text, first)
	# Our own send advances the thread head; without this the next send in a
	# two-message sequence trips the unread guard on a message we wrote.
	watermark.mark("imsg", a.target, thread_head(a.target)[0])
	print(("OK " if st == "ok" else ".. ") + msg)
	if st == "ok":
	sys.exit(0)
	# Only a hard failure earns a retry. "unknown" means it may already be out,
	# and resending on unknown is exactly how a message gets delivered twice.
	if st == "failed" and first == "iMessage":
	print(".. retrying over SMS")
	st, msg = deliver("participant", a.target, a.text, "SMS")
	print(("OK " if st == "ok" else st.upper() + " ") + msg)
	sys.exit(0 if st == "ok" else 1)

	if __name__ == "__main__":
	main()
	#!/usr/bin/env python3
	"""Per-reader watermarks: what has THIS agent session actually read in a thread?

	Why: drafting a message off a thread you read ten minutes ago sends a reply that
	ignores what they said since. Instructions ("re-read before sending") only bind a
	careful reader; this binds a careless one. Design is the .seen-<reader> watermark
	from slima4/agent-message rather than an ETag/If-Match token, so the caller never
	has to carry an id around - reading records the watermark as a side effect.

	Keyed by CLAUDE_CODE_SESSION_ID so parallel agent sessions never clobber each
	other's idea of what they've seen. No session id (a human at a shell) -> "manual".

	Watermarks live in ~/.claude/.msg-watermarks/<session>.json:
	{"imsg:+15551234567": 402931, "wa:!AbC:beeper.local": 1755712345000}

	The value is whatever monotonic cursor that backend has: chat.db ROWID for
	iMessage, an epoch-ms timestamp for Beeper. Only compared within one key.
	"""
	import json, os, sys, tempfile

	DIR = os.path.expanduser("~/.claude/.msg-watermarks")


	def session_id():
	return os.environ.get("CLAUDE_CODE_SESSION_ID") or "manual"


	def _path():
	return os.path.join(DIR, session_id() + ".json")


	def key(kind, peer):
	"""kind: 'imsg' \| 'wa' (any beeper network). peer: handle, chat guid, or chat id.

	Phone numbers are normalized to their last 10 digits so +1555..., 555...,
	and (555) ... all land on the same watermark - the same normalization
	send.py already does when it looks up a handle."""
	peer = (peer or "").strip()
	digits = "".join(c for c in peer if c.isdigit())
	if kind == "imsg" and len(digits) >= 10 and not peer.startswith("imsg#"):
	peer = digits[-10:]
	return "%s:%s" % (kind, peer)


	def _load():
	try:
	with open(_path()) as f:
	return json.load(f)
	except (OSError, ValueError):
	return {}


	def mark(kind, peer, cursor):
	"""Record that this session has seen up to `cursor` in a thread. Monotonic:
	a later read of an older slice never walks the watermark backwards."""
	if cursor is None:
	return
	data = _load()
	k = key(kind, peer)
	prev = data.get(k)
	# Cursors are ints (chat.db ROWID) or ISO strings (Beeper); never mix within a key.
	if prev is not None and type(prev) is type(cursor) and prev >= cursor:
	return
	data[k] = cursor
	os.makedirs(DIR, exist_ok=True)
	fd, tmp = tempfile.mkstemp(dir=DIR)
	try:
	with os.fdopen(fd, "w") as f:
	json.dump(data, f)
	os.replace(tmp, _path())
	except Exception:
	os.path.exists(tmp) and os.unlink(tmp)
	raise


	def seen(kind, peer):
	"""Cursor this session has read up to, or None if it has never read here."""
	return _load().get(key(kind, peer))


	def check(kind, peer, current, force=False, hint=""):
	"""Guard a send. Exits 3 unless this session has read the thread's newest
	message. Returns quietly when current is None (nothing to compare)."""
	if current is None:
	return
	s = seen(kind, peer)
	if s is not None and type(s) is type(current) and s >= current:
	return
	why = ("you have not read this thread in this session"
	if s is None else "new messages arrived since you read it")
	msg = ("UNREAD %s: %s.\n"
	" Read it first%s, then send. The reply you drafted may be\n"
	" answering something they have already moved past." % (peer, why, hint))
	if force:
	print("(--force: " + why + ")")
	return
	print(msg, file=sys.stderr)
	sys.exit(3)
	#!/usr/bin/env python3
	"""Which platform should I message this person on?

	where.py "Jane Doe" every channel, most recently active first
	where.py +15551234567 a phone number works too

	Answers the question that has to come BEFORE drafting: a quiet iMessage thread
	usually means the conversation moved to WhatsApp/Instagram/etc, not that the two
	of you rarely talk. Sending into the stale one also gets the content wrong, since
	the recent context lives in the other thread.

	Reads only. Combines msgs.py (iMessage/SMS/RCS via chat.db) with beeper.py
	(everything Beeper bridges). Beeper Desktop must be running for its half; if it
	isn't, that is reported instead of silently showing an iMessage-only picture.
	"""
	import argparse, importlib.util, sqlite3, sys, time, datetime, pathlib

	HERE = pathlib.Path(__file__).resolve().parent
	CHAT_DB = pathlib.Path.home() / "Library/Messages/chat.db"
	APPLE_EPOCH = 978307200
	WINDOW_DAYS = 180 # how far back "how much do we actually talk here" looks

	def load(mod):
	spec = importlib.util.spec_from_file_location(mod, HERE / (mod + ".py"))
	m = importlib.util.module_from_spec(spec)
	spec.loader.exec_module(m)
	return m

	def imessage_rows(who):
	"""One row per matching handle: last activity + how much they actually write here."""
	msgs = load("msgs")
	try:
	matches = msgs.resolve(who)
	except Exception as e:
	return [], "chat.db lookup failed: %s" % e
	cutoff = (time.time() - WINDOW_DAYS * 86400 - APPLE_EPOCH) * 1e9
	out = []
	c = sqlite3.connect("file:%s?mode=ro" % CHAT_DB, uri=True)
	c.create_function("digits", 1, msgs.digits)
	for full_name, num in matches:
	# 1:1 chats only. Group threads share a handle and would otherwise
	# present as "a channel we use", which they are not.
	row = c.execute(
	"SELECT MAX(m.date), MAX(CASE WHEN m.is_from_me=0 THEN m.date END), "
	" SUM(CASE WHEN m.is_from_me=0 AND m.date > ? THEN 1 ELSE 0 END), "
	" COUNT(*) "
	"FROM message m "
	"JOIN chat_message_join cmj ON m.ROWID=cmj.message_id "
	"JOIN chat ch ON ch.ROWID=cmj.chat_id "
	"WHERE digits(ch.chat_identifier) LIKE ? "
	" AND (SELECT COUNT(*) FROM chat_handle_join chj WHERE chj.chat_id=ch.ROWID) = 1",
	(cutoff, "%" + num[-10:])).fetchone()
	if not row or not row[0]:
	continue
	svc = c.execute(
	"SELECT m.service FROM message m "
	"JOIN chat_message_join j ON m.ROWID=j.message_id "
	"JOIN chat ch ON ch.ROWID=j.chat_id "
	"WHERE digits(ch.chat_identifier) LIKE ? ORDER BY m.date DESC LIMIT 1",
	("%" + num[-10:],)).fetchone()
	to_ts = lambda d: d / 1e9 + APPLE_EPOCH if d else None
	out.append({"channel": (svc[0] if svc else None) or "iMessage",
	"title": full_name or num, "who": num,
	"any": to_ts(row[0]), "inbound": to_ts(row[1]),
	"count": row[2] or 0,
	"send": 'send.py "%s" "..."' % num})
	c.close()
	return out, None


	def beeper_rows(who):
	try:
	b = load("beeper")
	chats = b.find_chats(who)
	digits = "".join(c for c in who if c.isdigit())
	if not chats and len(digits) >= 7:
	# Beeper stores participant numbers in several formats; a raw
	# "+1940..." query often matches none of them.
	seen = {c["id"] for c in chats}
	for variant in ("+" + digits, digits, digits[-10:]):
	for c in b.find_chats(variant):
	if c["id"] not in seen:
	seen.add(c["id"]); chats.append(c)
	except SystemExit as e:
	return [], "beeper unavailable (is Beeper Desktop running?)"
	except Exception as e:
	return [], "beeper lookup failed: %s" % e
	out = []
	cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=WINDOW_DAYS)
	for ch in chats:
	if ch.get("type") != "single":
	continue # group chats are not "a channel we use" for one person
	if (ch.get("network") or "").lower() == "imessage":
	continue # Beeper bridges iMessage too; chat.db already covers it
	try:
	import urllib.parse
	msgs = b.api("/v1/chats/%s/messages" % urllib.parse.quote(ch["id"], safe=""),
	params={"limit": 100}).get("items", [])
	except Exception:
	continue
	if not msgs:
	continue
	iso = lambda s: datetime.datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
	inbound = [m for m in msgs if not m.get("isSender")]
	recent = sum(1 for m in inbound
	if datetime.datetime.fromisoformat(
	m["timestamp"].replace("Z", "+00:00")) > cutoff)
	out.append({"channel": ch.get("network") or "?", "title": ch.get("title") or "?",
	"who": ch.get("id"), "any": iso(msgs[0]["timestamp"]),
	"inbound": iso(inbound[0]["timestamp"]) if inbound else None,
	"count": recent,
	"send": "beeper.py --send \"%s\" \"...\"" % (ch.get("title") or ch["id"])})
	return out, None

	def ago(ts):
	if not ts:
	return "never"
	d = int((time.time() - ts) / 86400)
	stamp = time.strftime("%Y-%m-%d", time.localtime(ts))
	return "%s (%dd)" % (stamp, d)

	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("who")
	a = ap.parse_args()

	rows, warn1 = imessage_rows(a.who)
	brows, warn2 = beeper_rows(a.who)
	rows += brows

	for w in (warn1, warn2):
	if w:
	print("!! " + w, file=sys.stderr)
	if not rows:
	print("no threads found for: %s" % a.who, file=sys.stderr)
	sys.exit(1)

	# Rank on how much THEY write here over the window, recency only as tiebreak.
	# Volume, not last-touch: an occasional stray iMessage into a relationship that
	# otherwise lives on WhatsApp should not move the conversation.
	rows.sort(key=lambda r: (r["count"], r["inbound"] or 0), reverse=True)
	by_recency = sorted(rows, key=lambda r: (r["inbound"] or 0), reverse=True)

	print("%-10s %-8s %-16s %-16s %s" % ("CHANNEL", "THEIRS", "LAST ANY", "LAST FROM THEM", "TITLE"))
	for i, r in enumerate(rows):
	mark = " <- USE THIS" if i == 0 else ""
	print("%-10s %-8s %-16s %-16s %s%s" % (
	r["channel"], "%d/%dd" % (r["count"], WINDOW_DAYS),
	ago(r["any"]), ago(r["inbound"]), r["title"], mark))

	print("\nsend here: %s" % rows[0]["send"])
	if len(rows) > 1:
	print("read it first, the recent context lives in that thread.")
	if by_recency[0] is not rows[0]:
	print("note: they replied more recently on %s, but %s is where you two actually "
	"talk. Keep it in one place unless that last message needs answering there."
	% (by_recency[0]["channel"], rows[0]["channel"]))

	if __name__ == "__main__":
	main()