Crafter.Debug/implementations/Crafter.Debug-Session.cpp

// SPDX-License-Identifier: LGPL-3.0-only
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

module;
#include <signal.h>
#include <sys/prctl.h>
#include <unistd.h>
module Crafter.Debug:Session_impl;
import std;
import Crafter.Thread;
import Crafter.Debug.Core;
import :Registry;
import :Timeline;
import :Condition;
import :Session;
import :Transport;

using namespace Crafter::Debug;
using Crafter::Debug::Detail::Exchange;

// Named, not anonymous: Session::State's members take these types, and a
// type with internal linkage in the signature of one with module linkage is
// an exposure clang warns about. Nothing here is exported.
namespace Crafter::Debug::SessionImpl {
    using Clock = std::chrono::steady_clock;

    // Request parameters, decoded through Codec so a mistyped request is
    // answered with every mistake in it, as an app command would be.
    struct ListParams {
        std::string glob = "*";
    };

    struct LogParams {
        std::uint32_t last = 50;
        std::string level = "info";
        std::string category;
        std::string grep;
        std::optional<std::uint64_t> sinceFrame;
        std::uint64_t afterSequence = 0;
        std::vector<std::string> kinds;
        bool follow = false;
        std::uint32_t forMs = 30 * 60 * 1000;
    };

    struct PauseParams {
        std::optional<std::string> when;
        bool onCheck = false;
        bool trap = false;
        bool clear = false;
    };

    struct StepParams {
        std::uint32_t count = 1;
    };

    struct WaitParams {
        std::optional<std::string> condition;
        std::optional<std::uint32_t> frames;
        std::optional<std::string> event;
        bool paused = false;
    };

    struct QuitParams {
        bool force = false;
    };

    struct GetParams {
        std::vector<std::string> paths;
    };

    struct SetParams {
        Required<std::string> path;
        Required<Value> value;
    };

    struct TraceParams {
        std::vector<std::string> paths;
        std::uint32_t frames = 60;
        bool history = false;
    };

    struct Job {
        std::string method;
        Value params;
        std::shared_ptr<Exchange> exchange;
        Clock::time_point deadline;
    };

    struct Pending {
        std::string method;
        std::function<std::optional<Result<Value>>()> poll;
        std::shared_ptr<Exchange> exchange;
        Clock::time_point deadline;
    };

    enum class WaitKind : std::uint8_t {
        Frames,
        Condition,
        Event,
        Paused,
        Steps,
    };

    struct Waiter {
        WaitKind kind = WaitKind::Frames;
        std::shared_ptr<Exchange> exchange;
        Clock::time_point deadline;
        Clock::time_point since;
        std::uint64_t targetFrame = 0;
        std::optional<Condition> condition;
        std::string event;
    };

    struct TraceRun {
        std::vector<std::string> paths;
        std::uint32_t remaining = 0;
        std::vector<std::uint64_t> frames;
        std::vector<std::vector<Value>> series;
        std::shared_ptr<Exchange> exchange;
        Clock::time_point deadline;
    };

    struct Follower {
        Detail::TimelineQuery query;
        std::shared_ptr<Exchange> exchange;
        Clock::time_point until;
    };

    // The screenshot source, which an app may set before its Session exists.
    std::mutex captureMutex;
    std::function<Async<std::vector<std::byte>>()> captureSource;

    // The one Session's state, for the free functions.
    std::atomic<Session::State*> current{ nullptr };

    template<class P>
    std::optional<P> DecodeParams(const Value& params, Exchange& exchange) {
        P decoded{};
        DecodeContext context;
        Crafter::Debug::Decode(params.IsNull() ? Value::MakeObject() : params, decoded, context);
        if (!context.Ok()) {
            exchange.Reply(MakeErrorReply(Codes::invalidParams, context.Report()));
            return std::nullopt;
        }
        return decoded;
    }

    void ReplyWith(Exchange& exchange, Result<Value> result) {
        exchange.Reply(result.Ok() ? MakeReply(std::move(result.Get())) : MakeErrorReply(result.GetError().code, result.GetError().message));
    }

    Value Object(std::initializer_list<std::pair<std::string_view, Value>> members) {
        Value object = Value::MakeObject();
        for (const auto& [key, value] : members) object.Set(key, value);
        return object;
    }

    std::string ProgramName() {
        std::error_code ec;
        const std::filesystem::path exe = std::filesystem::read_symlink("/proc/self/exe", ec);
        return ec ? std::string("app") : exe.filename().string();
    }

    std::string NewToken() {
        std::random_device random;
        return std::format("{:08x}{:08x}{:08x}{:08x}", random(), random(), random(), random());
    }

    // A tracer (a debugger) is attached to this process: TracerPid in
    // /proc/self/status is nonzero.
    bool TracerAttached() {
        std::ifstream status("/proc/self/status");
        std::string line;
        while (std::getline(status, line)) {
            if (line.starts_with("TracerPid:")) return line.find_first_not_of("0\t ", 10) != std::string::npos;
        }
        return false;
    }

    std::string FileName(std::string_view path) {
        const std::size_t slash = path.rfind('/');
        return std::string(slash == std::string_view::npos ? path : path.substr(slash + 1));
    }

    std::string Abbreviated(std::string text, std::size_t limit = 200) {
        if (text.size() <= limit) return text;
        text.resize(limit);
        return text + "...";
    }

    Value EntryValue(const TimelineEntry& entry) {
        Value value = Object({
            { "sequence", Value(entry.sequence) },
            { "frame", Value(entry.frame) },
            { "seconds", Value(entry.seconds) },
            { "kind", Value(EntryKindName(entry.kind)) },
            { "category", Value(entry.category) },
            { "message", Value(entry.message) },
            { "file", Value(entry.file) },
            { "line", Value(entry.line) },
        });
        if (!entry.payload.IsNull()) value.Set("payload", entry.payload);
        if (entry.kind == EntryKind::Check) value.Set("count", Value(entry.count));
        return value;
    }

    std::expected<Value, std::string> ResolvePath(std::string_view path) {
        Detail::Resolved resolved = Detail::Registry::Instance().Resolve(path);
        if (!resolved.Ok()) return std::unexpected(std::move(resolved.error));
        return std::move(resolved.value);
    }

    // min, max, mean, first, last and how often the value changed, for a
    // traced series; numbers only for the arithmetic ones.
    Value Stats(std::span<const std::uint64_t> frames, std::span<const Value> values) {
        Value stats = Value::MakeObject();
        if (values.empty()) return stats;
        stats.Set("first", values.front());
        stats.Set("last", values.back());
        std::uint64_t changes = 0;
        std::optional<std::uint64_t> firstChange;
        for (std::size_t i = 1; i < values.size(); ++i) {
            if (!(values[i] == values[i - 1])) {
                ++changes;
                if (!firstChange) firstChange = frames[i];
            }
        }
        stats.Set("changes", Value(changes));
        if (firstChange) stats.Set("firstChangeFrame", Value(*firstChange));
        if (std::ranges::all_of(values, [](const Value& value) { return value.IsNumber(); })) {
            double sum = 0;
            double low = values.front().AsNumber();
            double high = low;
            for (const Value& value : values) {
                const double number = value.AsNumber();
                sum += number;
                low = std::min(low, number);
                high = std::max(high, number);
            }
            stats.Set("min", Value(low));
            stats.Set("max", Value(high));
            stats.Set("mean", Value(sum / static_cast<double>(values.size())));
        }
        return stats;
    }

    constexpr std::array<std::string_view, 13> builtinMethods = {
        "debug.hello", "debug.status", "debug.list", "debug.log", "debug.pause", "debug.resume", "debug.step",
        "debug.wait", "debug.quit", "debug.get", "debug.set", "debug.trace", "debug.screenshot",
    };
}

using namespace Crafter::Debug::SessionImpl;

namespace Crafter::Debug {
    struct Session::State {
        SessionOptions options;
        std::string token = NewToken();
        std::string name;
        Clock::time_point started = Clock::now();
        std::int64_t pid = ::getpid();

        // Guards what transport threads, the Pump thread and logging threads share.
        std::mutex mutex;
        std::deque<Job> jobs;
        std::vector<Waiter> waiters;
        std::vector<Follower> followers;
        std::optional<Condition> armedWhen;
        bool armedOnCheck = false;
        bool armedTrap = false;
        std::string pauseReason;
        // The frame the pause began at. Frames go on counting while paused,
        // so the current one says nothing about when it stopped.
        std::uint64_t pausedAt = 0;
        bool shutDown = false;

        // The Pump thread's own.
        std::vector<Pending> pendings;
        std::vector<TraceRun> traces;
        bool quitHandled = false;
        std::optional<Clock::time_point> lastPump;

        std::atomic<bool> paused{ false };
        std::atomic<std::uint32_t> steps{ 0 };
        std::atomic<bool> pauseSupported{ false };
        std::atomic<bool> quitRequested{ false };
        std::atomic<std::uint64_t> frame{ 0 };
        std::atomic<std::int64_t> lastPumpNanos{ -1 };
        std::atomic<double> fps{ 0 };

        std::optional<Group> builtins;

        // When listening: the transport, and the discovery record that
        // tells clients where it is.
        std::unique_ptr<Detail::Server> server;
        std::optional<Registration> registration;

        explicit State(SessionOptions given) : options(std::move(given)) {
            name = options.name.empty() ? ProgramName() : options.name;
        }

        void Start() {
            builtins.emplace("debug");
            builtins->Watch("frame", [this] { return frame.load(); }, { .help = "frames pumped so far" });
            builtins->Watch("paused", [this] { return paused.load(); }, { .help = "whether Advance() holds the app" });
            builtins->Watch("fps", [this] { return fps.load(); }, { .help = "Pumps per second, smoothed" });
        }

        // Last in the constructor: once the record is written, agents find
        // this instance and send it requests.
        void Listen(const std::shared_ptr<State>& self) {
            // Crafter.Network hands new streams to the pool, and this hands
            // each request on; Start(1) gives the pool no worker to run them.
            if (Crafter::ThreadPool::activeThreads.load() == 0) {
                throw std::logic_error("Crafter.Debug: the Session needs Crafter.Thread's pool with a worker thread. "
                                       "Call ThreadPool::Start() before creating it (Start(1) starts no worker).");
            }
            const std::weak_ptr<State> weak = self;
            server = std::make_unique<Detail::Server>(options.port, [weak](const Value& request, std::shared_ptr<Exchange> exchange) {
                if (const std::shared_ptr<State> alive = weak.lock()) return alive->Handle(request, exchange);
                exchange->Reply(MakeErrorReply(Codes::shuttingDown, "the app is shutting down"));
            });
            registration.emplace(InstanceRecord{
                .name = name,
                .port = server->Port(),
                .token = token,
                .protocol = protocolVersion,
                .sourceId = std::string(Detail::SourceId()),
                .client = std::string(Detail::ClientPath()),
            });
        }

        void Agent(std::string_view method, const Value& params) {
            const std::string text = params.IsNull() ? std::string(method) : std::format("{} {}", method, Abbreviated(params.Dump()));
            Detail::Timeline::Instance().Append(Detail::MakeEntry(EntryKind::Agent, "agent", text, std::source_location::current()));
        }

        void PauseNow(std::string reason, bool trap) {
            {
                std::lock_guard lock(mutex);
                pauseReason = reason;
                pausedAt = frame.load();
            }
            paused.store(true);
            Detail::Timeline::Instance().Append(Detail::MakeEntry(EntryKind::Pause, "debug", std::format("paused: {}", reason), std::source_location::current()));
            if (trap && TracerAttached()) ::raise(SIGTRAP);
        }

        // ---- requests, on transport threads ----

        void Handle(const Value& request, const std::shared_ptr<Exchange>& exchange) {
            try {
                Dispatch(request, exchange);
            } catch (const std::exception& error) {
                exchange->Reply(MakeErrorReply(Codes::exception, error.what()));
            } catch (...) {
                exchange->Reply(MakeErrorReply(Codes::exception, "the session threw something that is not a std::exception"));
            }
        }

        void Dispatch(const Value& request, const std::shared_ptr<Exchange>& exchange) {
            if (!request.IsObject()) {
                return exchange->Reply(MakeErrorReply(Codes::invalidRequest, R"(a request is a JSON object: {"token", "method", "params", "timeoutMs"})"));
            }
            const Value* tokenValue = request.Find("token");
            if (tokenValue == nullptr || !tokenValue->IsString() || tokenValue->AsString() != token) {
                return exchange->Reply(MakeErrorReply(Codes::unauthorized, "wrong or missing token: crafter-debug reads it from the instance's discovery file"));
            }
            const Value* methodValue = request.Find("method");
            if (methodValue == nullptr || !methodValue->IsString()) {
                return exchange->Reply(MakeErrorReply(Codes::invalidRequest, "a request needs a method"));
            }
            const std::string method = methodValue->AsString();
            const Value* paramsValue = request.Find("params");
            const Value params = paramsValue != nullptr ? *paramsValue : Value();
            std::int64_t timeoutMs = 30'000;
            if (const Value* timeout = request.Find("timeoutMs"); timeout != nullptr && timeout->IsInteger() && timeout->AsInteger() > 0) {
                timeoutMs = timeout->AsInteger();
            }
            const Clock::time_point deadline = Clock::now() + std::chrono::milliseconds(timeoutMs);
            {
                std::lock_guard lock(mutex);
                if (shutDown) return exchange->Reply(MakeErrorReply(Codes::shuttingDown, "the app is shutting down"));
            }

            if (method == "debug.hello") return exchange->Reply(MakeReply(Hello()));
            if (method == "debug.status") return exchange->Reply(MakeReply(Status()));
            if (method == "debug.list") return List(params, *exchange);
            if (method == "debug.log") return Logs(params, exchange, deadline);
            if (method == "debug.pause") return Pause(params, *exchange);
            if (method == "debug.resume") return Resume(*exchange);
            if (method == "debug.step") return Step(params, exchange, deadline);
            if (method == "debug.wait") return Wait(params, exchange, deadline);
            if (method == "debug.quit") return Quit(params, exchange);

            const bool builtin = std::ranges::find(builtinMethods, method) != builtinMethods.end();
            if (method.starts_with("debug.") && !builtin) {
                const auto suggestion = Suggest(method, builtinMethods);
                return exchange->Reply(MakeErrorReply(Codes::unknownMethod, suggestion ? std::format("no method '{}'; did you mean '{}'?", method, *suggestion)
                                                                                      : std::format("no method '{}'", method)));
            }
            if (!builtin && Detail::Registry::Instance().FindCommand(method) == nullptr) {
                const auto suggestion = Detail::Registry::Instance().SuggestCommand(method);
                return exchange->Reply(MakeErrorReply(Codes::unknownMethod, suggestion ? std::format("no command '{}'; did you mean '{}'?", method, *suggestion)
                                                                                      : std::format("no command '{}'; crafter-debug cmds lists them", method)));
            }
            std::lock_guard lock(mutex);
            jobs.push_back({ method, params, exchange, deadline });
        }

        Value Hello() const {
            return Object({
                { "protocol", Value(protocolVersion) },
                { "library", Value(libraryVersion) },
                { "name", Value(name) },
                { "pid", Value(pid) },
            });
        }

        Value Status() {
            Value status = Object({
                { "name", Value(name) },
                { "pid", Value(pid) },
                { "frame", Value(frame.load()) },
                { "fps", Value(fps.load()) },
                { "paused", Value(paused.load()) },
                { "pauseSupported", Value(pauseSupported.load()) },
                { "quitRequested", Value(quitRequested.load()) },
            });
            const std::int64_t last = lastPumpNanos.load();
            const std::int64_t sinceStart = std::chrono::duration_cast<std::chrono::nanoseconds>(Clock::now() - started).count();
            status.Set("lastPumpAgoMs", last < 0 ? Value() : Value((sinceStart - last) / 1'000'000));
            {
                std::lock_guard lock(mutex);
                status.Set("pauseReason", Value(pauseReason));
                if (paused.load()) status.Set("pausedAt", Value(pausedAt));
                status.Set("queued", Value(jobs.size()));
                status.Set("waiting", Value(waiters.size()));
                status.Set("following", Value(followers.size()));
                Value armed = Value::MakeObject();
                if (armedWhen) armed.Set("when", Value(armedWhen->Text()));
                if (armedOnCheck) armed.Set("onCheck", Value(true));
                status.Set("armed", std::move(armed));
            }
            std::uint64_t failures = 0;
            Value sites = Value::MakeArray();
            for (const Detail::CheckSite& site : Detail::Timeline::Instance().CheckSites()) {
                failures += site.failures;
                sites.Push(Object({ { "file", Value(FileName(site.file)) }, { "line", Value(site.line) }, { "category", Value(site.category) }, { "failures", Value(site.failures) } }));
            }
            status.Set("checkFailures", Value(failures));
            status.Set("checkSites", std::move(sites));
            status.Set("commands", Value(Detail::Registry::Instance().Commands().size()));
            status.Set("watches", Value(Detail::Registry::Instance().Watches().size()));
            return status;
        }

        void List(const Value& params, Exchange& exchange) {
            const auto decoded = DecodeParams<ListParams>(params, exchange);
            if (!decoded) return;
            Value commands = Value::MakeArray();
            for (const auto& command : Detail::Registry::Instance().Commands(decoded->glob)) {
                commands.Push(Object({ { "name", Value(command->name) }, { "description", Value(command->description) }, { "args", command->args }, { "returns", command->returns } }));
            }
            Value watches = Value::MakeArray();
            for (const auto& watch : Detail::Registry::Instance().Watches(decoded->glob)) {
                watches.Push(Object({ { "path", Value(watch->path) }, { "type", watch->type }, { "help", Value(watch->help) }, { "readOnly", Value(!watch->set) } }));
            }
            Value samples = Value::MakeArray();
            for (const std::string& path : Detail::Registry::Instance().SamplePaths(decoded->glob)) samples.Push(Value(path));
            exchange.Reply(MakeReply(Object({ { "commands", std::move(commands) }, { "watches", std::move(watches) }, { "samples", std::move(samples) } })));
        }

        void Logs(const Value& params, const std::shared_ptr<Exchange>& exchange, Clock::time_point) {
            const auto decoded = DecodeParams<LogParams>(params, *exchange);
            if (!decoded) return;
            Detail::TimelineQuery query;
            query.afterSequence = decoded->afterSequence;
            query.sinceFrame = decoded->sinceFrame;
            query.category = decoded->category;
            query.grep = decoded->grep;
            if (decoded->level == "info") {
                query.minLevel = 0;
            } else if (decoded->level == "warn") {
                query.minLevel = 1;
            } else if (decoded->level == "error") {
                query.minLevel = 2;
            } else {
                return exchange->Reply(MakeErrorReply(Codes::invalidParams, std::format("level: '{}' is not one of info, warn, error", decoded->level)));
            }
            for (const std::string& kind : decoded->kinds) {
                bool known = false;
                for (std::uint8_t k = 0; k <= static_cast<std::uint8_t>(EntryKind::Stdio); ++k) {
                    if (EntryKindName(static_cast<EntryKind>(k)) == kind) {
                        query.kinds.push_back(static_cast<EntryKind>(k));
                        known = true;
                    }
                }
                if (!known) return exchange->Reply(MakeErrorReply(Codes::invalidParams, std::format("kinds: '{}' is not one of log, warn, error, check, event, agent, pause, stdio", kind)));
            }
            // `last` 0 means none, as with tail -n 0: a follower that only
            // wants what comes next. (A TimelineQuery's 0 means all.)
            Detail::TimelineQuery backlog = query;
            backlog.last = decoded->last;
            const auto recent = [&] { return decoded->last == 0 ? std::vector<TimelineEntry>() : Detail::Timeline::Instance().Query(backlog); };
            if (!decoded->follow) {
                Value entries = Value::MakeArray();
                for (const TimelineEntry& entry : recent()) entries.Push(EntryValue(entry));
                return exchange->Reply(MakeReply(Object({ { "entries", std::move(entries) }, { "lastSequence", Value(Detail::Timeline::Instance().LastSequence()) } })));
            }
            // The backlog and the registration happen under one lock, which
            // the listener also takes before streaming, so a follower sees
            // every entry once and in order. Stream never blocks.
            std::lock_guard lock(mutex);
            const std::uint64_t upTo = Detail::Timeline::Instance().LastSequence();
            for (const TimelineEntry& entry : recent()) {
                if (entry.sequence > upTo) continue;
                if (!exchange->Stream(Object({ { "entry", EntryValue(entry) } }))) return;
            }
            Follower follower{ query, exchange, Clock::now() + std::chrono::milliseconds(decoded->forMs) };
            follower.query.afterSequence = std::max(query.afterSequence, upTo);
            followers.push_back(std::move(follower));
        }

        void Pause(const Value& params, Exchange& exchange) {
            const auto decoded = DecodeParams<PauseParams>(params, exchange);
            if (!decoded) return;
            if (decoded->clear) {
                {
                    std::lock_guard lock(mutex);
                    armedWhen.reset();
                    armedOnCheck = false;
                    armedTrap = false;
                }
                Agent("pause --clear", Value());
                return exchange.Reply(MakeReply(Object({ { "armed", Value(false) } })));
            }
            if (!pauseSupported.load()) {
                return exchange.Reply(MakeErrorReply(Codes::pauseUnsupported, "this app never calls Debug::Advance() or Debug::Paused(), so pausing would hold nothing; "
                                                                             "gate the simulation on them (see the README)"));
            }
            if (decoded->trap && !decoded->when && !decoded->onCheck) {
                return exchange.Reply(MakeErrorReply(Codes::invalidParams, "trap needs when or onCheck: it stops the app where that happens"));
            }
            if (decoded->when) {
                std::optional<Condition> condition;
                try {
                    condition = Condition::Parse(*decoded->when);
                } catch (const ConditionError& error) {
                    return exchange.Reply(MakeErrorReply(Codes::conditionError, error.what()));
                }
                {
                    std::lock_guard lock(mutex);
                    armedWhen = std::move(condition);
                    armedTrap = decoded->trap;
                }
                Agent("pause --when", Value(*decoded->when));
                return exchange.Reply(MakeReply(Object({ { "armed", Value("when") }, { "condition", Value(*decoded->when) } })));
            }
            if (decoded->onCheck) {
                {
                    std::lock_guard lock(mutex);
                    armedOnCheck = true;
                    armedTrap = decoded->trap;
                }
                Agent("pause --on-check", Value());
                return exchange.Reply(MakeReply(Object({ { "armed", Value("onCheck") } })));
            }
            Agent("pause", Value());
            PauseNow("an agent asked", false);
            exchange.Reply(MakeReply(Object({ { "paused", Value(true) }, { "frame", Value(frame.load()) } })));
        }

        void Resume(Exchange& exchange) {
            Agent("resume", Value());
            steps.store(0);
            paused.store(false);
            {
                std::lock_guard lock(mutex);
                pauseReason.clear();
            }
            Detail::Timeline::Instance().Append(Detail::MakeEntry(EntryKind::Pause, "debug", "resumed", std::source_location::current()));
            exchange.Reply(MakeReply(Object({ { "paused", Value(false) }, { "frame", Value(frame.load()) } })));
        }

        void Step(const Value& params, const std::shared_ptr<Exchange>& exchange, Clock::time_point deadline) {
            const auto decoded = DecodeParams<StepParams>(params, *exchange);
            if (!decoded) return;
            if (!pauseSupported.load()) {
                return exchange->Reply(MakeErrorReply(Codes::pauseUnsupported, "this app never calls Debug::Advance() or Debug::Paused(), so there is nothing to step"));
            }
            if (!paused.load()) return exchange->Reply(MakeErrorReply(Codes::notPaused, "step works while paused: run crafter-debug pause first"));
            if (decoded->count == 0) return exchange->Reply(MakeErrorReply(Codes::invalidParams, "count: step at least 1"));
            Agent("step", Object({ { "count", Value(decoded->count) } }));
            std::lock_guard lock(mutex);
            steps.fetch_add(decoded->count);
            waiters.push_back({ .kind = WaitKind::Steps, .exchange = exchange, .deadline = deadline, .since = Clock::now() });
        }

        void Wait(const Value& params, const std::shared_ptr<Exchange>& exchange, Clock::time_point deadline) {
            const auto decoded = DecodeParams<WaitParams>(params, *exchange);
            if (!decoded) return;
            const std::int32_t asked = (decoded->condition ? 1 : 0) + (decoded->frames ? 1 : 0) + (decoded->event ? 1 : 0) + (decoded->paused ? 1 : 0);
            if (asked != 1) return exchange->Reply(MakeErrorReply(Codes::invalidParams, "wait for one thing: condition, frames, event or paused"));
            Waiter waiter{ .exchange = exchange, .deadline = deadline, .since = Clock::now() };
            if (decoded->condition) {
                try {
                    waiter.condition = Condition::Parse(*decoded->condition);
                } catch (const ConditionError& error) {
                    return exchange->Reply(MakeErrorReply(Codes::conditionError, error.what()));
                }
                waiter.kind = WaitKind::Condition;
            } else if (decoded->frames) {
                waiter.kind = WaitKind::Frames;
                waiter.targetFrame = frame.load() + *decoded->frames;
            } else if (decoded->event) {
                waiter.kind = WaitKind::Event;
                waiter.event = *decoded->event;
            } else {
                waiter.kind = WaitKind::Paused;
                if (paused.load()) {
                    std::lock_guard lock(mutex);
                    return exchange->Reply(MakeReply(Object({ { "frame", Value(frame.load()) }, { "pausedAt", Value(pausedAt) }, { "reason", Value(pauseReason) } })));
                }
            }
            std::lock_guard lock(mutex);
            waiters.push_back(std::move(waiter));
        }

        void Quit(const Value& params, const std::shared_ptr<Exchange>& exchange) {
            const auto decoded = DecodeParams<QuitParams>(params, *exchange);
            if (!decoded) return;
            Agent(decoded->force ? "quit --force" : "quit", Value());
            if (decoded->force) {
                quitRequested.store(true);
                // From the thread the reply leaves on, so the main loop
                // need not run: --force is for an app whose loop is stuck.
                return exchange->ReplyThen(MakeReply(Object({ { "quitting", Value(true) }, { "force", Value(true) } })), [] { std::_Exit(0); });
            }
            // The reply before the flag: a loop that sees the flag may end
            // the Session at once, and the transport delivers a reply given
            // before its connections close, not one given after.
            exchange->Reply(MakeReply(Object({ { "quitting", Value(true) } })));
            quitRequested.store(true);
        }

        // ---- the Pump thread ----

        void Pump() {
            const Clock::time_point now = Clock::now();
            const std::uint64_t current = frame.fetch_add(1) + 1;
            Detail::SetFrame(current);
            if (lastPump) {
                const double seconds = std::chrono::duration<double>(now - *lastPump).count();
                if (seconds > 0) {
                    const double previous = fps.load();
                    fps.store(previous == 0 ? 1 / seconds : previous * 0.9 + 0.1 / seconds);
                }
            }
            lastPump = now;
            lastPumpNanos.store(std::chrono::duration_cast<std::chrono::nanoseconds>(now - started).count());

            RunJobs(now);
            PollPendings(now);
            CheckArmed();
            ServeWaiters(now, current);
            RecordTraces(now, current);
            ExpireFollowers(now);
            if (quitRequested.load() && !quitHandled) {
                quitHandled = true;
                if (options.onQuit) options.onQuit();
            }
        }

        void RunJobs(Clock::time_point now) {
            std::deque<Job> taken;
            {
                std::lock_guard lock(mutex);
                taken.swap(jobs);
            }
            for (Job& job : taken) {
                if (now > job.deadline) {
                    job.exchange->Reply(MakeErrorReply(Codes::timeout, "not run: the app did not Pump before the request's timeout"));
                    continue;
                }
                try {
                    Execute(job);
                } catch (const std::exception& error) {
                    job.exchange->Reply(MakeErrorReply(Codes::exception, error.what()));
                } catch (...) {
                    job.exchange->Reply(MakeErrorReply(Codes::exception, "something that is not a std::exception was thrown"));
                }
            }
        }

        void Execute(Job& job) {
            Agent(job.method, job.params);
            if (job.method == "debug.get") return Get(job);
            if (job.method == "debug.set") return Set(job);
            if (job.method == "debug.trace") return StartTrace(job);
            if (job.method == "debug.screenshot") return Screenshot(job);
            const auto command = Detail::Registry::Instance().FindCommand(job.method);
            if (!command) return job.exchange->Reply(MakeErrorReply(Codes::unknownMethod, std::format("'{}' was unregistered before it ran", job.method)));
            Detail::CommandOutcome outcome = command->invoke(job.params);
            if (outcome.done) return ReplyWith(*job.exchange, std::move(*outcome.done));
            pendings.push_back({ job.method, std::move(outcome.pending), job.exchange, job.deadline });
        }

        void Get(Job& job) {
            const auto decoded = DecodeParams<GetParams>(job.params, *job.exchange);
            if (!decoded) return;
            if (decoded->paths.empty()) return job.exchange->Reply(MakeErrorReply(Codes::invalidParams, "paths: name at least one path, or a glob such as 'fort1.*'"));
            Detail::Registry& registry = Detail::Registry::Instance();
            Value values = Value::MakeObject();
            Value errors = Value::MakeObject();
            const auto read = [&](std::string_view path) {
                Detail::Resolved resolved = registry.Resolve(path);
                if (resolved.Ok()) {
                    values.Set(path, std::move(resolved.value));
                } else {
                    errors.Set(path, Value(resolved.error));
                }
            };
            for (const std::string& path : decoded->paths) {
                if (path.find_first_of("*?") == std::string::npos) {
                    read(path);
                    continue;
                }
                const auto watches = registry.Watches(path);
                const auto samples = registry.SamplePaths(path);
                if (watches.empty() && samples.empty()) errors.Set(path, Value("no watch or sample matches"));
                for (const auto& watch : watches) read(watch->path);
                for (const std::string& sample : samples) {
                    if (values.Find(sample) == nullptr) read(sample);
                }
            }
            if (values.Size() == 0 && errors.Size() != 0) {
                std::string message;
                for (const auto& [path, error] : errors.Members()) message += message.empty() ? error.AsString() : "; " + error.AsString();
                return job.exchange->Reply(MakeErrorReply(Codes::unknownPath, message));
            }
            Value result = Object({ { "values", std::move(values) } });
            if (errors.Size() != 0) result.Set("errors", std::move(errors));
            job.exchange->Reply(MakeReply(std::move(result)));
        }

        void Set(Job& job) {
            const auto decoded = DecodeParams<SetParams>(job.params, *job.exchange);
            if (!decoded) return;
            if (const auto error = Detail::Registry::Instance().Set(*decoded->path, *decoded->value)) {
                return job.exchange->Reply(MakeErrorReply(error->code, error->message));
            }
            Detail::Resolved now = Detail::Registry::Instance().Resolve(*decoded->path);
            job.exchange->Reply(MakeReply(Object({ { "path", Value(*decoded->path) }, { "value", now.Ok() ? now.value : *decoded->value } })));
        }

        void StartTrace(Job& job) {
            const auto decoded = DecodeParams<TraceParams>(job.params, *job.exchange);
            if (!decoded) return;
            if (decoded->paths.empty()) return job.exchange->Reply(MakeErrorReply(Codes::invalidParams, "paths: name at least one path"));
            Detail::Registry& registry = Detail::Registry::Instance();
            if (decoded->history) {
                Value series = Value::MakeObject();
                for (const std::string& path : decoded->paths) {
                    const std::vector<Detail::SamplePoint> points = registry.SampleHistory(path);
                    if (points.empty()) return job.exchange->Reply(MakeErrorReply(Codes::unknownPath, std::format("no sample history for '{}'; history is kept for Sample() paths", path)));
                    std::vector<std::uint64_t> frames;
                    std::vector<Value> values;
                    for (const Detail::SamplePoint& point : points) {
                        frames.push_back(point.frame);
                        values.push_back(point.value);
                    }
                    Value framesValue = Value::MakeArray();
                    for (const std::uint64_t f : frames) framesValue.Push(Value(f));
                    series.Set(path, Object({ { "frames", std::move(framesValue) }, { "values", Value::MakeArray(values) }, { "stats", Stats(frames, values) } }));
                }
                return job.exchange->Reply(MakeReply(Object({ { "series", std::move(series) } })));
            }
            if (decoded->frames == 0) return job.exchange->Reply(MakeErrorReply(Codes::invalidParams, "frames: trace at least 1 frame"));
            for (const std::string& path : decoded->paths) {
                const Detail::Resolved resolved = registry.Resolve(path);
                if (!resolved.Ok()) return job.exchange->Reply(MakeErrorReply(Codes::unknownPath, resolved.error));
            }
            traces.push_back({ decoded->paths, decoded->frames, {}, std::vector<std::vector<Value>>(decoded->paths.size()), job.exchange, job.deadline });
        }

        void Screenshot(Job& job) {
            std::function<Async<std::vector<std::byte>>()> source;
            {
                std::lock_guard lock(captureMutex);
                source = captureSource;
            }
            if (!source) {
                return job.exchange->Reply(MakeErrorReply(Codes::noCapture, "this app has no screenshot source: call Debug::SetCapture (the README has the Crafter.Graphics version)"));
            }
            pendings.push_back({
                "debug.screenshot",
                [async = source()]() mutable -> std::optional<Result<Value>> {
                    std::optional<Result<std::vector<std::byte>>> polled = async.Poll();
                    if (!polled) return std::nullopt;
                    if (!polled->Ok()) return Result<Value>(polled->GetError());
                    const std::vector<std::byte>& png = polled->Get();
                    return Result<Value>(Object({ { "png", Value(Base64Encode(png)) }, { "bytes", Value(png.size()) } }));
                },
                job.exchange,
                job.deadline,
            });
        }

        void PollPendings(Clock::time_point now) {
            for (auto it = pendings.begin(); it != pendings.end();) {
                try {
                    std::optional<Result<Value>> polled = it->poll();
                    if (polled) {
                        ReplyWith(*it->exchange, std::move(*polled));
                        it = pendings.erase(it);
                        continue;
                    }
                } catch (const std::exception& error) {
                    it->exchange->Reply(MakeErrorReply(Codes::exception, std::format("the result threw: {}", error.what())));
                    it = pendings.erase(it);
                    continue;
                } catch (...) {
                    it->exchange->Reply(MakeErrorReply(Codes::exception, "the result threw something that is not a std::exception"));
                    it = pendings.erase(it);
                    continue;
                }
                if (now > it->deadline) {
                    it->exchange->Reply(MakeErrorReply(Codes::timeout, std::format("{} ran, but its result was not ready before the request's timeout", it->method)));
                    it = pendings.erase(it);
                    continue;
                }
                ++it;
            }
        }

        void CheckArmed() {
            std::optional<Condition> when;
            bool trap = false;
            {
                std::lock_guard lock(mutex);
                if (!armedWhen) return;
                when = armedWhen;
                trap = armedTrap;
            }
            const auto result = when->Evaluate(ResolvePath);
            if (result && !*result) return;
            {
                std::lock_guard lock(mutex);
                armedWhen.reset();
            }
            if (!result) {
                Detail::Timeline::Instance().Append(Detail::MakeEntry(EntryKind::Warn, "debug", std::format("pause --when '{}' disarmed: {}", when->Text(), result.error()), std::source_location::current()));
                return;
            }
            PauseNow(std::format("'{}' became true", when->Text()), trap);
        }

        std::string TimeoutMessage(const Waiter& waiter) const {
            const auto waited = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now() - waiter.since).count();
            switch (waiter.kind) {
            case WaitKind::Frames: return std::format("frame {} was not reached within {} ms; the app is at frame {}", waiter.targetFrame, waited, frame.load());
            case WaitKind::Paused: return std::format("the app did not pause within {} ms", waited);
            case WaitKind::Steps: return std::format("the steps were not taken within {} ms: does the app call Debug::Advance() or TakeSteps() every frame?", waited);
            case WaitKind::Event: return std::format("no '{}' event within {} ms", waiter.event, waited);
            case WaitKind::Condition: {
                std::string values;
                for (const std::string& path : waiter.condition->Paths()) {
                    const auto value = ResolvePath(path);
                    if (!values.empty()) values += ", ";
                    values += value ? std::format("{} is {}", path, Abbreviated(value->Dump(), 80)) : std::format("{}: {}", path, value.error());
                }
                return std::format("'{}' was not true within {} ms{}", waiter.condition->Text(), waited, values.empty() ? std::string() : std::format(" ({})", values));
            }
            }
            return "timed out";
        }

        void ServeWaiters(Clock::time_point now, std::uint64_t current) {
            // Event waiters stay in the list, where the listener finds them.
            std::vector<Waiter> taken;
            {
                std::lock_guard lock(mutex);
                for (auto it = waiters.begin(); it != waiters.end();) {
                    if (it->kind == WaitKind::Event) {
                        if (now > it->deadline) {
                            it->exchange->Reply(MakeErrorReply(Codes::timeout, TimeoutMessage(*it)));
                            it = waiters.erase(it);
                        } else {
                            ++it;
                        }
                        continue;
                    }
                    taken.push_back(std::move(*it));
                    it = waiters.erase(it);
                }
            }
            std::vector<Waiter> remaining;
            for (Waiter& waiter : taken) {
                bool done = false;
                switch (waiter.kind) {
                case WaitKind::Frames:
                    if (current >= waiter.targetFrame) {
                        waiter.exchange->Reply(MakeReply(Object({ { "frame", Value(current) } })));
                        done = true;
                    }
                    break;
                case WaitKind::Condition: {
                    const auto result = waiter.condition->Evaluate(ResolvePath);
                    if (!result) {
                        waiter.exchange->Reply(MakeErrorReply(Codes::conditionError, result.error()));
                        done = true;
                    } else if (*result) {
                        waiter.exchange->Reply(MakeReply(Object({ { "frame", Value(current) }, { "condition", Value(waiter.condition->Text()) } })));
                        done = true;
                    }
                    break;
                }
                case WaitKind::Paused:
                    if (paused.load()) {
                        std::string reason;
                        std::uint64_t since = 0;
                        {
                            std::lock_guard lock(mutex);
                            reason = pauseReason;
                            since = pausedAt;
                        }
                        waiter.exchange->Reply(MakeReply(Object({ { "frame", Value(current) }, { "pausedAt", Value(since) }, { "reason", Value(reason) } })));
                        done = true;
                    }
                    break;
                case WaitKind::Steps:
                    if (steps.load() == 0) {
                        waiter.exchange->Reply(MakeReply(Object({ { "frame", Value(current) }, { "paused", Value(paused.load()) } })));
                        done = true;
                    }
                    break;
                case WaitKind::Event: break;
                }
                if (!done && now > waiter.deadline) {
                    waiter.exchange->Reply(MakeErrorReply(Codes::timeout, TimeoutMessage(waiter)));
                    done = true;
                }
                if (!done) remaining.push_back(std::move(waiter));
            }
            std::lock_guard lock(mutex);
            for (Waiter& waiter : remaining) waiters.push_back(std::move(waiter));
        }

        void RecordTraces(Clock::time_point now, std::uint64_t current) {
            for (auto it = traces.begin(); it != traces.end();) {
                TraceRun& run = *it;
                run.frames.push_back(current);
                for (std::size_t i = 0; i < run.paths.size(); ++i) {
                    const auto value = ResolvePath(run.paths[i]);
                    run.series[i].push_back(value ? *value : Value());
                }
                const bool truncated = now > run.deadline;
                if (--run.remaining != 0 && !truncated) {
                    ++it;
                    continue;
                }
                Value frames = Value::MakeArray();
                for (const std::uint64_t f : run.frames) frames.Push(Value(f));
                Value series = Value::MakeObject();
                for (std::size_t i = 0; i < run.paths.size(); ++i) {
                    series.Set(run.paths[i], Object({ { "values", Value::MakeArray(run.series[i]) }, { "stats", Stats(run.frames, run.series[i]) } }));
                }
                Value result = Object({ { "frames", std::move(frames) }, { "series", std::move(series) } });
                if (truncated) result.Set("truncated", Value(true));
                run.exchange->Reply(MakeReply(std::move(result)));
                it = traces.erase(it);
            }
        }

        void ExpireFollowers(Clock::time_point now) {
            std::vector<std::shared_ptr<Exchange>> ended;
            {
                std::lock_guard lock(mutex);
                for (auto it = followers.begin(); it != followers.end();) {
                    if (now > it->until) {
                        ended.push_back(it->exchange);
                        it = followers.erase(it);
                    } else {
                        ++it;
                    }
                }
            }
            for (const auto& exchange : ended) exchange->Reply(MakeReply(Object({ { "ended", Value("the follow time elapsed") } })));
        }

        // ---- every timeline entry, on the thread that recorded it ----

        void OnEntry(const TimelineEntry& entry) {
            if (entry.kind == EntryKind::Check) {
                bool trigger = false;
                bool trap = false;
                {
                    std::lock_guard lock(mutex);
                    if (armedOnCheck) {
                        armedOnCheck = false;
                        trigger = true;
                        trap = armedTrap;
                    }
                }
                if (trigger) PauseNow(std::format("check failed at {}:{}: {}", FileName(entry.file), entry.line, entry.message), trap);
            }
            if (entry.kind == EntryKind::Event) {
                std::vector<std::shared_ptr<Exchange>> matched;
                {
                    std::lock_guard lock(mutex);
                    for (auto it = waiters.begin(); it != waiters.end();) {
                        if (it->kind == WaitKind::Event && it->event == entry.category) {
                            matched.push_back(it->exchange);
                            it = waiters.erase(it);
                        } else {
                            ++it;
                        }
                    }
                }
                for (const auto& exchange : matched) {
                    exchange->Reply(MakeReply(Object({ { "frame", Value(entry.frame) }, { "event", Value(entry.category) }, { "payload", entry.payload } })));
                }
            }
            if (entry.sequence == 0) return;
            std::lock_guard lock(mutex);
            std::erase_if(followers, [&](Follower& follower) {
                if (!Detail::Matches(follower.query, entry)) return false;
                return !follower.exchange->Stream(Object({ { "entry", EntryValue(entry) } }));
            });
        }

        // ---- the end, on the thread destroying the Session ----

        void Shutdown() {
            Detail::Timeline::Instance().SetListener(nullptr);
            std::deque<Job> queued;
            std::vector<Waiter> waiting;
            std::vector<Follower> following;
            {
                std::lock_guard lock(mutex);
                shutDown = true;
                queued.swap(jobs);
                waiting.swap(waiters);
                following.swap(followers);
            }
            const Value goodbye = MakeErrorReply(Codes::shuttingDown, "the app is shutting down");
            for (const Job& job : queued) job.exchange->Reply(goodbye);
            for (const Waiter& waiter : waiting) waiter.exchange->Reply(goodbye);
            for (const Follower& follower : following) follower.exchange->Reply(goodbye);
            for (const Pending& pending : pendings) pending.exchange->Reply(goodbye);
            for (const TraceRun& run : traces) run.exchange->Reply(goodbye);
            pendings.clear();
            traces.clear();
            builtins.reset();
        }
    };
}

Session::Session(SessionOptions options) : state(std::make_shared<State>(std::move(options))) {
    State* expected = nullptr;
    if (!current.compare_exchange_strong(expected, state.get())) {
        throw std::logic_error("Crafter.Debug: a Session already exists; there is one per process");
    }
    try {
        state->Start();
        std::weak_ptr<State> weak = state;
        Detail::Timeline::Instance().SetListener([weak](const TimelineEntry& entry) {
            if (const std::shared_ptr<State> alive = weak.lock()) alive->OnEntry(entry);
        });
        // A debugger the agent starts is never this process's ancestor,
        // which Yama's ptrace_scope 1 would otherwise require.
        ::prctl(PR_SET_PTRACER, PR_SET_PTRACER_ANY, 0, 0, 0);
        if (state->options.listen) state->Listen(state);
    } catch (...) {
        state->registration.reset();
        state->server.reset();
        state->Shutdown();
        current.store(nullptr);
        throw;
    }
}

Session::~Session() {
    // Out of the listing first, so no agent picks an instance on its way
    // out; then every waiting request answered; then the connections
    // closed, which lets those answers leave first.
    state->registration.reset();
    state->Shutdown();
    state->server.reset();
    current.store(nullptr);
}

void Session::Handle(const Value& request, std::shared_ptr<Detail::Exchange> exchange) {
    state->Handle(request, exchange);
}

const std::string& Session::Token() const {
    return state->token;
}

void Crafter::Debug::Pump() {
    if (Session::State* session = current.load()) session->Pump();
}

bool Crafter::Debug::Advance() {
    Session::State* session = current.load();
    if (session == nullptr) return true;
    session->pauseSupported.store(true, std::memory_order_relaxed);
    if (!session->paused.load()) return true;
    std::uint32_t available = session->steps.load();
    while (available > 0) {
        if (session->steps.compare_exchange_weak(available, available - 1)) return true;
    }
    return false;
}

bool Crafter::Debug::Paused() {
    Session::State* session = current.load();
    if (session == nullptr) return false;
    session->pauseSupported.store(true, std::memory_order_relaxed);
    return session->paused.load();
}

std::uint32_t Crafter::Debug::TakeSteps() {
    Session::State* session = current.load();
    if (session == nullptr) return 0;
    session->pauseSupported.store(true, std::memory_order_relaxed);
    return session->steps.exchange(0);
}

bool Crafter::Debug::QuitRequested() {
    Session::State* session = current.load();
    return session != nullptr && session->quitRequested.load();
}

void Crafter::Debug::SetCapture(std::function<Async<std::vector<std::byte>>()> capture) {
    std::lock_guard lock(captureMutex);
    captureSource = std::move(capture);
}