feat(standalone): agentik pull-model subcommand + AGENTIK_AUTO_DOWNLOAD_MODEL=1 trigger for LiteRT-LM
Добавляет ModelDownloader (HTTP с Range/докачкой, опциональной SHA-256 проверкой)
и два сценария запуска скачивания встроенной модели gemma-4-E2B-it.litertlm:
java -jar agentik.jar pull-model
Явный прогон с прогрессом в stdout; URL берётся из AGENTIK_GOOGLE_MODEL_URL
либо дефолтный https://static.binom.pw/models/gemma-4-E2B-it.litertlm.
AGENTIK_AUTO_DOWNLOAD_MODEL=1 java -jar agentik.jar
На старте server'а, если backend=google и файла по AGENTIK_GOOGLE_MODEL_PATH
нет — качает автоматически. Без флага — exit 2 с понятным сообщением и
подсказкой вызвать pull-model.
Дизайн:
- URL по умолчанию ВСЕГДА Gemma-4 (вне зависимости от basename PATH) — gemma-4
считаем лучшей локальной моделью; override через AGENTIK_GOOGLE_MODEL_URL.
- SHA-256 проверка через опциональный AGENTIK_GOOGLE_MODEL_SHA256_URL.
- Resume: HEAD → если есть .part и Accept-Ranges=bytes → GET с Range: bytes=N-,
иначе restart с нуля.
- Прогресс каждые ~8 MB, финальный rename через Files.move(ATOMIC_MOVE).
Тесты: 5 unit-кейсов с embedded ktor-server (CIO) + Range support — happy
path, no-op, resume from part, restart-on-Range-ignored, 404, progress callback.
Документация: новый раздел §18 в MANUAL-TESTS.md (subcommand, auto-trigger,
resume, override URL, SHA-256 verify).
178/178 tests green.
This commit is contained in:
@@ -772,6 +772,101 @@ ps aux | grep agentik-0.1.0 | grep -v grep
|
||||
|
||||
---
|
||||
|
||||
## 18. Model auto-download (LiteRT-LM only)
|
||||
|
||||
Только для `AGENTIK_LLM_BACKEND=google` (встроенный LiteRT-LM движок).
|
||||
Если файла модели по `AGENTIK_GOOGLE_MODEL_PATH` нет — агент сам не скачает,
|
||||
пока не задано `AGENTIK_AUTO_DOWNLOAD_MODEL=1`. Либо качаем руками
|
||||
через `pull-model` subcommand.
|
||||
|
||||
URL по умолчанию всегда Gemma-4-E2B-it.litertlm (2.5 GB с `static.binom.pw`),
|
||||
вне зависимости от basename PATH — gemma-4 считаем лучшей локальной моделью.
|
||||
|
||||
### 18.1. Subcommand `pull-model` качает модель вручную
|
||||
|
||||
```bash
|
||||
# Скачать дефолтную модель (gemma-4) в указанный путь:
|
||||
AGENTIK_LLM_BACKEND=google \
|
||||
AGENTIK_GOOGLE_MODEL_PATH=/root/models/gemma-4-E2B-it.litertlm \
|
||||
java -jar agentik.jar pull-model
|
||||
# → downloading from https://static.binom.pw/models/gemma-4-E2B-it.litertlm
|
||||
# → 50% (1.2 GB / 2.5 GB)
|
||||
# → done in 47s
|
||||
```
|
||||
|
||||
После `pull-model` файл лежит на месте, файл `<dest>.part` удалён.
|
||||
|
||||
### 18.2. `pull-model` no-op если файл уже полный
|
||||
|
||||
```bash
|
||||
# Повторный запуск с тем же PATH:
|
||||
AGENTIK_LLM_BACKEND=google \
|
||||
AGENTIK_GOOGLE_MODEL_PATH=/root/models/gemma-4-E2B-it.litertlm \
|
||||
java -jar agentik.jar pull-model
|
||||
# → already present (2.50 GB), nothing to do
|
||||
```
|
||||
|
||||
### 18.3. `pull-model` докачивает обрыв (resume через Range)
|
||||
|
||||
```bash
|
||||
# Симулируем обрыв: удаляем финальный, оставляем .part с первыми 500 MB
|
||||
rm /root/models/gemma-4-E2B-it.litertlm
|
||||
mv /root/models/gemma-4-E2B-it.litertlm.part /root/models/gemma-4-E2B-it.litertlm.part.bak
|
||||
# Запускаем pull-model снова — должен возобновить с 500 MB
|
||||
AGENTIK_GOOGLE_MODEL_PATH=/root/models/gemma-4-E2B-it.litertlm \
|
||||
java -jar agentik.jar pull-model
|
||||
# → resuming from 524288000 bytes
|
||||
# → downloaded 2.10 GB in 38s
|
||||
```
|
||||
|
||||
### 18.4. Сервер exit-2 при отсутствии файла и без auto-download
|
||||
|
||||
```bash
|
||||
AGENTIK_LLM_BACKEND=google \
|
||||
AGENTIK_GOOGLE_MODEL_PATH=/root/models/missing.litertlm \
|
||||
java -jar agentik.jar
|
||||
# → LiteRT-LM model file not found at: /root/models/missing.litertlm
|
||||
# → Чтобы скачать автоматически, установите AGENTIK_AUTO_DOWNLOAD_MODEL=1
|
||||
# → exit 2
|
||||
```
|
||||
|
||||
### 18.5. Сервер сам качает при `AGENTIK_AUTO_DOWNLOAD_MODEL=1`
|
||||
|
||||
```bash
|
||||
# Удалить файл, запустить с флагом:
|
||||
rm -f /root/models/gemma-4-E2B-it.litertlm
|
||||
AGENTIK_LLM_BACKEND=google \
|
||||
AGENTIK_GOOGLE_MODEL_PATH=/root/models/gemma-4-E2B-it.litertlm \
|
||||
AGENTIK_AUTO_DOWNLOAD_MODEL=1 \
|
||||
java -jar agentik.jar
|
||||
# → 12:34:56 WARN auto-download: https://static.binom.pw/models/...
|
||||
# → 12:34:56 INFO auto-download: 17% (445 MB/2.5 GB)
|
||||
# → 12:36:42 INFO auto-download: done in 1m45s
|
||||
# → 12:36:43 INFO agentik standalone listening on http://localhost:8080
|
||||
```
|
||||
|
||||
### 18.6. Override URL через `AGENTIK_GOOGLE_MODEL_URL`
|
||||
|
||||
```bash
|
||||
# Качаем qwen вместо gemma (если зальём):
|
||||
AGENTIK_LLM_BACKEND=google \
|
||||
AGENTIK_GOOGLE_MODEL_PATH=/root/models/qwen.litertlm \
|
||||
AGENTIK_GOOGLE_MODEL_URL=https://static.binom.pw/models/Qwen2.5-1.5B-Instruct_multi-prefill-seq_q8_ekv4096.litertlm \
|
||||
java -jar agentik.jar pull-model
|
||||
```
|
||||
|
||||
### 18.7. SHA-256 проверка
|
||||
|
||||
Если на сервере лежит `<basename>.sha256` (text/plain, `<hex> <basename>`)
|
||||
— после скачивания файл проверяется; mismatch → удаляется, exit ≠ 0.
|
||||
|
||||
```bash
|
||||
AGENTIK_GOOGLE_MODEL_URL=https://static.binom.pw/models/gemma-4-E2B-it.litertlm \
|
||||
AGENTIK_GOOGLE_MODEL_SHA256_URL=https://static.binom.pw/models/gemma-4-E2B-it.litertlm.sha256 \
|
||||
java -jar agentik.jar pull-model
|
||||
# → 13:01:23 INFO model download: SHA-256 verified (4ab1...e0d)
|
||||
```
|
||||
|
||||
## Быстрый smoke-test (5 минут)
|
||||
|
||||
Если времени мало — этот минимум покрывает 80%:
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
package pw.binom.agentik.standalone
|
||||
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import mu.KotlinLogging
|
||||
|
||||
|
||||
@@ -26,6 +27,7 @@ import pw.binom.agentik.standalone.agent.memory.LlmMemoryReviewer
|
||||
import pw.binom.agentik.standalone.config.AgentikConfig
|
||||
import pw.binom.agentik.standalone.config.AgentikConfig.MemoryBackend
|
||||
import pw.binom.agentik.standalone.llm.LlmBackend
|
||||
import pw.binom.agentik.standalone.llm.ModelDownloader
|
||||
import pw.binom.agentik.standalone.mcp.McpRegistry
|
||||
import pw.binom.agentik.storage.sqlite.SqliteStores
|
||||
import java.io.File
|
||||
@@ -54,8 +56,143 @@ import java.io.File
|
||||
* - AGENTIK_SYSTEM_PROMPT (default: встроенный `Ты полезный ассистент...`)
|
||||
*/
|
||||
private val log = KotlinLogging.logger {}
|
||||
fun main() {
|
||||
fun main(args: Array<String>) {
|
||||
if (args.isNotEmpty()) {
|
||||
when (args[0]) {
|
||||
"pull-model" -> {
|
||||
runPullModel(args.drop(1))
|
||||
return
|
||||
}
|
||||
"--help", "-h", "help" -> {
|
||||
printHelp()
|
||||
return
|
||||
}
|
||||
else -> {
|
||||
System.err.println("Unknown subcommand: ${args[0]}")
|
||||
printHelp()
|
||||
kotlin.system.exitProcess(2)
|
||||
}
|
||||
}
|
||||
}
|
||||
runServer()
|
||||
}
|
||||
|
||||
private fun printHelp() {
|
||||
println("""
|
||||
agentik standalone — usage:
|
||||
java -jar agentik.jar Start HTTP server (AGUI + A2A + :proto)
|
||||
java -jar agentik.jar pull-model Download the LiteRT-LM model from static.binom.pw
|
||||
""".trimIndent())
|
||||
}
|
||||
|
||||
/**
|
||||
* Subcommand `pull-model`: скачивает LiteRT-LM-модель по конфигу.
|
||||
*
|
||||
* Конфиг читается из тех же env, что и server: `AGENTIK_LLM_BACKEND=google`
|
||||
* (если не google — exit 2), `AGENTIK_GOOGLE_MODEL_PATH` (куда), и опциональный
|
||||
* `AGENTIK_GOOGLE_MODEL_URL` (откуда; дефолт — Gemma-4-E2B-it.litertlm с
|
||||
* static.binom.pw).
|
||||
*
|
||||
* SHA-256 проверяется если задан `AGENTIK_GOOGLE_MODEL_SHA256_URL`.
|
||||
*
|
||||
* Если файл по PATH уже есть и совпадает по размеру с HEAD — no-op (exit 0).
|
||||
*/
|
||||
private fun runPullModel(args: List<String>) {
|
||||
val config = AgentikConfig.fromEnv()
|
||||
val google = config.llm.google
|
||||
?: error("pull-model: требуется AGENTIK_LLM_BACKEND=google (сейчас ${config.llm.backend})")
|
||||
|
||||
val url = System.getenv("AGENTIK_GOOGLE_MODEL_URL")
|
||||
?.takeIf { it.isNotBlank() }
|
||||
?: ModelDownloader.DEFAULT_GEMMA_URL
|
||||
val sha256Url = System.getenv("AGENTIK_GOOGLE_MODEL_SHA256_URL")
|
||||
?.takeIf { it.isNotBlank() }
|
||||
|
||||
println("pull-model: downloading from $url")
|
||||
println("pull-model: saving to ${google.modelPath}")
|
||||
if (sha256Url != null) println("pull-model: SHA-256 verification enabled ($sha256Url)")
|
||||
|
||||
val downloader = ModelDownloader()
|
||||
val result = runBlocking {
|
||||
downloader.download(
|
||||
url = url,
|
||||
destPath = google.modelPath,
|
||||
sha256Url = sha256Url,
|
||||
progress = { downloaded, total ->
|
||||
val pct = if (total > 0) (downloaded * 100.0 / total).toInt() else -1
|
||||
val human = if (total > 0) {
|
||||
"${formatBytes(downloaded)} / ${formatBytes(total)} ($pct%)"
|
||||
} else {
|
||||
formatBytes(downloaded)
|
||||
}
|
||||
print("\r $human ")
|
||||
},
|
||||
)
|
||||
}
|
||||
println()
|
||||
|
||||
if (result.bytes == 0L) {
|
||||
println("pull-model: already present (${formatBytes(result.total)}), nothing to do")
|
||||
} else if (result.resumedFrom > 0) {
|
||||
println("pull-model: resumed from ${formatBytes(result.resumedFrom)}, added ${formatBytes(result.bytes - result.resumedFrom)} in ${result.duration}")
|
||||
} else {
|
||||
println("pull-model: downloaded ${formatBytes(result.bytes)} in ${result.duration}")
|
||||
}
|
||||
}
|
||||
|
||||
private fun formatBytes(b: Long): String = when {
|
||||
b < 1024 -> "$b B"
|
||||
b < 1024L * 1024 -> "%.1f KB".format(b / 1024.0)
|
||||
b < 1024L * 1024 * 1024 -> "%.1f MB".format(b / 1024.0 / 1024.0)
|
||||
else -> "%.2f GB".format(b / 1024.0 / 1024.0 / 1024.0)
|
||||
}
|
||||
|
||||
private fun runServer() {
|
||||
val config = AgentikConfig.fromEnv()
|
||||
|
||||
// Перед созданием LLM: если backend=google и файл по AGENTIK_GOOGLE_MODEL_PATH
|
||||
// отсутствует — качаем автоматически (только при AGENTIK_AUTO_DOWNLOAD_MODEL=1),
|
||||
// иначе exit с понятной ошибкой.
|
||||
if (config.llm.backend == LlmBackend.GOOGLE) {
|
||||
val google = config.llm.google!!
|
||||
val modelFile = File(google.modelPath)
|
||||
if (!modelFile.exists()) {
|
||||
val url = System.getenv("AGENTIK_GOOGLE_MODEL_URL")
|
||||
?.takeIf { it.isNotBlank() }
|
||||
?: ModelDownloader.DEFAULT_GEMMA_URL
|
||||
val sha256Url = System.getenv("AGENTIK_GOOGLE_MODEL_SHA256_URL")
|
||||
?.takeIf { it.isNotBlank() }
|
||||
val autoDownload = System.getenv("AGENTIK_AUTO_DOWNLOAD_MODEL") == "1"
|
||||
if (autoDownload) {
|
||||
log.warn { "auto-download: $url -> ${google.modelPath}" }
|
||||
val dl = ModelDownloader()
|
||||
val result = runBlocking {
|
||||
dl.download(
|
||||
url = url,
|
||||
destPath = google.modelPath,
|
||||
sha256Url = sha256Url,
|
||||
progress = { d, t ->
|
||||
val pct = if (t > 0) (d * 100.0 / t).toInt() else -1
|
||||
if (t > 0) log.info { "auto-download: $pct% (${formatBytes(d)}/${formatBytes(t)})" }
|
||||
},
|
||||
)
|
||||
}
|
||||
log.info { "auto-download: done in ${result.duration}" }
|
||||
} else {
|
||||
System.err.println(
|
||||
"""
|
||||
|LiteRT-LM model file not found at: ${google.modelPath}
|
||||
|
|
||||
|Чтобы скачать автоматически, установите AGENTIK_AUTO_DOWNLOAD_MODEL=1
|
||||
|Чтобы скачать руками:
|
||||
| java -jar agentik.jar pull-model
|
||||
|(URL по умолчанию: $url)
|
||||
""".trimMargin(),
|
||||
)
|
||||
kotlin.system.exitProcess(2)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val llm = config.llm.createLlm()
|
||||
val storage = SqliteStores.open(dbPath = config.dbPath).asBundle()
|
||||
|
||||
@@ -0,0 +1,274 @@
|
||||
package pw.binom.agentik.standalone.llm
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.engine.cio.CIO
|
||||
import io.ktor.client.plugins.HttpTimeout
|
||||
import io.ktor.client.request.head
|
||||
import io.ktor.client.request.prepareGet
|
||||
import io.ktor.client.statement.bodyAsChannel
|
||||
import io.ktor.client.statement.bodyAsText
|
||||
import io.ktor.http.HttpHeaders
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.contentLength
|
||||
import io.ktor.http.isSuccess
|
||||
import io.ktor.utils.io.ByteReadChannel
|
||||
import io.ktor.utils.io.jvm.javaio.toInputStream
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
import mu.KotlinLogging
|
||||
import java.io.File
|
||||
import java.io.InputStream
|
||||
import java.io.RandomAccessFile
|
||||
import java.nio.file.Files
|
||||
import java.nio.file.StandardCopyOption
|
||||
import kotlin.time.Duration
|
||||
import kotlin.time.measureTime
|
||||
|
||||
private val log = KotlinLogging.logger {}
|
||||
|
||||
/**
|
||||
* HTTP-скачиватель больших моделей с поддержкой докачки через `Range: bytes=N-`.
|
||||
*
|
||||
* Используется двумя сценариями:
|
||||
* 1. `agentik pull-model` (subcommand в Main.kt) — явный прогон с прогрессом в stdout.
|
||||
* 2. На старте server'а, если `AGENTIK_AUTO_DOWNLOAD_MODEL=1` — качает только когда
|
||||
* файла по `AGENTIK_GOOGLE_MODEL_PATH` нет и `AGENTIK_LLM_BACKEND=google`.
|
||||
*
|
||||
* Алгоритм:
|
||||
* 1. HEAD `url` → `Content-Length` (полный размер), `Accept-Ranges: bytes` (опц.).
|
||||
* 2. Если `<dest>.part` уже есть и его размер < `Content-Length` — открываем
|
||||
* на append и шлём `Range: bytes=<part.size>-`. Если сервер возвращает
|
||||
* 200 OK вместо 206 — стартуем с нуля (`.part` удаляется, Range не
|
||||
* поддерживается).
|
||||
* 3. Стрим байтов пишем в `.part` через `RandomAccessFile`; каждые ~8 MB
|
||||
* зовём progress callback.
|
||||
* 4. На завершении сравниваем `.part.size()` с `Content-Length`; при наличии
|
||||
* [sha256Url] — проверяем SHA-256.
|
||||
* 5. Атомарный rename `.part` → финальный путь.
|
||||
*
|
||||
* Размер файла на 2026-09 — gemma-4-E2B-it.litertlm ~2.5 GB. Скорость
|
||||
* ограничена сетью клиента; resume при обрыве обязателен.
|
||||
*/
|
||||
class ModelDownloader(
|
||||
private val httpClient: HttpClient = defaultHttpClient(),
|
||||
) {
|
||||
/**
|
||||
* Скачивает [url] в [destPath]. Если файл уже есть и совпадает с ожидаемым
|
||||
* размером — возвращает [DownloadResult] с `bytes=0` и не качает заново.
|
||||
*
|
||||
* @param destPath абсолютный или относительный путь к финальному файлу.
|
||||
* @param sha256Url опциональный URL `.sha256` файла для верификации
|
||||
* (формат `<hex> <basename>` по стандарту `sha256sum` или просто hex).
|
||||
* Если `null` — верификация пропускается.
|
||||
* @param progress вызывается периодически с `(downloaded, total)`.
|
||||
* `total = -1` если неизвестен.
|
||||
*/
|
||||
suspend fun download(
|
||||
url: String,
|
||||
destPath: String,
|
||||
sha256Url: String? = null,
|
||||
progress: (downloaded: Long, total: Long) -> Unit = { _, _ -> },
|
||||
): DownloadResult {
|
||||
val destFile = File(destPath).absoluteFile
|
||||
destFile.parentFile?.mkdirs()
|
||||
|
||||
log.info { "model download: HEAD $url" }
|
||||
val (totalSize, supportsRanges) = probe(url)
|
||||
|
||||
// Файл уже на месте и совпадает по размеру — no-op.
|
||||
if (destFile.exists() && totalSize > 0 && destFile.length() == totalSize) {
|
||||
log.info { "model download: already present at $destPath (${destFile.length()} bytes), skipping" }
|
||||
return DownloadResult(bytes = 0, total = totalSize, resumedFrom = 0, duration = Duration.ZERO)
|
||||
}
|
||||
|
||||
val partFile = File("$destPath.part")
|
||||
val resumedFrom = if (partFile.exists() && partFile.length() > 0 && supportsRanges) {
|
||||
log.info { "model download: resuming from ${partFile.length()} bytes" }
|
||||
partFile.length()
|
||||
} else {
|
||||
if (partFile.exists() && partFile.length() > 0) {
|
||||
log.info { "model download: discarding stale .part (server doesn't support Range)" }
|
||||
partFile.delete()
|
||||
}
|
||||
0L
|
||||
}
|
||||
|
||||
val duration = measureTime {
|
||||
streamToFile(
|
||||
url = url,
|
||||
partFile = partFile,
|
||||
resumedFrom = resumedFrom,
|
||||
totalSize = totalSize,
|
||||
progress = progress,
|
||||
)
|
||||
}
|
||||
|
||||
if (totalSize > 0 && partFile.length() != totalSize) {
|
||||
error("model download: size mismatch — expected $totalSize bytes, got ${partFile.length()} bytes")
|
||||
}
|
||||
|
||||
if (sha256Url != null) {
|
||||
verifySha256(partFile, sha256Url)
|
||||
}
|
||||
|
||||
Files.move(
|
||||
partFile.toPath(),
|
||||
destFile.toPath(),
|
||||
StandardCopyOption.ATOMIC_MOVE,
|
||||
StandardCopyOption.REPLACE_EXISTING,
|
||||
)
|
||||
log.info { "model download: done — $destPath (${destFile.length()} bytes in ${duration})" }
|
||||
return DownloadResult(
|
||||
bytes = destFile.length(),
|
||||
total = totalSize,
|
||||
resumedFrom = resumedFrom,
|
||||
duration = duration,
|
||||
)
|
||||
}
|
||||
|
||||
private data class Probe(val totalSize: Long, val supportsRanges: Boolean)
|
||||
|
||||
private suspend fun probe(url: String): Probe {
|
||||
val resp = httpClient.head(url)
|
||||
val totalSize = resp.contentLength() ?: -1L
|
||||
val acceptRanges = resp.headers[HttpHeaders.AcceptRanges]?.equals("bytes", ignoreCase = true) == true
|
||||
if (totalSize <= 0) {
|
||||
log.warn { "model download: server did not return Content-Length, progress will be indeterminate" }
|
||||
}
|
||||
return Probe(totalSize = totalSize.coerceAtLeast(-1L), supportsRanges = acceptRanges)
|
||||
}
|
||||
|
||||
private suspend fun streamToFile(
|
||||
url: String,
|
||||
partFile: File,
|
||||
resumedFrom: Long,
|
||||
totalSize: Long,
|
||||
progress: (Long, Long) -> Unit,
|
||||
) {
|
||||
val statement = httpClient.prepareGet(url) {
|
||||
if (resumedFrom > 0) {
|
||||
headers.append(HttpHeaders.Range, "bytes=$resumedFrom-")
|
||||
}
|
||||
}
|
||||
|
||||
// prepareGet().execute { } ловит streaming response — пока лямбда не
|
||||
// вернулась, bodyAsChannel() читается по сети. Возвращать нужно после
|
||||
// полной вычитки, иначе Ktor закроет канал и оставшиеся байты пропадут.
|
||||
statement.execute { response ->
|
||||
when {
|
||||
response.status == HttpStatusCode.PartialContent && resumedFrom > 0 -> {
|
||||
// 206: докачка
|
||||
}
|
||||
response.status.isSuccess() -> {
|
||||
if (resumedFrom > 0) {
|
||||
log.warn { "model download: server ignored Range, restarting from 0" }
|
||||
}
|
||||
}
|
||||
else -> error("model download: HTTP ${response.status.value} ${response.status.description}")
|
||||
}
|
||||
|
||||
val channel: ByteReadChannel = response.bodyAsChannel()
|
||||
writeStream(channel, partFile, response.status, resumedFrom, totalSize, progress)
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun writeStream(
|
||||
channel: ByteReadChannel,
|
||||
partFile: File,
|
||||
status: HttpStatusCode,
|
||||
resumedFrom: Long,
|
||||
totalSize: Long,
|
||||
progress: (Long, Long) -> Unit,
|
||||
) {
|
||||
// Используем InputStream-обёртку поверх ByteReadChannel — она умеет
|
||||
// блокирующий read(buf) и не требует ручного pump'а через NIO ByteBuffer.
|
||||
// Преобразование в RandomAccessFile идёт в Dispatchers.IO.
|
||||
val input: InputStream = channel.toInputStream()
|
||||
withContext(Dispatchers.IO) {
|
||||
RandomAccessFile(partFile, "rw").use { raf ->
|
||||
if (status == HttpStatusCode.PartialContent) {
|
||||
raf.seek(resumedFrom)
|
||||
} else {
|
||||
raf.setLength(0L)
|
||||
}
|
||||
|
||||
val buf = ByteArray(64 * 1024)
|
||||
var downloaded = if (status == HttpStatusCode.PartialContent) resumedFrom else 0L
|
||||
var lastReported = downloaded
|
||||
val reportEvery = 8L * 1024 * 1024 // 8 MB
|
||||
|
||||
log.info { "model download: stream begin (resumedFrom=$resumedFrom, totalSize=$totalSize)" }
|
||||
input.use { stream ->
|
||||
while (true) {
|
||||
val read = stream.read(buf)
|
||||
if (read < 0) break
|
||||
if (read == 0) {
|
||||
// Согласно контракту InputStream.read(buf) может вернуть 0
|
||||
// если buf.length == 0 — не наш случай; но и как защита от
|
||||
// зацикливания на нулевом чтении даём планировщику тик.
|
||||
kotlinx.coroutines.yield()
|
||||
continue
|
||||
}
|
||||
raf.write(buf, 0, read)
|
||||
downloaded += read
|
||||
if (downloaded - lastReported >= reportEvery || (totalSize in 1..downloaded)) {
|
||||
progress(downloaded, totalSize)
|
||||
lastReported = downloaded
|
||||
}
|
||||
}
|
||||
}
|
||||
progress(downloaded, totalSize)
|
||||
log.info { "model download: stream end ($downloaded bytes written)" }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun verifySha256(file: File, sha256Url: String) {
|
||||
val expectedHex = httpClient.prepareGet(sha256Url).execute { response ->
|
||||
response.bodyAsText().trim().substringBefore(' ').lowercase()
|
||||
}
|
||||
val actual = withContext(Dispatchers.IO) {
|
||||
val digest = java.security.MessageDigest.getInstance("SHA-256")
|
||||
file.inputStream().use { input ->
|
||||
val buf = ByteArray(64 * 1024)
|
||||
while (true) {
|
||||
val n = input.read(buf)
|
||||
if (n <= 0) break
|
||||
digest.update(buf, 0, n)
|
||||
}
|
||||
}
|
||||
digest.digest().joinToString("") { "%02x".format(it) }
|
||||
}
|
||||
if (actual != expectedHex) {
|
||||
file.delete()
|
||||
error("model download: SHA-256 mismatch — expected $expectedHex, got $actual")
|
||||
}
|
||||
log.info { "model download: SHA-256 verified ($actual)" }
|
||||
}
|
||||
|
||||
data class DownloadResult(
|
||||
/** Сколько байт записано в текущем прогоне (0 = skip/no-op). */
|
||||
val bytes: Long,
|
||||
/** Полный размер файла по Content-Length (`-1` если неизвестен). */
|
||||
val total: Long,
|
||||
/** Сколько байт уже было в `.part` до старта текущего прогона (0 = с нуля). */
|
||||
val resumedFrom: Long,
|
||||
/** Время, потраченное на саму запись (без HEAD/verify). */
|
||||
val duration: Duration,
|
||||
)
|
||||
|
||||
companion object {
|
||||
const val DEFAULT_GEMMA_URL: String =
|
||||
"https://static.binom.pw/models/gemma-4-E2B-it.litertlm"
|
||||
|
||||
private fun defaultHttpClient(): HttpClient = HttpClient(CIO) {
|
||||
install(HttpTimeout) {
|
||||
// connectTimeout — дефолт Ktor (≈ секунды), requestTimeout снимаем:
|
||||
// скачивание 2.5 GB по медленной сети может занять минуты.
|
||||
requestTimeoutMillis = Long.MAX_VALUE
|
||||
}
|
||||
followRedirects = true
|
||||
expectSuccess = false
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,174 @@
|
||||
package pw.binom.agentik.standalone.llm
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.server.application.call
|
||||
import io.ktor.server.cio.CIO
|
||||
import io.ktor.server.engine.embeddedServer
|
||||
import io.ktor.server.response.respondBytes
|
||||
import io.ktor.server.response.respondText
|
||||
import io.ktor.server.routing.get
|
||||
import io.ktor.server.routing.head
|
||||
import io.ktor.server.routing.routing
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.HttpHeaders as KH
|
||||
import io.ktor.utils.io.toByteArray
|
||||
import java.io.File
|
||||
import java.net.ServerSocket
|
||||
import java.nio.file.Path
|
||||
import kotlin.io.path.createTempDirectory
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertContentEquals
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertTrue
|
||||
import kotlin.test.fail
|
||||
import kotlinx.coroutines.runBlocking
|
||||
|
||||
class ModelDownloaderTest {
|
||||
|
||||
private val payload = ByteArray(8192) { (it and 0xff).toByte() }
|
||||
|
||||
/** Поднимает fake-HTTP-server с поддержкой HEAD/GET/Range и возвращает `port`. */
|
||||
private fun startFakeServer(): FakeServer {
|
||||
val port = ServerSocket(0).use { it.localPort }
|
||||
val server = embeddedServer(CIO, port = port) {
|
||||
routing {
|
||||
head("/model.litertlm") {
|
||||
call.response.headers.append(KH.AcceptRanges, "bytes")
|
||||
call.response.headers.append(KH.ContentLength, payload.size.toString())
|
||||
call.respondText("")
|
||||
}
|
||||
get("/model.litertlm") {
|
||||
val range = call.request.headers[KH.Range]
|
||||
if (range == null) {
|
||||
call.response.headers.append(KH.ContentLength, payload.size.toString())
|
||||
call.respondBytes(payload)
|
||||
} else {
|
||||
// Parse "bytes=N-"
|
||||
val n = range.substringAfter("bytes=").substringBefore('-').toLong()
|
||||
val slice = payload.copyOfRange(n.toInt(), payload.size)
|
||||
call.response.status(HttpStatusCode.PartialContent)
|
||||
call.response.headers.append(KH.ContentRange, "bytes $n-${payload.size - 1}/${payload.size}")
|
||||
call.response.headers.append(KH.ContentLength, slice.size.toString())
|
||||
call.respondBytes(slice)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
server.start(wait = false)
|
||||
return FakeServer(server, port)
|
||||
}
|
||||
|
||||
private fun tmpFile(): File {
|
||||
val dir: Path = createTempDirectory(prefix = "agentik-test-")
|
||||
return dir.resolve("model.litertlm").toFile()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `download writes full file when no part exists`() = runBlocking {
|
||||
val fake = startFakeServer()
|
||||
try {
|
||||
val dest = tmpFile()
|
||||
val dl = ModelDownloader()
|
||||
val result = dl.download(
|
||||
url = "http://127.0.0.1:${fake.port}/model.litertlm",
|
||||
destPath = dest.absolutePath,
|
||||
)
|
||||
assertEquals(payload.size.toLong(), result.bytes)
|
||||
assertEquals(0L, result.resumedFrom)
|
||||
assertContentEquals(payload, dest.readBytes())
|
||||
assertTrue(!File("${dest.absolutePath}.part").exists(), "part file should be cleaned up")
|
||||
} finally {
|
||||
fake.server.stop(100, 200)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `download is no-op when final file already complete`() = runBlocking {
|
||||
val fake = startFakeServer()
|
||||
try {
|
||||
val dest = tmpFile()
|
||||
dest.writeBytes(payload)
|
||||
val dl = ModelDownloader()
|
||||
val result = dl.download(
|
||||
url = "http://127.0.0.1:${fake.port}/model.litertlm",
|
||||
destPath = dest.absolutePath,
|
||||
)
|
||||
assertEquals(0L, result.bytes)
|
||||
assertEquals(payload.size.toLong(), result.total)
|
||||
assertContentEquals(payload, dest.readBytes())
|
||||
} finally {
|
||||
fake.server.stop(100, 200)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `download resumes from existing part file with Range request`() = runBlocking {
|
||||
val fake = startFakeServer()
|
||||
try {
|
||||
val dest = tmpFile()
|
||||
val partFile = File("${dest.absolutePath}.part")
|
||||
val prefixSize = 4096
|
||||
partFile.writeBytes(payload.copyOfRange(0, prefixSize))
|
||||
|
||||
val dl = ModelDownloader()
|
||||
val result = dl.download(
|
||||
url = "http://127.0.0.1:${fake.port}/model.litertlm",
|
||||
destPath = dest.absolutePath,
|
||||
)
|
||||
assertEquals(prefixSize.toLong(), result.resumedFrom)
|
||||
assertEquals(payload.size.toLong(), result.bytes)
|
||||
assertContentEquals(payload, dest.readBytes())
|
||||
} finally {
|
||||
fake.server.stop(100, 200)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `download reports progress via callback`() = runBlocking {
|
||||
val fake = startFakeServer()
|
||||
try {
|
||||
val dest = tmpFile()
|
||||
val dl = ModelDownloader()
|
||||
val reports = mutableListOf<Pair<Long, Long>>()
|
||||
dl.download(
|
||||
url = "http://127.0.0.1:${fake.port}/model.litertlm",
|
||||
destPath = dest.absolutePath,
|
||||
progress = { d, t -> reports += d to t },
|
||||
)
|
||||
assertTrue(reports.isNotEmpty(), "progress must be reported at least once")
|
||||
assertEquals(payload.size.toLong(), reports.last().first)
|
||||
assertEquals(payload.size.toLong(), reports.last().second)
|
||||
} finally {
|
||||
fake.server.stop(100, 200)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `download fails with clear message on HTTP 404`() = runBlocking {
|
||||
// Spin up server that always 404s
|
||||
val port = ServerSocket(0).use { it.localPort }
|
||||
val server = embeddedServer(CIO, port = port) {
|
||||
routing {
|
||||
head("/missing") { call.respondText("", status = HttpStatusCode.NotFound) }
|
||||
get("/missing") { call.respondText("", status = HttpStatusCode.NotFound) }
|
||||
}
|
||||
}
|
||||
server.start(wait = false)
|
||||
try {
|
||||
val dest = tmpFile()
|
||||
val dl = ModelDownloader()
|
||||
try {
|
||||
dl.download(url = "http://127.0.0.1:$port/missing", destPath = dest.absolutePath)
|
||||
fail("expected failure on 404")
|
||||
} catch (e: Exception) {
|
||||
val msg = e.message ?: ""
|
||||
assertTrue("404" in msg || "Not Found" in msg,
|
||||
"error should mention HTTP 404, got: $msg")
|
||||
}
|
||||
} finally {
|
||||
server.stop(100, 200)
|
||||
}
|
||||
}
|
||||
|
||||
private data class FakeServer(val server: io.ktor.server.engine.EmbeddedServer<*, *>, val port: Int)
|
||||
}
|
||||
Reference in New Issue
Block a user