#!/usr/bin/env python3
"""Report one hook event to the local receiver, and tell a context its name.

A hook that breaks is worse than a hook that learns nothing, so every path
here ends in exit 0, and with no output unless there is something to say.

**What is sent.** Field names and identifiers, and three values. A payload
carries prompts, tool arguments and file paths, and none of that answers any
question `plan/collection.md` asks. The three values are the exceptions
`plan/collection.md` names. `cwd` is the session's working directory.
`seen_request_uris` is the opaque, short-lived OAuth `request_uri` of each
authorize URL a tool call was given or printed; the daemon shows a context only
the sign-ins this list names. `browser_sign_in` is the sign-in page a browser
tool is about to open, or failed to open, an origin and a path.

**Signing the agent's browser in.** A web app's sign-in code goes only to a
browser signed in as the agent. When a browser tool is about to open a
server's sign-in page, the daemon answers with a one-time link, and this hook
writes it into the tool's input (`updatedInput`), so the model never holds
it. The link's ticket and the browser's session cookie are removed from every
browser tool's output (`updatedToolOutput`) before the model reads it. The
harness takes no replacement for a failed tool's error, which can name the
link. So when a browser tool fails holding a link, the hook asks for a new one
and throws it away: the server keeps one live ticket per account, so the new
one ends the ticket the model saw.

**What comes back.** The receiver answers, and the answer is sometimes a name,
sometimes a sign-in waiting on a decision, sometimes both, and sometimes
trouble: why it could not act, such as why the context has no account. A
context that has just been given an identity is told so through
`additionalContext`, which is the only channel that reaches the model itself.
Saying an identity or a refusal twice is the receiver's job to prevent, not
this program's: a hook that tried to remember what it had already said would
be a hook keeping state across processes, and there is one of these per event.
A pending sign-in has no such protection and is rendered on every event in
`SHOWN` while the receiver still offers it — the receiver stops offering it
once it is approved, declined or expired.

**Which local app is asking.** Every local client has the origin
`http://localhost`, so a decision alone cannot tell one from another. The
redirect port that does is in the `client_id` of the authorize URL the client
printed. Every rendering reads the authorize URLs in the acting context's
transcript and in this tool call, sends their `request_uri` values once more,
in a `show`, and matches the decisions that come back by token to ones
already pending. So a sign-in shown again a turn later names the same port.

**Which context is acting.** `PreToolUse` is the only place the harness names
the acting agent, so a subagent that makes a tool call without having been
announced is still identified. `agent_id` is absent when the call comes from
the main conversation, which is how a session is told apart from its children.

**Which context started it.** A subagent's account is created beneath the
subagent that started it, which the report names as `spawner`. Claude Code
records that in a `.meta.json` beside the subagent's transcript, written only
after `SubagentStart`'s hooks return. So a `SubagentStart` that cannot say how
deep its subagent is reports `noted`, and the account is created on the
subagent's first tool call, once the file is there.
"""

import json
import os
import re
import socket
import stat
import struct
import sys
import time
from datetime import datetime, timezone
from urllib.parse import parse_qs, unquote, urlsplit

#: Which plugin is asking. The receiver counts askers separately, so a second
#: plugin on these events is answered rather than met with the silence that
#: means "already told".
ASKER = "didbot"

#: The wire version this program writes. A receiver that does not know it says
#: so rather than going quiet, so a mismatch shows up in the log below.
#:
#: Version 2 added sign-in decisions: the answer's `pending` list and the
#: report's `seen_request_uris`. Version 3 added the report's `operator`, which
#: this hook never sends, leaving where an account goes to the daemon's
#: setting; within version 3, `spawner`.
VERSION = 3

#: What each hook event means to a receiver that thinks in contexts rather than
#: in harness events. Everything not named here is `noted`: it happened, and it
#: does not change what the context is owed.
OBSERVED = {
    "SessionStart": "began",
    "SubagentStart": "began",
    "PreToolUse": "acted",
    "Stop": "rested",
    "SubagentStop": "rested",
    "SessionEnd": "ended",
}

#: The events whose `additionalContext` the harness hands to the model while
#: it works. On any other event nothing is printed: `Stop` and `SubagentStop`
#: would re-open a finished turn, and the rest fail the harness's output
#: schema. Read from Claude Code 2.1.241's hook-output schema, not a run.
SHOWN = frozenset(
    {
        "SessionStart",
        "SubagentStart",
        "UserPromptSubmit",
        "PreToolUse",
        "PostToolUse",
        "PostToolUseFailure",
    }
)

#: Where the receiver listens, unless told otherwise. Not a secret — reaching
#: the socket is the permission, and its own directory is what enforces that.
def socket_path():
    override = os.environ.get("DIDBOT_SOCK")
    if override:
        return override
    runtime = os.environ.get("XDG_RUNTIME_DIR") or "/tmp"
    return os.path.join(runtime, "didbot-agent", "agent.sock")


#: Values safe to record: identifiers with no content in them, and the names
#: of mechanisms.
SCALARS = (
    "hook_event_name",
    "session_id",
    "agent_id",
    "agent_type",
    "tool_use_id",
    "tool_name",
    "prompt_id",
    "permission_mode",
    "source",
    "reason",
    "spawn_depth",
)

#: An `agent_id` as the harness writes it, and the only shape this hook puts
#: into a file path.
AGENT_ID_RE = re.compile(r"[A-Za-z0-9_-]{1,128}")

#: The most of a subagent's `.meta.json` this hook reads. Observed at about
#: 200 bytes.
MAX_META = 64 * 1024

#: The browser tools whose navigation this hook rewrites, by the field their
#: input names the URL in. chrome-devtools-mcp's, under any server name that
#: ends in `chrome-devtools`, a plugin's scoped name included.
BROWSER_NAVIGATION_RE = re.compile(r"^mcp__.*chrome-devtools__(new_page|navigate_page)$")

#: Every tool a browser MCP server offers, whose output may show a page, a
#: URL or a request's headers.
BROWSER_TOOL_RE = re.compile(r"^mcp__.*chrome-devtools__")

#: A server's browser sign-in page. Matched on the path alone, as an
#: authorize URL is: the daemon mints a link only for its own server's page.
SIGN_IN_PATH = "/oauth/browser"

#: What stands in for a secret this hook removed from a tool's output.
REMOVED = "[removed]"

#: A sign-in link, from its path to the end of its query. The ticket is found
#: inside the query in a second step, so a long run of links is read once
#: rather than once per link.
SIGN_IN_LINK_RE = re.compile(r"/oauth/browser\?[^\s\"'<>#]*")

#: A ticket parameter's value, inside a sign-in link's query.
TICKET_RE = re.compile(r"([?&]ticket=)[A-Za-z0-9_-]+")

#: The browser session cookie's value, in a `Cookie` or `Set-Cookie` header or
#: anywhere else a page prints it.
COOKIE_RE = re.compile(r"(__Host-didbot_browser=)[A-Za-z0-9_-]+")

#: How many `request_uri` values one report carries. The daemon refuses a
#: message over 64 KiB, and a sign-in waiting on a decision is one of a few.
MAX_SEEN = 64

#: A client's own authorize page, printed rather than opened, carrying the
#: pushed-authorization-request URI as a query parameter. Matched loosely —
#: only the path and the parameter matter, not the host — because the daemon
#: fetches the record by `request_uri` itself and does not need the URL.
AUTHORIZE_PATH = "/oauth/authorize?"

#: How much of the end of a context's transcript is read for the authorize
#: URLs it was shown. A sign-in expires minutes after its URL is printed, so
#: a URL worth reading is near the end.
TRANSCRIPT_TAIL = 4 * 1024 * 1024

#: What ends a URL in a tool's text. Text is split on these before any URL is
#: looked for, so a long run of base64 costs one pass rather than one per
#: character. A backslash is one, because a browser tool prints its result as
#: JSON, where `\n` follows a URL as two characters.
URL_BOUNDARY_RE = re.compile(r"[\s\"'<>\\]+")

#: A shell command that runs `didbot inbox`, whose output is what other
#: accounts wrote. A stranger can push a sign-in request for this account and
#: post its authorize URL, so that output is never read for sign-ins.
INBOX_COMMAND_RE = re.compile(r"(?:^|[\s;&|(`])didbot(?:-|\s+)inbox\b")

#: What ends a sentence after a URL, as in "navigated to <url>.". Taken off
#: the end of a URL before it is read.
URL_TRAILER = ".,;:!?)]"


class Untrusted(Exception):
    """The socket, or the directory holding it, could belong to someone else."""


def check_owned(path):
    """Refuse a socket this user does not hold alone.

    Whoever listens here reads every report and writes text the model reads,
    so a socket another user could have planted is refused before connecting.
    The directory must be a real directory, not a symlink, owned by this user
    and closed to everyone else. The daemon holds its own directory to the
    same rule. `FileNotFoundError` means nobody is listening.
    """
    uid = os.getuid()
    directory = os.path.dirname(path) or "."
    held = os.lstat(directory)
    if not stat.S_ISDIR(held.st_mode):
        raise Untrusted(f"{directory} is not a directory")
    if held.st_uid != uid:
        raise Untrusted(f"{directory} belongs to uid {held.st_uid}")
    if held.st_mode & 0o077:
        raise Untrusted(f"{directory} is mode {stat.S_IMODE(held.st_mode):04o}")
    listening = os.lstat(path)
    if not stat.S_ISSOCK(listening.st_mode):
        raise Untrusted(f"{path} is not a socket")
    if listening.st_uid != uid:
        raise Untrusted(f"{path} belongs to uid {listening.st_uid}")


def check_peer(sock):
    """Refuse a listener run by another user, where the system says who it is.

    Linux has `SO_PEERCRED`. Elsewhere the directory check stands alone.
    """
    option = getattr(socket, "SO_PEERCRED", None)
    if option is None:
        return
    peer = sock.getsockopt(socket.SOL_SOCKET, option, struct.calcsize("3i"))
    _, uid, _ = struct.unpack("3i", peer)
    if uid != os.getuid():
        raise Untrusted(f"the listener runs as uid {uid}")


def ask(record):
    """Hand one record to the receiver and read its answer.

    Nobody listening is the ordinary case — the plugin is installed and the
    daemon is not running — so that is silent. A listener that fails
    `check_owned` or `check_peer` is logged and sent nothing. The two seconds
    is a ceiling on how long a session will wait for a name, not an
    expectation: the answer normally involves one local request. A listener
    that takes longer is logged, because every event pays that wait.
    """
    path = socket_path()
    try:
        check_owned(path)
        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
            sock.settimeout(2.0)
            sock.connect(path)
            check_peer(sock)
            sock.sendall((json.dumps(record, sort_keys=True) + "\n").encode())
            reply = sock.makefile().readline()
    except Untrusted as error:
        complain(error)
        return {}
    except TimeoutError:
        complain(TimeoutError(f"{path} did not answer within 2 s"))
        return {}
    except OSError:
        return {}
    try:
        answer = json.loads(reply)
    except ValueError:
        return {}
    return answer if isinstance(answer, dict) else {}


def effort(payload):
    """`effort` arrives as an object on some events and is absent on others."""
    value = payload.get("effort")
    if isinstance(value, dict):
        return value.get("level")
    return value if isinstance(value, str) else None


def lineage(payload):
    """How deep the acting subagent was spawned, and the subagent that did it.

    Returns `(depth, spawner)`, each `None` when unknown. `spawner` is set
    only from depth 2 on: a subagent at depth 1 was started by the session.
    The payload is read first, under the names Claude Code documents. Claude
    Code 2.1.284 sends neither name, so the harness's own `.meta.json` for
    the subagent is read instead.
    """
    agent = payload.get("agent_id")
    if not isinstance(agent, str) or not AGENT_ID_RE.fullmatch(agent):
        return None, None
    depth = payload.get("spawn_depth")
    parent = payload.get("parent_agent_id")
    if not isinstance(depth, int):
        meta = harness_meta(payload.get("transcript_path"), agent)
        depth = meta.get("spawnDepth")
        parent = meta.get("parentAgentId")
    if isinstance(depth, bool) or not isinstance(depth, int):
        return None, None
    if depth > 1 and isinstance(parent, str) and parent and parent != agent:
        return depth, parent
    return depth, None


def harness_meta(transcript, agent):
    """The `.meta.json` Claude Code writes for subagent `agent`, or `{}`.

    It sits at `<transcript without .jsonl>/subagents/agent-<id>.meta.json`,
    where `transcript` is the session's own, on a subagent's events too.
    """
    if not isinstance(transcript, str) or not transcript.endswith(".jsonl"):
        return {}
    path = os.path.join(
        transcript[: -len(".jsonl")], "subagents", f"agent-{agent}.meta.json"
    )
    try:
        with open(path, "rb") as handle:
            meta = json.loads(handle.read(MAX_META))
    except (OSError, ValueError):
        return {}
    return meta if isinstance(meta, dict) else {}


def strings_in(value):
    """Every string in a tool's input or output, however deeply it is nested.

    Walked with a stack rather than by recursion, so output nested past
    Python's recursion limit is read like any other.
    """
    stack = [value]
    while stack:
        value = stack.pop()
        if isinstance(value, str):
            yield value
        elif isinstance(value, dict):
            stack.extend(reversed(list(value.values())))
        elif isinstance(value, list):
            stack.extend(reversed(value))


def seen_requests(*parts):
    """The `request_uri` values named in a tool call's input or output, each
    with the `client_id` its URL named.

    A client waiting to be signed in prints its authorize URL rather than
    opening it, and that URL carries the pushed-authorization-request's
    `request_uri` as a query parameter. The daemon shows a context only the
    sign-ins this names. A model that opens an authorize URL itself has seen
    it, so `tool_input` counts as well as `tool_response`.

    Each arrives as a string on some tools and as a dict or list on others,
    so every string inside is searched as the text it is, with no JSON
    escapes in it. The `client_id` is None where the URL named none, one
    more than once, or a different one from another mention.

    At most `MAX_SEEN` are returned, the last ones named, so a log full of
    authorize URLs cannot push the report past the daemon's line limit.
    """
    found = {}
    for text in strings_in(list(parts)):
        if AUTHORIZE_PATH not in text:
            continue
        for word in URL_BOUNDARY_RE.split(text):
            if AUTHORIZE_PATH not in word:
                continue
            query = parse_qs(urlsplit(word.rstrip(URL_TRAILER)).query)
            client_ids = query.get("client_id", [])
            client_id = client_ids[0] if len(client_ids) == 1 else None
            for value in query.get("request_uri", []):
                named = found.pop(value, client_id)
                # Moved to the end, so the last mention decides the order.
                found[value] = client_id if named == client_id else None
    return dict(list(found.items())[-MAX_SEEN:])


def runs_inbox(tool_input):
    """Whether a tool call is a shell command that runs `didbot inbox`."""
    command = tool_input.get("command") if isinstance(tool_input, dict) else None
    return isinstance(command, str) and bool(INBOX_COMMAND_RE.search(command))


def transcript_entries(payload):
    """The entries near the end of the acting context's transcript that name
    an authorize URL: a subagent's own transcript, or else the session's.

    The first line read may be cut off, and is skipped like any line that is
    not JSON.
    """
    path = payload.get("agent_transcript_path") or payload.get("transcript_path")
    if not isinstance(path, str) or not path.endswith(".jsonl"):
        return []
    try:
        with open(path, "rb") as handle:
            handle.seek(0, os.SEEK_END)
            handle.seek(max(0, handle.tell() - TRANSCRIPT_TAIL))
            tail = handle.read()
    except OSError:
        return []
    entries = []
    marker = AUTHORIZE_PATH.encode()
    for line in tail.split(b"\n"):
        if marker not in line:
            continue
        try:
            entries.append(json.loads(line))
        except ValueError:
            continue
    return entries


def shown_requests(payload):
    """`seen_requests` over the acting context's transcript and this tool
    call: every authorize URL the context has been shown lately."""
    return seen_requests(
        transcript_entries(payload),
        payload.get("tool_input"),
        payload.get("tool_response"),
    )


def is_sign_in_page(url):
    """Whether `url` is a browser sign-in page, with nothing after its path."""
    if not isinstance(url, str):
        return False
    try:
        parts = urlsplit(url)
    except ValueError:
        return False
    return (
        parts.scheme == "https"
        and bool(parts.netloc)
        and parts.path == SIGN_IN_PATH
        and not parts.query
        and not parts.fragment
    )


def navigation(payload):
    """The sign-in page a browser tool in this call is about to open, if any."""
    if not BROWSER_NAVIGATION_RE.match(str(payload.get("tool_name") or "")):
        return None
    tool_input = payload.get("tool_input")
    if not isinstance(tool_input, dict):
        return None
    url = tool_input.get("url")
    return url if is_sign_in_page(url) else None


def signed_in_input(tool_input, page, link):
    """The tool's input with its URL replaced by `link`, or None.

    Only a link on the very page the tool was opening is written in: the
    daemon's answer decides nothing about where the browser goes.
    """
    if not isinstance(link, str) or not link.startswith(page + "?"):
        return None
    try:
        query = parse_qs(urlsplit(link).query)
    except ValueError:
        return None
    if not query.get("ticket"):
        return None
    rewritten = dict(tool_input)
    rewritten["url"] = link
    return rewritten


def failed_sign_in(payload):
    """The sign-in page whose link a failed browser tool held, or None.

    The link is looked for in the tool's input and in its error, which names
    the URL a navigation failed on.
    """
    if not BROWSER_TOOL_RE.match(str(payload.get("tool_name") or "")):
        return None
    for text in strings_in([payload.get("tool_input"), payload.get("error")]):
        if SIGN_IN_PATH not in text:
            continue
        for word in URL_BOUNDARY_RE.split(text):
            try:
                parts = urlsplit(word.rstrip(URL_TRAILER))
                ticket = parse_qs(parts.query).get("ticket")
            except ValueError:
                continue
            page = f"{parts.scheme}://{parts.netloc}{parts.path}"
            if ticket and is_sign_in_page(page):
                return page
    return None


def tickets_in(value):
    """The ticket of each sign-in link in a tool's input."""
    found = []
    for text in strings_in(value):
        for match in SIGN_IN_LINK_RE.finditer(text):
            for ticket in parse_qs(urlsplit(match.group(0)).query).get("ticket", []):
                if ticket and ticket not in found:
                    found.append(ticket)
    return found


def scrub(value, tickets):
    """`value` with every ticket and browser session cookie value removed.

    The shape is kept, key for key and item for item, because the harness
    takes a replacement output only in the shape the tool gave.
    """
    if isinstance(value, str):
        value = SIGN_IN_LINK_RE.sub(
            lambda link: TICKET_RE.sub(
                lambda ticket: ticket.group(1) + REMOVED, link.group(0)
            ),
            value,
        )
        value = COOKIE_RE.sub(lambda match: match.group(1) + REMOVED, value)
        for ticket in tickets:
            value = value.replace(ticket, REMOVED)
        return value
    if isinstance(value, list):
        return [scrub(item, tickets) for item in value]
    if isinstance(value, dict):
        return {key: scrub(item, tickets) for key, item in value.items()}
    return value


def utc_clock(rfc3339):
    """`HH:MM UTC` from an RFC 3339 timestamp, or the timestamp itself if it
    will not parse — a hook must say something rather than raise."""
    if not isinstance(rfc3339, str):
        return "an unknown time"
    try:
        moment = datetime.fromisoformat(rfc3339.replace("Z", "+00:00"))
    except ValueError:
        return rfc3339
    return moment.astimezone(timezone.utc).strftime("%H:%M UTC")


#: The origin of the atproto profile's local client, `http://localhost`, as
#: the server reports it. didbot's `didbot_policy::is_local_origin` is the
#: same test.
LOCAL_ORIGIN = "http://localhost"

#: What a URL parser strips from either end of a URL: C0 controls and space.
URL_TRIM = "".join(map(chr, range(0x21)))

#: One part of an IPv4 address as a URL parser reads it: hex, octal or
#: decimal.
IPV4_PART_RE = re.compile(r"0[xX][0-9a-fA-F]*|0[0-7]*|[1-9][0-9]*")

#: `[::1]`, packed, which every spelling of that address packs to.
IPV6_LOOPBACK = socket.inet_pton(socket.AF_INET6, "::1")


def http_url(url):
    """`(host, port, path, query)` of an `http` URL as didbot's parser reads
    it, or None.

    Hand-rolled because `urlsplit` reads `http:localhost`, `http:///localhost`
    and `http://localhost\\` differently from a WHATWG parser such as Rust's
    `url`. `host` is as `canonical_host` gives it. `port` is None when there
    is none and when it is 80, which the parser drops as the default.
    """
    url = url.strip(URL_TRIM).replace("\t", "").replace("\n", "").replace("\r", "")
    scheme, colon, rest = url.partition(":")
    if not colon or scheme.lower() != "http":
        return None
    rest, _, query = rest.partition("#")[0].lstrip("/\\").partition("?")
    authority, slash, path = rest.replace("\\", "/").partition("/")
    authority = authority.rpartition("@")[2]
    if authority.startswith("["):
        host, bracket, port = authority.partition("]")
        host += bracket
        if port and not port.startswith(":"):
            return None
        port = port[1:]
    else:
        host, _, port = authority.partition(":")
    if port and not (port.isascii() and port.isdigit() and int(port) <= 65535):
        return None
    port = int(port) if port and int(port) != 80 else None
    return canonical_host(host), port, slash + path, query


def canonical_host(host):
    """`localhost`, `127.0.0.1` or `[::1]`, for any spelling a URL parser
    reads as one of them, and None for any other host."""
    if host.startswith("[") and host.endswith("]"):
        try:
            packed = socket.inet_pton(socket.AF_INET6, host[1:-1])
        except OSError:
            return None
        return "[::1]" if packed == IPV6_LOOPBACK else None
    host = unquote(host)
    if not host.isascii():
        # Full-width digits and ideographic dots map to ASCII, as they do
        # in a browser.
        try:
            host = host.encode("idna").decode()
        except UnicodeError:
            return None
    host = host.lower()
    if host == "localhost":
        return host
    parts = host.split(".")
    if len(parts) > 1 and parts[-1] == "":
        parts.pop()
    if not 1 <= len(parts) <= 4 or not all(map(IPV4_PART_RE.fullmatch, parts)):
        return None
    numbers = []
    for part in parts:
        if part[:2] == "0x":
            numbers.append(int(part[2:] or "0", 16))
        else:
            numbers.append(int(part, 8 if part[0] == "0" else 10))
    if any(number > 255 for number in numbers[:-1]):
        return None
    address = numbers[-1]
    for index, number in enumerate(numbers[:-1]):
        address += number << (8 * (3 - index))
    return "127.0.0.1" if address == 0x7F000001 else None


def is_root(path):
    """Whether `path` is `/` once its dot segments are resolved."""
    segments = []
    for segment in path.split("/")[1:]:
        dots = segment.lower().replace("%2e", ".")
        if dots == "..":
            if segments:
                segments.pop()
        elif dots != ".":
            segments.append(segment)
    return segments in ([], [""])


def local_redirect_ports(client_id):
    """The redirect ports a local `client_id` names, sorted, or None if it is
    not a local `client_id` didbot would accept.

    Read as didbot's `parse_loopback_client_id` reads it: `http://localhost`
    with no port and a path of `/`, and each `redirect_uri` an `http` URL on
    `127.0.0.1` or `[::1]`. A redirect with no port names none.
    """
    local = http_url(client_id)
    if local is None:
        return None
    host, port, path, query = local
    if host != "localhost" or port is not None or not is_root(path):
        return None
    ports = set()
    for redirect_uri in parse_qs(query).get("redirect_uri", []):
        redirect = http_url(redirect_uri)
        if redirect is None or redirect[0] not in ("127.0.0.1", "[::1]"):
            return None
        if redirect[1] is not None:
            ports.add(redirect[1])
    return sorted(ports)


def redirect_ports(pending, payload):
    """The redirect ports of each local sign-in in `pending`, by token, where
    the context was shown its authorize URL (`shown_requests`).

    Each local URL is looked up by its `request_uri` alone. Only a decision
    already in `pending` is labelled, so the lookup shows nothing new.
    """
    waiting = {
        decision.get("token")
        for decision in pending
        if decision.get("token") and decision.get("clientOrigin") == LOCAL_ORIGIN
    }
    ports = {}
    if not waiting:
        return ports
    asked = 0
    for request_uri, client_id in reversed(shown_requests(payload).items()):
        if not waiting or asked == MAX_RENDERED:
            break
        named = local_redirect_ports(client_id) if client_id else None
        if named is None:
            continue
        asked += 1
        answer = ask({"asks": "show", "version": VERSION, "url": request_uri})
        for decision in answer.get("pending") or []:
            token = decision.get("token") if isinstance(decision, dict) else None
            if token in waiting and decision.get("clientOrigin") == LOCAL_ORIGIN:
                ports[token] = named
                waiting.discard(token)
    return ports


#: How many pending sign-ins get their own block before the rest are just
#: counted. Plain and factual either way — a context with a dozen of these
#: waiting does not need a dozen renderings of the same three lines.
MAX_RENDERED = 5


def render_decision(decision, ports):
    """One block of `additionalContext` for one pending sign-in.

    Plain and factual: what asked, what for, what policy did about it, and
    how to answer. No urging, no reassurance, no claim that the request is
    legitimate — that is not this hook's to say. `ports` is `redirect_ports`.
    """
    lines = ["An app asked to sign in as you."]

    origin = decision.get("clientOrigin") or "an unknown client"
    token = decision.get("token")
    if origin == LOCAL_ORIGIN:
        # Any program on this machine can send a local client's id, so the
        # server never says it saw one before. Its redirect port is what tells
        # one local app from another.
        named = ports.get(token)
        if not token:
            lines.append("  client: unverified local app")
        elif named is None:
            lines.append("  client: unverified local app, redirect port unknown")
        elif not named:
            lines.append("  client: unverified local app, no redirect port named")
        else:
            plural = "s" if len(named) > 1 else ""
            listed = ", ".join(map(str, named))
            lines.append(f"  client: unverified local app, redirect port{plural} {listed}")
    else:
        seen = "first time for this account" if decision.get("firstTime") else "seen before"
        lines.append(f"  client: {origin} ({seen})")

    requested = ", ".join(decision.get("requested") or [])
    lines.append(f"  asked for: {requested}")

    verdict = decision.get("verdict")
    rule = decision.get("rule")
    if verdict == "allow":
        lines.append("  policy: allowed in full")
    elif verdict == "narrow":
        granted = ", ".join(decision.get("granted") or [])
        cut = ", ".join(decision.get("cut") or [])
        by_rule = f', by rule "{rule}"' if rule else ""
        lines.append(f"  policy: narrowed to {granted}; cut: {cut}{by_rule}")
    elif verdict == "deny":
        # `reason` is newer than `rule`: a daemon that has it puts the human
        # explanation first and the rule id alongside, in parentheses; a
        # daemon that predates it still puts the rule alone where it was.
        reason = decision.get("reason")
        if reason:
            by_rule = f' (rule "{rule}")' if rule else ""
            lines.append(f"  policy: refused: {reason}{by_rule}")
        elif rule:
            lines.append(f"  policy: refused: {rule}")
        else:
            lines.append("  policy: refused")
    else:
        # A verdict this hook does not know is rendered rather than dropped —
        # a future word from a newer daemon should still say something.
        lines.append(f"  policy: {verdict}")

    if token:
        lines.append(
            f"Approve with: didbot oauth approve {token}        "
            f"Decline with: didbot oauth decline {token}"
        )
        lines.append(
            "Nothing happens until you choose; the request expires at "
            f"{utc_clock(decision.get('expiresAt'))}."
        )
    else:
        # A deny carries no token: there is nothing to approve or decline,
        # and the app was already answered.
        lines.append("The app was already told no; there is nothing to approve.")

    return "\n".join(lines)


def render_pending(pending, ports):
    """`additionalContext` for every sign-in the daemon is holding, capped."""
    shown = pending[:MAX_RENDERED]
    blocks = [render_decision(decision, ports) for decision in shown]
    rest = len(pending) - len(shown)
    if rest > 0:
        blocks.append(
            f"{rest} more pending sign-in{'s' if rest != 1 else ''}, not shown here."
        )
    return "\n\n".join(blocks)


def main():
    try:
        payload = json.load(sys.stdin)
    except Exception:
        payload = {}
    if not isinstance(payload, dict):
        payload = {}

    event = payload.get("hook_event_name")
    observed = OBSERVED.get(event, "noted")

    # Only a report that begins or acts can create an account, so only those
    # look for the subagent that started this one. A `SubagentStart` that
    # cannot say yet would create the account beneath the session instead.
    depth, spawner = None, None
    if observed in ("began", "acted"):
        depth, spawner = lineage(payload)
    if event == "SubagentStart" and depth is None:
        observed = "noted"

    # The receiver's vocabulary first, then everything observed alongside it.
    # A receiver reads the fields it knows and ignores the rest, which is what
    # lets one wire carry both a question and a measurement.
    record = {
        "asks": "report",
        "asker": ASKER,
        "version": VERSION,
        "observed": observed,
        "session": payload.get("session_id") or "",
        "context": payload.get("agent_id"),
        "kind": payload.get("agent_type"),
        "call": payload.get("tool_use_id"),
        "spawner": spawner,
        "epoch": round(time.time(), 3),
        "keys": sorted(payload),
        "effort": effort(payload),
        "cwd": payload.get("cwd"),
    }
    # An identifier the harness did not send is left off rather than sent as
    # null, which is how the daemon's own fixtures write it.
    for field in ("context", "kind", "call", "spawner"):
        if record[field] is None:
            del record[field]
    for name in SCALARS:
        value = payload.get(name)
        if isinstance(value, (str, int, bool)):
            record[name] = value
    if depth is not None:
        record.setdefault("spawn_depth", depth)

    # A subagent's own transcript is written at a path derived from its id,
    # and an internal fork of the harness is given an id and no transcript.
    # Whether the promised file exists is what tells the two apart.
    promised = payload.get("agent_transcript_path")
    if isinstance(promised, str) and promised:
        record["agent_transcript_exists"] = os.path.exists(promised)

    # The sign-in page a browser tool is about to open. The daemon answers
    # with a link only when it is its own server's page.
    opening = navigation(payload) if event == "PreToolUse" else None
    if opening:
        record["browser_sign_in"] = opening

    # A failed browser tool's error reaches the model as it is. Reported as
    # opening the page again, it has the daemon mint a link that ends the
    # ticket the error may show. The new link is never written anywhere.
    failed = failed_sign_in(payload) if event == "PostToolUseFailure" else None
    if failed:
        record["observed"] = "acted"
        record["browser_sign_in"] = failed

    # What the daemon will show: the sign-ins whose authorize URL this tool
    # call was given or printed. It fetches any its own poll has not found.
    requests = {}
    if event == "PostToolUse":
        tool_input = payload.get("tool_input")
        tool_response = payload.get("tool_response")
        if runs_inbox(tool_input):
            tool_response = None
        requests = seen_requests(tool_input, tool_response)
        if requests:
            record["seen_request_uris"] = list(requests)

    answer = ask(record)

    # A receiver that could not do what was asked says why. Swallowing that
    # leaves an agent with no name and no way to find out that it should have
    # had one, so the model is shown it as well as the log. It is the
    # receiver's sentence, naming an operator's fix where there is one, kept
    # to one line.
    pieces = []
    trouble = answer.get("trouble")
    if trouble:
        complain(RuntimeError(trouble))
        pieces.append("didbot: " + " ".join(str(trouble).split()))

    if answer.get("identity"):
        # Both halves, because they are one thing: who the agent is, and what
        # happens here when something asks to sign in as it.
        identity = (
            "Your agent identity on this machine is "
            f"{answer['identity']}. It signs in to atproto apps (the "
            "Atmosphere) that use OAuth.\n"
            "When an app asks to sign in as you, the request appears here "
            "with what it asked for and what policy allows; approve it with "
            "`didbot oauth approve <token>` or decline with `didbot oauth decline <token>`."
        )
        # The page alone, never a link: a link carries a ticket. The examples
        # are web apps, so they are named only where a browser can sign in.
        page = answer.get("browser_sign_in")
        if is_sign_in_page(page):
            identity += (
                "\nA web app, such as pdsls.dev or Leaflet, finishes its sign-in "
                f"only in a browser signed in as you. Open {page} with your "
                "browser tool to sign it in."
            )
        pieces.append(identity)

    pending = answer.get("pending")
    if isinstance(pending, list) and pending:
        pieces.append(
            render_pending(pending, redirect_ports(pending, payload))
        )

    output = {}
    if pieces and event in SHOWN:
        output["additionalContext"] = "\n\n".join(pieces)

    # The link goes into the tool's input and nowhere else.
    if opening:
        rewritten = signed_in_input(
            payload["tool_input"], opening, answer.get("browser_sign_in")
        )
        if rewritten is not None:
            output["updatedInput"] = rewritten

    # A browser tool's output can show the link it was given, a page, or a
    # request's headers. Neither the ticket nor the cookie reaches the model.
    if event == "PostToolUse" and BROWSER_TOOL_RE.match(
        str(payload.get("tool_name") or "")
    ):
        response = payload.get("tool_response")
        scrubbed = scrub(response, tickets_in(payload.get("tool_input")))
        if scrubbed != response:
            output["updatedToolOutput"] = scrubbed

    if output:
        output["hookEventName"] = event
        print(json.dumps({"hookSpecificOutput": output}))


#: How large the error log grows before it is moved aside to
#: `hook-errors.log.1`, replacing the one before. The two files together stay
#: under twice this.
MAX_LOG_BYTES = 1024 * 1024


def complain(error):
    """Write down why this hook could not do its job.

    A hook must not break a session, so it cannot raise. But a hook that
    swallows its own failure is indistinguishable from a harness that never
    called it, which is a much worse place to debug from than a file with a
    traceback in it.
    """
    try:
        # Imported here: on Python 3.14 it costs a third of the hook's start-up,
        # and only a failing hook needs it.
        import traceback

        base = os.environ.get("CLAUDE_PLUGIN_DATA") or os.environ.get("TMPDIR") or "/tmp"
        os.makedirs(base, exist_ok=True)
        log = os.path.join(base, "hook-errors.log")
        try:
            if os.path.getsize(log) >= MAX_LOG_BYTES:
                os.replace(log, log + ".1")
        except OSError:
            pass
        with open(log, "a") as handle:
            handle.write(
                time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                + " " + repr(error) + "\n"
                + "".join(traceback.format_exception(error))
            )
    except Exception:
        pass


if __name__ == "__main__":
    try:
        main()
    except Exception as error:  # noqa: BLE001 - a hook may not raise
        complain(error)