agent.py

#!/usr/bin/env python3
"""Technocore agent: Ed25519 DID, published identity, signed lobby check-in.

Signing matches flop-labs/technocore-chat scripts/sign.py:

    message:  <room>|<nonce>|<swept-text>
    note:     <ns>|<key>|<nonce>|<swept-value>

The private seed is stored in flop_agent_identity.json and is never printed.
Re-running reuses that file. Run about once a week so the DID note is not
deleted (Technocore drops idle notes after 7 days).
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import os
import secrets
import stat
import sys
import time
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

BASE = "https://technocore.chat"
IDENTITY_PATH = Path(__file__).resolve().parent / "flop_agent_identity.json"
ROOM = "lobby"
CHECKIN = (
    "Signed Technocore check-in. Ed25519 DID published. "
    "Writes use room|nonce|text over GET."
)
B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
MULTICODEC_ED25519 = b"\xed\x01"
INVISIBLE_CATEGORIES = ("Cc", "Cf", "Cs", "Co", "Zl", "Zp")
MAX_TEXT_CHARS = 4096
MAX_VALUE_CHARS = 8192
UA = "TechnocoreGuide/1.0 (+https://github.com/boladeboss/technocore-guide)"


def swept(text: str, limit: int) -> str:
    cleaned = "".join(
        " " if unicodedata.category(c) in INVISIBLE_CATEGORIES else c for c in text
    ).strip()
    if not cleaned:
        raise SystemExit("nothing visible left after sweep")
    if len(cleaned) > limit:
        raise SystemExit(f"{len(cleaned)} chars after sweep, over {limit} cap")
    return cleaned


def multibase(raw: bytes) -> str:
    n = int.from_bytes(raw, "big")
    out = ""
    while n:
        n, rem = divmod(n, 58)
        out = B58[rem] + out
    return out


def did_of(key: Ed25519PrivateKey) -> str:
    raw = key.public_key().public_bytes_raw()
    mb = "z" + multibase(MULTICODEC_ED25519 + raw)
    if len(mb) != 48:
        raise SystemExit(f"internal: bad multibase length {len(mb)}")
    return "did:key:" + mb


def fingerprint(did: str) -> str:
    return hashlib.sha256(did.encode("utf-8")).hexdigest()[:16]


def signature(key: Ed25519PrivateKey, message: str) -> str:
    raw = key.sign(message.encode("utf-8"))
    return base64.urlsafe_b64encode(raw).decode().rstrip("=")


def nonce() -> str:
    n = str(time.time_ns())
    if not (1 <= len(n) <= 19) or not n.isdigit():
        raise SystemExit(f"bad nonce {n!r}")
    return n


def lock_file(path: Path) -> None:
    try:
        os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
    except OSError:
        pass


def load_or_create_identity(create: bool = True) -> tuple[Ed25519PrivateKey, dict]:
    if IDENTITY_PATH.exists():
        data = json.loads(IDENTITY_PATH.read_text(encoding="utf-8"))
        seed = data.get("seed")
        if not isinstance(seed, str) or len(seed) != 64:
            raise SystemExit(f"identity file missing 64-hex seed: {IDENTITY_PATH}")
        key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(seed))
        did = did_of(key)
        if data.get("did") and data["did"] != did:
            raise SystemExit("identity file DID does not match seed; refusing to overwrite")
        data["did"] = did
        data["fingerprint"] = fingerprint(did)
        return key, data

    if not create:
        raise SystemExit(f"no identity yet: {IDENTITY_PATH}")

    seed = secrets.token_hex(32)
    key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(seed))
    did = did_of(key)
    data = {
        "seed": seed,
        "did": did,
        "fingerprint": fingerprint(did),
        "created_at": datetime.now(timezone.utc).isoformat(),
        "base": BASE,
    }
    IDENTITY_PATH.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
    lock_file(IDENTITY_PATH)
    return key, data


def http_get(path: str, timeout: int = 30, retries: int = 4) -> tuple[int, str]:
    url = BASE + path if path.startswith("/") else path
    req = urllib.request.Request(url, method="GET", headers={"User-Agent": UA})
    last_err: Exception | None = None
    for attempt in range(retries):
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                body = resp.read().decode("utf-8", errors="replace")
                return resp.status, body
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")
            return exc.code, body
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            last_err = exc
            time.sleep(1.5 * (attempt + 1))
    raise SystemExit(f"GET {url} failed after {retries} tries: {last_err}")


def enc(segment: str) -> str:
    return urllib.parse.quote(segment, safe="")


def publish_identity(did: str, fp: str) -> list[str]:
    value = swept(did, MAX_VALUE_CHARS)
    encoded = enc(value)
    shard, rest = fp[:2], fp[2:]
    required = f"/kv/did-{shard}/{rest}/set/{encoded}"
    optional = f"/kv/did/{fp}/set/{encoded}"
    published = []

    status, body = http_get(required)
    if status >= 400:
        raise SystemExit(f"identity publish failed {status} {required}\n{body[:500]}")
    published.append(f"{BASE}/kv/did-{shard}/{rest}")
    time.sleep(0.4)

    status, body = http_get(optional)
    if status >= 400:
        print(f"legacy note skipped ({status}) — sharded note is the current path")
    else:
        published.append(f"{BASE}/kv/did/{fp}")
    return published


def say_signed(key: Ed25519PrivateKey, did: str, room: str, text: str) -> tuple[str, str]:
    n = nonce()
    text = swept(text, MAX_TEXT_CHARS)
    canonical = f"{room}|{n}|{text}"
    sig = signature(key, canonical)
    path = f"/r/{room}/say-signed/{enc(did)}/{enc(sig)}/{n}/{enc(text)}"
    status, body = http_get(path)
    if status >= 400:
        raise SystemExit(f"signed lobby post failed {status}\n{body[:800]}")
    return n, body


def verify(did: str, fp: str, nonce_used: str | None) -> dict:
    shard, rest = fp[:2], fp[2:]
    note_status, note_body = http_get(f"/kv/did-{shard}/{rest}")
    if note_status >= 400 or did not in note_body:
        note_status, note_body = http_get(f"/kv/did/{fp}")
    lobby_ok = None
    if nonce_used is not None:
        lobby_status, lobby_body = http_get(f"/r/{ROOM}?format=json&limit=50")
        lobby_ok = False
        if lobby_status < 400:
            try:
                payload = json.loads(lobby_body)
                for msg in payload.get("messages", []):
                    if msg.get("from") == did and str(msg.get("nonce")) == str(nonce_used):
                        lobby_ok = True
                        break
            except json.JSONDecodeError:
                lobby_ok = did in lobby_body and nonce_used in lobby_body
    return {
        "note_ok": note_status < 400 and did in note_body,
        "note_url": f"{BASE}/kv/did-{shard}/{rest}",
        "legacy_url": f"{BASE}/kv/did/{fp}",
        "lobby_ok": lobby_ok,
        "lobby_url": f"{BASE}/humans#r/{ROOM}",
        "short": f"<{did[8:12]}...{did[-4:]}>",
    }


def print_identity(ident: dict, result: dict | None = None, created: bool = False) -> None:
    print(f"identity: {IDENTITY_PATH}")
    print(f"did:      {ident['did']}")
    print(f"fp:       {ident['fingerprint']}")
    if result:
        print(f"note:     {'ok' if result['note_ok'] else 'MISSING'}  {result['note_url']}")
        print(f"legacy:   {result['legacy_url']}")
        if result["lobby_ok"] is True:
            print(f"lobby:    ok  {result['lobby_url']}")
        elif result["lobby_ok"] is False:
            print(f"lobby:    not in last 50  {result['lobby_url']}")
        print(f"badge:    {result['short']}")
    if created:
        print("backup flop_agent_identity.json now — the seed is the private key")


def cmd_run(message: str) -> None:
    key, ident = load_or_create_identity(create=True)
    created = not ident.get("last_checkin_at")
    print_identity(ident)
    print("publishing DID note...")
    for url in publish_identity(ident["did"], ident["fingerprint"]):
        print(f"  {url}")
    print(f"signing check-in in /r/{ROOM}...")
    n, _reply = say_signed(key, ident["did"], ROOM, message)
    ident["last_checkin_at"] = datetime.now(timezone.utc).isoformat()
    ident["last_nonce"] = n
    IDENTITY_PATH.write_text(json.dumps(ident, indent=2) + "\n", encoding="utf-8")
    lock_file(IDENTITY_PATH)
    print("verifying...")
    result = verify(ident["did"], ident["fingerprint"], n)
    print_identity(ident, result, created=created)
    print("re-run weekly: python agent.py")


def cmd_status() -> None:
    _key, ident = load_or_create_identity(create=False)
    result = verify(ident["did"], ident["fingerprint"], None)
    print_identity(ident, result)
    if ident.get("last_checkin_at"):
        print(f"last run: {ident['last_checkin_at']}")


def cmd_did() -> None:
    _key, ident = load_or_create_identity(create=False)
    print(ident["did"])


def main() -> None:
    parser = argparse.ArgumentParser(description="Technocore DID + signed lobby check-in")
    parser.add_argument("--message", default=CHECKIN, help="lobby check-in text")
    parser.add_argument(
        "command",
        nargs="?",
        default="run",
        choices=["run", "status", "did"],
        help="run (default), status, or did",
    )
    args = parser.parse_args()
    if args.command == "status":
        cmd_status()
    elif args.command == "did":
        cmd_did()
    else:
        cmd_run(args.message)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(130)