@file:JvmName("StressKt")

// Same-language Kotlin stress / benchmark harness.
//
// 고성능 병렬 운용 기준선 Milestone의 lang-baseline Task에서, TypeScript stress.ts와 같은 출력
// 계약(ROW|/SKIP|/SUMMARY|)으로 Kotlin same-language TCP/WS baseline을 남긴다. 절대 성능 합격선은
// 고정하지 않고 local 환경 baseline(throughput, p50/p95/p99 latency, used heap)을 기록하며, 안정성
// 합격선만 hard fail로 둔다: timeout 0, nonce mismatch 0, response type mismatch 0, connection별
// FIFO 위반 0, 종료 후 pending leak 0.
//
// 실행: ./gradlew run -PmainClass=com.tokilabs.proto_socket.crosstest.StressKt --args="--quick --transport=tcp,ws"

package com.tokilabs.proto_socket.crosstest

import com.tokilabs.proto_socket.Communicator
import com.tokilabs.proto_socket.ParserMap
import com.tokilabs.proto_socket.TcpClient
import com.tokilabs.proto_socket.TcpServer
import com.tokilabs.proto_socket.WsClient
import com.tokilabs.proto_socket.WsServer
import com.tokilabs.proto_socket.DialTcp
import com.tokilabs.proto_socket.DialWs
import com.tokilabs.proto_socket.addListenerTyped
import com.tokilabs.proto_socket.addRequestListenerTyped
import com.tokilabs.proto_socket.packets.TestData
import com.tokilabs.proto_socket.typeNameOf
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
import kotlinx.coroutines.withTimeoutOrNull
import java.net.InetAddress
import java.net.ServerSocket
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicLong
import kotlin.system.exitProcess

private const val HOST = "127.0.0.1"
private const val LANGUAGE = "Kotlin"
private const val WS_PATH = "/"
private val ALL_PROFILES = listOf("roundtrip", "burst", "sustained", "parallel", "slow-mix", "payload")

// payload size matrix 축. quick은 작은 샘플(1KB/64KB), full은 Milestone 기준(1KB/64KB/1MB)을 측정한다.
private const val PAYLOAD_SEED = "0123456789abcdef"
private val PAYLOAD_SIZES_QUICK = listOf(1024, 65536)
private val PAYLOAD_SIZES_FULL = listOf(1024, 65536, 1048576)

private val rowsEmitted = AtomicInteger(0)
private val totalViolations = AtomicLong(0)

/** 안정성 위반 카운터. 0이 아니면 해당 row는 FAIL이고 프로세스도 non-zero exit한다. */
private class StressStability {
    val timeouts = AtomicLong(0)
    val nonceMismatch = AtomicLong(0)
    val typeMismatch = AtomicLong(0)
    val fifoViolations = AtomicLong(0)
    val pendingLeak = AtomicLong(0)

    fun violations(): Long =
        timeouts.get() + nonceMismatch.get() + typeMismatch.get() + fifoViolations.get() + pendingLeak.get()
}

private class StressServerHandle(
    val port: Int,
    val stop: () -> Unit,
    private val connected: Channel<Unit>,
) {
    suspend fun waitForConnections(count: Int = 1) {
        repeat(count) {
            if (withTimeoutOrNull(3_000L) { connected.receive() } == null) {
                throw IllegalStateException("server did not finish client setup")
            }
        }
    }
}

private class StressClientHandle(
    val comm: Communicator,
    val close: () -> Unit,
    // [0]=maxStaged, [1]=droppedCount, [2]=totalCopyNanos, [3]=receivedCount
    val wsMetrics: (() -> LongArray)? = null,
)

private suspend fun stopServer(handle: StressServerHandle) {
    handle.stop()
    delay(50)
}

private fun parserMap(): ParserMap = mapOf(typeNameOf<TestData>() to { bytes: ByteArray -> TestData.parseFrom(bytes) })

private fun log(line: String) = System.err.println(line)

private fun memMb(): Double {
    val rt = Runtime.getRuntime()
    return (rt.totalMemory() - rt.freeMemory()).toDouble() / (1024 * 1024)
}

private fun freePort(): Int = ServerSocket(0, 50, InetAddress.getByName(HOST)).use { it.localPort }

private fun testDataPayloadBytes(message: String): Int =
    TestData.newBuilder().setIndex(0).setMessage(message).build().toByteArray().size

// payloadFiller는 size 바이트 길이의 deterministic filler string을 만든다.
private fun payloadFiller(size: Int): String {
    if (size <= 0) return ""
    val reps = (size + PAYLOAD_SEED.length - 1) / PAYLOAD_SEED.length
    return PAYLOAD_SEED.repeat(reps).substring(0, size)
}

// payloadLabel은 payload 크기를 사람이 읽는 축 이름으로 바꾼다(1024 -> 1KB, 1048576 -> 1MB).
private fun payloadLabel(bytes: Int): String = when {
    bytes >= 1048576 && bytes % 1048576 == 0 -> "${bytes / 1048576}MB"
    bytes >= 1024 && bytes % 1024 == 0 -> "${bytes / 1024}KB"
    else -> "${bytes}B"
}

private fun percentile(sorted: List<Double>, p: Double): Double {
    if (sorted.isEmpty()) return 0.0
    var rank = Math.ceil(p / 100.0 * sorted.size).toInt() - 1
    if (rank < 0) rank = 0
    if (rank > sorted.size - 1) rank = sorted.size - 1
    return sorted[rank]
}

private fun classifyRequestError(error: Throwable, s: StressStability) {
    val msg = (error.message ?: error.toString()).lowercase()
    when {
        msg.contains("timeout") -> s.timeouts.incrementAndGet()
        msg.contains("expected") || msg.contains("mismatch") -> s.typeMismatch.incrementAndGet()
        else -> s.nonceMismatch.incrementAndGet()
    }
}

private fun emitRow(
    profile: String,
    axis: String,
    transport: String,
    payloadBytes: Int,
    clientCount: Int,
    requests: Int,
    throughput: Double,
    p50: Double,
    p95: Double,
    p99: Double,
    mem: Double,
    s: StressStability,
    queueBacklog: Long = 0L,
    gatewayBacklog: Long = 0L,
) {
    val violations = s.violations()
    val status = if (violations == 0L) "PASS" else "FAIL"
    totalViolations.addAndGet(violations)
    rowsEmitted.incrementAndGet()
    val fields = listOf(
        "ROW",
        profile,
        axis,
        LANGUAGE,
        transport,
        payloadBytes.toString(),
        clientCount.toString(),
        requests.toString(),
        String.format("%.1f", throughput),
        String.format("%.3f", p50),
        String.format("%.3f", p95),
        String.format("%.3f", p99),
        s.timeouts.get().toString(),
        s.nonceMismatch.get().toString(),
        s.typeMismatch.get().toString(),
        s.fifoViolations.get().toString(),
        s.pendingLeak.get().toString(),
        queueBacklog.toString(), // queueBacklog: WS=maxStaged, TCP=0 (same-language baseline에는 inbound gateway가 없다)
        gatewayBacklog.toString(), // gatewayBacklog: WS=droppedCount, TCP=0
        "off",
        String.format("%.1f", mem),
        status,
    )
    println(fields.joinToString("|"))
    // fixed-latency 진단용 stderr diagnostic. stdout ROW 계약은 그대로 두고, transport별 p50/p99를
    // grep 가능한 형태로 남겨 ws fixed latency 분석에서 Kotlin WS row를 쉽게 추출한다.
    log(
        "DIAG fixed-latency profile=$profile transport=$transport " +
            "p50=${String.format("%.3f", p50)} p99=${String.format("%.3f", p99)}",
    )
}

private fun emitSkip(profile: String, axis: String, transport: String, reason: String) {
    println("SKIP|$profile|$axis|$LANGUAGE|$transport|$reason")
}

@Suppress("UNUSED_PARAMETER")
private fun deferredBottleneck(transport: String, profile: String, concurrency: Int?): String? {
    if (profile == "slow-mix") {
        return "deferred: optimize Epic — Kotlin slow-mix fires same-client requests concurrently, so response completion order is coroutine-scheduling dependent"
    }
    return null
}

// TCP parallel clients>=128 deferred skip은 제거됐다. TcpClient가 read 전용 dispatcher와 soTimeout
// 기반 read poll을 쓰도록 고쳐(read/write 스레드 풀 분리) thread starvation deadlock이 사라졌고,
// clients=128/512/1024 row를 실제로 측정한다. 남는 deferred는 deferredBottleneck(slow-mix)뿐이다.
@Suppress("UNUSED_PARAMETER")
private fun parallelDeadlockBottleneck(transport: String, clientCount: Int): String? = null

private fun summarize(
    profile: String,
    axis: String,
    transport: String,
    latencies: List<Double>,
    elapsedMs: Double,
    clientCount: Int,
    s: StressStability,
    mem: Double,
) {
    val sorted = latencies.sorted()
    val throughput = if (elapsedMs > 0) latencies.size / elapsedMs * 1000.0 else 0.0
    emitRow(
        profile, axis, transport, testDataPayloadBytes("req-0"), clientCount, latencies.size,
        throughput, percentile(sorted, 50.0), percentile(sorted, 95.0), percentile(sorted, 99.0), mem, s,
    )
}

private fun emitThroughput(
    profile: String,
    axis: String,
    transport: String,
    count: Int,
    elapsedMs: Double,
    clientCount: Int,
    payloadBytes: Int,
    s: StressStability,
    mem: Double,
) {
    val throughput = if (elapsedMs > 0) count / elapsedMs * 1000.0 else 0.0
    emitRow(profile, axis, transport, payloadBytes, clientCount, count, throughput, 0.0, 0.0, 0.0, mem, s)
}

private fun startServer(transport: String, port: Int, onConnected: (Communicator) -> Unit): StressServerHandle {
    val connected = Channel<Unit>(Channel.UNLIMITED)
    when (transport) {
        "tcp" -> {
            val srv = TcpServer(HOST, port) { socket -> TcpClient.fromSocket(socket, 0, 0, parserMap()) }
            srv.onClientConnected = { client ->
                onConnected(client.communicator)
                connected.trySend(Unit)
            }
            srv.start()
            return StressServerHandle(srv.port(), { srv.stop() }, connected)
        }
        "ws" -> {
            val srv = WsServer(HOST, port, WS_PATH) { conn -> WsClient.forServer(conn, 0, 0, parserMap()) }
            srv.onClientConnected = { client ->
                onConnected(client.communicator)
                connected.trySend(Unit)
            }
            srv.start()
            return StressServerHandle(srv.port(), { srv.stop() }, connected)
        }
        else -> throw IllegalArgumentException("unknown transport $transport")
    }
}

// request-response 축에서는 client가 concurrency C로 동시에 send하므로 frame 도착 순서가 결정적이지
// 않다. 따라서 여기서는 FIFO를 측정하지 않고 응답 정확성/timeout만 본다. per-connection FIFO는 단일
// connection 순차 send인 burst 축에서 검증한다.
private fun startEchoServer(transport: String, port: Int): StressServerHandle =
    startServer(transport, port) { comm ->
        addRequestListenerTyped<TestData, TestData>(comm) { req ->
            TestData.newBuilder().setIndex(req.index * 2).setMessage("echo:" + req.message).build()
        }
    }

private fun startSlowMixServer(transport: String, port: Int, slowEvery: Int, delayMs: Long): StressServerHandle =
    startServer(transport, port) { comm ->
        addRequestListenerTyped<TestData, TestData>(comm) { req ->
            if ((req.index + 1) % slowEvery == 0) delay(delayMs)
            TestData.newBuilder().setIndex(req.index * 2).setMessage("echo:" + req.message).build()
        }
    }

private suspend fun dial(transport: String, port: Int): StressClientHandle = when (transport) {
    "tcp" -> {
        val c = DialTcp(HOST, port, 0, 0, parserMap())
        StressClientHandle(comm = c.communicator, close = { c.close() })
    }
    "ws" -> {
        val c = DialWs(HOST, port, WS_PATH, 0, 0, parserMap())
        StressClientHandle(comm = c.communicator, close = { c.close() }, wsMetrics = { c.getMetricsDiagnostics() })
    }
    else -> throw IllegalArgumentException("unknown transport $transport")
}

private suspend fun dialWithRetry(transport: String, port: Int): StressClientHandle {
    val deadline = System.nanoTime() + 3_000_000_000L
    var lastError: Throwable? = null
    while (System.nanoTime() < deadline) {
        try {
            return dial(transport, port)
        } catch (error: Throwable) {
            lastError = error
            withContext(Dispatchers.IO) { Thread.sleep(50) }
        }
    }
    throw IllegalStateException("connect $transport:$port timed out: $lastError")
}

private suspend fun sendOne(comm: Communicator, index: Int, prefix: String, timeoutMs: Long, latencies: ConcurrentLinkedQueue<Double>, s: StressStability) {
    val started = System.nanoTime()
    try {
        val res: TestData = com.tokilabs.proto_socket.sendRequestTyped(
            comm, TestData.newBuilder().setIndex(index).setMessage("$prefix-$index").build(), timeoutMs,
        )
        latencies.add((System.nanoTime() - started) / 1e6)
        if (res.index != index * 2 || res.message != "echo:$prefix-$index") {
            s.nonceMismatch.incrementAndGet()
        }
    } catch (error: Throwable) {
        classifyRequestError(error, s)
    }
}

private suspend fun sendPayloadOne(
    comm: Communicator,
    index: Int,
    filler: String,
    expectedEcho: String,
    timeoutMs: Long,
    latencies: ConcurrentLinkedQueue<Double>,
    s: StressStability,
) {
    val started = System.nanoTime()
    try {
        val res: TestData = com.tokilabs.proto_socket.sendRequestTyped(
            comm, TestData.newBuilder().setIndex(index).setMessage(filler).build(), timeoutMs,
        )
        latencies.add((System.nanoTime() - started) / 1e6)
        if (res.index != index * 2 || res.message != expectedEcho) {
            s.nonceMismatch.incrementAndGet()
        }
    } catch (error: Throwable) {
        classifyRequestError(error, s)
    }
}

private suspend fun sendSlowMixOne(
    comm: Communicator,
    clientId: Int,
    index: Int,
    timeoutMs: Long,
    completedByClient: Array<AtomicInteger>,
    latencies: ConcurrentLinkedQueue<Double>,
    s: StressStability,
) {
    val message = "sm-$clientId-$index"
    val started = System.nanoTime()
    try {
        val res: TestData = com.tokilabs.proto_socket.sendRequestTyped(
            comm, TestData.newBuilder().setIndex(index).setMessage(message).build(), timeoutMs,
        )
        latencies.add((System.nanoTime() - started) / 1e6)
        if (res.index != index * 2 || res.message != "echo:$message") {
            s.nonceMismatch.incrementAndGet()
        }
        if (index <= completedByClient[clientId].get()) {
            s.fifoViolations.incrementAndGet()
        }
        completedByClient[clientId].set(index)
    } catch (error: Throwable) {
        classifyRequestError(error, s)
    }
}

private suspend fun runRequestLoad(comm: Communicator, total: Int, concurrency: Int, timeoutMs: Long, s: StressStability): List<Double> {
    val latencies = ConcurrentLinkedQueue<Double>()
    var issued = 0
    var nextIndex = 0
    while (issued < total) {
        val batchSize = minOf(concurrency, total - issued)
        coroutineScope {
            for (i in 0 until batchSize) {
                val index = nextIndex++
                launch(Dispatchers.Default) { sendOne(comm, index, "req", timeoutMs, latencies, s) }
            }
        }
        issued += batchSize
    }
    return latencies.toList()
}

private suspend fun profileRoundtrip(transport: String, mode: String) {
    val concurrencies = listOf(1, 16, 64, 256)
    val batches = if (mode == "quick") 2 else 20
    val timeoutMs = if (mode == "quick") 5_000L else 15_000L
    log("[roundtrip] transport=$transport mode=$mode concurrencies=1,16,64,256 batches/level=$batches")
    for (concurrency in concurrencies) {
        val deferred = deferredBottleneck(transport, "roundtrip", concurrency)
        if (deferred != null) {
            emitSkip("roundtrip", "concurrency=$concurrency", transport, deferred)
            continue
        }
        val s = StressStability()
        val total = concurrency * batches
        val srv = startEchoServer(transport, freePort())
        try {
            val handle = dialWithRetry(transport, srv.port)
            srv.waitForConnections()
            try {
                val started = System.nanoTime()
                val latencies = runRequestLoad(handle.comm, total, concurrency, timeoutMs, s)
                val snap = handle.wsMetrics?.invoke()
                val queueBacklog = snap?.get(0) ?: 0L
                val gatewayBacklog = snap?.get(1) ?: 0L
                if (snap != null) {
                    val copyMs = snap[2] / 1_000_000.0
                    log(
                        "[roundtrip] transport=$transport concurrency=$concurrency " +
                            "wsCopyMs=${String.format("%.3f", copyMs)} " +
                            "wsMaxStaged=${snap[0]} wsDropped=${snap[1]} " +
                            "wsReceived=${snap[3]}"
                    )
                }
                val sorted = latencies.sorted()
                val elapsedMs = (System.nanoTime() - started) / 1e6
                val throughput = if (elapsedMs > 0) latencies.size / elapsedMs * 1000.0 else 0.0
                emitRow(
                    "roundtrip", "concurrency=$concurrency", transport,
                    testDataPayloadBytes("req-0"), 1, latencies.size,
                    throughput, percentile(sorted, 50.0), percentile(sorted, 95.0), percentile(sorted, 99.0),
                    memMb(), s, queueBacklog, gatewayBacklog,
                )
            } finally {
                handle.close()
                if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
            }
        } catch (error: Throwable) {
            s.timeouts.incrementAndGet()
            log("[roundtrip] concurrency=$concurrency error=$error")
            emitRow("roundtrip", "concurrency=$concurrency", transport, testDataPayloadBytes("req-0"), 1, 0, 0.0, 0.0, 0.0, 0.0, memMb(), s)
        } finally {
            stopServer(srv)
        }
        log("[roundtrip] transport=$transport concurrency=$concurrency done violations=${s.violations()}")
    }
}

private suspend fun profileBurst(transport: String, mode: String) {
    val counts = if (mode == "quick") listOf(200) else listOf(1000, 10000, 100000)
    log("[burst] transport=$transport mode=$mode counts=$counts")
    val deferred = deferredBottleneck(transport, "burst", null)
    if (deferred != null) {
        for (count in counts) emitSkip("burst", "count=$count", transport, deferred)
        return
    }
    for (count in counts) {
        val s = StressStability()
        val received = AtomicInteger(0)
        val lastIndex = AtomicInteger(-1)
        val done = CompletableDeferred<Unit>()
        val srv = startServer(transport, freePort()) { comm ->
            addListenerTyped<TestData>(comm) { data ->
                if (data.index <= lastIndex.get()) s.fifoViolations.incrementAndGet()
                lastIndex.set(data.index)
                if (received.incrementAndGet() >= count) done.complete(Unit)
            }
        }
        var handle: StressClientHandle? = null
        try {
            handle = dialWithRetry(transport, srv.port)
            srv.waitForConnections()
            val started = System.nanoTime()
            for (i in 0 until count) {
                handle.comm.send(TestData.newBuilder().setIndex(i).setMessage("b-$i").build())
            }
            val timeoutMs = if (mode == "quick") 10_000L else 60_000L
            if (withTimeoutOrNull(timeoutMs) { done.await() } == null) {
                s.timeouts.incrementAndGet()
                log("[burst] count=$count dispatch timeout received=${received.get()}")
            }
            val elapsed = (System.nanoTime() - started) / 1e6
            if (received.get() != count) s.pendingLeak.incrementAndGet()
            emitThroughput("burst", "count=$count", transport, count, elapsed, 1, testDataPayloadBytes("b-0"), s, memMb())
        } catch (error: Throwable) {
            s.timeouts.incrementAndGet()
            log("[burst] count=$count error=$error")
            emitThroughput("burst", "count=$count", transport, received.get(), 0.0, 1, testDataPayloadBytes("b-0"), s, memMb())
        } finally {
            handle?.let {
                it.close()
                if (it.comm.isAlive()) s.pendingLeak.incrementAndGet()
            }
            stopServer(srv)
        }
        log("[burst] transport=$transport count=$count received=${received.get()} violations=${s.violations()}")
    }
}

private suspend fun profileSustained(transport: String, mode: String) {
    val durationsMs = if (mode == "quick") listOf(2000L) else listOf(30000L, 300000L, 1800000L)
    val concurrency = 16
    val timeoutMs = 15_000L
    log("[sustained] transport=$transport mode=$mode durations=$durationsMs ms concurrency=$concurrency")
    val deferred = deferredBottleneck(transport, "sustained", concurrency)
    if (deferred != null) {
        for (durationMs in durationsMs) emitSkip("sustained", "duration=${durationMs}ms", transport, deferred)
        return
    }
    for (durationMs in durationsMs) {
        val s = StressStability()
        var peak = memMb()
        val srv = startEchoServer(transport, freePort())
        try {
            val handle = dialWithRetry(transport, srv.port)
            srv.waitForConnections()
            try {
                val latencies = ConcurrentLinkedQueue<Double>()
                var nextIndex = 0
                val deadline = System.nanoTime() + durationMs * 1_000_000L
                while (System.nanoTime() < deadline) {
                    coroutineScope {
                        for (i in 0 until concurrency) {
                            val index = nextIndex++
                            launch(Dispatchers.Default) { sendOne(handle.comm, index, "s", timeoutMs, latencies, s) }
                        }
                    }
                    val cur = memMb()
                    if (cur > peak) peak = cur
                }
                summarize("sustained", "duration=${durationMs}ms", transport, latencies.toList(), durationMs.toDouble(), 1, s, peak)
            } finally {
                handle.close()
                if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
            }
        } catch (error: Throwable) {
            s.timeouts.incrementAndGet()
            log("[sustained] duration=${durationMs}ms error=$error")
            summarize("sustained", "duration=${durationMs}ms", transport, emptyList(), 0.0, 1, s, peak)
        } finally {
            stopServer(srv)
        }
        log("[sustained] transport=$transport duration=${durationMs}ms done peakMemMb=${String.format("%.1f", peak)} violations=${s.violations()}")
    }
}

private suspend fun profileParallel(transport: String, mode: String) {
    val clientCounts = if (mode == "quick") listOf(4) else listOf(16, 128, 512, 1024)
    val perClient = if (mode == "quick") 50 else 500
    val concurrency = 8
    val timeoutMs = 15_000L
    log("[parallel] transport=$transport mode=$mode clients=$clientCounts perClient=$perClient")
    val deferred = deferredBottleneck(transport, "parallel", concurrency)
    if (deferred != null) {
        for (clientCount in clientCounts) emitSkip("parallel", "clients=$clientCount", transport, deferred)
        return
    }
    for (clientCount in clientCounts) {
        val parallelDeferred = parallelDeadlockBottleneck(transport, clientCount)
        if (parallelDeferred != null) {
            emitSkip("parallel", "clients=$clientCount", transport, parallelDeferred)
            continue
        }
        val s = StressStability()
        val srv = startEchoServer(transport, freePort())
        val handles = mutableListOf<StressClientHandle>()
        try {
            for (i in 0 until clientCount) {
                try {
                    handles.add(dialWithRetry(transport, srv.port))
                } catch (error: Throwable) {
                    s.timeouts.incrementAndGet()
                    log("[parallel] clients=$clientCount dial error=$error")
                }
            }
            srv.waitForConnections(handles.size)
            val allLatencies = ConcurrentLinkedQueue<Double>()
            val started = System.nanoTime()
            coroutineScope {
                for (handle in handles) {
                    launch(Dispatchers.Default) {
                        allLatencies.addAll(runRequestLoad(handle.comm, perClient, concurrency, timeoutMs, s))
                    }
                }
            }
            summarize("parallel", "clients=$clientCount", transport, allLatencies.toList(), (System.nanoTime() - started) / 1e6, clientCount, s, memMb())
        } finally {
            for (handle in handles) {
                handle.close()
                if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
            }
            stopServer(srv)
        }
        log("[parallel] transport=$transport clients=$clientCount done violations=${s.violations()}")
    }
}

private suspend fun profileSlowMix(transport: String, mode: String) {
    val clientCount = if (mode == "quick") 4 else 16
    val perClient = if (mode == "quick") 20 else 200
    val slowEvery = if (mode == "quick") 5 else 10
    val delayMs = if (mode == "quick") 50L else 100L
    val timeoutMs = if (mode == "quick") 10_000L else 30_000L
    val axis = "clients=$clientCount,slowEvery=$slowEvery,delayMs=$delayMs"
    log("[slow-mix] transport=$transport mode=$mode $axis perClient=$perClient")
    val deferred = deferredBottleneck(transport, "slow-mix", perClient)
    if (deferred != null) {
        emitSkip("slow-mix", axis, transport, deferred)
        return
    }

    val s = StressStability()
    var peak = memMb()
    val srv = startSlowMixServer(transport, freePort(), slowEvery, delayMs)
    val handles = mutableListOf<Pair<Int, StressClientHandle>>()
    val latencies = ConcurrentLinkedQueue<Double>()
    var elapsedMs = 0.0
    try {
        for (clientId in 0 until clientCount) {
            try {
                handles.add(clientId to dialWithRetry(transport, srv.port))
            } catch (error: Throwable) {
                s.timeouts.incrementAndGet()
                log("[slow-mix] client=$clientId dial error=$error")
            }
        }
        srv.waitForConnections(handles.size)
        val completedByClient = Array(clientCount) { AtomicInteger(-1) }
        val started = System.nanoTime()
        coroutineScope {
            for ((clientId, handle) in handles) {
                for (index in 0 until perClient) {
                    launch(Dispatchers.Default) {
                        sendSlowMixOne(handle.comm, clientId, index, timeoutMs, completedByClient, latencies, s)
                    }
                }
            }
        }
        val cur = memMb()
        if (cur > peak) peak = cur
        elapsedMs = (System.nanoTime() - started) / 1e6
    } finally {
        for ((_, handle) in handles) {
            handle.close()
            if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
        }
        stopServer(srv)
    }
    val sorted = latencies.toList().sorted()
    val throughput = if (elapsedMs > 0) latencies.size / elapsedMs * 1000.0 else 0.0
    emitRow(
        "slow-mix", axis, transport, testDataPayloadBytes("sm-0-0"), clientCount, latencies.size,
        throughput, percentile(sorted, 50.0), percentile(sorted, 95.0), percentile(sorted, 99.0), peak, s,
    )
    log("[slow-mix] transport=$transport clients=$clientCount done peakMemMb=${String.format("%.1f", peak)} violations=${s.violations()}")
}

// profilePayload는 payload size matrix를 측정한다. 동일 connection request-response를 1KB/64KB/1MB
// payload에서 돌리고, message field를 deterministic filler로 채워 실제 serialized payload bytes,
// payload별 p50/p95/p99 latency, throughput, used heap, stability counters를 결과 row에 남긴다.
private suspend fun profilePayload(transport: String, mode: String) {
    val sizes = if (mode == "quick") PAYLOAD_SIZES_QUICK else PAYLOAD_SIZES_FULL
    val concurrency = if (mode == "quick") 4 else 8
    val requests = if (mode == "quick") 20 else 100
    val timeoutMs = if (mode == "quick") 10_000L else 30_000L
    log("[payload] transport=$transport mode=$mode sizes=${sizes.map(::payloadLabel)} concurrency=$concurrency requests/size=$requests")
    val deferred = deferredBottleneck(transport, "payload", concurrency)
    if (deferred != null) {
        for (size in sizes) emitSkip("payload", "payload=${payloadLabel(size)}", transport, deferred)
        return
    }
    for (size in sizes) {
        val s = StressStability()
        val filler = payloadFiller(size)
        val expectedEcho = "echo:$filler"
        val payloadBytes = testDataPayloadBytes(filler)
        val axis = "payload=${payloadLabel(size)}"
        var peak = memMb()
        val srv = startEchoServer(transport, freePort())
        try {
            val handle = dialWithRetry(transport, srv.port)
            srv.waitForConnections()
            try {
                val latencies = ConcurrentLinkedQueue<Double>()
                var issued = 0
                var nextIndex = 0
                val started = System.nanoTime()
                while (issued < requests) {
                    val batchSize = minOf(concurrency, requests - issued)
                    coroutineScope {
                        for (i in 0 until batchSize) {
                            val index = nextIndex++
                            launch(Dispatchers.Default) {
                                sendPayloadOne(handle.comm, index, filler, expectedEcho, timeoutMs, latencies, s)
                            }
                        }
                    }
                    issued += batchSize
                    val cur = memMb()
                    if (cur > peak) peak = cur
                }
                val elapsedMs = (System.nanoTime() - started) / 1e6
                val sorted = latencies.toList().sorted()
                val throughput = if (elapsedMs > 0) latencies.size / elapsedMs * 1000.0 else 0.0
                val snap = handle.wsMetrics?.invoke()
                val queueBacklog = snap?.get(0) ?: 0L
                val gatewayBacklog = snap?.get(1) ?: 0L
                if (snap != null) {
                    val copyMs = snap[2] / 1_000_000.0
                    log(
                        "[payload] transport=$transport size=${payloadLabel(size)} " +
                            "wsCopyMs=${String.format("%.3f", copyMs)} " +
                            "wsMaxStaged=${snap[0]} wsDropped=${snap[1]} " +
                            "wsReceived=${snap[3]} peakMemMb=${String.format("%.1f", peak)}"
                    )
                }
                emitRow(
                    "payload", axis, transport, payloadBytes, 1, latencies.size, throughput,
                    percentile(sorted, 50.0), percentile(sorted, 95.0), percentile(sorted, 99.0),
                    peak, s, queueBacklog, gatewayBacklog,
                )
            } finally {
                handle.close()
                if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
            }
        } catch (error: Throwable) {
            s.timeouts.incrementAndGet()
            log("[payload] size=${payloadLabel(size)} error=$error")
            emitRow("payload", axis, transport, payloadBytes, 1, 0, 0.0, 0.0, 0.0, 0.0, peak, s)
        } finally {
            stopServer(srv)
        }
        log("[payload] transport=$transport size=${payloadLabel(size)} bytes=$payloadBytes peakMemMb=${String.format("%.1f", peak)} violations=${s.violations()}")
    }
}

private fun parseList(args: Array<String>, prefix: String, def: List<String>): List<String> {
    for (arg in args) {
        if (arg.startsWith(prefix)) {
            val items = arg.substring(prefix.length).split(",").map { it.trim() }.filter { it.isNotEmpty() }
            if (items.isNotEmpty()) return items
        }
    }
    return def
}

fun main(args: Array<String>) = runBlocking {
    var mode = "quick"
    for (arg in args) {
        if (arg == "--full" || arg == "--mode=full") mode = "full"
        if (arg == "--quick" || arg == "--mode=quick") mode = "quick"
    }
    var transports = parseList(args, "--transport=", listOf("tcp", "ws"))
    transports = parseList(args, "--transports=", transports)
    var profiles = parseList(args, "--profiles=", ALL_PROFILES)
    profiles = parseList(args, "--profile=", profiles)

    log(
        "INFO stress harness language=$LANGUAGE mode=$mode transports=${transports.joinToString(",")} " +
            "profiles=${profiles.joinToString(",")} typeName=${typeNameOf<TestData>()}",
    )

    val runners: Map<String, suspend (String, String) -> Unit> = mapOf(
        "roundtrip" to ::profileRoundtrip,
        "burst" to ::profileBurst,
        "sustained" to ::profileSustained,
        "parallel" to ::profileParallel,
        "slow-mix" to ::profileSlowMix,
        "payload" to ::profilePayload,
    )

    for (transport in transports) {
        if (transport != "tcp" && transport != "ws") {
            emitSkip("all", "transport", transport, "unsupported transport for Kotlin same-language baseline")
            continue
        }
        for (profile in profiles) {
            val runner = runners[profile]
            if (runner == null) {
                emitSkip(profile, "profile", transport, "profile not part of Kotlin same-language baseline (gateway is TypeScript-specific)")
                continue
            }
            runner(transport, mode)
        }
    }

    val status = if (totalViolations.get() == 0L) "PASS" else "FAIL"
    println(
        "SUMMARY|status=$status|language=$LANGUAGE|mode=$mode|transports=${transports.joinToString(",")}|" +
            "profiles=${profiles.joinToString(",")}|rows=${rowsEmitted.get()}|stability_violations=${totalViolations.get()}",
    )
    System.out.flush()
    if (status != "PASS") exitProcess(1)
}
