Official Cheater CATF 2026 DFIR Writeup

The script takes PID=path arguments and writes its JSON report to standard output. It uses field names as anchors, then validates the surrounding formats, none of the constants contains the recovered prompt or response. The same tested file is available as compare_renderers.py.

python3 compare_renderers.py "1472=renderer-1472/pid.1472.dmp" "6612=renderer-6612/pid.6612.dmp"

#!/usr/bin/env python3
"""Compare raw Chrome renderer dumps without searching for the prompt or answer."""

import argparse
import json
import mmap
import re
import struct
from html.parser import HTMLParser
from pathlib import Path
from urllib.parse import parse_qs, urlsplit
from uuid import UUID


COMPOSER_KEY = b"oai/apps/lightweight-web/composerDraft/v1"
TERMINAL = b'data-conversation-control="terminal-received"'
RESUME = b'data-conversation-control="resume-token"'
STARTED = b'data-conversation-control="assistant-content-started"'
STREAM = b'data-assistant-stream-block=""'
NOAUTH_ROUTE = b"chatgpt.com/unauth-mweb/conversation?lightweight_authenticated=0"
SUBMISSION_OWNER = b"conversationRetryOwner="

UUID_TEXT = rb"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
PENDING_FOR = re.compile(
    rb'for="assistant-pending-(' + UUID_TEXT + rb')-(committed-tail|pending)"'
)
TURN_STATE = re.compile(
    rb'\x22\x12conversationTurnId\x22\x24(' + UUID_TEXT +
    rb')\x22\x04kind\x22\x1cweb-mobile-conversation-turn'
    rb'\x22\x09streaming([FT])'
)


class FirstStartTag(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.name = None
        self.attrs = {}

    def handle_starttag(self, tag, attrs):
        if self.name is None:
            self.name = tag
            self.attrs = dict(attrs)


def offsets(data, needle):
    at = 0
    while (at := data.find(needle, at)) != -1:
        yield at
        at += 1


def canonical_uuid(value):
    try:
        parsed = str(UUID(value))
    except (ValueError, AttributeError):
        return None
    return parsed if parsed == value.lower() else None


def parse_start_tag(data, anchor, name, look_back=512, look_ahead=2048):
    start = data.rfind(b"<" + name.encode(), max(0, anchor - look_back), anchor + 1)
    end = data.find(b">", anchor, min(len(data), anchor + look_ahead))
    if start == -1 or end == -1:
        return None

    parser = FirstStartTag()
    try:
        parser.feed(data[start : end + 1].decode("latin-1"))
    except Exception:
        return None
    if parser.name != name:
        return None
    return start, end + 1, parser.attrs


def decode_storage_array(data, start):
    if start < 0 or start + 9 > len(data) or start % 8:
        return None
    num_bytes, num_elements = struct.unpack_from("<II", data, start)
    if num_elements < 1 or num_bytes != 8 + num_elements:
        return None
    end = start + num_bytes
    if end > len(data):
        return None

    payload = data[start + 8 : end]
    marker, raw = payload[0], payload[1:]
    try:
        if marker == 1:
            return raw.decode("latin-1"), start + 9
        if marker == 0 and len(raw) % 2 == 0:
            return raw.decode("utf-16le"), start + 9
    except UnicodeDecodeError:
        pass
    return None


def composer_pairs(data):
    hits = list(offsets(data, COMPOSER_KEY))
    pairs = []

    for hit in hits:
        key_array = hit - 9  # 8-byte array header + 1-byte encoding marker
        decoded_key = decode_storage_array(data, key_array)
        if not decoded_key or decoded_key[0].encode("latin-1") != COMPOSER_KEY:
            continue

        first_slot = max(8, key_array - 0x80)
        first_slot += (-first_slot) % 8
        for key_slot in range(first_slot, key_array, 8):
            key_relative = struct.unpack_from("<Q", data, key_slot)[0]
            if key_slot + key_relative != key_array:
                continue

            record = key_slot - 8
            num_bytes, version = struct.unpack_from("<II", data, record)
            if version != 0 or num_bytes < 0x18 or num_bytes % 8:
                continue
            if record + num_bytes != key_array:
                continue

            value_slot = key_slot + 8
            value_relative = struct.unpack_from("<Q", data, value_slot)[0]
            value_array = value_slot + value_relative
            decoded_value = decode_storage_array(data, value_array)
            if not decoded_value:
                continue

            value, value_offset = decoded_value
            pairs.append(
                {
                    "record_offset": hex(record),
                    "key_offset": hex(hit),
                    "value_offset": hex(value_offset),
                    "value_characters": len(value),
                    "value": value,
                }
            )
            break

    return hits, pairs


def printable_allocation(data, anchor, maximum=0x10000):
    """Bound one printable ASCII run around an anchor in raw memory."""
    start = anchor
    floor = max(0, anchor - maximum)
    while start > floor and 0x20 <= data[start - 1] <= 0x7E:
        start -= 1

    end = anchor
    ceiling = min(len(data), anchor + maximum)
    while end < ceiling and 0x20 <= data[end] <= 0x7E:
        end += 1
    return start, end


def submission_records(data):
    """Decode form fragments containing complete owner and message-ID fields."""
    grouped = {}

    for anchor in offsets(data, SUBMISSION_OWNER):
        start, end = printable_allocation(data, anchor)
        try:
            fields = parse_qs(
                data[start:end].decode("ascii"),
                keep_blank_values=True,
                strict_parsing=False,
            )
            owner_text = fields["conversationRetryOwner"][-1]
            assistant_text = fields["assistantMessageId"][-1]
            user_text = fields["userMessageId"][-1]
            owner = json.loads(owner_text)
        except (UnicodeDecodeError, KeyError, IndexError, json.JSONDecodeError):
            continue

        if not assistant_text.startswith("pending-"):
            continue
        pending_id = canonical_uuid(assistant_text.removeprefix("pending-"))
        user_message_id = canonical_uuid(user_text)
        if pending_id is None or user_message_id is None or not isinstance(owner, dict):
            continue
        if owner.get("mode") != "anonymous" or owner.get("sessionEpoch") is not None:
            continue

        key = (pending_id, user_message_id, json.dumps(owner, sort_keys=True))
        record = grouped.setdefault(
            key,
            {
                "allocation_offset": hex(start),
                "allocation_end": hex(end),
                "owner_field_offset": hex(anchor),
                "owner": owner,
                "pending_assistant_id": pending_id,
                "user_message_id": user_message_id,
                "copy_offsets": [],
            },
        )
        record["copy_offsets"].append(hex(anchor))

    return list(grouped.values())


def history_url_arrays(data, conversation_id):
    """Validate exact /uc/ URLs as length-delimited Mojo arrays."""
    if conversation_id is None:
        return []

    encoded = f"https://chatgpt.com/uc/{conversation_id}".encode("ascii")
    records = []
    for url_offset in offsets(data, encoded):
        header = url_offset - 8
        if header < 0 or header % 8:
            continue
        num_bytes, num_elements = struct.unpack_from("<II", data, header)
        if num_bytes != 8 + len(encoded) or num_elements != len(encoded):
            continue
        records.append(
            {
                "array_header_offset": hex(header),
                "url_offset": hex(url_offset),
                "conversation_id_offset": hex(url_offset + encoded.index(conversation_id.encode("ascii"))),
                "num_bytes": num_bytes,
                "num_elements": num_elements,
                "url": encoded.decode("ascii"),
            }
        )
    return records


def conversation_turn_states(data):
    """Parse serialized navigation-state records keyed by user message UUID."""
    grouped = {}
    for match in TURN_STATE.finditer(data):
        conversation_turn_id = canonical_uuid(match.group(1).decode("ascii"))
        if conversation_turn_id is None:
            continue
        state = "true" if match.group(2) == b"T" else "false"
        record = grouped.setdefault(
            conversation_turn_id,
            {
                "conversation_turn_id": conversation_turn_id,
                "kind": "web-mobile-conversation-turn",
                "streaming_true_offsets": [],
                "streaming_false_offsets": [],
            },
        )
        record[f"streaming_{state}_offsets"].append(hex(match.start()))
    return list(grouped.values())


def storage_page_links(url_arrays, pairs):
    """Keep only pages containing both the pivot URL and a valid composer pair."""
    links = []
    for url_record in url_arrays:
        page = int(url_record["array_header_offset"], 16) & ~0xFFF
        page_pairs = [
            pair
            for pair in pairs
            if page <= int(pair["record_offset"], 16) < page + 0x1000
        ]
        if page_pairs:
            links.append(
                {
                    "page": hex(page),
                    "url_array_header": url_record["array_header_offset"],
                    "composer_records": page_pairs,
                }
            )
    return links


def controls(data, needle, kind, required):
    parsed = []
    for anchor in offsets(data, needle):
        tag = parse_start_tag(data, anchor, "span")
        if not tag:
            continue
        start, end, attrs = tag
        if attrs.get("data-conversation-control") != kind:
            continue

        values = {}
        valid = True
        for attr in required:
            value = canonical_uuid(attrs.get(attr))
            if value is None:
                valid = False
                break
            values[attr.removeprefix("data-").replace("-", "_")] = value
        if valid:
            values.update({"anchor": anchor, "tag_start": start, "tag_end": end})
            parsed.append(values)
    return parsed


def stream_blocks(data):
    blocks = []
    for attr_offset in offsets(data, STREAM):
        tag = parse_start_tag(data, attr_offset, "p", 256, 1024)
        if not tag:
            continue
        start, open_end, attrs = tag
        try:
            index = int(attrs["data-assistant-stream-block-index"])
        except (KeyError, ValueError):
            continue

        close = data.find(b"</p>", open_end, min(len(data), open_end + 4096))
        if close == -1:
            continue
        body = data[open_end:close].split(b"<", 1)[0]
        try:
            text = body.decode("utf-8")
        except UnicodeDecodeError:
            continue

        # The allocation boundary can cut off the beginning of <template>, so
        # use the last complete `for=` attribute before the paragraph.
        context_start = max(0, start - 2048)
        matches = list(PENDING_FOR.finditer(data[context_start:start]))
        if not matches:
            continue
        match = matches[-1]

        pending_id = canonical_uuid(match.group(1).decode("ascii"))
        state = match.group(2).decode("ascii")
        blocks.append(
            {
                "anchor": attr_offset,
                "end": close + 4,
                "block_index": index,
                "text": text,
                "pending_id": pending_id,
                "committed": state == "committed-tail",
            }
        )
    return blocks


def pending_target_after(data, control):
    end = min(len(data), control["tag_end"] + 4096)
    match = PENDING_FOR.search(data[control["tag_end"] : end])
    if not match or match.group(2) != b"pending":
        return None
    return canonical_uuid(match.group(1).decode("ascii"))


def linked_turns(data, terminals, blocks, resumes, started):
    turns = []
    for block in (item for item in blocks if item["committed"]):
        nearby_resumes = [r for r in resumes if 0 < r["anchor"] - block["end"] < 0x8000]
        for resume in nearby_resumes:
            nearby_started = [
                s for s in started if 0 < s["anchor"] - resume["anchor"] < 0x8000
            ]
            for began in nearby_started:
                if pending_target_after(data, began) != block["pending_id"]:
                    continue
                for terminal in terminals:
                    if not (0 < block["anchor"] - terminal["anchor"] < 0x10000):
                        continue
                    if not (
                        terminal["conversation_id"] == resume["conversation_id"]
                        and terminal["operation_id"] == resume["operation_id"]
                        and terminal["operation_id"] == began["operation_id"]
                        and terminal["message_id"] == began["message_id"]
                    ):
                        continue
                    turns.append(
                        {
                            "terminal_control": hex(terminal["anchor"]),
                            "committed_stream_block": hex(block["anchor"]),
                            "conversation_id": terminal["conversation_id"],
                            "pending_assistant_id": block["pending_id"],
                            "final_assistant_id": terminal["message_id"],
                            "operation_id": terminal["operation_id"],
                            "response_text": block["text"],
                        }
                    )
    return turns


def attach_submissions(turns, submissions):
    for turn in turns:
        matches = [
            record
            for record in submissions
            if record["pending_assistant_id"] == turn["pending_assistant_id"]
        ]
        if len(matches) == 1:
            turn["submission"] = matches[0]


def attach_turn_states(submissions, states):
    by_id = {state["conversation_turn_id"]: state for state in states}
    for submission in submissions:
        state = by_id.get(submission["user_message_id"])
        if state is not None:
            submission["navigation_state"] = state


def inspect_dump(pid, path, conversation_id=None):
    with path.open("rb") as stream, mmap.mmap(
        stream.fileno(), 0, access=mmap.ACCESS_READ
    ) as data:
        key_hits, pairs = composer_pairs(data)
        terminals = controls(
            data,
            TERMINAL,
            "terminal-received",
            ("data-conversation-id", "data-message-id", "data-operation-id"),
        )
        resumes = controls(
            data, RESUME, "resume-token", ("data-conversation-id", "data-operation-id")
        )
        started = controls(
            data, STARTED, "assistant-content-started", ("data-message-id", "data-operation-id")
        )
        blocks = stream_blocks(data)
        turns = linked_turns(data, terminals, blocks, resumes, started)
        submissions = submission_records(data)
        turn_states = conversation_turn_states(data)
        attach_turn_states(submissions, turn_states)
        attach_submissions(turns, submissions)
        url_arrays = history_url_arrays(data, conversation_id)
        page_links = storage_page_links(url_arrays, pairs)

        route_offsets = []
        for at in offsets(data, NOAUTH_ROUTE):
            route = urlsplit("https://" + NOAUTH_ROUTE.decode("ascii"))
            if (
                route.hostname == "chatgpt.com"
                and route.path == "/unauth-mweb/conversation"
                and parse_qs(route.query) == {"lightweight_authenticated": ["0"]}
            ):
                route_offsets.append(hex(at))

    matching_turns = [
        turn
        for turn in turns
        if conversation_id is None or turn["conversation_id"] == conversation_id
    ]

    return {
        "pid": pid,
        "history_pivot": {
            "conversation_id": conversation_id,
            "validated_url_arrays": url_arrays,
            "storage_page_links": page_links,
            "matching_completed_turn_count": len(matching_turns),
        },
        "composer_key_candidates": {
            "count": len(key_hits),
            "offsets": [hex(item) for item in key_hits],
        },
        "valid_mojo_composer_pairs": pairs,
        "terminal_controls": [
            {
                "anchor_offset": hex(item["anchor"]),
                "conversation_id": item["conversation_id"],
                "message_id": item["message_id"],
                "operation_id": item["operation_id"],
            }
            for item in terminals
        ],
        "assistant_stream_blocks": [
            {
                "anchor_offset": hex(item["anchor"]),
                "text_characters": len(item["text"]),
                "committed": item["committed"],
                "pending_assistant_id": item["pending_id"],
                "text": item["text"],
            }
            for item in blocks
        ],
        "submission_records": submissions,
        "conversation_turn_states": turn_states,
        "valid_noauth_route_literals": route_offsets,
        "linked_completed_turns": turns,
        "complete_turn": bool(matching_turns),
    }


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--conversation-id",
        help="conversation UUID recovered from Chrome History",
    )
    parser.add_argument("dump", nargs="+", metavar="PID=PATH")
    args = parser.parse_args()

    conversation_id = None
    if args.conversation_id is not None:
        conversation_id = canonical_uuid(args.conversation_id)
        if conversation_id is None:
            parser.error("--conversation-id must be a canonical UUID")

    renderers = []
    for item in args.dump:
        pid_text, separator, path_text = item.partition("=")
        if not separator:
            parser.error(f"expected PID=PATH, got {item!r}")
        renderers.append(
            inspect_dump(int(pid_text), Path(path_text), conversation_id)
        )

    completed = [r["pid"] for r in renderers if r["complete_turn"]]
    report = {
        "report": "Cheater Chrome renderer comparison",
        "source": "Volatility windows.memmap --dump output",
        "method": {
            "history_pivot": "Validate the supplied History conversation UUID in exact /uc/ URL arrays and typed terminal controls.",
            "response_join": "Join terminal, committed stream, resume-token, and assistant-content-started records by their named UUID fields.",
            "submission_join": "Match the committed stream's pending assistant UUID to assistantMessageId in a decoded form allocation.",
            "prompt_join": "Require the validated /uc/ URL array and a valid Mojo composer KeyValue record to occupy the same 4 KiB storage page.",
        },
        "history_conversation_id": conversation_id,
        "renderers": renderers,
        "selected_pid": completed[0] if len(completed) == 1 else None,
    }
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Reconstruct Discord evidence from a dumped Chromium renderer.

The parser deliberately starts without a channel ID, username, message text, or
known offset.  It validates complete Discord Gateway ETF dispatches, correlates
the recovered DM channel with Discord's cached selected-channel state, and then
extracts DraftStore revisions for that same channel.
"""

from __future__ import annotations

import argparse
import collections
import datetime as dt
import json
import mmap
import struct
from pathlib import Path
from typing import Any, Iterator


DISCORD_EPOCH_MS = 1_420_070_400_000
MAX_TERM_BYTES = 8 * 1024 * 1024
MAX_COLLECTION_ITEMS = 100_000
MAX_DEPTH = 64


class ETFError(Exception):
    """A candidate is not a complete, supported ETF term."""


class ETFReader:
    def __init__(self, data: mmap.mmap, pos: int, limit: int):
        self.data = data
        self.pos = pos
        self.limit = min(limit, len(data))

    def take(self, size: int) -> bytes:
        if size < 0 or self.pos + size > self.limit:
            raise ETFError("term runs outside the candidate bounds")
        value = self.data[self.pos : self.pos + size]
        self.pos += size
        return value

    def u8(self) -> int:
        return self.take(1)[0]

    def u16(self) -> int:
        return struct.unpack(">H", self.take(2))[0]

    def u32(self) -> int:
        return struct.unpack(">I", self.take(4))[0]

    def checked_count(self, count: int) -> int:
        if count > MAX_COLLECTION_ITEMS:
            raise ETFError(f"unreasonable collection length: {count}")
        return count

    def atom(self, size: int) -> Any:
        value = self.take(size).decode("utf-8", errors="strict")
        return {"true": True, "false": False, "nil": None}.get(value, value)

    def binary(self, size: int) -> Any:
        value = self.take(size)
        try:
            return value.decode("utf-8", errors="strict")
        except UnicodeDecodeError:
            return {"$binary_hex": value.hex()}

    @staticmethod
    def map_key(value: Any) -> Any:
        if isinstance(value, (str, int, float, bool, bytes, type(None))):
            return value
        return json.dumps(value, ensure_ascii=False, sort_keys=True)

    def term(self, depth: int = 0) -> Any:
        if depth > MAX_DEPTH:
            raise ETFError("maximum ETF nesting depth exceeded")

        tag = self.u8()

        if tag == 70:  # NEW_FLOAT_EXT
            return struct.unpack(">d", self.take(8))[0]
        if tag == 97:  # SMALL_INTEGER_EXT
            return self.u8()
        if tag == 98:  # INTEGER_EXT
            return struct.unpack(">i", self.take(4))[0]
        if tag == 99:  # FLOAT_EXT (legacy, 31-byte string)
            raw = self.take(31).split(b"\x00", 1)[0]
            try:
                return float(raw.decode("ascii"))
            except ValueError as exc:
                raise ETFError("invalid FLOAT_EXT") from exc
        if tag in (100, 118):  # ATOM_EXT / ATOM_UTF8_EXT
            return self.atom(self.u16())
        if tag in (115, 119):  # SMALL_ATOM_EXT / SMALL_ATOM_UTF8_EXT
            return self.atom(self.u8())
        if tag == 104:  # SMALL_TUPLE_EXT
            return [self.term(depth + 1) for _ in range(self.checked_count(self.u8()))]
        if tag == 105:  # LARGE_TUPLE_EXT
            return [self.term(depth + 1) for _ in range(self.checked_count(self.u32()))]
        if tag == 106:  # NIL_EXT
            return []
        if tag == 107:  # STRING_EXT
            return self.take(self.u16()).decode("latin-1")
        if tag == 108:  # LIST_EXT
            count = self.checked_count(self.u32())
            values = [self.term(depth + 1) for _ in range(count)]
            tail = self.term(depth + 1)
            if tail != []:
                return {"items": values, "tail": tail}
            return values
        if tag == 109:  # BINARY_EXT
            return self.binary(self.u32())
        if tag == 77:  # BIT_BINARY_EXT
            size = self.u32()
            significant_bits = self.u8()
            value = self.binary(size)
            return {"bits_in_last_byte": significant_bits, "value": value}
        if tag in (110, 111):  # SMALL_BIG_EXT / LARGE_BIG_EXT
            size = self.u8() if tag == 110 else self.checked_count(self.u32())
            sign = self.u8()
            magnitude = int.from_bytes(self.take(size), "little")
            return -magnitude if sign else magnitude
        if tag == 116:  # MAP_EXT
            count = self.checked_count(self.u32())
            result = {}
            for _ in range(count):
                key = self.map_key(self.term(depth + 1))
                result[key] = self.term(depth + 1)
            return result

        raise ETFError(f"unsupported ETF tag {tag}")


def scan_gateway_events(data: mmap.mmap) -> list[dict[str, Any]]:
    """Find and validate uncompressed ETF maps beginning with version 131."""
    events: list[dict[str, Any]] = []
    seen: set[str] = set()
    cursor = 0

    while True:
        offset = data.find(b"\x83\x74", cursor)
        if offset == -1:
            break
        cursor = offset + 1
        reader = ETFReader(data, offset + 1, offset + MAX_TERM_BYTES)

        try:
            event = reader.term()
        except (ETFError, UnicodeDecodeError, struct.error):
            continue

        if not (
            isinstance(event, dict)
            and event.get("op") == 0
            and isinstance(event.get("t"), str)
            and "d" in event
            and "s" in event
        ):
            continue

        fingerprint = json.dumps(event, ensure_ascii=False, sort_keys=True)
        if fingerprint in seen:
            continue
        seen.add(fingerprint)
        events.append(
            {
                "offset": offset,
                "end_offset": reader.pos,
                "event": event,
            }
        )

    return events


def bounded_json_object(data: mmap.mmap, start: int, maximum: int = 1024 * 1024) -> tuple[Any, int]:
    """Return one balanced UTF-8 JSON object and its exclusive end offset."""
    if start < 0 or start >= len(data) or data[start] != ord("{"):
        raise ValueError("JSON object does not start with an opening brace")

    depth = 0
    in_string = False
    escaped = False
    end_limit = min(len(data), start + maximum)

    for pos in range(start, end_limit):
        byte = data[pos]
        if in_string:
            if escaped:
                escaped = False
            elif byte == ord("\\"):
                escaped = True
            elif byte == ord('"'):
                in_string = False
            continue

        if byte == ord('"'):
            in_string = True
        elif byte == ord("{"):
            depth += 1
        elif byte == ord("}"):
            depth -= 1
            if depth == 0:
                raw = data[start : pos + 1]
                return json.loads(raw.decode("utf-8")), pos + 1

    raise ValueError("unterminated JSON object")


def scan_selected_channels(data: mmap.mmap) -> list[dict[str, Any]]:
    marker = b'{"selectedChannelId"'
    records: list[dict[str, Any]] = []
    seen: set[str] = set()
    cursor = 0

    while True:
        json_offset = data.find(marker, cursor)
        if json_offset == -1:
            break
        cursor = json_offset + 1

        if json_offset < 8:
            continue
        length, flag = struct.unpack("<II", data[json_offset - 8 : json_offset])
        if flag != 1 or length < len(marker) or length > 1024 * 1024:
            continue

        try:
            raw = data[json_offset : json_offset + length]
            value = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            continue
        if not isinstance(value, dict) or "selectedChannelId" not in value:
            continue

        fingerprint = json.dumps(value, ensure_ascii=False, sort_keys=True)
        if fingerprint in seen:
            continue
        seen.add(fingerprint)
        records.append(
            {
                "record_offset": json_offset - 8,
                "json_offset": json_offset,
                "length": length,
                "string_flag": flag,
                "value": value,
            }
        )

    return records


def walk_draft_state(value: Any, path: tuple[str, ...] = ()) -> Iterator[dict[str, Any]]:
    if isinstance(value, dict):
        if isinstance(value.get("timestamp"), int) and isinstance(value.get("draft"), str):
            yield {
                "path": list(path),
                "timestamp_ms": value["timestamp"],
                "draft": value["draft"],
            }
        for key, child in value.items():
            yield from walk_draft_state(child, path + (str(key),))
    elif isinstance(value, list):
        for index, child in enumerate(value):
            yield from walk_draft_state(child, path + (str(index),))


def scan_draft_store(data: mmap.mmap) -> list[dict[str, Any]]:
    marker = b"DraftStore"
    drafts: list[dict[str, Any]] = []
    seen: set[tuple[str, str, int, str]] = set()
    cursor = 0

    while True:
        anchor = data.find(marker, cursor)
        if anchor == -1:
            break
        cursor = anchor + 1
        json_offset = data.find(b"{", anchor + len(marker), anchor + len(marker) + 32)
        if json_offset == -1:
            continue

        try:
            value, _ = bounded_json_object(data, json_offset)
        except (ValueError, UnicodeDecodeError, json.JSONDecodeError):
            continue
        if not isinstance(value, dict) or not isinstance(value.get("_state"), dict):
            continue

        for leaf in walk_draft_state(value["_state"]):
            path = leaf.pop("path")
            if len(path) < 2:
                continue
            user_id, channel_id = path[0], path[1]
            key = (user_id, channel_id, leaf["timestamp_ms"], leaf["draft"])
            if key in seen:
                continue
            seen.add(key)
            drafts.append(
                {
                    "record_offset": anchor,
                    "json_offset": json_offset,
                    "user_id": user_id,
                    "channel_id": channel_id,
                    **leaf,
                }
            )

    drafts.sort(key=lambda item: (item["timestamp_ms"], item["record_offset"]))
    return drafts


def iso_from_ms(timestamp_ms: int) -> str:
    value = dt.datetime.fromtimestamp(timestamp_ms / 1000, tz=dt.timezone.utc)
    return value.isoformat(timespec="milliseconds").replace("+00:00", "Z")


def snowflake_ms(value: str | int) -> int:
    return (int(value) >> 22) + DISCORD_EPOCH_MS


def event_id(entry: dict[str, Any]) -> str | None:
    payload = entry["event"].get("d")
    if isinstance(payload, dict) and payload.get("id") is not None:
        return str(payload["id"])
    return None


def choose_dm_channel(
    events: list[dict[str, Any]],
    selected: list[dict[str, Any]],
    drafts: list[dict[str, Any]],
) -> dict[str, Any]:
    messages_per_channel: collections.Counter[str] = collections.Counter()
    for entry in events:
        event = entry["event"]
        payload = event.get("d")
        if event.get("t") == "MESSAGE_CREATE" and isinstance(payload, dict):
            if payload.get("channel_id") is not None:
                messages_per_channel[str(payload["channel_id"])] += 1

    selected_ids = {
        str(record["value"].get("selectedChannelId"))
        for record in selected
        if record["value"].get("selectedChannelId") is not None
    }
    draft_counts: collections.Counter[str] = collections.Counter(
        draft["channel_id"] for draft in drafts
    )

    candidates = []
    for entry in events:
        event = entry["event"]
        payload = event.get("d")
        if event.get("t") != "CHANNEL_CREATE" or not isinstance(payload, dict):
            continue
        if payload.get("type") != 1 or payload.get("id") is None:
            continue
        channel_id = str(payload["id"])
        candidates.append(
            {
                "entry": entry,
                "channel_id": channel_id,
                "message_count": messages_per_channel[channel_id],
                "selected": channel_id in selected_ids,
                "draft_count": draft_counts[channel_id],
            }
        )

    if not candidates:
        raise RuntimeError("no validated type-1 CHANNEL_CREATE dispatch was recovered")

    corroborated = [
        candidate
        for candidate in candidates
        if candidate["message_count"] > 0
        and candidate["selected"]
        and candidate["draft_count"] > 0
    ]
    if len(corroborated) != 1:
        details = ", ".join(
            f"{candidate['channel_id']} "
            f"(messages={candidate['message_count']}, selected={candidate['selected']}, "
            f"drafts={candidate['draft_count']})"
            for candidate in candidates
        )
        raise RuntimeError(f"expected one fully corroborated DM channel; recovered: {details}")
    return corroborated[0]["entry"]


def compact_channel(payload: dict[str, Any]) -> dict[str, Any]:
    recipients = []
    for recipient in payload.get("recipients", []):
        if isinstance(recipient, dict):
            recipients.append(
                {
                    key: recipient.get(key)
                    for key in ("id", "username", "global_name")
                    if recipient.get(key) is not None
                }
            )
    return {"id": str(payload["id"]), "type": payload.get("type"), "recipients": recipients}


def compact_message(entry: dict[str, Any]) -> dict[str, Any]:
    payload = entry["event"]["d"]
    author = payload.get("author") if isinstance(payload.get("author"), dict) else {}
    identifier = str(payload["id"])
    derived_ms = snowflake_ms(identifier)
    supplied = str(payload.get("timestamp", ""))
    supplied_ms = int(dt.datetime.fromisoformat(supplied.replace("Z", "+00:00")).timestamp() * 1000)
    return {
        "event_offset": f"0x{entry['offset']:x}",
        "id": identifier,
        "channel_id": str(payload.get("channel_id")),
        "timestamp": supplied,
        "snowflake_timestamp": iso_from_ms(derived_ms),
        "snowflake_delta_ms": derived_ms - supplied_ms,
        "author": {
            key: author.get(key)
            for key in ("id", "username", "global_name")
            if author.get(key) is not None
        },
        "content": payload.get("content"),
    }


def final_draft_before(
    drafts: list[dict[str, Any]], timestamp_ms: int, user_id: str
) -> dict[str, Any] | None:
    candidates = [
        draft
        for draft in drafts
        if draft["user_id"] == user_id and draft["timestamp_ms"] <= timestamp_ms
    ]
    if not candidates:
        return None
    candidate = max(candidates, key=lambda item: item["timestamp_ms"])
    if timestamp_ms - candidate["timestamp_ms"] > 60_000:
        return None
    return candidate


def build_report(path: Path, data: mmap.mmap) -> dict[str, Any]:
    events = scan_gateway_events(data)
    selected = scan_selected_channels(data)
    drafts = scan_draft_store(data)
    channel_entry = choose_dm_channel(events, selected, drafts)
    channel = compact_channel(channel_entry["event"]["d"])
    channel_id = channel["id"]

    channel_messages = [
        compact_message(entry)
        for entry in events
        if entry["event"].get("t") == "MESSAGE_CREATE"
        and isinstance(entry["event"].get("d"), dict)
        and str(entry["event"]["d"].get("channel_id")) == channel_id
    ]
    channel_messages.sort(key=lambda item: item["timestamp"])

    selected_matches = [
        record for record in selected if str(record["value"].get("selectedChannelId")) == channel_id
    ]
    relevant_drafts = [draft for draft in drafts if draft["channel_id"] == channel_id]

    draft_user_counts = collections.Counter(draft["user_id"] for draft in relevant_drafts)
    message_author_ids = {str(item["author"].get("id")) for item in channel_messages}
    local_user_id = next(
        (user_id for user_id, _ in draft_user_counts.most_common() if user_id in message_author_ids),
        None,
    )

    message_comparisons = []
    for message in channel_messages:
        if local_user_id is None or str(message["author"].get("id")) != local_user_id:
            continue
        message_ms = int(
            dt.datetime.fromisoformat(message["timestamp"].replace("Z", "+00:00")).timestamp() * 1000
        )
        draft = final_draft_before(relevant_drafts, message_ms, local_user_id)
        if draft is None:
            continue
        draft_words = draft["draft"].split()
        sent_words = str(message["content"]).split()
        removed = [word for word in draft_words if word not in sent_words]
        message_comparisons.append(
            {
                "message_id": message["id"],
                "final_draft": draft["draft"],
                "sent_content": message["content"],
                "removed_words": removed,
                "draft_to_send_ms": message_ms - draft["timestamp_ms"],
            }
        )

    event_counts = collections.Counter(entry["event"]["t"] for entry in events)
    recipient_ids = [str(item["id"]) for item in channel["recipients"] if item.get("id")]

    return {
        "report": "Cheater Discord renderer reconstruction",
        "source": {"file": path.name, "size_bytes": path.stat().st_size},
        "method": {
            "gateway": "complete ETF terms beginning with VERSION_MAGIC + MAP_EXT",
            "selected_channel": "complete little-endian length-prefixed JSON records",
            "drafts": "complete DraftStore JSON states",
            "deduplication": "identical decoded Gateway events and identical draft revisions",
        },
        "gateway": {
            "validated_dispatch_count": len(events),
            "event_type_counts": dict(sorted(event_counts.items())),
            "dm_channel": {
                "event_offset": f"0x{channel_entry['offset']:x}",
                **channel,
            },
            "messages": channel_messages,
        },
        "client_state": {
            "selected_channel_records": [
                {
                    "record_offset": f"0x{record['record_offset']:x}",
                    "json_offset": f"0x{record['json_offset']:x}",
                    "length": record["length"],
                    "string_flag": record["string_flag"],
                    "value": record["value"],
                }
                for record in selected_matches
            ],
            "draft_revisions": [
                {
                    "record_offset": f"0x{draft['record_offset']:x}",
                    "user_id": draft["user_id"],
                    "channel_id": draft["channel_id"],
                    "timestamp": iso_from_ms(draft["timestamp_ms"]),
                    "timestamp_ms": draft["timestamp_ms"],
                    "draft": draft["draft"],
                }
                for draft in relevant_drafts
            ],
        },
        "correlation": {
            "channel_id": channel_id,
            "local_user_id": local_user_id,
            "recipient_ids": recipient_ids,
            "message_count": len(channel_messages),
            "all_messages_match_channel": all(
                message["channel_id"] == channel_id for message in channel_messages
            ),
            "all_snowflake_checks_zero_delta": all(
                message["snowflake_delta_ms"] == 0 for message in channel_messages
            ),
            "draft_to_sent": message_comparisons,
        },
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("renderer", type=Path, help="renderer dump, for example pid.4080.dmp")
    parser.add_argument("-o", "--output", type=Path, help="write the JSON report here")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    with args.renderer.open("rb") as stream:
        data = mmap.mmap(stream.fileno(), 0, access=mmap.ACCESS_READ)
        try:
            report = build_report(args.renderer, data)
        finally:
            data.close()

    rendered = json.dumps(report, indent=2, ensure_ascii=False) + "\n"
    if args.output:
        args.output.write_text(rendered, encoding="utf-8")
    else:
        print(rendered, end="")

    print(
        f"Recovered {report['gateway']['validated_dispatch_count']} Gateway dispatches, "
        f"{report['correlation']['message_count']} messages, and "
        f"{len(report['client_state']['draft_revisions'])} draft revisions "
        f"for channel {report['correlation']['channel_id']}."
    )


if __name__ == "__main__":
    main()