From cbb336f501ac781afa83bf87c6ece79390377897 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 22:03:27 +0000 Subject: [PATCH 1/2] cultmesh-kotlin: decode write_admitted.v0 and error.v1, content hash, WriteTracker, vector replay Strict decode of the two receipt frames (and error.v0), hello capability test, the dedupe content hash and the client settle rules, ported from cultnet-ts. write-receipt-vectors.json replays 38/38 (constants, contentHash, wire, schemaInvalid, scenarios, v0 limit_reached); dedupe and blankFields are server inputs and not run. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultmesh-kotlin/README.md | 11 + packages/cultmesh-kotlin/build.ps1 | 3 + .../kotlin/org/gamecult/cultmesh/CultMesh.kt | 1 + .../org/gamecult/cultmesh/WriteReceipt.kt | 232 ++++++++++++++++++ .../gamecult/cultmesh/WriteReceiptVectors.kt | 224 +++++++++++++++++ 5 files changed, 471 insertions(+) create mode 100644 packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceipt.kt create mode 100644 packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceiptVectors.kt diff --git a/packages/cultmesh-kotlin/README.md b/packages/cultmesh-kotlin/README.md index 3da68679..ff69c346 100644 --- a/packages/cultmesh-kotlin/README.md +++ b/packages/cultmesh-kotlin/README.md @@ -474,3 +474,14 @@ powershell -ExecutionPolicy Bypass -File .\packages\cultmesh-kotlin\build.ps1 The build writes `artifacts/cultmesh-kotlin/cultmesh-kotlin.jar`. It also runs the built-in RUDP packet fixture and localhost UDP socket self-test. + +## Write Receipt + +`WriteReceipt.kt` decodes `cultnet.write_admitted.v0` and `cultnet.error.v1` (and `error.v0`) strictly against their +closed schemas (`decodeWriteAdmitted`, `decodeErrorFrame`), computes the dedupe `writeContentHash`, tests a hello with +`helloAdvertisesWriteReceipts`, and settles sent writes as admitted, refused or unconfirmed in `WriteTracker` +(pure state machine, abstract clock; the caller owns transport and timers). Kotlin has no write client yet, so nothing +sends a put and waits on this. `WriteReceiptVectors.kt` replays +`contracts/cultnet/interop/write-receipt-vectors.json` (constants, contentHash, wire, schemaInvalid, scenarios; the +server-side `dedupe` and `blankFields` sections are not run); `build.ps1` runs it after the self-test, or run +`CultMeshKt write-receipt-vectors `. diff --git a/packages/cultmesh-kotlin/build.ps1 b/packages/cultmesh-kotlin/build.ps1 index aabacbcd..42c22358 100644 --- a/packages/cultmesh-kotlin/build.ps1 +++ b/packages/cultmesh-kotlin/build.ps1 @@ -26,4 +26,7 @@ $sources = Get-ChildItem $src -Recurse -Filter *.kt | Select-Object -ExpandPrope if ($LASTEXITCODE -ne 0) { throw "kotlinc failed with exit code $LASTEXITCODE" } & $java -cp "$jarPath;$kotlinStdlib" org.gamecult.cultmesh.CultMeshKt if ($LASTEXITCODE -ne 0) { throw "cultmesh-kotlin self-test failed with exit code $LASTEXITCODE" } +$vectors = Join-Path $repoRoot "contracts\cultnet\interop\write-receipt-vectors.json" +& $java -cp "$jarPath;$kotlinStdlib" org.gamecult.cultmesh.CultMeshKt write-receipt-vectors $vectors +if ($LASTEXITCODE -ne 0) { throw "write-receipt vector replay failed with exit code $LASTEXITCODE" } Write-Host "Built $jarPath" diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index 7d052785..d9cdfad7 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt @@ -3424,6 +3424,7 @@ fun main(args: Array) { "rudp-dial-message-once" -> rudpDialSchemaMessageOnce(options) "serve" -> interopServe(options) "dial" -> interopDial(options) + "write-receipt-vectors" -> if (!replayWriteReceiptVectors(args.getOrNull(1) ?: error("usage: write-receipt-vectors "))) kotlin.system.exitProcess(1) else -> error("Unknown mode ${args[0]}") } } diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceipt.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceipt.kt new file mode 100644 index 00000000..4060f08a --- /dev/null +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceipt.kt @@ -0,0 +1,232 @@ +package org.gamecult.cultmesh + +// The write receipt (contracts/cultnet/interop/write-receipt.md): the two frames, their strict decode, the dedupe +// content hash, the hello capability test and the client settle rules. Mirrors packages/cultnet-ts (contracts.ts, +// write-tracker.ts) and packages/cultnet-rs (write_content_hash). The replay of write-receipt-vectors.json lives in +// WriteReceiptVectors.kt. Server rules (dedupe, limits, blank fields) are not a decoder's business and are not here. + +import java.io.IOException +import java.security.MessageDigest + +const val WRITE_ADMITTED_VERSION = "cultnet.write_admitted.v0" +const val ERROR_V1_VERSION = "cultnet.error.v1" +const val ERROR_V0_VERSION = "cultnet.error.v0" + +/** A connection is receipt-capable only when its hello lists both frames. */ +val CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS: List = listOf(WRITE_ADMITTED_VERSION, ERROR_V1_VERSION) + +const val RECEIPT_RESULT_WINDOW_MS = 5000L +const val LEGACY_RESULT_WINDOW_MS = 500L + +fun resultWindowMs(serverAdvertisesReceipts: Boolean): Long = + if (serverAdvertisesReceipts) RECEIPT_RESULT_WINDOW_MS else LEGACY_RESULT_WINDOW_MS + +/** The closed refusal-code set of error.v0; error.v1 adds `limit_reached`. */ +val CULTNET_ERROR_V0_CODES: Set = setOf( + "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", + "unowned_schema", "write_not_authorized", "unsupported_schema_version", "message_invalid", +) +val CULTNET_ERROR_V1_CODES: Set = CULTNET_ERROR_V0_CODES + "limit_reached" + +fun helloAdvertisesWriteReceipts(hello: CultNetMessage): Boolean { + val versions = hello.body["supportedMessageVersions"] as? List<*> ?: return false + return CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS.all { it in versions } +} + +data class CultNetWriteAdmitted(val messageId: String, val operation: String, val schemaId: String, val recordKey: String) { + fun toMessage() = CultNetMessage( + WRITE_ADMITTED_VERSION, + linkedMapOf("messageId" to messageId, "operation" to operation, "schemaId" to schemaId, "recordKey" to recordKey), + ) +} + +data class CultNetErrorDetails(val field: String?, val value: String?, val asOf: Long?, val current: Long?) + +/** error.v0 (`messageId` always null, `version` 0) or error.v1 (`version` 1). */ +data class CultNetErrorFrame( + val version: Int, + val error: String, + val routingHint: Map?, + val code: String?, + val details: CultNetErrorDetails?, + val messageId: String?, +) { + fun toMessage(): CultNetMessage { + val body = linkedMapOf( + "error" to error, + "routingHint" to routingHint, + "code" to code, + "details" to details?.let { + linkedMapOf("field" to it.field, "value" to it.value, "asOf" to it.asOf, "current" to it.current) + }, + ) + if (version == 1) body["messageId"] = messageId + return CultNetMessage(if (version == 1) ERROR_V1_VERSION else ERROR_V0_VERSION, body) + } +} + +/** + * Strict decode of a write_admitted.v0 message against its closed schema: exactly five keys, a non-empty string each, + * operation put|delete. + */ +fun decodeWriteAdmitted(message: CultNetMessage): CultNetWriteAdmitted { + if (message.schemaVersion != WRITE_ADMITTED_VERSION) throw IOException("not a $WRITE_ADMITTED_VERSION message") + closedKeys(message, setOf("messageId", "operation", "schemaId", "recordKey")) + val operation = nonEmpty(message, "operation") + if (operation != "put" && operation != "delete") throw IOException("operation must be put or delete") + return CultNetWriteAdmitted(nonEmpty(message, "messageId"), operation, nonEmpty(message, "schemaId"), nonEmpty(message, "recordKey")) +} + +/** Strict decode of error.v0 (no messageId, no limit_reached) or error.v1 (messageId required, string or nil, never empty). */ +fun decodeErrorFrame(message: CultNetMessage): CultNetErrorFrame { + val version = when (message.schemaVersion) { + ERROR_V1_VERSION -> 1 + ERROR_V0_VERSION -> 0 + else -> throw IOException("not an error message") + } + closedKeys(message, setOf("error", "routingHint", "code", "details") + if (version == 1) setOf("messageId") else emptySet()) + val error = nonEmpty(message, "error") + val codeValue = message.body["code"] + if (codeValue != null && (codeValue !is String || codeValue !in (if (version == 1) CULTNET_ERROR_V1_CODES else CULTNET_ERROR_V0_CODES))) { + throw IOException("unknown error code") + } + var messageId: String? = null + if (version == 1) { + val id = message.body["messageId"] + if (id != null && (id !is String || id.isEmpty())) throw IOException("messageId must be a non-empty string or nil") + messageId = id as String? + } + return CultNetErrorFrame(version, error, routingHintOf(message.body["routingHint"]), codeValue as String?, detailsOf(message.body["details"]), messageId) +} + +private fun closedKeys(message: CultNetMessage, required: Set) { + if (!message.body.keys.containsAll(required)) throw IOException("${message.schemaVersion} is missing ${required - message.body.keys}") + val unknown = message.body.keys - required + if (unknown.isNotEmpty()) throw IOException("${message.schemaVersion} carries unknown keys $unknown") +} + +private fun nonEmpty(message: CultNetMessage, key: String): String { + val value = message.body[key] + if (value !is String || value.isEmpty()) throw IOException("$key must be a non-empty string") + return value +} + +private fun routingHintOf(value: Any?): Map? { + if (value == null) return null + if (value !is Map<*, *> || value.keys != setOf("shard", "reason")) throw IOException("routingHint must be nil or {shard, reason}") + if (value["shard"] != null && value["shard"] !is Map<*, *>) throw IOException("routingHint.shard must be an object or nil") + if (value["reason"] !is String) throw IOException("routingHint.reason must be a string") + @Suppress("UNCHECKED_CAST") + return value as Map +} + +private fun detailsOf(value: Any?): CultNetErrorDetails? { + if (value == null) return null + if (value !is Map<*, *> || value.keys != setOf("field", "value", "asOf", "current")) { + throw IOException("details must be nil or {field, value, asOf, current}") + } + fun text(key: String): String? = value[key].let { if (it == null || it is String) it as String? else throw IOException("details.$key must be a string or nil") } + fun cursor(key: String): Long? = value[key].let { + if (it == null) null else if (it is Long && it >= 0) it else throw IOException("details.$key must be a non-negative integer or nil") + } + return CultNetErrorDetails(text("field"), text("value"), cursor("asOf"), cursor("current")) +} + +/** sha256 of the MessagePack array [operation, schemaId, recordKey, payload]: payload a bin for a put, nil for a delete. Lowercase hex. */ +fun writeContentHash(operation: String, schemaId: String, recordKey: String, payload: ByteArray?): String { + val writer = MessagePackWriter().array(4).string(operation).string(schemaId).string(recordKey) + if (payload == null) writer.nil() else writer.binary(payload) + return MessageDigest.getInstance("SHA-256").digest(writer.toByteArray()).joinToString("") { "%02x".format(it) } +} + +/** Why nothing settled a write. */ +enum class WriteUnconfirmedReason(val wire: String) { WINDOW_ELAPSED("window_elapsed"), SESSION_ENDED("session_ended"), LOCAL_SEND_FAILED("local_send_failed") } + +/** What a late answer is recorded as on a write that already settled unconfirmed. */ +enum class WriteRecorded(val wire: String) { ADMITTED_LATE("admitted_late"), REFUSED_LATE("refused_late") } + +/** What the caller was told. Settled once; never changed by a later frame. */ +sealed class WriteSettlement { + object Admitted : WriteSettlement() + data class Refused(val code: String?, val error: String, val response: CultNetErrorFrame) : WriteSettlement() + data class Unconfirmed(val reason: WriteUnconfirmedReason) : WriteSettlement() +} + +/** + * The client half of the receipt rules: a pure state machine with an abstract clock, one decision, how each sent write + * settles. Transport, timers and decoding belong to the caller; `frame` takes a message and decodes it strictly. + */ +class WriteTracker(private val windowMs: Long, private val onSettle: (String, WriteSettlement) -> Unit = { _, _ -> }) { + private class Entry(val sentAtMs: Long) { + var settlement: WriteSettlement? = null + var recorded: WriteRecorded? = null + } + + private val writes = linkedMapOf() + + /** Registers a write sent at [atMs]. Re-sending a known id is the same write and does not restart its window. */ + fun send(messageId: String, atMs: Long) { + tick(atMs) + writes.getOrPut(messageId) { Entry(atMs) } + } + + /** Feeds one received message. Only write_admitted, error.v1 and error.v0 answer; anything else is ignored. */ + fun frame(message: CultNetMessage, atMs: Long) { + tick(atMs) + when (message.schemaVersion) { + WRITE_ADMITTED_VERSION -> { + answer(decodeWriteAdmitted(message).messageId, WriteSettlement.Admitted, WriteRecorded.ADMITTED_LATE) + return + } + ERROR_V1_VERSION, ERROR_V0_VERSION -> { + val error = decodeErrorFrame(message) + val refusal = WriteSettlement.Refused(error.code, error.error, error) + if (error.messageId != null) { + answer(error.messageId, refusal, WriteRecorded.REFUSED_LATE) + return + } + // An error that names no write is attributable only when exactly one write is unanswered. + val unanswered = writes.filterValues { it.settlement == null } + if (unanswered.size == 1) settle(unanswered.keys.first(), unanswered.values.first(), refusal) + } + else -> return + } + } + + /** Every unanswered write whose window, counted from its own send, has run settles unconfirmed. */ + fun tick(atMs: Long) { + for ((id, entry) in writes) { + if (entry.settlement == null && atMs - entry.sentAtMs >= windowMs) settle(id, entry, WriteSettlement.Unconfirmed(WriteUnconfirmedReason.WINDOW_ELAPSED)) + } + } + + fun sessionEnd(atMs: Long) { + tick(atMs) + for ((id, entry) in writes) if (entry.settlement == null) settle(id, entry, WriteSettlement.Unconfirmed(WriteUnconfirmedReason.SESSION_ENDED)) + } + + /** This client's own send of [messageId] failed: that write settles unconfirmed; the others are untouched. */ + fun sendFailed(messageId: String, atMs: Long) { + val entry = writes[messageId] + if (entry != null && entry.settlement == null) settle(messageId, entry, WriteSettlement.Unconfirmed(WriteUnconfirmedReason.LOCAL_SEND_FAILED)) + tick(atMs) + } + + /** The settlement told to the caller, or null while unanswered or unknown. */ + fun settlement(messageId: String): WriteSettlement? = writes[messageId]?.settlement + + /** The late answer recorded on an unconfirmed write. */ + fun recorded(messageId: String): WriteRecorded? = writes[messageId]?.recorded + + private fun answer(messageId: String, settlement: WriteSettlement, late: WriteRecorded) { + val entry = writes[messageId] ?: return // names a write that is not in flight: settles nothing + val current = entry.settlement + if (current == null) settle(messageId, entry, settlement) + else if (current is WriteSettlement.Unconfirmed && entry.recorded == null) entry.recorded = late + } + + private fun settle(messageId: String, entry: Entry, settlement: WriteSettlement) { + entry.settlement = settlement + onSettle(messageId, settlement) + } +} diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceiptVectors.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceiptVectors.kt new file mode 100644 index 00000000..033ef8fd --- /dev/null +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceiptVectors.kt @@ -0,0 +1,224 @@ +package org.gamecult.cultmesh + +// Replays contracts/cultnet/interop/write-receipt-vectors.json against this runtime: constants, contentHash, every +// `wire` frame (decode to the vector's fields, re-encode to its bytes), every `schemaInvalid` frame (must be refused), +// and every `scenarios` event replay through WriteTracker. `dedupe` and `blankFields` are server inputs and are not +// run here. Mode: `java -cp ... org.gamecult.cultmesh.CultMeshKt write-receipt-vectors `. + +import java.io.File +import java.io.IOException + +fun replayWriteReceiptVectors(path: String): Boolean { + @Suppress("UNCHECKED_CAST") + val vectors = JsonReader(File(path).readText(Charsets.UTF_8)).readDocument() as Map + val results = linkedMapOf>() + fun check(name: String, run: () -> List) { + results[name] = try { run() } catch (e: Throwable) { listOf("threw: $e") } + } + + val constants = obj(vectors["constants"]) + check("constants") { + listOfNotNull( + "RECEIPT window".takeIf { constants["RECEIPT_RESULT_WINDOW_MS"] != RECEIPT_RESULT_WINDOW_MS }, + "LEGACY window".takeIf { constants["LEGACY_RESULT_WINDOW_MS"] != LEGACY_RESULT_WINDOW_MS }, + ) + } + + for (case in list(obj(vectors["contentHash"])["cases"])) { + val c = obj(case) + check("contentHash/${c["name"]}") { + val got = writeContentHash(c["operation"] as String, c["schemaId"] as String, c["recordKey"] as String, (c["payloadHex"] as String?)?.let(::unhex)) + if (got == c["sha256"]) emptyList() else listOf("$got != ${c["sha256"]}") + } + } + + for (case in list(vectors["wire"])) { + val c = obj(case) + check("wire/${c["name"]}") { + val bytes = unhex(c["messagePackHex"] as String) + val problems = mutableListOf() + val message = decodeReceiptFrame(parseCultNetMessage(bytes)) + if (message.toWireMap() != c["message"]) problems += "decoded ${message.toWireMap()} != ${c["message"]}" + val again = message.toBytes().joinToString("") { "%02X".format(it) } + if (again != (c["messagePackHex"] as String).uppercase()) problems += "re-encoded bytes $again != captured ${c["messagePackHex"]}" + problems + } + } + + for (case in list(vectors["schemaInvalid"])) { + val c = obj(case) + check("schemaInvalid/${c["name"]}") { + try { + decodeReceiptFrame(parseCultNetMessage(unhex(c["messagePackHex"] as String))) + listOf("accepted a frame that must be refused (${c["why"]})") + } catch (_: IOException) { + emptyList() + } + } + } + + // Not a vector: `limit_reached` is in error.v1's enum only (write-receipt.md), as the TS and Rust suites also pin. + check("error_v0_refuses_the_v1_only_limit_reached_code") { + val body = linkedMapOf("error" to "x", "routingHint" to null, "code" to "limit_reached", "details" to null) + val v1 = decodeErrorFrame(CultNetMessage(ERROR_V1_VERSION, body + ("messageId" to null))).code + try { + decodeErrorFrame(CultNetMessage(ERROR_V0_VERSION, body)) + listOf("error.v0 accepted limit_reached") + } catch (_: IOException) { + if (v1 == "limit_reached") emptyList() else listOf("error.v1 lost limit_reached") + } + } + + for (case in list(obj(vectors["scenarios"])["cases"])) { + val scenario = obj(case) + check("scenario/${scenario["name"]}") { + val problems = mutableListOf() + // The fake server states its capability the only way a real one does: in its hello. + val hello = cultNetHello( + "fake-server", "fake", + supportedMessageVersions = if (scenario["server"] == "receipt") CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS else emptyList(), + ) + val windowMs = resultWindowMs(helloAdvertisesWriteReceipts(hello)) + if (windowMs != constants[scenario["window"] as String]) problems += "window $windowMs != ${scenario["window"]}" + val told = linkedMapOf() + val tracker = WriteTracker(windowMs) { id, settlement -> + if (id in told) problems += "$id settled twice" + told[id] = settlement + } + for (event in list(scenario["events"]).map(::obj).sortedBy { it["atMs"] as Long }) { + val at = event["atMs"] as Long + when { + "send" in event -> tracker.send(event["send"] as String, at) + // A frame reaches the client as bytes: the fake server encodes it, the client decodes it. + "frame" in event -> tracker.frame(parseCultNetMessage(MessagePackWriter().value(event["frame"]).toByteArray()), at) + "tick" in event -> tracker.tick(at) + "sessionEnd" in event -> tracker.sessionEnd(at) + // transportAck: delivery is not an answer; the tracker has no such input and must not need one. + } + } + val expect = obj(scenario["expect"]) + for ((id, wantAny) in expect) { + val want = obj(wantAny) + val got = told[id] + if (got == null) { problems += "$id: never settled, wanted ${want["outcome"]}"; continue } + val (outcome, code, reason) = when (got) { + is WriteSettlement.Admitted -> Triple("admitted", null, null) + is WriteSettlement.Refused -> Triple("refused", got.code, null) + is WriteSettlement.Unconfirmed -> Triple("unconfirmed", null, got.reason.wire) + } + if (outcome != want["outcome"]) problems += "$id: told $outcome, wanted ${want["outcome"]}" + if ("code" in want && code != want["code"]) problems += "$id: code $code, wanted ${want["code"]}" + if ("reason" in want && reason != want["reason"]) problems += "$id: reason $reason, wanted ${want["reason"]}" + if (tracker.recorded(id)?.wire != want["recorded"]) problems += "$id: recorded ${tracker.recorded(id)?.wire}, wanted ${want["recorded"]}" + } + for (id in told.keys) if (id !in expect) problems += "$id: settled but the scenario expects nothing for it" + problems + } + } + + for ((name, problems) in results) { + println(if (problems.isEmpty()) "PASS $name" else "FAIL $name") + problems.forEach { println(" $it") } + } + val failed = results.values.count { it.isNotEmpty() } + println("${results.size - failed}/${results.size} passed") + return failed == 0 +} + +/** Strict decode of whichever receipt-family frame this is, returned as the message its typed form encodes; anything else is refused. */ +private fun decodeReceiptFrame(message: CultNetMessage): CultNetMessage = when (message.schemaVersion) { + WRITE_ADMITTED_VERSION -> decodeWriteAdmitted(message).toMessage() + ERROR_V1_VERSION, ERROR_V0_VERSION -> decodeErrorFrame(message).toMessage() + else -> throw IOException("not a receipt frame: ${message.schemaVersion}") +} + +@Suppress("UNCHECKED_CAST") +private fun obj(value: Any?): Map = value as Map +private fun list(value: Any?): List = value as List + +private fun unhex(text: String): ByteArray = ByteArray(text.length / 2) { text.substring(it * 2, it * 2 + 2).toInt(16).toByte() } + +/** A small JSON reader for the vector file: objects, arrays, strings, integers, true/false/null. */ +private class JsonReader(private val text: String) { + private var at = 0 + + fun readDocument(): Any? = readValue().also { skipSpace(); if (at != text.length) throw IOException("trailing JSON") } + + private fun skipSpace() { while (at < text.length && text[at].isWhitespace()) at++ } + + private fun readValue(): Any? { + skipSpace() + return when (val c = text[at]) { + '{' -> { + at++ + val map = linkedMapOf() + skipSpace() + if (text[at] == '}') { at++; return map } + do { + skipSpace() + val key = readString() + skipSpace(); expect(':') + map[key] = readValue() + skipSpace() + } while (closes('}')) + map + } + '[' -> { + at++ + val items = mutableListOf() + skipSpace() + if (text[at] == ']') { at++; return items } + do { + items += readValue() + skipSpace() + } while (closes(']')) + items + } + '"' -> readString() + 't' -> literal("true", true) + 'f' -> literal("false", false) + 'n' -> literal("null", null) + else -> { + if (c != '-' && !c.isDigit()) throw IOException("unexpected '$c' at $at") + val start = at++ + while (at < text.length && text[at].isDigit()) at++ + text.substring(start, at).toLong() + } + } + } + + /** After an element: false when [close] ends the container, true after a comma. */ + private fun closes(close: Char): Boolean { + val c = text[at++] + if (c == close) return false + if (c != ',') throw IOException("expected , or $close at ${at - 1}") + return true + } + + private fun literal(word: String, value: Any?): Any? { + if (!text.startsWith(word, at)) throw IOException("bad literal at $at") + at += word.length + return value + } + + private fun expect(c: Char) { if (text[at++] != c) throw IOException("expected $c at ${at - 1}") } + + private fun readString(): String { + expect('"') + val out = StringBuilder() + while (true) { + val c = text[at++] + when (c) { + '"' -> return out.toString() + '\\' -> when (val e = text[at++]) { + 'n' -> out.append('\n'); 't' -> out.append('\t'); 'r' -> out.append('\r') + 'b' -> out.append('\b'); 'f' -> out.append('\u000c') + 'u' -> { out.append(text.substring(at, at + 4).toInt(16).toChar()); at += 4 } + '"', '\\', '/' -> out.append(e) + else -> throw IOException("bad escape \\$e") + } + else -> out.append(c) + } + } + } +} -- 2.47.3 From 6ca9c455b71c07591603e7b07fd08dfc762dae7b Mon Sep 17 00:00:00 2001 From: Libby Date: Mon, 9 Nov 2026 10:37:48 +0000 Subject: [PATCH 2/2] cultmesh-kotlin: sendFailed vectors, a tracker that ends with its session, White_Space blank The replay runner handles the sendFailed event, fails on an event key it does not know and on a vector count the file did not declare. WriteTracker settles a write sent after sessionEnd as session_ended. Source ids on a raw document record are dropped when blank by Unicode White_Space (isBlankText), not Kotlin's isNullOrBlank, which keeps U+0085 and drops U+001C to U+001F; the vector file's JSON reader skips JSON whitespace only. Co-Authored-By: Claude Sonnet 5.5 --- .../kotlin/org/gamecult/cultmesh/CultMesh.kt | 7 ++--- .../org/gamecult/cultmesh/WriteReceipt.kt | 27 ++++++++++++++++++- .../gamecult/cultmesh/WriteReceiptVectors.kt | 21 ++++++++++++++- 3 files changed, 50 insertions(+), 5 deletions(-) diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index d9cdfad7..4a06d7f5 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt @@ -1100,9 +1100,9 @@ data class CultNetRawDocumentRecord( "payloadEncoding" to payloadEncoding, "payload" to payload.copyOf(), ) - if (!sourceRuntimeId.isNullOrBlank()) wire["sourceRuntimeId"] = sourceRuntimeId - if (!sourceAgentId.isNullOrBlank()) wire["sourceAgentId"] = sourceAgentId - if (!sourceRole.isNullOrBlank()) wire["sourceRole"] = sourceRole + if (!isBlankText(sourceRuntimeId)) wire["sourceRuntimeId"] = sourceRuntimeId + if (!isBlankText(sourceAgentId)) wire["sourceAgentId"] = sourceAgentId + if (!isBlankText(sourceRole)) wire["sourceRole"] = sourceRole if (tags.isNotEmpty()) wire["tags"] = tags return wire } @@ -3413,6 +3413,7 @@ fun main(args: Array) { rudpClientBindsLoopbackOnlyForALoopbackEndpoint() rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames() rudpSocketTransportCarriesFragmentedReliableOrderedSchemaFrames() + rawDocumentRecordDropsOnlyWhiteSpaceSourceIds() return } diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceipt.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceipt.kt index 4060f08a..4b38d942 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceipt.kt +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceipt.kt @@ -163,11 +163,16 @@ class WriteTracker(private val windowMs: Long, private val onSettle: (String, Wr } private val writes = linkedMapOf() + private var ended = false /** Registers a write sent at [atMs]. Re-sending a known id is the same write and does not restart its window. */ fun send(messageId: String, atMs: Long) { tick(atMs) - writes.getOrPut(messageId) { Entry(atMs) } + if (messageId in writes) return + val entry = Entry(atMs) + writes[messageId] = entry + // On a session that already ended nothing can answer this write. + if (ended) settle(messageId, entry, WriteSettlement.Unconfirmed(WriteUnconfirmedReason.SESSION_ENDED)) } /** Feeds one received message. Only write_admitted, error.v1 and error.v0 answer; anything else is ignored. */ @@ -200,8 +205,10 @@ class WriteTracker(private val windowMs: Long, private val onSettle: (String, Wr } } + /** The session ended: every unanswered write settles unconfirmed, and so does any write sent afterwards. */ fun sessionEnd(atMs: Long) { tick(atMs) + ended = true for ((id, entry) in writes) if (entry.settlement == null) settle(id, entry, WriteSettlement.Unconfirmed(WriteUnconfirmedReason.SESSION_ENDED)) } @@ -230,3 +237,21 @@ class WriteTracker(private val windowMs: Long, private val onSettle: (String, Wr onSettle(messageId, settlement) } } + +/** Unicode White_Space (PropList), the one definition of "blank" in write-receipt.md rule 5; Kotlin's own isBlank takes U+001C to U+001F. */ +private val WHITE_SPACE_CODE_POINTS: Set = (0x0009..0x000D).toSet() + 0x0020 + 0x0085 + 0x00A0 + 0x1680 + + (0x2000..0x200A).toSet() + 0x2028 + 0x2029 + 0x202F + 0x205F + 0x3000 + +/** True for null, empty, or only White_Space code points. */ +fun isBlankText(text: String?): Boolean = text == null || text.codePoints().allMatch { it in WHITE_SPACE_CODE_POINTS } + +/** Self-test: blank is White_Space, not Kotlin's isBlank (rule 5): U+001C to U+001F name something, U+00A0 and U+0085 do not. */ +fun rawDocumentRecordDropsOnlyWhiteSpaceSourceIds() { + fun wireKeys(id: String) = CultNetRawDocumentRecord("note.v0", "note:1", "2026-11-08T00:00:00Z", payload = byteArrayOf(1), sourceRuntimeId = id, sourceAgentId = id, sourceRole = id).toWireMap().keys + for (blank in listOf("", "\u0085", "\u00a0\u2003", "\u3000", "\t\r\n ")) { + check(isBlankText(blank) && "sourceRuntimeId" !in wireKeys(blank)) { "blank ${blank.map { it.code }} must be dropped" } + } + for (named in listOf("\u001c", "\u001f", "\u200b", "\ufeff", "\u180e", "a")) { + check(!isBlankText(named) && "sourceRuntimeId" in wireKeys(named) && "sourceRole" in wireKeys(named)) { "${named.map { it.code }} names something and must be kept" } + } +} diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceiptVectors.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceiptVectors.kt index 033ef8fd..e527692d 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceiptVectors.kt +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/WriteReceiptVectors.kt @@ -8,6 +8,9 @@ package org.gamecult.cultmesh import java.io.File import java.io.IOException +// The event actions scenarios.semantics names; anything else is refused, not skipped. +private val EVENT_KEYS = listOf("send", "frame", "tick", "sessionEnd", "sendFailed", "transportAck") + fun replayWriteReceiptVectors(path: String): Boolean { @Suppress("UNCHECKED_CAST") val vectors = JsonReader(File(path).readText(Charsets.UTF_8)).readDocument() as Map @@ -87,12 +90,15 @@ fun replayWriteReceiptVectors(path: String): Boolean { } for (event in list(scenario["events"]).map(::obj).sortedBy { it["atMs"] as Long }) { val at = event["atMs"] as Long + val actions = event.keys - "atMs" + if (actions.size != 1 || actions.first() !in EVENT_KEYS) { problems += "event at $at has keys $actions, wanted atMs and exactly one of $EVENT_KEYS"; continue } when { "send" in event -> tracker.send(event["send"] as String, at) // A frame reaches the client as bytes: the fake server encodes it, the client decodes it. "frame" in event -> tracker.frame(parseCultNetMessage(MessagePackWriter().value(event["frame"]).toByteArray()), at) "tick" in event -> tracker.tick(at) "sessionEnd" in event -> tracker.sessionEnd(at) + "sendFailed" in event -> tracker.sendFailed(event["sendFailed"] as String, at) // transportAck: delivery is not an answer; the tracker has no such input and must not need one. } } @@ -116,6 +122,19 @@ fun replayWriteReceiptVectors(path: String): Boolean { } } + // A runner that read fewer vectors than the file declares has silently skipped some. + check("counts") { + val declared = obj(vectors["counts"]) + fun ran(prefix: String) = results.keys.count { it.startsWith(prefix) } + val loaded = mapOf( + "wire" to list(vectors["wire"]).size, "schemaInvalid" to list(vectors["schemaInvalid"]).size, + "contentHashCases" to list(obj(vectors["contentHash"])["cases"]).size, "scenarioCases" to list(obj(vectors["scenarios"])["cases"]).size, + ) + val executed = mapOf("wire" to ran("wire/"), "schemaInvalid" to ran("schemaInvalid/"), "contentHashCases" to ran("contentHash/"), "scenarioCases" to ran("scenario/")) + loaded.keys.filter { loaded[it] != (declared[it] as Long).toInt() || executed[it] != loaded[it] } + .map { "$it: declared ${declared[it]}, loaded ${loaded[it]}, ran ${executed[it]}" } + } + for ((name, problems) in results) { println(if (problems.isEmpty()) "PASS $name" else "FAIL $name") problems.forEach { println(" $it") } @@ -144,7 +163,7 @@ private class JsonReader(private val text: String) { fun readDocument(): Any? = readValue().also { skipSpace(); if (at != text.length) throw IOException("trailing JSON") } - private fun skipSpace() { while (at < text.length && text[at].isWhitespace()) at++ } + private fun skipSpace() { while (at < text.length && text[at] in " \t\r\n") at++ } private fun readValue(): Any? { skipSpace() -- 2.47.3