1 Commits

Author SHA1 Message Date
subochev dba52d22fd feat: вычисление сессии по истории (LCP) для providers[].session_header
- SHA-256 на commonMain + SessionRegistry (LCP по префикс-хэшам, LRU 1000/6ч, Mutex)
- providers[].session_header: прокси сам ставит/перезаписывает заголовок сессии
- цепочка хэшей стартует с первого user-сообщения (system не склеивает сессии)
- CONFIG.md + тесты (SHA-256, LCP, LRU/TTL, парсинг, стабильность префиксов)
2026-09-11 05:27:15 +03:00
4 changed files with 370 additions and 3 deletions
+35
View File
@@ -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
@@ -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<String, ProviderConf>,
upstreamsById: Map<String, UpstreamConf>,
active: Map<String, UpstreamCounter>,
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<String, ProviderConf>,
upstreamsById: Map<String, UpstreamConf>,
active: Map<String, UpstreamCounter>,
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"),
)
}
@@ -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<String> {
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<String>(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<String>) {
var lastAccess: Long = 0
}
private val mutex = Mutex()
private val prefixToSession = mutableMapOf<String, String>()
/** В порядке доступа: голова — самая давняя, хвост — свежая. */
private val sessions = LinkedHashMap<String, Entry>()
suspend fun resolve(prefixHashes: List<String>): String? =
mutex.withLock { resolveLocked(prefixHashes, Clock.System.now().toEpochMilliseconds()) }
/** Синхронная (без блокировки) версия — для тестов и вызовов под mutex. */
internal fun resolveLocked(prefixHashes: List<String>, 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()
}
}
}
@@ -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)
}
}