diff --git a/CONFIG.md b/CONFIG.md index cec8634..7417c4d 100644 --- a/CONFIG.md +++ b/CONFIG.md @@ -73,6 +73,7 @@ providers: url: "https://routerai.ru/api/v1" key: "sk-..." # Bearer-ключ; можно подставлять из env max_concurrency: 4 # опционально; лимит по умолчанию для апстримов + session_header: x-opencode-session # опционально; прокси считает сессию из истории patch: # уровень провайдера: ко всем его запросам provider: allow_fallbacks: false @@ -131,6 +132,39 @@ models: `patch` опционален на **любом** уровне (`providers` / `upstreams` / `models`): если ни одного нет — запрос проксируется как есть (исходное тело клиента). +### Сессия по истории (`session_header`) + +`providers[].session_header` (опционально) — имя HTTP-заголовка, который прокси +**вычисляет сам** из истории сообщений и ставит в запрос к этому провайдеру. +Нужно для API, требующих стабильный идентификатор сессии (например, +`x-opencode-session`), когда клиент его не шлёт или шлёт не то. + +```yaml +providers: + - id: some-provider + url: "https://.../v1" + session_header: x-opencode-session +``` + +Как считается id: + +1. Берётся финальное тело запроса (после всех `patch`), из него — `messages`. +2. Цепочка **инкрементальных SHA-256 префикс-хэшей** начинается с первого + `user`-сообщения (ведущий `system`-промпт игнорируется: он обычно одинаков + у всех сессий клиента и как признак сессии бесполезен). +3. В реестре сессий ищется **наибольший общий префикс** (LCP) с уже виденной + историей. Нашли — используется id той сессии; не нашли — создаётся новая + (`id` = хэш всей истории на первом ходу). +4. Заголовок ставится **всегда** (клиентское значение перезаписывается). + +Итог: пока история одной сессии растёт (дописываются assistant/user-сообщения), +id не меняется; разные диалоги получают разные id. + +> **Ограничения.** Реестр живёт в памяти (LRU: 1000 сессий / 6 часов) — при +> рестарте прокси активные сессии получат новый id. Обрезка/суммаризация +> истории рвёт общий префикс → сессия распадётся на новую. Диалоги с +> одинаковым первым `user`-сообщением неразличимы (склеятся). + ### Пример сборки тела (многослойный `patch`) Берём модель `my-gpt` (из примера выше), маршрут уходит на апстрим @@ -211,6 +245,7 @@ data class ProviderConf( val key: String = "", val max_concurrency: Int? = null, // лимит по умолчанию для апстримов провайдера val patch: JsonObject? = null, // ко всем запросам провайдера + val session_header: String? = null, // заголовок-сессия, считается из истории ) @Serializable diff --git a/src/commonMain/kotlin/pw/binom/llmproxy/Main.kt b/src/commonMain/kotlin/pw/binom/llmproxy/Main.kt index 05ac8fb..2bcccde 100644 --- a/src/commonMain/kotlin/pw/binom/llmproxy/Main.kt +++ b/src/commonMain/kotlin/pw/binom/llmproxy/Main.kt @@ -97,8 +97,9 @@ fun main() { } val http = createHttpClient() + val sessions = SessionRegistry() startServer(config.server.host, config.server.port) { - proxyModule(config, providersById, upstreamsById, active, http) + proxyModule(config, providersById, upstreamsById, active, sessions, http) } } @@ -109,11 +110,12 @@ fun Application.proxyModule( providersById: Map, upstreamsById: Map, active: Map, + sessions: SessionRegistry, http: HttpClient, ) { routing { post("/v1/chat/completions") { - handleChat(call, config, providersById, upstreamsById, active, http) + handleChat(call, config, providersById, upstreamsById, active, sessions, http) } get("/v1/models") { handleModels(call, config) @@ -127,6 +129,7 @@ private suspend fun handleChat( providersById: Map, upstreamsById: Map, active: Map, + sessions: SessionRegistry, http: HttpClient, ) { log.info { "[llm-proxy] chat ${call.request.httpMethod.value} ${call.request.uri} headers: ${formatHeadersForLog(call.request.headers)}" } @@ -186,13 +189,19 @@ private suspend fun handleChat( val url = provider.url.trimEnd('/') + "/chat/completions" val providerKey = resolveEnv(provider.key) + val sessionHeader = provider.session_header + // Клиентский одноимённый заголовок не пробрасываем: при метке + // значение x-opencode-session всегда вычисляем сами (LCP по истории). val forwardedHeaders = headersToForward(call.request.headers) + .filterKeys { sessionHeader == null || !it.equals(sessionHeader, ignoreCase = true) } + val sessionId = sessionHeader?.let { sessions.resolve(sessionPrefixHashes(forwarded)) } val outgoingHeaders = forwardedHeaders.toMutableMap().apply { this["Content-Type"] = listOf("application/json") if (providerKey.isNotEmpty()) this["Authorization"] = listOf("Bearer $providerKey") + if (sessionHeader != null && sessionId != null) this[sessionHeader] = listOf(sessionId) } log.info { - "[llm-proxy] chat model=$modelName upstream=${up.id} → $url headers: ${formatHeadersForLog(outgoingHeaders)}" + "[llm-proxy] chat model=$modelName upstream=${up.id} session=${sessionId ?: "-"} → $url headers: ${formatHeadersForLog(outgoingHeaders)}" } var failover = false var responded = false @@ -210,6 +219,9 @@ private suspend fun handleChat( set("Authorization", "Bearer $providerKey") } set("Content-Type", "application/json") + if (sessionHeader != null && sessionId != null) { + set(sessionHeader, sessionId) + } } setBody(forwarded.toString()) }.execute { resp -> @@ -575,6 +587,7 @@ data class ProviderConf( val key: String = "", val max_concurrency: Int? = null, val patch: JsonObject? = null, + val session_header: String? = null, ) data class UpstreamConf( @@ -636,6 +649,7 @@ internal fun parseConfig(root: YamlElement): Config { key = m.strOrNull("key") ?: "", max_concurrency = m.strOrNull("max_concurrency")?.toIntOrNull(), patch = m.yamlMapOrNull("patch")?.let { yamlToJson(it) as JsonObject }, + session_header = m.strOrNull("session_header"), ) } diff --git a/src/commonMain/kotlin/pw/binom/llmproxy/Session.kt b/src/commonMain/kotlin/pw/binom/llmproxy/Session.kt new file mode 100644 index 0000000..8710b04 --- /dev/null +++ b/src/commonMain/kotlin/pw/binom/llmproxy/Session.kt @@ -0,0 +1,206 @@ +package pw.binom.llmproxy + +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlinx.datetime.Clock +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive + +/** + * SHA-256, чистая реализация на commonMain (без внешних зависимостей — + * KMP-зеркало их не отдаёт). Нужен для стабильного хэша префиксов истории. + */ +internal object Sha256 { + private val K = intArrayOf( + 0x428a2f98, 0x71374491, 0xb5c0fbcf.toInt(), 0xe9b5dba5.toInt(), 0x3956c25b, 0x59f111f1, 0x923f82a4.toInt(), 0xab1c5ed5.toInt(), + 0xd807aa98.toInt(), 0x12835b01, 0x243185be, 0x550c7dc3, 0x72be5d74, 0x80deb1fe.toInt(), 0x9bdc06a7.toInt(), 0xc19bf174.toInt(), + 0xe49b69c1.toInt(), 0xefbe4786.toInt(), 0x0fc19dc6, 0x240ca1cc, 0x2de92c6f, 0x4a7484aa, 0x5cb0a9dc, 0x76f988da, + 0x983e5152.toInt(), 0xa831c66d.toInt(), 0xb00327c8.toInt(), 0xbf597fc7.toInt(), 0xc6e00bf3.toInt(), 0xd5a79147.toInt(), 0x06ca6351, 0x14292967, + 0x27b70a85, 0x2e1b2138, 0x4d2c6dfc, 0x53380d13, 0x650a7354, 0x766a0abb, 0x81c2c92e.toInt(), 0x92722c85.toInt(), + 0xa2bfe8a1.toInt(), 0xa81a664b.toInt(), 0xc24b8b70.toInt(), 0xc76c51a3.toInt(), 0xd192e819.toInt(), 0xd6990624.toInt(), 0xf40e3585.toInt(), 0x106aa070, + 0x19a4c116, 0x1e376c08, 0x2748774c, 0x34b0bcb5, 0x391c0cb3, 0x4ed8aa4a, 0x5b9cca4f, 0x682e6ff3, + 0x748f82ee, 0x78a5636f, 0x84c87814.toInt(), 0x8cc70208.toInt(), 0x90befffa.toInt(), 0xa4506ceb.toInt(), 0xbef9a3f7.toInt(), 0xc67178f2.toInt(), + ) + + private fun rotr(x: Int, n: Int): Int = (x ushr n) or (x shl (32 - n)) + + private fun pad(input: ByteArray): ByteArray { + val total = ((input.size + 9 + 63) / 64) * 64 + val out = ByteArray(total) + input.copyInto(out) + out[input.size] = 0x80.toByte() + val bits = input.size.toLong() * 8 + for (k in 0 until 8) { + out[total - 1 - k] = (bits ushr (8 * k)).toByte() + } + return out + } + + fun hash(input: ByteArray): ByteArray { + val msg = pad(input) + val h = intArrayOf( + 0x6a09e667, 0xbb67ae85.toInt(), 0x3c6ef372, 0xa54ff53a.toInt(), + 0x510e527f, 0x9b05688c.toInt(), 0x1f83d9ab, 0x5be0cd19, + ) + val w = IntArray(64) + var i = 0 + while (i < msg.size) { + for (j in 0 until 16) { + w[j] = ((msg[i + j * 4].toInt() and 0xff) shl 24) or + ((msg[i + j * 4 + 1].toInt() and 0xff) shl 16) or + ((msg[i + j * 4 + 2].toInt() and 0xff) shl 8) or + (msg[i + j * 4 + 3].toInt() and 0xff) + } + for (j in 16 until 64) { + val s0 = rotr(w[j - 15], 7) xor rotr(w[j - 15], 18) xor (w[j - 15] ushr 3) + val s1 = rotr(w[j - 2], 17) xor rotr(w[j - 2], 19) xor (w[j - 2] ushr 10) + w[j] = w[j - 16] + s0 + w[j - 7] + s1 + } + var a = h[0]; var b = h[1]; var c = h[2]; var d = h[3] + var e = h[4]; var f = h[5]; var g = h[6]; var hh = h[7] + for (j in 0 until 64) { + val s1 = rotr(e, 6) xor rotr(e, 11) xor rotr(e, 25) + val ch = (e and f) xor (e.inv() and g) + val t1 = hh + s1 + ch + K[j] + w[j] + val s0 = rotr(a, 2) xor rotr(a, 13) xor rotr(a, 22) + val maj = (a and b) xor (a and c) xor (b and c) + val t2 = s0 + maj + hh = g; g = f; f = e; e = d + t1; d = c; c = b; b = a; a = t1 + t2 + } + h[0] += a; h[1] += b; h[2] += c; h[3] += d + h[4] += e; h[5] += f; h[6] += g; h[7] += hh + i += 64 + } + val out = ByteArray(32) + for (j in 0 until 8) { + out[j * 4] = (h[j] ushr 24).toByte() + out[j * 4 + 1] = (h[j] ushr 16).toByte() + out[j * 4 + 2] = (h[j] ushr 8).toByte() + out[j * 4 + 3] = h[j].toByte() + } + return out + } +} + +private const val HEX = "0123456789abcdef" + +/** SHA-256 строки (UTF-8) в нижнем hex. */ +internal fun sha256Hex(text: String): String { + val bytes = Sha256.hash(text.encodeToByteArray()) + val sb = StringBuilder(bytes.size * 2) + for (b in bytes) { + val v = b.toInt() and 0xff + sb.append(HEX[v ushr 4]).append(HEX[v and 0xf]) + } + return sb.toString() +} + +/** + * Инкрементальные префикс-хэши истории: H_i = sha256(H_{i-1} + "\u0000" + msg_i). + * + * Цепочка начинается с первого `user`-сообщения: system-промпт обычно + * идентичен у всех сессий одного клиента и как признак сессии бесполезен + * (иначе LCP склеивает все сессии на общем префиксе `[system]`). + * Если `messages` нет/пусто — fallback на хэш всего тела. + */ +internal fun sessionPrefixHashes(body: JsonObject): List { + val messages = body["messages"] as? JsonArray + if (messages == null || messages.isEmpty()) { + return listOf(sha256Hex(body.toString())) + } + + fun roleAt(i: Int): String? = + ((messages[i] as? JsonObject)?.get("role") as? JsonPrimitive)?.content + + var start = messages.indices.firstOrNull { roleAt(it) == "user" } + ?: messages.indices.firstOrNull { roleAt(it) != "system" } + ?: 0 + + val res = ArrayList(messages.size - start) + var prev = "" + for (i in start until messages.size) { + prev = sha256Hex(prev + "\u0000" + messages[i].toString()) + res.add(prev) + } + return res +} + +/** + * Реестр сессий: id сессии определяется наибольшим общим префиксом (LCP) + * присланной истории. Для префиксов, которые уже встречались, возвращается + * id исходной сессии; иначе создаётся новая (id = хэш всей истории). + * + * Таблица ограничена по размеру (LRU) и времени жизни (TTL). Доступ под + * Mutex: параллельные запросы одной сессии не должны гонять состояние. + */ +class SessionRegistry( + private val maxSessions: Int = 1000, + private val ttlMillis: Long = 6 * 60 * 60 * 1000L, +) { + private class Entry(val prefixes: MutableSet) { + var lastAccess: Long = 0 + } + + private val mutex = Mutex() + private val prefixToSession = mutableMapOf() + + /** В порядке доступа: голова — самая давняя, хвост — свежая. */ + private val sessions = LinkedHashMap() + + suspend fun resolve(prefixHashes: List): String? = + mutex.withLock { resolveLocked(prefixHashes, Clock.System.now().toEpochMilliseconds()) } + + /** Синхронная (без блокировки) версия — для тестов и вызовов под mutex. */ + internal fun resolveLocked(prefixHashes: List, now: Long): String? { + if (prefixHashes.isEmpty()) return null + evictExpired(now) + + var found: String? = null + for (i in prefixHashes.indices.reversed()) { + val s = prefixToSession[prefixHashes[i]] + if (s != null) { + found = s + break + } + } + val id = found ?: prefixHashes.last() + + val entry = sessions.remove(id) ?: Entry(mutableSetOf()) + for (h in prefixHashes) { + val prev = prefixToSession.put(h, id) + if (prev != null && prev != id) { + sessions[prev]?.prefixes?.remove(h) + } + entry.prefixes.add(h) + } + entry.lastAccess = now + sessions[id] = entry + evictOverflow() + return id + } + + private fun drop(sessionId: String, entry: Entry) { + entry.prefixes.forEach { h -> if (prefixToSession[h] == sessionId) prefixToSession.remove(h) } + } + + private fun evictExpired(now: Long) { + val it = sessions.entries.iterator() + while (it.hasNext()) { + val e = it.next() + if (now - e.value.lastAccess <= ttlMillis) break + drop(e.key, e.value) + it.remove() + } + } + + private fun evictOverflow() { + val it = sessions.entries.iterator() + while (sessions.size > maxSessions && it.hasNext()) { + val e = it.next() + drop(e.key, e.value) + it.remove() + } + } +} diff --git a/src/commonTest/kotlin/pw/binom/llmproxy/ConfigLogicTest.kt b/src/commonTest/kotlin/pw/binom/llmproxy/ConfigLogicTest.kt index bef175a..f020559 100644 --- a/src/commonTest/kotlin/pw/binom/llmproxy/ConfigLogicTest.kt +++ b/src/commonTest/kotlin/pw/binom/llmproxy/ConfigLogicTest.kt @@ -382,4 +382,116 @@ class ConfigLogicTest { val err = out["error"]?.jsonObject ?: error("error block missing") assertEquals("Unsupported field 'foo'", err["message"]?.jsonPrimitive?.content) } + + @Test + fun sha256MatchesKnownVectors() { + assertEquals( + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + sha256Hex(""), + ) + assertEquals( + "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad", + sha256Hex("abc"), + ) + assertEquals( + "248d6a61d20638b8e5c026930c3e6039a33ce45964ff2167f6ecedd419db06c1", + sha256Hex("abcdbcdecdefdefgefghfghighijhijkijkljklmklmnlmnomnopnopq"), + ) + } + + @Test + fun sessionPrefixHashesAreStableWhenHistoryGrows() { + val first = Json.parseToJsonElement( + """{"messages":[{"role":"system","content":"S"},{"role":"user","content":"hi"}]}""", + ).jsonObject + val second = Json.parseToJsonElement( + """{"messages":[{"role":"system","content":"S"},{"role":"user","content":"hi"},{"role":"assistant","content":"yo"},{"role":"user","content":"again"}]}""", + ).jsonObject + + val h1 = sessionPrefixHashes(first) + val h2 = sessionPrefixHashes(second) + + // Цепочка начинается с первого user — system-преамбула не хэшируется. + assertEquals(1, h1.size) + assertEquals(3, h2.size) + assertEquals(h1, h2.take(1)) + assertEquals(h1.last(), h2[0]) + } + + @Test + fun sessionPrefixHashesIgnoreLeadingSystemSoSharedPromptDoesNotCollide() { + val a = Json.parseToJsonElement( + """{"messages":[{"role":"system","content":"SAME"},{"role":"user","content":"session A"}]}""", + ).jsonObject + val b = Json.parseToJsonElement( + """{"messages":[{"role":"system","content":"SAME"},{"role":"user","content":"session B"}]}""", + ).jsonObject + // разные первые user-сообщения → разные хэши, несмотря на общий system + assertFalse(sessionPrefixHashes(a) == sessionPrefixHashes(b)) + } + + @Test + fun sessionPrefixHashesFallBackToWholeBodyWithoutMessages() { + val body = Json.parseToJsonElement("""{"model":"m"}""").jsonObject + assertEquals(listOf(sha256Hex(body.toString())), sessionPrefixHashes(body)) + } + + @Test + fun sessionRegistryReusesIdByLongestCommonPrefix() { + val reg = SessionRegistry() + val first = reg.resolveLocked(listOf("H1"), 0) + assertEquals("H1", first) + + // история выросла: [H1, H2, H3] — самый длинный известный префикс H1 + val next = reg.resolveLocked(listOf("H1", "H2", "H3"), 1) + assertEquals("H1", next) + + // и дальше — id не меняется + val deep = reg.resolveLocked(listOf("H1", "H2", "H3", "H4"), 2) + assertEquals("H1", deep) + } + + @Test + fun sessionRegistryCreatesNewIdForDifferentHistory() { + val reg = SessionRegistry() + assertEquals("A1", reg.resolveLocked(listOf("A1"), 0)) + assertEquals("B1", reg.resolveLocked(listOf("B1"), 0)) + assertEquals("A1", reg.resolveLocked(listOf("A1", "A2"), 0)) + } + + @Test + fun sessionRegistryEvictsLruWhenOverCapacity() { + val reg = SessionRegistry(maxSessions = 2, ttlMillis = Long.MAX_VALUE) + reg.resolveLocked(listOf("A1"), 0) + reg.resolveLocked(listOf("B1"), 1) + reg.resolveLocked(listOf("C1"), 2) // A вытеснена + // a1 больше неизвестен → новая сессия с id = хэш всей истории A2 + assertEquals("A2", reg.resolveLocked(listOf("A1", "A2"), 3)) + } + + @Test + fun sessionRegistryEvictsExpiredByTtl() { + val reg = SessionRegistry(maxSessions = 100, ttlMillis = 1000) + reg.resolveLocked(listOf("A1"), 0) + reg.resolveLocked(listOf("B1"), 2000) // A протухла + assertEquals("A2", reg.resolveLocked(listOf("A1", "A2"), 2001)) + } + + @Test + fun parseConfigReadsSessionHeader() { + val yaml = """ + providers: + - id: p1 + url: "https://x.ru/api/v1" + session_header: x-opencode-session + - id: p2 + url: "https://y.ru/api/v1" + models: + - name: m1 + upstreams: [] + """.trimIndent() + val cfg = parseConfig(Yaml.decodeYamlFromString(yaml)) + assertEquals("x-opencode-session", cfg.providers[0].session_header) + assertEquals(null, cfg.providers[1].session_header) + } }