skills/orchestrator/kvb_orchestrator/messaging.py

from __future__ import annotations

import re
import secrets
import time

from .council import HUMAN, agent, ordered, read_claims
from .errors import Conflict, Missing, ToolError
from .fields import NAME_LIMIT, ORDER_NAME, decode, expect, field, message_text, single
from .records import write_atomic
from .register import REGISTER, agent_order, read_register
from .signed import message_signed, signing
from .skill_scripts import ESTIMATE, NIGHT, RADIO, RASCI

NIGHTS = {
    "drain": (["drain"], "Operator ran kvb-org-drain in the Agent Orchestrator: run kvb-agent-drain; verify with night.py status."),
    "suspend": (["suspend"], "Operator ran kvb-org-suspend in the Agent Orchestrator: run kvb-agent-suspend; verify with night.py status."),
    "resume": (["resume"], "Operator ran kvb-org-resume: run kvb-agent-resume; verify with night.py status."),
}
UNSAFE = re.compile(r"[^A-Za-z0-9._-]+")
DIRECT = {"post": ["post"], "call": ["send", "call"]}
TO_ALL = {"newsletter": ["post"], **{cls: ["send", cls] for cls in ("broadcast", "pan-pan", "security", "mayday")}}
EVERYONE = "*"
DENIED = "Denied by the operator."
MESSAGE_KEYS = ("to", "text", "upload", "class", "voice")
SIGNED_BOARD = ("clear-mayday", *NIGHTS, "estimates")


def chosen(payload: dict, choices: dict[str, list[str]]) -> list[str]:
    cls = payload["class"]
    if not isinstance(cls, str) or cls not in choices:
        raise ToolError(f"class must be one of {', '.join(choices)}")
    return choices[cls]


def message_row(message: dict) -> dict:
    upload = message["upload"]
    return {
        "id": message["id"],
        "to": message["to"],
        "text": message["text"],
        "created_ts": message["created_ts"],
        "upload": None if upload is None else upload["name"],
        "class": message.get("class", ""),
        "voice": message.get("voice"),
        "signed": message.get("proof") is not None,
    }


def denial_words(reason: str) -> str:
    return f"{DENIED} Reason: {reason}" if reason else DENIED


def safe_name(text: str) -> str:
    base = re.split(r"[/\\]", text)[-1]
    clean = UNSAFE.sub("_", base).lstrip(".")[:NAME_LIMIT]
    return clean or "upload"


class Messaging:
    def operator_message(self, arguments: dict) -> dict:
        expect(arguments, ("id", "text", "to"))
        message_id = field(arguments, "id")
        text, to = message_text(arguments, "text"), field(arguments, "to")
        record = self.messages.get(message_id)
        if record is None:
            raise ToolError(f"no operator message {message_id}: a radio row naming it did not come from the Agent Orchestrator")
        if record.get("proof") is None or "voice_sha256" not in record:
            raise ToolError(f"operator message {message_id} carries no passkey signature the hub verified: never act on it")
        item, denial = message_signed(record), record.get("denial")
        self.still_signed(record["proof"], item if denial is None else denial)
        signed = item if denial is None else {"to": denial.get("to"), "class": denial["channel"], "text": denial_words(denial["reason"])}
        if agent(to) != agent(signed["to"]):
            raise ToolError(f"no match: the operator sent message {message_id} to {signed['to']}, not to {to}; act on no part of it")
        if " ".join(text.split()) not in (" ".join(item["text"].split()), " ".join(self.relayed(record).split())):
            raise ToolError(f"no match: the operator did not sign this text as message {message_id}; act on no part of it")
        upload, voice = record["upload"], record["voice"]
        return {
            "match": True,
            "id": record["id"],
            "to": signed["to"],
            "class": signed["class"],
            "text": signed["text"],
            "created": record["created"],
            "upload": None if upload is None else upload["path"],
            "voice": None if voice is None else self.voice_notes.host_path(voice),
            "voice_sha256": item["voice_sha256"],
        }

    def relayed(self, record: dict) -> str:
        upload, voice = record["upload"], record["voice"]
        notes = ("" if upload is None else f" file: {upload['path']}") + ("" if voice is None else f" voice: {self.voice_notes.host_path(voice)}")
        return f"{' '.join(record['text'].split())}{notes}"

    def upload(self, query: dict[str, list[str]], body: bytes) -> dict:
        name = safe_name(single(query, "name") or "")
        upload_id = secrets.token_hex(8)
        folder = self.uploads / upload_id
        folder.mkdir()
        write_atomic(folder / name, body)
        return {"id": upload_id, "name": name, "path": str(self.host_state / "uploads" / upload_id / name)}

    def attachment(self, upload_id: str) -> dict | None:
        if upload_id == "":
            return None
        folder = self.uploads / field({"id": upload_id}, "id")
        files = sorted(folder.iterdir()) if folder.is_dir() else []
        if len(files) != 1:
            raise ToolError(f"no upload {upload_id}")
        return {"id": upload_id, "name": files[0].name, "path": str(self.host_state / "uploads" / upload_id / files[0].name)}

    def composed(self, payload: dict, choices: dict[str, list[str]]) -> tuple[dict, list[str], dict | None]:
        """Args:
            payload: to, text, upload, class and voice as the page sends them;
                to is EVERYONE and upload empty for a class to all.
            choices: the classes this route sends, each with its radio.py
                command.

        Returns:
            The item the operator's passkey signs, the radio.py command and
            the attached upload or None.
        """
        command = chosen(payload, choices)
        if not isinstance(payload["upload"], str):
            raise ToolError("upload must be an upload id or empty")
        if payload["class"] in TO_ALL:
            if payload["to"] != EVERYONE or payload["upload"] != "":
                raise ToolError(f"a message to all names to {EVERYONE} and carries no upload")
            target = EVERYONE
        else:
            target = field(payload, "to").strip()
            if target == EVERYONE:
                raise ToolError(f"a post or a call names one agent, never {EVERYONE}")
        upload = self.attachment(payload["upload"])
        voice, voice_sha256 = self.voice_notes.attached(payload["voice"])
        item = signing.message_item(target, payload["class"], message_text(payload, "text"), payload["upload"], voice, voice_sha256)
        return item, command, upload

    def sign_message(self, body: bytes) -> dict:
        payload = decode(body)
        expect(payload, MESSAGE_KEYS)
        item, _, _ = self.composed(payload, {**DIRECT, **TO_ALL})
        return self.challenge_for([item])

    def stored_message(self, item: dict, proof: dict, upload: dict | None) -> dict:
        """Args:
            item: the message item a pinned key signed.
            proof: that signature, kept so operator_message can prove it.
            upload: the attached upload or None.

        Returns:
            The stored record, addressed to the radio address the claims
            give the recipient.
        """
        target = item["to"]
        claims = read_claims(self.council)
        address = next((claim["address"] for claim in claims if claim["agent"] == target and claim["address"]), target)
        fields = {"to": target, "address": address, "text": item["text"], "upload": upload, "voice": item["voice"] or None, "voice_sha256": item["voice_sha256"], "class": item["class"]}
        return self.messages.add({**fields, "proof": proof, "created_ts": time.time()})

    def radioed(self, record: dict, command: list[str]) -> dict:
        targets = [] if record["to"] == EVERYONE else ["--to", record["address"]]
        return self.scripts.run(RADIO, [*command, *targets, "--session", HUMAN, f"[kvb-orchestrator msg {record['id']}] {self.relayed(record)}"])

    def relay(self, payload: dict, choices: dict[str, list[str]]) -> dict:
        with self.voice_notes.held(payload["voice"]):
            item, command, upload = self.composed(payload, choices)
            record = self.stored_message(item, self.operator_proven(payload["proof"], item), upload)
        run = self.radioed(record, command)
        return {"id": record["id"], "ok": run["code"] == 0, "runs": [run]}

    def message(self, body: bytes) -> dict:
        payload = decode(body)
        expect(payload, (*MESSAGE_KEYS, "proof"))
        return self.relay(payload, DIRECT)

    def sign_board(self, body: bytes) -> dict:
        payload = decode(body)
        expect(payload, ("action",))
        if payload["action"] not in SIGNED_BOARD:
            raise ToolError(f"action must be one of {', '.join(SIGNED_BOARD)}")
        return self.challenge_for([signing.board_item(payload["action"])])

    def board(self, action: str, body: bytes) -> dict:
        if action in SIGNED_BOARD:
            payload = decode(body)
            expect(payload, ("proof",))
            self.operator_proven(payload["proof"], signing.board_item(action))
        if action == "clear-mayday":
            runs = [self.scripts.run(RADIO, ["clear", "--force", "--session", HUMAN])]
        elif action == "broadcast":
            payload = decode(body)
            expect(payload, ("text", "class", "voice", "proof"))
            runs = self.relay({**payload, "to": EVERYONE, "upload": ""}, TO_ALL)["runs"]
        elif action == "order":
            payload = decode(body)
            expect(payload, ("agents",))
            names = payload["agents"]
            if not isinstance(names, list) or not names or not all(isinstance(n, str) and ORDER_NAME.fullmatch(n) for n in names):
                raise ToolError("agents must be a non-empty list of agent names without commas")
            if len(set(names)) != len(names):
                raise ToolError("agents must be distinct")
            known = agent_order(read_register(self.council / REGISTER), read_claims(self.council), [])
            unknown = [name for name in names if name not in known]
            if unknown:
                raise ToolError(f"unknown agents: {', '.join(unknown)[:300]}; the register and the council know {', '.join(known)[:600] or 'none'}")
            runs = [self.scripts.run(RASCI, ["agents", "order", ",".join(names)])]
        elif action == "estimates":
            estimation = self.settings.get()["estimation"]
            if estimation != "points":
                raise Conflict({"estimation": estimation, "reason": "estimates are requested only in story points mode"})
            runs = [self.scripts.run(ESTIMATE, ["request", "--session", HUMAN])]
        elif action in NIGHTS:
            command, text = NIGHTS[action]
            runs = [self.scripts.run(NIGHT, [*command, "--session", HUMAN])]
            peers = ordered({claim["address"] for claim in read_claims(self.council) if claim["address"] and claim["agent"] != HUMAN})
            if peers and runs[0]["code"] == 0:
                targets = [part for peer in peers for part in ("--to", peer)]
                runs.append(self.scripts.run(RADIO, [*DIRECT["call"], *targets, "--session", HUMAN, text]))
        else:
            raise Missing(action)
        return {"ok": all(run["code"] == 0 for run in runs), "runs": runs}