9 Commits

Author SHA1 Message Date
Jaakko Vanhala
80806498e0 Remote start stop control 2026-04-11 19:14:20 +03:00
Jaakko Vanhala
660e80c2bc natiivinodehommajuttuja 2026-04-11 18:14:08 +03:00
Jaakko Vanhala
591cfcb04b Päivitetyt kipina-node-binäärit: macOS, Linux x86/ARM, Windows
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-11 18:04:53 +03:00
Jaakko Vanhala
3cda57f0bc Hub: solmujen mallilistaus muistiin + /api/tags palauttaa verkon mallit
Natiivisolmun auth-viestistä tallennetaan mallilistaus node_models-mappiin.
/api/tags priorisoi verkon solmujen malleja lokaalin Ollaman edelle.
api_hardware käyttää tietokannan litteää rakennetta.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-11 18:04:41 +03:00
Jaakko Vanhala
23e7b92d03 kipina-node: auth-viesti välittää mallinimen ja Ollama-mallilistauksen hubille
build_auth_message käyttää nyt oikeaa mallinimeä hardkoodatun sijaan.
Lisäksi natiivisolmu hakee Ollaman mallilistauksen ja lähettää sen
auth-viestissä hubille.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-11 18:04:23 +03:00
Jaakko Vanhala
9f58febe21 Deploy-putki: Windows-build + automaattinen binäärikäännös
build-binaries.sh: lisätty Windows x86_64 (mingw-w64) neljänneksi
kohteeksi. deploy.sh: binäärit käännetään automaattisesti ennen
Docker-buildia, jolloin ne päätyvät Astron kautta kipina.studioon.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-11 18:03:53 +03:00
Jaakko Vanhala
b1de0d37f7 lisätty admin laitteistonäkymä 2026-04-11 17:42:17 +03:00
Jaakko Vanhala
4ff626ab88 broadcastit pois 2026-04-11 17:37:16 +03:00
Jaakko Vanhala
a45616046d Hub: broadcast-viestittely korvattu kohdennetulla reitityksellä
API-vastaukset käyttävät nyt oneshot-kanavaa broadcast-suodatuksen
sijaan, ja user_text lähetetään vain lähettäjäsolmulle. Stats-broadcast
säilyy UI:lle ja adminille.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-11 17:36:24 +03:00
15 changed files with 500 additions and 161 deletions

View File

@@ -4,19 +4,28 @@ set -e
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
OUT="$SCRIPT_DIR/frontend/public/download"
HASH_FILE="$OUT/.build-hash"
mkdir -p "$OUT"
# Tarkistetaan onko native-node muuttunut edellisen buildin jälkeen
CURRENT_HASH=$(git -C "$SCRIPT_DIR" log -1 --format=%H -- native-node/ Cargo.toml Cargo.lock)
if [ -f "$HASH_FILE" ] && [ "$(cat "$HASH_FILE")" = "$CURRENT_HASH" ]; then
echo "=== Kipinä Node — ei muutoksia, ohitetaan build ==="
ls -lh "$OUT"/kipina-node-* 2>/dev/null || true
exit 0
fi
echo "=== Kipinä Node — Binary Build ==="
# macOS ARM (natiivi)
echo "[1/3] macOS ARM64..."
echo "[1/4] macOS ARM64..."
cd "$SCRIPT_DIR"
cargo build --release -p native-node --no-default-features 2>&1 | tail -1
cp target/release/native-node "$OUT/kipina-node-macos-arm64"
echo " $(ls -lh "$OUT/kipina-node-macos-arm64" | awk '{print $5}')"
# Linux x86_64 (Docker)
echo "[2/3] Linux x86_64..."
echo "[2/4] Linux x86_64..."
docker run --rm \
-v "$SCRIPT_DIR":/app -w /app \
--platform linux/amd64 \
@@ -25,7 +34,7 @@ docker run --rm \
echo " $(ls -lh "$OUT/kipina-node-linux-x86_64" | awk '{print $5}')"
# Linux ARM64 (Docker)
echo "[3/3] Linux ARM64..."
echo "[3/4] Linux ARM64..."
docker run --rm \
-v "$SCRIPT_DIR":/app -w /app \
--platform linux/arm64 \
@@ -33,6 +42,18 @@ docker run --rm \
bash -c "apt-get update -qq && apt-get install -y -qq pkg-config libssl-dev >/dev/null 2>&1 && cargo build --release -p native-node --no-default-features 2>&1 | tail -1 && cp target/release/native-node /app/frontend/public/download/kipina-node-linux-arm64"
echo " $(ls -lh "$OUT/kipina-node-linux-arm64" | awk '{print $5}')"
# Windows x86_64 (Docker + mingw-w64)
echo "[4/4] Windows x86_64..."
docker run --rm \
-v "$SCRIPT_DIR":/app -w /app \
--platform linux/amd64 \
rust:slim \
bash -c "apt-get update -qq && apt-get install -y -qq gcc-mingw-w64-x86-64 pkg-config libssl-dev >/dev/null 2>&1 && rustup target add x86_64-pc-windows-gnu && cargo build --release -p native-node --no-default-features --target x86_64-pc-windows-gnu 2>&1 | tail -1 && cp target/x86_64-pc-windows-gnu/release/native-node.exe /app/frontend/public/download/kipina-node-windows-x86_64.exe"
echo " $(ls -lh "$OUT/kipina-node-windows-x86_64.exe" | awk '{print $5}')"
# Tallennetaan onnistuneen buildin hash
echo "$CURRENT_HASH" > "$HASH_FILE"
echo ""
echo "=== Binäärit valmiina ==="
ls -lh "$OUT"/kipina-node-*

View File

@@ -0,0 +1,13 @@
#!/bin/bash
# Deploy + native-node-binäärien käännös (jos muutoksia)
set -e
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
echo "=== Kipinä Studio Deploy (+ native binäärit) ==="
# Käännetään native-node-binäärit (ohittaa automaattisesti jos ei muutoksia)
"$SCRIPT_DIR/build-binaries.sh"
# Ajetaan normaali deploy
exec "$SCRIPT_DIR/deploy.sh"

View File

@@ -40,20 +40,20 @@ echo "[1/4] Rakennetaan image lokaalisti..."
docker build --platform linux/amd64 -f Dockerfile.prod -t kipina-agentic:latest .
# 2. Tallennetaan tiedostoon
echo "[2/5] Pakataan image..."
echo "[2/4] Pakataan image..."
docker save kipina-agentic:latest | gzip > /tmp/kipina-agentic.tar.gz
echo " Koko: $(du -h /tmp/kipina-agentic.tar.gz | cut -f1)"
# 3. Siirretään palvelimelle
echo "[3/5] Siirretään palvelimelle..."
echo "[3/4] Siirretään palvelimelle..."
scp $SSH_OPTS /tmp/kipina-agentic.tar.gz $SERVER:/tmp/
scp $SSH_OPTS docker-compose.prod.yml Caddyfile.prod $SERVER:$REMOTE_DIR/
# 4. Ladataan image ja käynnistetään
echo "[4/5] Ladataan image palvelimella..."
echo "[4/4] Ladataan image palvelimella..."
ssh $SSH_OPTS $SERVER "gunzip -c /tmp/kipina-agentic.tar.gz | docker load && rm /tmp/kipina-agentic.tar.gz"
echo "[5/5] Käynnistetään palvelut uudelleen..."
echo "[4/4] Käynnistetään palvelut uudelleen..."
ssh $SSH_OPTS $SERVER "cd $REMOTE_DIR && docker compose -f docker-compose.prod.yml down && docker compose -f docker-compose.prod.yml up -d"
echo "=== Valmis! https://kipina.studio ==="

View File

@@ -0,0 +1 @@
fc5fac8455bbb502ae98e22408bb92f9072cdf8a

Binary file not shown.

View File

@@ -4,7 +4,6 @@ set -e
BASE_URL="https://kipina.studio/download"
HUB_URL="${KIPINA_HUB:-wss://kipina.studio/ws}"
MODEL="${KIPINA_MODEL:-qwen2.5-coder:3b}"
OLLAMA_URL="${OLLAMA_URL:-http://localhost:11434}"
# Tunnista OS ja arkkitehtuuri
@@ -96,26 +95,35 @@ fi
echo ""
echo " Hub: $HUB_URL"
echo " Ollama: $OLLAMA_URL"
echo " Malli: $MODEL"
# Lataa malli (toimii sekä lokaalilla binäärillä että API:n kautta)
if ! curl -s "$OLLAMA_URL/api/tags" | grep -q "$MODEL"; then
echo " Ladataan $MODEL..."
curl -s "$OLLAMA_URL/api/pull" -d "{\"name\":\"$MODEL\"}" > /dev/null
if [ -n "$KIPINA_MODEL" ]; then
echo " Malli: $KIPINA_MODEL (Ympäristömuuttujasta)"
fi
echo " ✓ Malli $MODEL valmis"
# Lataa binääri
BIN_PATH="./kipina-node-bin"
if [ -f "$BIN_PATH" ]; then
echo ""
read -p " Löydettiin vanha kipina-node-bin lokaalisti. Haluatko poistaa sen ja ladata uusimman version? [y/N] " -r DEL_CHOICE
if [[ "$DEL_CHOICE" =~ ^[Yy]$ ]]; then
rm -f "$BIN_PATH"
echo " ✓ Vanha binääri poistettu."
fi
fi
if [ ! -f "$BIN_PATH" ]; then
echo " Ladataan $BINARY..."
echo " Ladataan tuorein $BINARY..."
curl -sSL "$BASE_URL/$BINARY" -o "$BIN_PATH"
chmod +x "$BIN_PATH"
fi
echo ""
echo " ✓ Yhdistetään laskentaverkkoon..."
echo " ✓ Siirrytään Kipinä Noden hallintaan..."
echo " Ctrl+C pysäyttää"
echo ""
HUB_URL="$HUB_URL" OLLAMA_URL="$OLLAMA_URL" OLLAMA_MODEL="$MODEL" exec "$BIN_PATH"
if [ -n "$KIPINA_MODEL" ]; then
export OLLAMA_MODEL="$KIPINA_MODEL"
fi
export HUB_URL="$HUB_URL"
export OLLAMA_URL="$OLLAMA_URL"
exec "$BIN_PATH"

View File

@@ -49,6 +49,13 @@ impl NodeDb {
INSERT INTO _schema_version VALUES (3);
");
}
if version < 4 {
let _ = conn.execute_batch("
ALTER TABLE node_sessions ADD COLUMN is_paused BOOLEAN DEFAULT 0;
DELETE FROM _schema_version;
INSERT INTO _schema_version VALUES (4);
");
}
conn.execute_batch("
CREATE TABLE IF NOT EXISTS node_sessions (
@@ -84,7 +91,10 @@ impl NodeDb {
has_webgpu BOOLEAN,
-- Tehtävätilastot
tasks_completed INTEGER DEFAULT 0
tasks_completed INTEGER DEFAULT 0,
-- Ohjaustilat
is_paused BOOLEAN DEFAULT 0
);
CREATE TABLE IF NOT EXISTS pair_results (
@@ -183,6 +193,14 @@ impl NodeDb {
);
}
pub fn update_session_status(&self, node_id: u64, is_paused: bool) {
let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
let _ = conn.execute(
"UPDATE node_sessions SET is_paused = ?1 WHERE node_id = ?2 AND disconnected_at IS NULL",
params![is_paused as i64, node_id as i64],
);
}
/// Sulkee saman IP:n viewer-sessiot kun aktiivinen node liittyy
pub fn close_viewers_by_ip(&self, ip: &str) {
let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
@@ -216,7 +234,7 @@ impl NodeDb {
"SELECT id, node_id, ip, node_type, connected_at, disconnected_at,
platform, hostname, os, cpu_cores, cpu_model, ram_mb,
gpu_name, gpu_vendor, gpu_backend, vram_total_mb, gpu_temp_c, gpu_util_pct,
allocated_gb, selected_task, has_webgpu, tasks_completed
allocated_gb, selected_task, has_webgpu, tasks_completed, is_paused
FROM node_sessions ORDER BY id DESC LIMIT ?1"
).unwrap();
@@ -244,6 +262,7 @@ impl NodeDb {
"selected_task": row.get::<_, Option<String>>(19)?,
"has_webgpu": row.get::<_, Option<bool>>(20)?,
"tasks_completed": row.get::<_, i64>(21)?,
"is_paused": row.get::<_, Option<bool>>(22)?.unwrap_or(false),
}))
}).unwrap().filter_map(|r| r.ok()).collect()
}

View File

@@ -25,7 +25,7 @@ const ALLOWED_ORIGINS: &[&str] = &[
];
// Sallitut viestityyypit clientilta
const ALLOWED_MSG_TYPES: &[&str] = &["auth", "result", "pair_done", "llm_chunk", "llm_done", "llm_error", "download_progress", "user_text", "single_tokenize_done"];
const ALLOWED_MSG_TYPES: &[&str] = &["auth", "result", "pair_done", "llm_chunk", "llm_done", "llm_error", "download_progress", "user_text", "single_tokenize_done", "status_update"];
struct AppState {
next_node_id: Mutex<u64>,
@@ -40,9 +40,12 @@ struct AppState {
node_ips: Mutex<HashMap<u64, IpAddr>>,
node_tasks: Mutex<HashMap<u64, String>>, // node_id → selected_task
node_types: Mutex<HashMap<u64, String>>, // node_id → "native" | "browser"
node_paused: Mutex<std::collections::HashSet<u64>>, // node_id → onko tauolla
node_busy: Mutex<std::collections::HashSet<u64>>, // Solmut joilla on aktiivinen tehtävä
pending_task_ids: Mutex<std::collections::HashSet<String>>, // Hubin jakamat task_id:t (gamification-validointi)
pending_responses: Mutex<HashMap<String, tokio::sync::oneshot::Sender<serde_json::Value>>>, // task_id → oneshot API-vastaukselle
api_rate_limits: Mutex<HashMap<IpAddr, (std::time::Instant, u32)>>, // IP → (ikkuna-alku, pyyntömäärä)
node_models: tokio::sync::RwLock<HashMap<u64, serde_json::Value>>, // node_id → ollama tags JSON
db: db::NodeDb,
}
@@ -80,6 +83,8 @@ tr:hover td { background:#1c2333; }
.table-wrap { overflow-x:auto; max-height:70vh; overflow-y:auto; }
.online { color:var(--green); }
.offline { color:#8b949e; }
.pause-btn { background:var(--panel); border:1px solid var(--border); color:var(--text); padding:4px 8px; border-radius:4px; cursor:pointer; font-size:12px; }
.pause-btn:hover { border-color:var(--yellow); }
</style>
</head>
<body>
@@ -91,6 +96,7 @@ tr:hover td { background:#1c2333; }
<div class="tabs">
<div class="tab active" onclick="showTab('sessions')">Sessiot</div>
<div class="tab" onclick="showTab('pairs')">Tokenisointiparit</div>
<div class="tab" onclick="showTab('hardware')">Laitteisto & Mallit</div>
</div>
<div id="sessions" class="panel active">
@@ -99,12 +105,12 @@ tr:hover td { background:#1c2333; }
<colgroup>
<col style="width:35px"><col style="width:85px"><col style="width:95px"><col style="width:65px"><col style="width:110px"><col style="width:80px">
<col style="width:65px"><col style="width:40px"><col style="width:70px"><col style="width:90px"><col style="width:60px">
<col style="width:65px"><col style="width:40px"><col style="width:130px"><col style="width:60px">
<col style="width:65px"><col style="width:40px"><col style="width:130px"><col style="width:60px"><col style="width:80px">
</colgroup>
<thead><tr>
<th>ID</th><th>Tila</th><th>Tehtävä</th><th>Tyyppi</th><th>IP</th><th>Alusta</th>
<th>OS</th><th>CPU</th><th>RAM</th><th>GPU</th><th>VRAM</th>
<th>WebGPU</th><th>Teht.</th><th>Yhdistetty</th><th>Kesto</th>
<th>WebGPU</th><th>Teht.</th><th>Yhdistetty</th><th>Kesto</th><th>Toiminnot</th>
</tr></thead><tbody id="sessions-body"></tbody></table>
</div>
</div>
@@ -118,6 +124,19 @@ tr:hover td { background:#1c2333; }
</div>
</div>
<div id="hardware" class="panel">
<div class="stats-grid" id="hardware-stats"></div>
<h2 style="margin-top: 10px; margin-bottom: 10px; color: var(--accent); font-size: 16px;">Käytettävissä olevat paikalliset kielimallit</h2>
<div class="table-wrap">
<table>
<thead><tr>
<th>Nimi</th><th>Koko</th><th>Parametrit</th>
</tr></thead>
<tbody id="models-body"></tbody>
</table>
</div>
</div>
<script>
function showTab(name) {
document.querySelectorAll('.panel').forEach(p => p.classList.remove('active'));
@@ -149,12 +168,16 @@ function duration(start, end) {
}
async function load() {
const [statsRes, sessionsRes, pairsRes] = await Promise.all([
fetch('/api/stats'), fetch('/api/sessions'), fetch('/api/pairs')
const [statsRes, sessionsRes, pairsRes, hwRes, modelsRes] = await Promise.all([
fetch('/api/stats'), fetch('/api/sessions'), fetch('/api/pairs'),
fetch('/api/v1/hardware').catch(() => ({json: async()=>({gpu_name:'', vram_mb:0, ram_mb:0})})),
fetch('/api/v1/ollama/tags').catch(() => ({json: async()=>({models:[]})}))
]);
const stats = await statsRes.json();
const sessions = await sessionsRes.json();
const pairs = await pairsRes.json();
const hw = await hwRes.json().catch(() => ({gpu_name:'', vram_mb:0, ram_mb:0}));
const modelsData = await modelsRes.json().catch(() => ({models:[]}));
// Versio
if (stats.version) document.getElementById('admin-version').textContent = 'v' + stats.version;
@@ -190,9 +213,17 @@ async function load() {
document.getElementById('sessions-body').innerHTML = sessions.map(s => {
const online = !s.disconnected_at;
const isViewer = s.selected_task === 'viewer';
const status = online
? (isViewer ? '<span style="color:#d29922">CONNECTED</span>' : '<span class="online">ACTIVE</span>')
: '<span class="offline">offline</span>';
let status;
if (!online) {
status = '<span class="offline">offline</span>';
} else if (isViewer) {
status = '<span style="color:#d29922">CONNECTED</span>';
} else if (s.is_paused) {
status = '<span style="color:#8b949e">PAUSED</span>';
} else {
status = '<span class="online">ACTIVE</span>';
}
const typeBadge = s.node_type === 'native' ? badge('native','blue') : badge('browser','yellow');
const taskColor = isViewer ? 'yellow' : s.selected_task === 'tokenize' ? 'green' : 'blue';
const taskBadge = badge(taskNames[s.selected_task] || s.selected_task || '?', taskColor);
@@ -205,11 +236,16 @@ async function load() {
const os = s.os || '-';
const time = s.connected_at ? new Date(s.connected_at).toLocaleString('fi-FI') : '';
const dur = duration(s.connected_at, s.disconnected_at);
const actionBtn = online && !isViewer
? `<button class="pause-btn" onclick="togglePause(${s.node_id}, ${s.is_paused})">${s.is_paused ? '▶ Työhön' : '⏸ Tauolle'}</button>`
: '';
return `<tr>
<td>${s.node_id}</td><td>${status}</td><td>${taskBadge}</td><td>${typeBadge}</td><td>${s.ip}</td>
<td>${plat}</td><td>${os}</td><td>${cores}</td><td>${ram}</td>
<td>${gpu}</td><td>${vram}</td><td>${gpuBadge}</td>
<td>${s.tasks_completed}</td><td>${time}</td><td>${dur}</td>
<td>${actionBtn}</td>
</tr>`;
}).join('');
@@ -229,6 +265,35 @@ async function load() {
<td>${p.duration_ms||0}ms</td>
</tr>`;
}).join('');
// Hardware
document.getElementById('hardware-stats').innerHTML = [
{v: hw.gpu_name || '-', l: 'Paikallinen GPU tila'},
{v: hw.vram_mb ? hw.vram_mb + ' MB' : '-', l: 'GPU Muisti (VRAM)'},
{v: hw.ram_mb ? hw.ram_mb + ' MB' : '-', l: 'RAM'},
].map(s => `<div class="stat-card"><div class="val">${s.v}</div><div class="label">${s.l}</div></div>`).join('');
// Models
document.getElementById('models-body').innerHTML = (modelsData.models || []).map(m => {
const sizeGb = (m.size / (1024*1024*1024)).toFixed(2) + ' GB';
const params = m.details?.parameter_size || '-';
return `<tr>
<td><strong>${m.name}</strong></td>
<td>${sizeGb}</td>
<td>${params}</td>
</tr>`;
}).join('');
}
async function togglePause(nodeId, isPaused) {
try {
await fetch('/api/v1/control/' + nodeId, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ action: isPaused ? 'resume' : 'pause' })
});
load(); // virkistetään
} catch(e) { console.error(e); }
}
load();
@@ -262,9 +327,12 @@ async fn main() {
node_ips: Mutex::new(HashMap::new()),
node_tasks: Mutex::new(HashMap::new()),
node_types: Mutex::new(HashMap::new()),
node_paused: Mutex::new(std::collections::HashSet::new()),
node_busy: Mutex::new(std::collections::HashSet::new()),
pending_task_ids: Mutex::new(std::collections::HashSet::new()),
pending_responses: Mutex::new(HashMap::new()),
api_rate_limits: Mutex::new(HashMap::new()),
node_models: tokio::sync::RwLock::new(HashMap::new()),
db: db::NodeDb::new(&std::env::var("DATABASE_PATH").unwrap_or_else(|_| "nodes.db".to_string())),
});
@@ -381,6 +449,7 @@ async fn main() {
.route("/api/pairs", get(api_pairs))
.route("/api/stats", get(api_stats))
.route("/api/v1/chat/completions", axum::routing::post(api_chat_completions))
.route("/api/v1/control/:id", axum::routing::post(api_control_node))
.route("/api/v1/model", axum::routing::post(api_change_model))
.route("/api/v1/hardware", get(api_hardware))
.route("/api/v1/ollama/tags", get(api_ollama_tags))
@@ -400,6 +469,26 @@ async fn main() {
axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>()).await.unwrap();
}
async fn api_control_node(
headers: axum::http::HeaderMap,
axum::extract::State(state): axum::extract::State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<u64>,
axum::Json(payload): axum::Json<serde_json::Value>,
) -> axum::response::Response {
if !check_admin_auth(&headers) { return admin_unauthorized(); }
let action = payload.get("action").and_then(|v| v.as_str()).unwrap_or("");
if action == "pause" || action == "resume" {
let msg = serde_json::json!({ "type": "control", "action": action });
let channels = state.node_channels.read().await;
if let Some(tx) = channels.get(&id) {
let _ = tx.send(msg.to_string());
tracing::info!("Lähetetty control: {} solmulle {}", action, id);
return axum::Json(serde_json::json!({"status": "ok"})).into_response();
}
}
(axum::http::StatusCode::BAD_REQUEST, "Invalid action or node offline").into_response()
}
async fn api_sessions(
headers: axum::http::HeaderMap,
axum::extract::State(state): axum::extract::State<Arc<AppState>>,
@@ -730,6 +819,9 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, ip: IpAddr) {
}
state.node_tasks.lock().unwrap().insert(node_id, selected_task);
state.node_types.lock().unwrap().insert(node_id, node_type.to_string());
// Uudelleen-kirjautuessa nollataan tauko
state.node_paused.lock().unwrap().remove(&node_id);
state.db.update_session_status(node_id, false);
if node_type == "native" {
let sys = json.get("system");
@@ -743,6 +835,12 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, ip: IpAddr) {
node_id, ip, hostname, os, cores, ram, allocated
);
// Tallennetaan välitetyt mallit muistiin
if let Some(models) = json.get("models") {
let mut nm = state.node_models.write().await;
nm.insert(node_id, models.clone());
}
if let Some(gpus) = json.get("gpus").and_then(|v| v.as_array()) {
for gpu in gpus {
tracing::info!(
@@ -780,6 +878,18 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, ip: IpAddr) {
});
let _ = state.stats_tx.send(join_msg.to_string());
} else if msg_type == "status_update" {
let status = json.get("status").and_then(|v| v.as_str()).unwrap_or("active");
if status == "paused" {
state.node_paused.lock().unwrap().insert(node_id);
state.db.update_session_status(node_id, true);
tracing::info!("Solmu {} ({}) asettui tauolle.", node_id, ip);
} else {
state.node_paused.lock().unwrap().remove(&node_id);
state.db.update_session_status(node_id, false);
tracing::info!("Solmu {} ({}) on taas aktiivinen.", node_id, ip);
}
broadcast_stats(&state).await;
} else if msg_type == "result" {
tracing::info!("Solmu {} sai tuloksen: {}", node_id, text);
{
@@ -875,11 +985,18 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, ip: IpAddr) {
} else if msg_type == "llm_done" {
// Vapautetaan solmu ja tarkistetaan task_id:n aitous
state.node_busy.lock().unwrap().remove(&node_id);
let valid_task = if let Some(tid) = json.get("task_id").and_then(|v| v.as_str()) {
state.pending_task_ids.lock().unwrap().remove(tid)
let task_id = json.get("task_id").and_then(|v| v.as_str()).map(|s| s.to_string());
let valid_task = if let Some(ref tid) = task_id {
state.pending_task_ids.lock().unwrap().remove(tid.as_str())
} else {
false
};
// Jos API-pyyntö odottaa tätä vastausta, reititetään suoraan oneshot-kanavaan
let api_sender = task_id.as_ref().and_then(|tid| {
state.pending_responses.lock().unwrap().remove(tid)
});
{
let mut json = json;
if let Some(obj) = json.as_object_mut() {
@@ -899,6 +1016,12 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, ip: IpAddr) {
state.db.increment_tasks(node_id);
obj.insert("node_id".to_string(), serde_json::json!(node_id));
}
if let Some(sender) = api_sender {
// API-pyyntö: reititetään vastaus suoraan odottajalle
let _ = sender.send(json.clone());
}
// UI-broadcast jatkuu normaalisti
let _ = state.stats_tx.send(json.to_string());
let active_incentives = state.feature_flags.read().await.get("Insentiivit").copied().unwrap_or(false);
@@ -908,7 +1031,7 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, ip: IpAddr) {
{
let mut task_count = state.total_tasks.lock().unwrap();
*task_count += 1;
if active_incentives && valid_task {
let mut tokens = state.nodes_tokens.lock().unwrap();
let balance = tokens.entry(node_id).or_insert(0);
@@ -916,7 +1039,7 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, ip: IpAddr) {
current_balance = *balance;
}
}
if active_incentives && ui_sync {
if let Some(tx) = state.node_channels.read().await.get(&node_id) {
let msg = serde_json::json!({
@@ -926,45 +1049,50 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, ip: IpAddr) {
let _ = tx.send(msg.to_string());
}
}
broadcast_stats(&state).await;
}
} else if msg_type == "llm_error" {
state.node_busy.lock().unwrap().remove(&node_id);
if let Some(tid) = json.get("task_id").and_then(|v| v.as_str()) {
state.pending_task_ids.lock().unwrap().remove(tid);
let task_id = json.get("task_id").and_then(|v| v.as_str()).map(|s| s.to_string());
if let Some(ref tid) = task_id {
state.pending_task_ids.lock().unwrap().remove(tid.as_str());
}
// Jos API-pyyntö odottaa, reititetään virhe oneshot-kanavaan
let api_sender = task_id.as_ref().and_then(|tid| {
state.pending_responses.lock().unwrap().remove(tid)
});
{
let mut json = json;
if let Some(obj) = json.as_object_mut() {
obj.insert("node_id".to_string(), serde_json::json!(node_id));
}
if let Some(sender) = api_sender {
let _ = sender.send(json.clone());
}
let _ = state.stats_tx.send(json.to_string());
}
} else if msg_type == "user_text" {
// Käyttäjän lähettämä teksti — broadcastataan pair_taskina ja llm_promptina
// Käyttäjän lähettämä teksti — kohdennettu reititys lähettäjäsolmulle
let text = json.get("text").and_then(|v| v.as_str()).unwrap_or("").to_string();
let task_type = json.get("task_type").and_then(|v| v.as_str()).unwrap_or("tokenize");
if !text.is_empty() {
let preview: String = text.chars().take(80).collect();
tracing::info!("Solmu {} lähetti oman tekstin ({}): \"{}\"", node_id, task_type, preview);
match task_type {
"tokenize" => {
let msg = serde_json::json!({
"type": "single_tokenize",
"text": text,
});
let _ = state.stats_tx.send(msg.to_string());
}
_ => {
// LLM-prompti: lähetetään VAIN valitulle mallille, ei kaikille (välttää turhaa ruuhkaa ja busy-tiloja)
let prompt = serde_json::json!({
"type": "llm_prompt",
"prompt": text,
"model": task_type,
});
let _ = state.stats_tx.send(prompt.to_string());
}
let msg = match task_type {
"tokenize" => serde_json::json!({
"type": "single_tokenize",
"text": text,
}),
_ => serde_json::json!({
"type": "llm_prompt",
"prompt": text,
"model": task_type,
}),
};
// Lähetetään takaisin lähettäjäsolmulle (käyttäjä haluaa oman tekstinsä tuloksen)
if let Some(tx) = state.node_channels.read().await.get(&node_id) {
let _ = tx.send(msg.to_string());
}
}
}
@@ -989,6 +1117,8 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, ip: IpAddr) {
vram.remove(&node_id);
}
state.node_types.lock().unwrap().remove(&node_id);
state.node_paused.lock().unwrap().remove(&node_id);
state.node_models.write().await.remove(&node_id);
tracing::info!("Solmu {} ({}) poistui verkosta.", node_id, ip);
broadcast_stats(&state).await;
sender_task.abort();
@@ -1009,7 +1139,16 @@ struct ChatCompletionResponse {
tokens_generated: u64,
}
async fn api_ollama_tags() -> axum::response::Response {
async fn api_ollama_tags(
axum::extract::State(state): axum::extract::State<Arc<AppState>>,
) -> axum::response::Response {
// Haetaan natiivisolmun tila muistista — priorisoidaan aito verkko-solmu
let node_models = state.node_models.read().await;
if let Some((_, models_json)) = node_models.iter().next() {
return axum::Json(models_json.clone()).into_response();
}
// Fallback: Haetaan lokaalista infra-Ollamasta ohjaimesta käsin (esim dev ympäristö)
let ollama_url = std::env::var("OLLAMA_URL").unwrap_or_else(|_| "http://ollama:11434".to_string());
match reqwest::get(format!("{}/api/tags", ollama_url)).await {
Ok(resp) => {
@@ -1033,11 +1172,10 @@ async fn api_hardware(
});
let (mut vram_mb, mut gpu_name, ram_mb) = if let Some(s) = native {
let gpus = s.get("gpus").and_then(|v| v.as_array());
let gpu = gpus.and_then(|g| g.first());
let vram = gpu.and_then(|g| g.get("vram_total_mb")).and_then(|v| v.as_u64()).unwrap_or(0);
let name = gpu.and_then(|g| g.get("name")).and_then(|v| v.as_str()).unwrap_or("").to_string();
let ram = s.get("system").and_then(|v| v.get("ram_total_mb")).and_then(|v| v.as_u64()).unwrap_or(0);
// Tieto on tietokannassa litteänä
let vram = s.get("vram_total_mb").and_then(|v| v.as_u64()).unwrap_or(0);
let name = s.get("gpu_name").and_then(|v| v.as_str()).unwrap_or("").to_string();
let ram = s.get("ram_mb").and_then(|v| v.as_u64()).unwrap_or(0);
(vram, name, ram)
} else {
(0, String::new(), 0)
@@ -1106,7 +1244,9 @@ async fn api_chat_completions(
let tasks = state.node_tasks.lock().unwrap();
let _busy = state.node_busy.lock().unwrap();
let node_types = state.node_types.lock().unwrap();
let matching: Vec<u64> = tasks.iter().filter(|(_, task)| {
let paused = state.node_paused.lock().unwrap();
let matching: Vec<u64> = tasks.iter().filter(|(k, task)| {
if paused.contains(k) { return false; } // Ei sallita tauotettuja
// Eksakti match tai qwen-perheen yhteensopivuus (selain: qwen-coder-05b, natiivi: qwen2.5-coder:7b)
let req_model = payload.model.to_lowercase();
let node_task = task.to_lowercase();
@@ -1161,8 +1301,9 @@ async fn api_chat_completions(
msg.as_object_mut().unwrap().insert("max_tokens".to_string(), serde_json::json!(mt));
}
// Odotuskanava valmiiksi (solmu palauttaa tuloksen stats_tx kautta)
let mut rx = state.stats_tx.subscribe();
// Oneshot-kanava: solmu palauttaa tuloksen suoraan tälle pyynnölle
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel::<serde_json::Value>();
state.pending_responses.lock().unwrap().insert(payload.task_id.clone(), resp_tx);
// Kohdennettu reititys: lähetetään AI-tehtävä suoraan VAIN valitulle solmulle
{
@@ -1171,48 +1312,34 @@ async fn api_chat_completions(
let _ = tx.send(msg.to_string());
tracing::info!("Reititettiin API-pyyntö solmulle {} (Malli: {})", target_node_id, payload.model);
} else {
state.pending_responses.lock().unwrap().remove(&payload.task_id);
return (axum::http::StatusCode::SERVICE_UNAVAILABLE, "Verkkovirhe: solmun yhteys katkesi reitityksen aikana").into_response();
}
}
let timeout = tokio::time::timeout(std::time::Duration::from_secs(600), async move {
loop {
let msg_str = match rx.recv().await {
Ok(msg) => msg,
Err(broadcast::error::RecvError::Lagged(n)) => {
tracing::debug!("API-kanava lagged {} viestiä", n);
continue;
}
Err(_) => return Ok(None), // Kanava suljettu
};
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&msg_str) {
if v["type"].as_str() == Some("llm_done") {
if let Some(tid) = v["task_id"].as_str() {
if tid == payload.task_id {
return Ok(Some(ChatCompletionResponse {
response: v["response"].as_str().unwrap_or("").to_string(),
model: v["model"].as_str().unwrap_or("").to_string(),
tokens_generated: v["tokens_generated"].as_u64().unwrap_or(0),
}));
}
}
} else if v["type"].as_str() == Some("llm_error") {
if let Some(tid) = v["task_id"].as_str() {
if tid == payload.task_id {
return Err(v["error"].as_str().unwrap_or("Määrittelemätön virhe solmussa").to_string());
}
}
}
}
}
#[allow(unreachable_code)]
Ok(None)
}).await;
let timeout = tokio::time::timeout(std::time::Duration::from_secs(600), resp_rx).await;
match timeout {
Ok(Ok(Some(res))) => axum::Json(res).into_response(),
Ok(Ok(None)) => (axum::http::StatusCode::INTERNAL_SERVER_ERROR, "Verkkovirhe: yhteys katkesi").into_response(),
Ok(Err(err)) => (axum::http::StatusCode::CONFLICT, err).into_response(),
Err(_) => (axum::http::StatusCode::GATEWAY_TIMEOUT, "Aikakatkaisu: solmu ei saanut tehtävää ajoissa valmiiksi").into_response(),
Ok(Ok(v)) => {
if v["type"].as_str() == Some("llm_error") {
let err = v["error"].as_str().unwrap_or("Määrittelemätön virhe solmussa").to_string();
(axum::http::StatusCode::CONFLICT, err).into_response()
} else {
axum::Json(ChatCompletionResponse {
response: v["response"].as_str().unwrap_or("").to_string(),
model: v["model"].as_str().unwrap_or("").to_string(),
tokens_generated: v["tokens_generated"].as_u64().unwrap_or(0),
}).into_response()
}
}
Ok(Err(_)) => {
// Oneshot-kanava sulkeutui (solmu katosi)
state.pending_responses.lock().unwrap().remove(&payload.task_id);
(axum::http::StatusCode::INTERNAL_SERVER_ERROR, "Verkkovirhe: yhteys katkesi").into_response()
}
Err(_) => {
state.pending_responses.lock().unwrap().remove(&payload.task_id);
(axum::http::StatusCode::GATEWAY_TIMEOUT, "Aikakatkaisu: solmu ei saanut tehtävää ajoissa valmiiksi").into_response()
}
}
}

View File

@@ -19,3 +19,5 @@ wgpu = { version = "24", optional = true }
reqwest = { version = "0.12", features = ["json"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
dialoguer = "0.12.0"
console = "0.16.3"

View File

@@ -9,8 +9,6 @@ pub struct LlmEngine {
impl LlmEngine {
pub async fn load() -> Result<Self, String> {
let model = std::env::var("OLLAMA_MODEL").unwrap_or_else(|_| "qwen2.5-coder:3b".to_string());
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(600))
.connect_timeout(std::time::Duration::from_secs(3))
@@ -48,6 +46,13 @@ impl LlmEngine {
})
};
// Kysytään malli TUI:lla jos ei pakotettu ympäristöstä
let model = if let Ok(m) = std::env::var("OLLAMA_MODEL") {
m
} else {
crate::tui::select_model(&ollama_url, &client).await?
};
tracing::info!("Ollama backend: {} | malli: {}", ollama_url, model);
Ok(LlmEngine { ollama_url, model: RefCell::new(model), client })
}
@@ -78,6 +83,20 @@ impl LlmEngine {
}
}
/// Hakee kaikki Ollamaan asennetut mallit
pub async fn fetch_models(&self) -> Result<serde_json::Value, String> {
let resp = self.client.get(format!("{}/api/tags", self.ollama_url))
.send()
.await
.map_err(|e| format!("Ollama tags fetch: {}", e))?;
if resp.status().is_success() {
resp.json().await.map_err(|e| format!("Ollama tags json: {}", e))
} else {
Err(format!("Ollama tags epäonnistui: {}", resp.status()))
}
}
pub async fn generate(&self, prompt: &str, max_tokens: usize) -> Result<GenerateResult, String> {
// System prompt tulee agentin konfiguraatiosta (frontend lähettää sen osana promptia).
// Tässä ei yliajeta sitä — Ollama saa vain prompt-kentän.

View File

@@ -5,6 +5,7 @@ use tokio_tungstenite::connect_async;
use tokio_tungstenite::tungstenite::Message;
mod inference;
mod tui;
/// GPU-tietorakenne — yhtenäinen kaikille valmistajille
struct GpuInfo {
@@ -222,7 +223,7 @@ fn collect_system_info() -> serde_json::Value {
}
/// Koko auth-viesti hubille
fn build_auth_message(allocated_gb: u32) -> String {
fn build_auth_message(allocated_gb: u32, model_name: &str, models_data: Option<serde_json::Value>) -> String {
let sys = collect_system_info();
let gpus = collect_all_gpus();
@@ -239,7 +240,7 @@ fn build_auth_message(allocated_gb: u32) -> String {
"status": "agent_ready",
"node_type": "native",
"allocated_gb": allocated_gb,
"selected_task": "qwen2.5-coder:7b",
"selected_task": model_name,
"system": sys,
});
@@ -251,6 +252,10 @@ fn build_auth_message(allocated_gb: u32) -> String {
msg.as_object_mut().unwrap().insert("gpus".to_string(), json!(gpu_json));
}
if let Some(models) = models_data {
msg.as_object_mut().unwrap().insert("models".to_string(), models);
}
msg.to_string()
}
@@ -321,6 +326,22 @@ async fn main() {
}
};
let active_model = llm.as_ref().map(|e| e.model_name()).unwrap_or_else(|| "unknown".to_string());
tracing::info!("Käytettävä kielimalli konfiguroitu (selected_task): {}", active_model);
// Haetaan paikalliset mallit hubille lähetettäväksi
let mut available_models = None;
if let Some(ref engine) = llm {
match engine.fetch_models().await {
Ok(models) => {
available_models = Some(models);
}
Err(e) => {
tracing::warn!("Mallilistauksen haku epäonnistui: {}", e);
}
}
}
// Yhdistetään hubiin
loop {
match connect_async(&hub_url).await {
@@ -328,80 +349,121 @@ async fn main() {
tracing::info!("Yhdistetty hubiin!");
let (mut write, mut read) = ws_stream.split();
let auth = build_auth_message(allocated_gb);
let auth = build_auth_message(allocated_gb, &active_model, available_models.clone());
if write.send(Message::Text(auth)).await.is_err() {
tracing::error!("Auth-viestin lähetys epäonnistui");
continue;
}
while let Some(Ok(msg)) = read.next().await {
if let Message::Text(text) = msg {
// LLM-promptit
if text.contains("llm_prompt") {
if let Ok(task) = serde_json::from_str::<serde_json::Value>(&text) {
let prompt = task.get("prompt").and_then(|v| v.as_str()).unwrap_or("");
let task_id = task.get("task_id").and_then(|v| v.as_str()).unwrap_or("?");
let msg_model = task.get("model").and_then(|v| v.as_str()).unwrap_or("");
if !prompt.is_empty() && (msg_model.starts_with("qwen-coder") || msg_model.starts_with("qwen2.5-coder")) {
use tokio::io::AsyncBufReadExt;
let mut stdin_lines = tokio::io::BufReader::new(tokio::io::stdin()).lines();
if let Some(ref engine) = llm {
let max_tokens = task.get("max_tokens").and_then(|v| v.as_u64()).unwrap_or(1024) as usize;
let prompt_lines = prompt.lines().count();
let prompt_last: String = prompt.lines().last().unwrap_or("").chars().take(60).collect();
tracing::info!("→ task_id:{} | {}r prompti | \"{}...\"", task_id, prompt_lines, prompt_last);
let model_name = engine.model_name();
match engine.generate(prompt, max_tokens).await {
Ok(result) => {
tracing::info!(
"✓ {} | {} tok | {:.0}ms | {:.1} tok/s",
model_name,
result.tokens_generated,
result.duration_ms,
result.tokens_per_sec,
);
// Lähetetään vain lyhyt prompti-esikatselu (ei koko kontekstia)
let prompt_short: String = prompt.lines().last().unwrap_or("").chars().take(100).collect();
let done = json!({
"type": "llm_done",
"prompt": prompt_short,
"model": format!("{} (Ollama)", model_name),
"response": result.text,
"tokens_generated": result.tokens_generated,
"duration_ms": result.duration_ms,
"tokens_per_sec": (result.tokens_per_sec * 10.0).round() / 10.0,
"load_time_ms": 0,
"task_id": task_id,
});
let _ = write.send(Message::Text(done.to_string())).await;
}
Err(e) => {
tracing::error!("Inferenssivirhe: {}", e);
}
}
}
loop {
tokio::select! {
line = stdin_lines.next_line() => {
if let Ok(Some(text)) = line {
let t = text.trim();
if t == "p" || t == "pause" {
tracing::info!("Tauotetaan solmun suoritus (Hub ei lähetä tehtäviä)...");
let req = json!({"type": "status_update", "status": "paused"});
let _ = write.send(Message::Text(req.to_string())).await;
} else if t == "r" || t == "resume" || t == "s" {
tracing::info!("Jatketaan solmun suoritusta...");
let req = json!({"type": "status_update", "status": "active"});
let _ = write.send(Message::Text(req.to_string())).await;
}
}
}
// Mallin vaihto lennossa
if text.contains("change_model") {
if let Ok(task) = serde_json::from_str::<serde_json::Value>(&text) {
if let Some(new_model) = task.get("model").and_then(|v| v.as_str()) {
if let Some(ref engine) = llm {
tracing::info!("Vaihdetaan malli: {}", new_model);
engine.set_model(new_model.to_string());
match engine.ensure_model().await {
Ok(()) => tracing::info!("Malli {} valmis!", new_model),
Err(e) => tracing::error!("Mallin lataus epäonnistui: {}", e),
ws_msg = read.next() => {
match ws_msg {
Some(Ok(Message::Text(text))) => {
// Hubin control-viestit
if text.contains(r#""type":"control""#) {
if let Ok(task) = serde_json::from_str::<serde_json::Value>(&text) {
if let Some(action) = task.get("action").and_then(|v| v.as_str()) {
if action == "pause" {
tracing::info!("Hub pakotti solmun tauolle (Pause)");
let req = json!({"type": "status_update", "status": "paused"});
let _ = write.send(Message::Text(req.to_string())).await;
} else if action == "resume" {
tracing::info!("Hub aktivoi solmun suorituksen (Resume)");
let req = json!({"type": "status_update", "status": "active"});
let _ = write.send(Message::Text(req.to_string())).await;
}
}
}
}
// LLM-promptit
if text.contains("llm_prompt") {
if let Ok(task) = serde_json::from_str::<serde_json::Value>(&text) {
let prompt = task.get("prompt").and_then(|v| v.as_str()).unwrap_or("");
let task_id = task.get("task_id").and_then(|v| v.as_str()).unwrap_or("?");
let msg_model = task.get("model").and_then(|v| v.as_str()).unwrap_or("");
if !prompt.is_empty() && (msg_model.starts_with("qwen-coder") || msg_model.starts_with("qwen2.5-coder") || msg_model.starts_with("phi")) {
if let Some(ref engine) = llm {
let max_tokens = task.get("max_tokens").and_then(|v| v.as_u64()).unwrap_or(1024) as usize;
let prompt_lines = prompt.lines().count();
let prompt_last: String = prompt.lines().last().unwrap_or("").chars().take(60).collect();
tracing::info!("→ task_id:{} | {}r prompti | \"{}...\"", task_id, prompt_lines, prompt_last);
let model_name = engine.model_name();
match engine.generate(prompt, max_tokens).await {
Ok(result) => {
tracing::info!(
"✓ {} | {} tok | {:.0}ms | {:.1} tok/s",
model_name,
result.tokens_generated,
result.duration_ms,
result.tokens_per_sec,
);
let prompt_short: String = prompt.lines().last().unwrap_or("").chars().take(100).collect();
let done = json!({
"type": "llm_done",
"prompt": prompt_short,
"model": format!("{} (Ollama)", model_name),
"response": result.text,
"tokens_generated": result.tokens_generated,
"duration_ms": result.duration_ms,
"tokens_per_sec": (result.tokens_per_sec * 10.0).round() / 10.0,
"load_time_ms": 0,
"task_id": task_id,
});
let _ = write.send(Message::Text(done.to_string())).await;
}
Err(e) => {
tracing::error!("Inferenssivirhe: {}", e);
}
}
}
}
}
}
// Mallin vaihto lennossa
if text.contains("change_model") {
if let Ok(task) = serde_json::from_str::<serde_json::Value>(&text) {
if let Some(new_model) = task.get("model").and_then(|v| v.as_str()) {
if let Some(ref engine) = llm {
tracing::info!("Vaihdetaan malli: {}", new_model);
engine.set_model(new_model.to_string());
match engine.ensure_model().await {
Ok(()) => tracing::info!("Malli {} valmis!", new_model),
Err(e) => tracing::error!("Mallin lataus epäonnistui: {}", e),
}
}
}
}
}
}
Some(Ok(_)) => {} // Muut viestityypit (binary/ping)
Some(Err(_)) | None => break, // Yhteys poikki
}
}
}
}
tracing::warn!("Yhteys hubiin katkesi — yritetään uudelleen 5s...");
}
Err(e) => {

View File

@@ -0,0 +1,67 @@
use dialoguer::{Select, Input, theme::ColorfulTheme};
use reqwest::Client;
pub async fn select_model(ollama_url: &str, client: &Client) -> Result<String, String> {
// 1. Hae tagit
let mut models = vec![];
println!(" Haetaan asennettuja malleja osoitteesta {}...", ollama_url);
if let Ok(resp) = client.get(&format!("{}/api/tags", ollama_url)).send().await {
if resp.status().is_success() {
if let Ok(json) = resp.json::<serde_json::Value>().await {
if let Some(arr) = json.get("models").and_then(|v| v.as_array()) {
for m in arr {
if let Some(name) = m.get("name").and_then(|v| v.as_str()) {
models.push(name.to_string());
}
}
}
}
}
}
let download_opt = "[ Lataa uusi malli internetistä]";
let mut options = vec![download_opt.to_string()];
options.extend(models);
// 2. Kysy käyttäjältä Selectillä
let theme = ColorfulTheme::default();
let selection = Select::with_theme(&theme)
.with_prompt("Valitse Ollama-malli Kipinä-verkkoa varten:")
.default(if options.len() > 1 { 1 } else { 0 })
.items(&options)
.interact()
.map_err(|e| format!("TUI virhe: {}", e))?;
let selected = &options[selection];
// 3. Jos käyttäjä haluaa uuden, kysy nimeä
if selected == download_opt {
let new_model: String = Input::with_theme(&theme)
.with_prompt("Syötä ladattavan mallin nimi (esim. llama3 tai qwen2.5-coder:3b)")
.interact_text()
.map_err(|e| format!("TUI virhe: {}", e))?;
let new_model = new_model.trim().to_string();
if new_model.is_empty() {
return Err("Mallin nimi ei voi olla tyhjä".to_string());
}
println!(" Ladataan malleja taustalla... Tämä voi kestää hetken ({})", new_model);
// Odotetaan että pull on valmis
let pull_body = serde_json::json!({ "name": &new_model });
let resp = client.post(&format!("{}/api/pull", ollama_url))
.json(&pull_body)
.send()
.await
.map_err(|e| format!("Pull req virhe: {}", e))?;
if resp.status().is_success() {
println!(" ✓ Malli {} ladattu onnistuneesti!", new_model);
return Ok(new_model);
} else {
return Err(format!("Ollama pull epäonnistui: {}", resp.status()));
}
}
Ok(selected.clone())
}