wire.hpp

// SPDX-License-Identifier: Apache-2.0
 
#pragma once
#include <algorithm>
#include <array>
#include <cstddef>
#include <cstdint>
#include <expected>
#include <functional>
#include <glaze/glaze.hpp>
#include <limits>
#include <optional>
#include <span>
#include <stdexcept>
#include <string>
#include <string_view>
#include <vector>
 
#include "../session/session.hpp"
#include "change_types.hpp"
#include "profiler.hpp"
 
namespace morph::wire {
 
/// @brief Maximum accepted serialized size, in bytes, of a single wire envelope.
///
/// `decode` rejects any input longer than this before handing it to the JSON
/// parser. The cap is a wire-layer denial-of-service backstop: the outer decode
/// treats `body` as one opaque string, so a deeply-nested or enormous JSON
/// payload smuggled *inside* `body` is invisible to any structural/depth check
/// and would only detonate when the action codec re-parses it later (the "body
/// double-parse"). Bounding the total message length caps that cost up
/// front. The default of 8 MiB is generous for legitimate action payloads while
/// keeping a single message's peak allocation bounded; transports that want a
/// tighter bound should enforce it before calling `decode`.
inline constexpr std::size_t kMaxEnvelopeBytes = 8u * 1024u * 1024u;
 
/// @brief Protocol version this build of `morph` speaks.
///
/// Carried in `Envelope::protocolVersion` and announced by `makeHello()`.
/// Bumped only on a **breaking** wire change (a field removal/retype, or a
/// control-flow semantic change) — see "Action-evolution policy" in
/// docs/spec/core/wire.md. A purely additive change (a new optional field on
/// the envelope or on an action/result struct) does not bump this constant.
inline constexpr std::uint32_t kProtocolVersion = 1;
 
/// @brief JSON wire envelope used between any client and `RemoteServer`.
///
/// Supersedes the legacy pipe-delimited protocol: a single struct carries all
/// request and reply variants, with the `kind` field acting as a discriminator.
/// Empty/zero fields are tolerated — callers populate only what their kind needs.
///
/// @par Discriminator values
/// - `"register"`  — client requests model creation. Uses `typeId`, and
///                   optionally `contextKey` (the new instance's stable
///                   identity, e.g. an account id — see `ServerConfig::logProvider`).
///                   With `shared` set, additionally uses `primary` and becomes
///                   a register-or-attach against the shared directory.
/// - `"attach"`    — client re-points at a different `primary` of `typeId`,
///                   releasing `modelId` if non-zero. Replies `ok` with the
///                   target instance's id in `modelId`.
/// - `"assign"`    — client files live `modelId` under `primary` of `typeId`.
/// - `"instances"` — client asks for the live shared primary keys of `typeId`.
///                   Replies `ok` with a JSON array of key strings in `body`.
/// - `"schemas"`   — client asks for the action descriptions of `typeId`.
///                   Replies `ok` with a `{actionType: schema}` JSON object in
///                   `body` — see "Serving action schemas" in
///                   docs/spec/core/wire.md.
/// - `"deregister"` — client destroys an instance. Uses `modelId`.
/// - `"execute"`   — client dispatches an action. Uses `callId`, `modelId`,
///                   `modelType`, `actionType`, `body`, and optionally `session`.
/// - `"ok"`        — server success reply. Uses `callId` if present and carries
///                   the serialized result in `body`. For `register` replies,
///                   `modelId` carries the new id.
/// - `"err"`       — server failure reply. Uses `callId` if present and `message`.
/// - `"hello"`     — client opens protocol-version negotiation, once per
///                   connection before any other kind. Uses `protocolVersion`;
///                   see `makeHello()` and `interpretHelloReply()`. A peer
///                   predating negotiation answers `err "unknown envelope kind:
///                   hello"`, which is how a legacy server is detected.
/// - `"event"`     — server-initiated notification, sent unprompted on a
///                   connection that negotiated events. `callId` is `0`; `body`
///                   is an `EventBody`. A client routes it by kind before reply
///                   matching; see `makeEvent()`.
/// - `"cancel"`    — client asks the server to stop one of its own in-flight
///                   `execute` calls. Uses `cancelCallId` (the execute's
///                   `callId`) and `session`; its own `callId` correlates the
///                   `ok` reply, which is the same whatever became of the call.
///                   Sent only to a server whose `hello` reply advertised
///                   `kCapabilityCancel`; see `makeCancel()`.
struct Envelope {
    /// @brief Discriminator — see class docstring for valid values.
    std::string kind;
 
    /// @brief Correlation id for matching async replies to their requests.
    uint64_t callId = 0;
 
    /// @brief Model type id for `register`.
    std::string typeId;
 
    /// @brief Stable identity of the model instance being registered or
    ///        attached to (e.g. an account id). Empty means "no identity" —
    ///        the server-side holder gets no action log attached even if a
    ///        `LogProvider` is configured. Carried on `register` and
    ///        `attach`; ignored on every other kind.
    std::string contextKey;
 
    /// @brief Primary key of the instance being registered or attached to.
    ///
    /// Carried on `register` (when `shared` is set) and on `attach`, always as
    /// the key's canonical string encoding (`morph::model::keyToString`)
    /// whatever its C++ type, so the server's directory needs one map type
    /// rather than one per key type. Empty means "no primary" — the instance is
    /// anonymous and cannot be shared. Ignored on every other kind.
    ///
    /// Distinct from `contextKey`, which continues to mean only "entity key for
    /// the action log". A keyed model will normally set both to the same value,
    /// but conflating the fields would change behaviour for callers already
    /// setting `contextKey` for journal purposes.
    std::string primary;
 
    /// @brief Whether a `register` should join the shared instance directory.
    ///
    /// `true` makes the request a *register-or-attach*: the server returns the
    /// existing instance for `(typeId, primary)` if one is live, otherwise
    /// creates it and enters it in the directory. `false` (the default, and the
    /// value every pre-existing client sends) is today's behaviour exactly — a
    /// private instance, invisible to the directory.
    bool shared = false;
 
    /// @brief Existing model instance id for `deregister`, `execute`, `ok(register)`.
    uint64_t modelId = 0;
 
    /// @brief Model type id for `execute` (routing key in `ActionDispatcher`).
    std::string modelType;
 
    /// @brief Action type id for `execute` (the second routing key).
    std::string actionType;
 
    /// @brief Serialized JSON of the action payload (`execute`) or of the
    ///        deserializable result (`ok`).
    std::string body;
 
    /// @brief Free-text error message on `err`. Empty on success.
    std::string message;
 
    /// @brief Session context for authorization and routing. Populated on
    ///        `execute`; ignored on every other kind.
    ::morph::session::Context session;
 
    /// @brief Protocol version the sender speaks.
    ///
    /// `0` means "unspecified / legacy peer" — the value on every envelope an
    /// old encoder (unaware of this field) produces, and the value an old
    /// decoder (unaware of this field) leaves untouched when it receives one
    /// from a newer peer (ignored via `decode`'s `error_on_unknown_keys =
    /// false`). Populated by `makeHello()` on `"hello"`; not otherwise
    /// inspected by other `kind`s today.
    std::uint32_t protocolVersion = 0;
 
    /// @brief The `callId` of the `execute` a `"cancel"` asks the server to stop.
    ///
    /// Carried in its own field rather than in `callId` because a `cancel` is
    /// a request with a reply of its own: were the target's id in `callId`, the
    /// server's `ok` to the cancel would be matched by the client against the
    /// still-pending `execute` and settle it with an empty result. `0` names no
    /// call. Ignored on every other kind.
    std::uint64_t cancelCallId = 0;
};
 
/// @brief The capability a server's `"hello"` reply lists when it honours
///        `"cancel"` envelopes.
///
/// A client sends `"cancel"` only to a server that advertised this; a server
/// that predates it never sees one. See "Cancelling a call" in
/// docs/spec/core/wire.md.
inline constexpr std::string_view kCapabilityCancel = "cancel";
 
/// @brief Inclusive protocol-version range a server advertises in reply to
///        `"hello"`.
///
/// Serialized as the `"ok"` reply's `body` (JSON) when a server accepts a
/// `"hello"` — see `ServerConfig::minProtocolVersion`/`maxProtocolVersion`.
struct ProtocolRange {
    /// @brief Oldest protocol version the server accepts.
    std::uint32_t min = kProtocolVersion;
 
    /// @brief Newest protocol version the server accepts.
    std::uint32_t max = kProtocolVersion;
 
    /// @brief Optional envelope kinds the server honours, beyond those every
    ///        server of this protocol version does (e.g. `kCapabilityCancel`).
    ///
    /// Additive: a reply from a server that predates the list has no such key,
    /// and decodes to an empty list — "nothing optional is supported".
    std::vector<std::string> capabilities;
};
 
/// @brief Builds a `register` envelope.
/// @param typeId     Model type id to register.
/// @param contextKey Optional stable identity for the new instance (default: none).
inline Envelope makeRegister(std::string typeId, std::string contextKey = {}) {
    Envelope env;
    env.kind = "register";
    env.typeId = std::move(typeId);
    env.contextKey = std::move(contextKey);
    return env;
}
 
/// @brief Builds a shared (register-or-attach) `register` envelope.
///
/// The server returns the live instance for `(typeId, primary)` if one exists,
/// otherwise creates it and enters it in the shared directory. A shared instance
/// is recorded with no owner principal, so `IAuthorizer::authorizeInstance`'s
/// documented `ownerPrincipal == ctx.principal` policy does not lock the second
/// client out of an instance the first created — see
/// docs/spec/core/shared_instances.md.
///
/// @param typeId     Model type id to register or attach to.
/// @param primary    Canonical string encoding of the instance's primary key.
/// @param contextKey Optional entity key for the action log (default: none).
inline Envelope makeRegisterShared(std::string typeId, std::string primary, std::string contextKey = {}) {
    Envelope env;
    env.kind = "register";
    env.typeId = std::move(typeId);
    env.primary = std::move(primary);
    env.contextKey = std::move(contextKey);
    env.shared = true;
    return env;
}
 
/// @brief Builds an `attach` envelope — re-points a client at a different primary.
///
/// Semantically a `deregister` + shared `register` pair, made a single request
/// so a re-pointing handler cannot lose its slot to `LimitPolicy::maxLiveModels`
/// between releasing the old instance and acquiring the new one.
///
/// @param typeId     Model type id.
/// @param primary    Canonical string encoding of the primary key to attach to.
/// @param modelId    Instance the client is currently attached to; `0` if none.
/// @param contextKey Optional entity key for the action log, carried the same
///                    way `register` carries it, so an instance created via
///                    its *first* `attach` (rather than a shared `register`)
///                    still gets a configured `LogProvider`'s log attached.
inline Envelope makeAttach(std::string typeId, std::string primary, uint64_t modelId = 0,
                           std::string contextKey = {}) {
    Envelope env;
    env.kind = "attach";
    env.typeId = std::move(typeId);
    env.primary = std::move(primary);
    env.modelId = modelId;
    env.contextKey = std::move(contextKey);
    env.shared = true;
    return env;
}
 
/// @brief Builds an `assign` envelope — files a live instance under a primary key.
///
/// The promotion half of keyed instances: an action that creates its own entity
/// runs on a not-yet-keyed instance, and only the reply carries the generated
/// key. Assigning promotes that same instance in place, so nothing the create
/// did is stranded on a throwaway.
/// @param typeId  Model type id.
/// @param primary Canonical string encoding of the key to file the instance under.
/// @param modelId Live instance to promote.
inline Envelope makeAssign(std::string typeId, std::string primary, uint64_t modelId) {
    Envelope env;
    env.kind = "assign";
    env.typeId = std::move(typeId);
    env.primary = std::move(primary);
    env.modelId = modelId;
    env.shared = true;
    return env;
}
 
/// @brief Builds an `instances` envelope — asks for the live shared keys of a type.
///
/// The reply's `body` is a JSON array of canonical key strings. Only instances
/// created through a shared `register`/`attach` are listed; a private instance
/// is invisible to the directory by construction.
/// @param typeId Model type id to enumerate.
inline Envelope makeInstances(std::string typeId) {
    Envelope env;
    env.kind = "instances";
    env.typeId = std::move(typeId);
    return env;
}
 
/// @brief Builds a `schemas` envelope — asks for a model type's action descriptions.
///
/// The reply's `body` is a `{actionType: schema}` JSON object: one JSON Schema
/// document per action registered for @p typeId, each carrying that action's
/// `x-payloadFingerprint` and `x-payloadShape`
/// (`morph::model::payloadFingerprint` / `payloadShapeString`) alongside the
/// usual form metadata. A client not linked against the model's C++ has no
/// other way to learn an action's shape — see "Serving action schemas" in
/// docs/spec/core/wire.md.
/// @param typeId Model type id whose action descriptions are wanted.
inline Envelope makeSchemas(std::string typeId) {
    Envelope env;
    env.kind = "schemas";
    env.typeId = std::move(typeId);
    return env;
}
 
/// @brief Builds a `cancel` envelope asking the server to stop the in-flight
///        `execute` filed under @p targetCallId on the sender's connection.
///
/// The caller still assigns the envelope's own `callId` and `session` — the
/// same session the `execute` carried, since the server honours a cancel only
/// from the principal and connection that made the call.
/// @param targetCallId `callId` of the `execute` to stop.
/// @return The envelope, with `cancelCallId` set.
inline Envelope makeCancel(uint64_t targetCallId) {
    Envelope env;
    env.kind = "cancel";
    env.cancelCallId = targetCallId;
    return env;
}
 
/// @brief Builds a `deregister` envelope.
inline Envelope makeDeregister(uint64_t modelId) {
    Envelope env;
    env.kind = "deregister";
    env.modelId = modelId;
    return env;
}
 
/// @brief Builds an `ok` reply envelope. `body` carries the result for executes,
///        `modelId` carries the new id for register-replies.
inline Envelope makeOk(uint64_t callId = 0, std::string body = {}, uint64_t modelId = 0) {
    Envelope env;
    env.kind = "ok";
    env.callId = callId;
    env.body = std::move(body);
    env.modelId = modelId;
    return env;
}
 
namespace detail {
 
/// @brief Replaces every ASCII control byte (0x00-0x1F, 0x7F) in @p text with a
///        printable `\xHH` textual placeholder.
///
/// Applied to `makeErr`'s `message` only, and no longer for JSON validity —
/// `detail::EscapingWriteOpts` now handles that for every field. What remains is
/// an output-sanitization concern specific to this one field: an `err` message
/// echoes untrusted content back (an unrecognized `Envelope::kind`, a caught
/// exception's `what()`) and is overwhelmingly destined for a log or a console,
/// where a raw 0x1B would carry an ANSI escape sequence into the reader's
/// terminal. `message` is diagnostic text rather than data that must round-trip
/// byte-for-byte, so replacing these bytes with a printable transcription costs
/// nothing and keeps them inert.
///
/// Originally added for the invalid-JSON failure found by the
/// `fuzz_dispatch_execute` harness via an `err` reply echoing an unrecognized
/// `kind` verbatim (see docs/spec/testing_strategy.md, "Known findings"); that
/// finding is now covered at the writer instead.
/// @param text Untrusted text that may contain raw control bytes.
/// @return @p text with every control byte replaced by `\xHH`.
[[nodiscard]] inline std::string sanitizeControlChars(std::string_view text) {
    std::string out;
    out.reserve(text.size());
    for (char c : text) {
        auto byte = static_cast<unsigned char>(c);
        if (byte < 0x20 || byte == 0x7F) {
            static constexpr char kHex[] = "0123456789abcdef";
            out += "\\x";
            out += kHex[(byte >> 4) & 0xF];
            out += kHex[byte & 0xF];
        } else {
            out += c;
        }
    }
    return out;
}
 
/// @brief Write options that escape ASCII control bytes as `\\uXXXX` sequences.
///
/// glaze 7.4 leaves control bytes unescaped by default, which breaks `encode`
/// two different ways depending on where the byte lands:
///
/// - **Invalid output.** RFC 8259 requires U+0000 through U+001F to be escaped,
///   and glaze's own reader enforces that, so an envelope carrying a raw 0x0B
///   anywhere serializes to JSON the peer's `decode` then throws on.
/// - **Silent corruption.** Worse, the writer's chunked fast path mangles such a
///   byte outright once the string also contains an escaped character: with a
///   backslash or double-quote character earlier in the same string, a 0x0B at certain offsets is written
///   as *two* 0x00 bytes. The payload is destroyed before it reaches the wire,
///   so no amount of post-processing on the serialized form can recover it —
///   the escaping has to happen inside the writer.
///
/// Enabling the option is glaze's documented remedy for exactly this (see its
/// README, "escape control characters"); it is off by default there only because
/// embedded nulls are hazardous for C APIs, which does not apply to a
/// `std::string` wire field. Escaping is lossless in both directions, so any
/// such byte round-trips unchanged rather than being replaced.
///
/// Applies to writing only. `decode` needs no counterpart: glaze's reader
/// already accepts `\\uXXXX` and turns it back into the original byte.
struct EscapingWriteOpts : glz::opts {
    /// @brief Emit control bytes as `\\uXXXX` rather than raw.
    // NOLINTNEXTLINE(readability-identifier-naming) — the name is glaze's, not ours; the option is matched by name.
    bool escape_control_characters = true;
};
 
/// @brief The `Envelope` keys `encode` omits when the member holds its default.
///
/// `Envelope` is a union-of-all-kinds struct: an `ok` reply uses three of its
/// thirteen members and a `deregister` request uses two, but glaze writes every
/// member of a struct it is handed, so without this a minimal reply carrying an
/// 8-byte payload costs 255 bytes on the wire, 213 of them fields the kind does
/// not use.
///
/// Omitting a member that holds its default is **not** a protocol change. The
/// decoder default-initialises the `Envelope` and reads with
/// `error_on_unknown_keys = false`; a key that is absent therefore leaves the
/// member at exactly the value the omitted key would have written. The
/// `kProtocolVersion` bump that `docs/spec/core/wire.md` requires for a field
/// removal or retype does not apply, because nothing is removed or retyped —
/// see "Omitted default fields" there, and `test_wire_omitted_fields.cpp`,
/// which decodes both forms and requires the results to be equal.
///
/// **Two fields are deliberately absent from this list, and for the same
/// reason: their default is not "unset".** The rule the rest of the list obeys
/// is that omitting a member is safe when its default means *nothing* — when
/// "absent" and "present, holding the default" are the same statement. That
/// holds for the payload and addressing fields: an empty `typeId` means no type
/// id, a `modelId` of `0` means no model id (`remote.hpp` tests exactly that),
/// and a `protocolVersion` of `0` is documented on the member as "unspecified /
/// legacy peer" — precisely what an encoder that had never heard of the field
/// would produce. For these two it does not hold:
///
/// - **`kind`** is the discriminator; an envelope without one is malformed.
///   Keeping it unconditional also means the first key of every message is
///   still `"kind"`.
/// - **`callId` is a correlation field, and `0` is a value rather than an
///   absence.** Both transports route on it: a reply is matched against the
///   pending request filed under its `callId`, and `0` names none, so such a
///   reply is dropped (`QtWebSocketBackend::onTextMessage`,
///   `SocketBackend::dispatchIncomingEnvelope`). A peer therefore cannot treat
///   an absent `callId` as "no information"; it has to *reconstruct* the
///   value before it can route the frame at all. morph's own `decode` does
///   that for free by default-initialising, so omitting it round-trips cleanly
///   in C++ and still breaks the scenario driver, a second decoder that reads
///   absence as `None`. Emitting correlation
///   fields unconditionally costs eleven bytes on the one message shape that
///   carries a zero id, and keeps "how do I route this frame" answerable from
///   the frame.
///
/// The general form of the rule, for anyone adding a member: **elide payload,
/// never elide identity.** If a peer has to know the field's value to decide
/// where the message goes, it is written even at its default.
inline constexpr std::array<std::string_view, 12> kOmittableEnvelopeKeys{
    "typeId",     "contextKey", "primary", "shared",  "modelId",         "modelType",
    "actionType", "body",       "message", "session", "protocolVersion", "cancelCallId"};
 
/// @brief Whether a session context carries nothing worth sending.
///
/// @param ctx Session context to inspect.
/// @return `true` when every member is empty, so the whole `session` object can
///         be left out of the envelope.
[[nodiscard]] inline bool isDefaultSession(const ::morph::session::Context& ctx) noexcept {
    // A field-by-field test rather than `ctx == Context{}`, because `Context` is
    // a plain aggregate with no `operator==` and giving it one would reach
    // outside this header. The assertion is what stops the two drifting: a new
    // member would otherwise be dropped from every envelope whose other session
    // fields happen to be empty, silently and only on the wire.
    static_assert(glz::reflect<::morph::session::Context>::size == 5,
                  "session::Context gained or lost a member: update isDefaultSession, or the new "
                  "member is silently dropped from every envelope with an otherwise-empty session.");
    return ctx.principal.empty() && ctx.token.empty() && ctx.requestId.empty() && ctx.locale.empty() &&
           ctx.metadata.empty();
}
 
/// @brief Collects the keys `encode` should leave out of @p env.
///
/// @param env Envelope about to be serialised.
/// @param out Storage for the key list; only the returned prefix is meaningful.
/// @return The keys to exclude, as a view into @p out.
[[nodiscard]] inline std::span<const std::string_view> defaultValuedKeys(
    const Envelope& env, std::array<std::string_view, kOmittableEnvelopeKeys.size()>& out) noexcept {
    // Every name below must be a real reflected key of `Envelope`: glaze's
    // exclude-write reports `unknown_key` for one that is not, which would turn
    // a member rename into a throw from *every* `encode` call at runtime. This
    // makes it a compile error instead.
    static_assert(glz::reflect<Envelope>::size == 14,
                  "Envelope gained or lost a member: decide whether it is omittable when default "
                  "and update kOmittableEnvelopeKeys. A new member left off that list is always "
                  "written, which is merely the old behaviour; one added with a wrong spelling is "
                  "caught by the assertion below.");
    static_assert(std::ranges::all_of(kOmittableEnvelopeKeys,
                                      [](std::string_view key) {
                                          return std::ranges::find(glz::reflect<Envelope>::keys, key) !=
                                                 std::ranges::end(glz::reflect<Envelope>::keys);
                                      }),
                  "kOmittableEnvelopeKeys names a key Envelope does not have — a rename, or a typo.");
 
    std::size_t count = 0;
    auto omit = [&out, &count](std::string_view key) noexcept { out.at(count++) = key; };
 
    // No `callId` arm, deliberately: see `kOmittableEnvelopeKeys`. It is a
    // correlation field whose zero is a routing sentinel, so it is written
    // unconditionally, exactly like `kind`.
    if (env.typeId.empty()) {
        omit("typeId");
    }
    if (env.contextKey.empty()) {
        omit("contextKey");
    }
    if (env.primary.empty()) {
        omit("primary");
    }
    if (!env.shared) {
        omit("shared");
    }
    if (env.modelId == 0) {
        omit("modelId");
    }
    if (env.modelType.empty()) {
        omit("modelType");
    }
    if (env.actionType.empty()) {
        omit("actionType");
    }
    if (env.body.empty()) {
        omit("body");
    }
    if (env.message.empty()) {
        omit("message");
    }
    if (isDefaultSession(env.session)) {
        omit("session");
    }
    if (env.protocolVersion == 0) {
        omit("protocolVersion");
    }
    if (env.cancelCallId == 0) {
        omit("cancelCallId");
    }
    return std::span<const std::string_view>{out.data(), count};
}
 
/// @brief Best-effort recovery of an envelope's `callId` without decoding it.
///
/// For the one case where a transport must answer a message it has decided
/// *not* to parse — a frame rejected for exceeding a transport-level size cap,
/// before `decode` is ever called. Such a reply still has to be addressed: a
/// client files every request under its `callId` and matches the reply to it,
/// and a reply carrying `0` matches nothing, so the client drops it and the
/// request it was meant for stays pending until the connection ends.
///
/// Scans at most @p maxScanBytes, so the size cap it serves keeps its value as
/// a cost bound; `callId` is the second field `encode` writes and is never
/// omitted (see `kOmittableEnvelopeKeys`), so for any envelope morph encoded it
/// lands well inside even a small window. A `"callId":` sequence cannot be
/// forged from within an earlier string field, because `encode` escapes any
/// embedded quote (yielding `\"callId\":`, which does not match).
///
/// The `0`-on-absence return is load-bearing: the input here is by definition a frame that was *not* produced under
/// morph's own invariants — it is whatever a peer sent, at a size this side already refused to parse — so "no `callId`
/// in the first KiB" remains a case that has to have an answer.
///
/// @param json         Raw, undecoded envelope text.
/// @param maxScanBytes Prefix length to search. Default 1 KiB.
/// @return The parsed `callId`, or `0` if absent, unparseable, or out of range —
///         an id that matches no pending call.
[[nodiscard]] inline std::uint64_t peekCallId(std::string_view json, std::size_t maxScanBytes = 1024) noexcept {
    static constexpr std::string_view kKey = "\"callId\":";
    const std::string_view window = json.substr(0, std::min(json.size(), maxScanBytes));
    const auto keyPos = window.find(kKey);
    if (keyPos == std::string_view::npos) {
        return 0;
    }
    std::size_t idx = keyPos + kKey.size();
    // idx is bounded by the same condition that guards the access.
    // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-avoid-unchecked-container-access)
    while (idx < window.size() && (window[idx] == ' ' || window[idx] == '\t')) {
        ++idx;
    }
    // No digits leaves `value` at 0, which is already the "no id" answer.
    std::uint64_t value = 0;
    for (; idx < window.size(); ++idx) {
        // idx is bounded by the loop condition.
        // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-avoid-unchecked-container-access)
        const char chr = window[idx];
        if (chr < '0' || chr > '9') {
            break;
        }
        const auto digit = static_cast<std::uint64_t>(chr - '0');
        if (value > (std::numeric_limits<std::uint64_t>::max() - digit) / 10U) {
            return 0;  // out of range — treat as unrecoverable rather than wrap
        }
        value = (value * 10U) + digit;
    }
    return value;
}
 
}  // namespace detail
 
/// @brief Builds an `err` reply envelope.
///
/// @p message may embed untrusted content (e.g. an unrecognized `Envelope::kind`
/// or a caught exception's `what()`) — any raw control byte it contains is
/// replaced with a `\xHH` placeholder. This is *output sanitization*, not JSON
/// validity: `encode`'s `detail::EscapingWriteOpts` already guarantees validity
/// for every field. What this stops is a raw control byte reaching a terminal or
/// log that interprets it (see `detail::sanitizeControlChars`).
inline Envelope makeErr(std::string message, uint64_t callId = 0) {
    Envelope env;
    env.kind = "err";
    env.callId = callId;
    env.message = detail::sanitizeControlChars(message);
    return env;
}
 
/// @brief The envelope kind of a server-initiated event.
inline constexpr std::string_view kEventKind = "event";
 
/// @brief The decoded `body` of an `event` envelope.
struct EventBody {
    /// @brief What the event is about (`changes`, `resync`).
    std::string topic;
    /// @brief The topic's payload: a JSON value, carried verbatim.
    glz::raw_json payload;
};
 
/// @brief Builds an `event` envelope: `callId` 0, `body` = `{"topic":…,"payload":…}`.
///
/// @param topic       Event topic.
/// @param payloadJson The payload as a JSON value; embedded verbatim, so it
///                    must be valid JSON (an empty string is sent as `null`).
/// @return The event envelope.
[[nodiscard]] inline Envelope makeEvent(std::string_view topic, std::string payloadJson) {
    Envelope env;
    env.kind = std::string{kEventKind};
    env.callId = 0;
    if (payloadJson.empty()) {
        payloadJson = "null";
    }
    EventBody body{.topic = std::string{topic}, .payload = glz::raw_json{std::move(payloadJson)}};
    env.body = glz::write_json(body).value_or(std::string{});
    return env;
}
 
/// @brief The `changes` event topic: a `ChangesBody` payload naming what changed.
inline constexpr std::string_view kChangesTopic = "changes";
 
/// @brief The `resync` event topic: events were withheld, so a client re-reads what it shows.
inline constexpr std::string_view kResyncTopic = "resync";
 
/// @brief The envelope kind that adds topics to a connection's watch set (or replaces it).
inline constexpr std::string_view kWatchKind = "watch";
 
/// @brief The envelope kind that removes topics from a connection's watch set.
inline constexpr std::string_view kUnwatchKind = "unwatch";
 
/// @brief The most topics (records or instances) one `watch`, `unwatch` or `changes` body may list, and
///        the most topics, records and instances together, one connection may watch.
///
/// A watch set is one entry per record type and open instance of a screen, so a list past this is a
/// malformed or hostile peer rather than a large application, and the bound keeps its allocation finite.
inline constexpr std::size_t kMaxWatchTopics = 4096;
 
/// @brief The longest record name, model type or primary key text in a watch or changes body.
///
/// The most keys one record entry of a `changes` body may list is `effects::kMaxChangeKeys`.
inline constexpr std::size_t kMaxTopicNameBytes = 1024;
 
/// @brief One instance topic: a model type, its directory key text and its instance id.
struct InstanceTopic {
    /// @brief The model's registered type name.
    std::string modelType;
    /// @brief The directory key's text for a keyed shared instance; empty for a private one.
    std::string primary;
    /// @brief The instance id; 0 where the topic is named by type and primary alone (a request).
    std::uint64_t instance = 0;
};
 
/// @brief The body of a `watch` or `unwatch` request.
struct WatchBody {
    /// @brief Record wire names to watch (or stop watching).
    std::vector<std::string> records;
    /// @brief Instance topics to watch (or stop watching).
    std::vector<InstanceTopic> instances;
    /// @brief For `watch`: replace the connection's whole watch set instead of adding to it.
    bool replace = false;
};
 
/// @brief The body of the `ok` reply to a `watch`: the topics the server refused.
struct WatchReply {
    /// @brief Record names the server does not publish or the caller may not read.
    std::vector<std::string> refusedRecords;
    /// @brief Instance topics the server refused.
    std::vector<InstanceTopic> refusedInstances;
};
 
/// @brief The payload of a `changes` event.
struct ChangesBody {
    /// @brief Record changes; a record with no `keys` changed as a whole.
    std::vector<effects::RecordChange> records;
    /// @brief Instance changes.
    std::vector<InstanceTopic> instances;
};
 
namespace detail {
 
template <class T>
[[nodiscard]] inline std::expected<T, std::string> readBody(std::string_view json, std::string_view what) {
    if (json.size() > kMaxEnvelopeBytes) {
        return std::unexpected(std::string{what} + " exceeds " + std::to_string(kMaxEnvelopeBytes) + " bytes");
    }
    T out{};
    static constexpr glz::opts kLenient{.null_terminated = false, .error_on_unknown_keys = false};
    if (auto const error = glz::read<kLenient>(out, json)) {
        return std::unexpected(std::string{what} + ": " + glz::format_error(error, json));
    }
    return out;
}
 
// NOLINTNEXTLINE(bugprone-easily-swappable-parameters): a name, then how a refusal describes it; swapped, only the message reads oddly.
[[nodiscard]] inline std::optional<std::string> checkTopicName(std::string_view name, std::string_view what) {
    if (name.size() > kMaxTopicNameBytes) {
        return std::string{what} + " is longer than " + std::to_string(kMaxTopicNameBytes) + " bytes";
    }
    return std::nullopt;
}
 
[[nodiscard]] inline std::optional<std::string> checkNames(const std::vector<std::string>& names,
                                                           std::string_view what) {
    if (names.size() > kMaxWatchTopics) {
        return std::string{what} + " list more than " + std::to_string(kMaxWatchTopics) + " topics";
    }
    for (const auto& name : names) {
        if (auto reason = checkTopicName(name, what)) {
            return reason;
        }
    }
    return std::nullopt;
}
 
[[nodiscard]] inline std::optional<std::string> checkInstances(const std::vector<InstanceTopic>& topics,
                                                               std::string_view what) {
    if (topics.size() > kMaxWatchTopics) {
        return std::string{what} + " list more than " + std::to_string(kMaxWatchTopics) + " topics";
    }
    for (const auto& topic : topics) {
        if (auto reason = checkTopicName(topic.modelType, "a model type")) {
            return reason;
        }
        if (auto reason = checkTopicName(topic.primary, "a primary key")) {
            return reason;
        }
    }
    return std::nullopt;
}
 
}  // namespace detail
 
/// @brief Reads the body of a `watch` or `unwatch` request.
///
/// Refused: input longer than `kMaxEnvelopeBytes`, text that is not a `WatchBody`, more than
/// `kMaxWatchTopics` records or instances, or a name longer than `kMaxTopicNameBytes`. Unknown keys are
/// ignored.
/// @param json The envelope `body`.
/// @return The body, or the reason it was refused.
[[nodiscard]] inline std::expected<WatchBody, std::string> readWatchBody(std::string_view json) {
    auto body = detail::readBody<WatchBody>(json, "the watch body");
    if (!body) {
        return body;
    }
    if (auto reason = detail::checkNames(body->records, "a record name")) {
        return std::unexpected(*reason);
    }
    if (auto reason = detail::checkInstances(body->instances, "an instance topic")) {
        return std::unexpected(*reason);
    }
    return body;
}
 
/// @brief Reads the body of the `ok` reply to a `watch`.
///
/// Bounded as `readWatchBody` is.
/// @param json The envelope `body`.
/// @return The reply, or the reason it was refused.
[[nodiscard]] inline std::expected<WatchReply, std::string> readWatchReply(std::string_view json) {
    auto body = detail::readBody<WatchReply>(json, "the watch reply");
    if (!body) {
        return body;
    }
    if (auto reason = detail::checkNames(body->refusedRecords, "a record name")) {
        return std::unexpected(*reason);
    }
    if (auto reason = detail::checkInstances(body->refusedInstances, "an instance topic")) {
        return std::unexpected(*reason);
    }
    return body;
}
 
/// @brief Reads the payload of a `changes` event.
///
/// Refused: input longer than `kMaxEnvelopeBytes`, text that is not a `ChangesBody`, more than
/// `kMaxWatchTopics` records or instances, more than `effects::kMaxChangeKeys` keys in one record, or a name longer
/// than `kMaxTopicNameBytes`.
/// @param json The event's `payload`.
/// @return The payload, or the reason it was refused.
[[nodiscard]] inline std::expected<ChangesBody, std::string> readChangesBody(std::string_view json) {
    auto body = detail::readBody<ChangesBody>(json, "the changes payload");
    if (!body) {
        return body;
    }
    if (body->records.size() > kMaxWatchTopics) {
        return std::unexpected("a record list more than " + std::to_string(kMaxWatchTopics) + " topics");
    }
    for (const auto& change : body->records) {
        if (auto reason = detail::checkTopicName(change.record, "a record name")) {
            return std::unexpected(*reason);
        }
        if (change.keys && change.keys->size() > effects::kMaxChangeKeys) {
            return std::unexpected("a record's keys list more than " + std::to_string(effects::kMaxChangeKeys) +
                                   " keys");
        }
    }
    if (auto reason = detail::checkInstances(body->instances, "an instance topic")) {
        return std::unexpected(*reason);
    }
    return body;
}
 
/// @brief The `err` reply message `RemoteServer` sends when
///        `LimitPolicy::executeTimeout` fires server-side (see
///        `RemoteServer::dispatchExecute`'s `_executeTimeouts` path).
///
/// `"timeout"` exactly is the documented wire contract (`docs/spec/core/
/// backend.md`'s `executeTimeout` row, `docs/spec/core/completion.md`'s
/// `TimeoutError` row, `docs/spec/error_handling.md`'s `"timeout"` row) —
/// real clients on the wire, not just this codebase's own backends, match
/// against this literal, so it cannot change. Callers matching a reply
/// against it should still use this constant rather than a hand-typed
/// literal, purely to keep the one producer and its consumers from drifting
/// apart by typo; it does not, on its own, rule out an application
/// exception whose `what()` happens to read exactly "timeout" being
/// misclassified the same way (see `makeErr`'s docs above on `message`
/// carrying caught exception text) — that is a narrow, pre-existing
/// ambiguity inherent to a free-text wire field, not something this
/// constant can close without breaking the documented contract.
inline constexpr std::string_view kExecuteTimeoutMessage = "timeout";
 
/// @brief Builds a `hello` envelope announcing the sender's protocol version.
///
/// Exchanged once per connection, before any `register`/`execute`. See
/// "Protocol version negotiation" in docs/spec/core/wire.md.
/// @param protocolVersion Protocol version the sender speaks (default: `kProtocolVersion`).
inline Envelope makeHello(std::uint32_t protocolVersion = kProtocolVersion) {
    Envelope env;
    env.kind = "hello";
    env.protocolVersion = protocolVersion;
    return env;
}
 
/// @brief The serialisation primitive `encode` calls, as an injectable strategy.
///
/// Same shape and rationale as `morph::core::FileIoOps`: the member defaults
/// to the real call it stands in for, so a default-constructed `WireCodecOps`
/// is byte-for-byte what `encode` did before this seam existed.
///
/// @par Why this exists
/// `encode`'s write-failure arm is unreachable through any `Envelope` value.
/// Glaze only sets a write-time error for `invalid_partial_key`/`unknown_key`
/// (its partial-write-by-key-list feature, which `encode` does not use) or
/// `invalid_variant_object` (a `std::variant` member, which `Envelope` does
/// not have). Confirmed by experiment as well as by reading glaze: five
/// flavours of invalid UTF-8, an embedded NUL, a raw control byte and an 8 MiB
/// payload all encode successfully.
///
/// That leaves a branch guarding a real invariant with no way to reach it.
/// Rather than reshape `Envelope` to make a test possible, or wait on a
/// fault-injection hook glaze does not offer, this puts the seam on morph's
/// side of the boundary — where the repository already puts it for the
/// identical problem with file I/O.
struct WireCodecOps {
    /// @brief Serialises @p env into @p out. Mirrors the `glz::write` call
    ///        `encode` would otherwise make directly.
    /// @return Glaze's error context; falsy on success.
    std::function<glz::error_ctx(const Envelope& env, std::string& out)> writeEnvelope = [](const Envelope& env,
                                                                                            std::string& out) {
        std::array<std::string_view, detail::kOmittableEnvelopeKeys.size()> omitted{};
        return glz::write_json_exclude<detail::EscapingWriteOpts{}>(env, detail::defaultValuedKeys(env, omitted), out);
    };
};
 
/// @brief The shared default `WireCodecOps`.
///
/// A function-local static, not a default-constructed temporary in `encode`'s
/// signature: `encode` runs on every outbound message, and building a fresh
/// `std::function` per call would put an allocation on that path purely to
/// support a test seam.
/// @return Reference to the process-wide default operations.
[[nodiscard]] inline const WireCodecOps& defaultWireCodecOps() {
    static const WireCodecOps ops;
    return ops;
}
 
/// @brief Encodes @p env as a single JSON line.
///
/// Writes with `detail::EscapingWriteOpts`, so raw ASCII control bytes in any
/// string field become `\uXXXX` escapes. Without it, a single 0x0B anywhere in
/// `body`, `modelType`, `contextKey`, `typeId`, `actionType`, or the session's
/// `principal`/`token` yields an envelope the peer's `decode` throws on — and,
/// where the same string also contains a `\` or `"`, one glaze silently
/// corrupts before it is ever sent. Both are payload-triggered transport
/// failures, not caller errors. The escape is lossless, so such bytes still
/// round-trip byte-for-byte.
///
/// @param env Envelope to serialize.
/// @param ops Serialisation strategy; defaults to the real `glz::write` call.
///        A test injects a failing one to exercise the throw path, which no
///        `Envelope` value can reach — see `WireCodecOps`.
/// @return The serialized envelope as valid, re-decodable JSON.
/// @throws std::runtime_error on serialisation failure (should never happen for valid input).
inline std::string encode(const Envelope& env, const WireCodecOps& ops = defaultWireCodecOps()) {
    MORPH_ZONE("wire::encode");
    MORPH_ZONE_TEXT(env.session.requestId);
    std::string out;
    if (auto errCode = ops.writeEnvelope(env, out)) {
        // Unreachable through any `Envelope` value -- see `WireCodecOps` for
        // why, and for how a test covers this arm anyway.
        throw std::runtime_error("envelope encode failed: " + glz::format_error(errCode, out));
    }
    return out;
}
 
/// @brief Decodes @p json into an `Envelope`.
///
/// Enforces a wire-layer size cap: inputs longer than `kMaxEnvelopeBytes` are
/// rejected before any parsing, bounding the cost of an oversized or
/// deeply-nested payload smuggled inside the opaque `body` string (see
/// `kMaxEnvelopeBytes`). This guard does **not** cap the *inner* re-parse of
/// `body` performed later by the action codec — that codec must impose its own
/// limits — but it does bound the total message a single `decode` will accept.
///
/// glaze 7.4 runs with its default options here. Note the parser does **not**
/// reject duplicate JSON object keys: a duplicated key is accepted and the last
/// occurrence wins. There is no glaze option to change this, so `decode` cannot
/// enforce rejection; callers must not treat duplicate-key rejection as a
/// security boundary (see docs/spec/core/wire.md).
///
/// @param json Serialized envelope JSON to decode.
/// @return The decoded `Envelope`.
/// @throws std::runtime_error if @p json exceeds `kMaxEnvelopeBytes` or is not a
///         valid envelope.
inline Envelope decode(std::string_view json) {
    MORPH_ZONE("wire::decode");
    if (json.size() > kMaxEnvelopeBytes) {
        throw std::runtime_error("envelope decode failed: input exceeds maximum size (" + std::to_string(json.size()) +
                                 " > " + std::to_string(kMaxEnvelopeBytes) + " bytes)");
    }
    Envelope env{};
    // Ignore unknown keys so the envelope is forward-compatible: a newer peer may
    // add fields a older peer does not know, and vice versa, without a hard parse
    // failure. Duplicate keys are still accepted last-wins (glaze offers no reject
    // option) — see docs/spec/core/wire.md "Parsing guarantees and hardening".
    // `null_terminated = false` is required, not cosmetic: `json` is a caller-supplied
    // `string_view` with no guarantee of a trailing '\0' (e.g. a raw socket read).
    // Left at glaze's default (true), `skip_ws` assumes it can scan past `end`
    // looking for a terminator that may not exist — a heap-buffer-overflow on
    // adversarial input (found by the wire_decode fuzz harness, see
    // docs/spec/testing_strategy.md). This costs nothing: it only disables an
    // optimization that never applied to us.
    static constexpr glz::opts kLenient{.null_terminated = false, .error_on_unknown_keys = false};
    if (auto errCode = glz::read<kLenient>(env, json)) {
        throw std::runtime_error("envelope decode failed: " + glz::format_error(errCode, json));
    }
    MORPH_ZONE_TEXT(env.session.requestId);
    return env;
}
 
/// @brief Outcome of a `"hello"` protocol-version negotiation attempt.
enum class ProtocolNegotiationResult : std::uint8_t {
    /// @brief The peer understood `"hello"` and accepted the announced version.
    Negotiated,
    /// @brief The peer does not understand `"hello"` (a pre-negotiation legacy peer).
    LegacyPeer,
};
 
/// @brief Classifies a decoded reply to a `"hello"` envelope.
///
/// @param reply Decoded reply envelope — the result of `decode()` on the
///        response to a `"hello"` round-trip.
/// @return `Negotiated` if @p reply's `kind` is `"ok"`; `LegacyPeer` if it is
///         an envelope whose `message` is exactly `"unknown envelope kind:
///         hello"` — the generic unrecognised-`kind` message a pre-negotiation
///         `RemoteServer` produces for a `kind` it does not switch on.
/// @throws std::runtime_error if @p reply is any other `"err"` (e.g.
///         `"protocol version unsupported"`), so the caller refuses to
///         proceed instead of silently treating an explicit rejection as a
///         legacy peer.
inline ProtocolNegotiationResult interpretHelloReply(const Envelope& reply) {
    if (reply.kind == "ok") {
        return ProtocolNegotiationResult::Negotiated;
    }
    // Matched on `message` alone: `kind` is deliberately not inspected, because a
    // peer old enough to not know `hello` is also a peer whose error shape we do
    // not want to depend on. Any envelope carrying exactly this message is
    // treated as a legacy peer.
    if (reply.message == "unknown envelope kind: hello") {
        return ProtocolNegotiationResult::LegacyPeer;
    }
    throw std::runtime_error("protocol negotiation failed: " +
                             (reply.message.empty() ? std::string{"malformed reply"} : reply.message));
}
 
/// @brief Whether a reply to `"hello"` advertises @p capability.
///
/// Reads the `ProtocolRange` in an `"ok"` reply's `body`, ignoring keys it does
/// not know. Anything else — an `"err"`, a legacy peer's refusal, a body that
/// does not parse, a server that predates the capability list — advertises
/// nothing, so a client falls back to not using the capability rather than to
/// sending a kind the server may not understand.
/// @param reply      Decoded reply to a `"hello"` envelope.
/// @param capability Capability to look for, e.g. `kCapabilityCancel`.
/// @return `true` only when the reply is `"ok"` and its capability list names
///         @p capability.
[[nodiscard]] inline bool helloAdvertises(const Envelope& reply, std::string_view capability) {
    if (reply.kind != "ok") {
        return false;
    }
    ProtocolRange range;
    static constexpr glz::opts kLenient{.null_terminated = false, .error_on_unknown_keys = false};
    if (glz::read<kLenient>(range, reply.body)) {
        return false;
    }
    return std::ranges::find(range.capabilities, capability) != range.capabilities.end();
}
 
}  // namespace morph::wire