From 2612e90dbf4d153d608c00a47e5175aa7b19e299 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Thu, 27 Aug 2026 22:37:06 +1000 Subject: [PATCH] fix: resume Ollama chat jobs across sessions --- README.md | 4 +- dashboard/dist/index.js | 35 ++++++++++++- dashboard/manifest.json | 2 +- dashboard/plugin_api.py | 89 +++++++++++++++++++++++++++----- plugin.yaml | 2 +- tests/test_validation_harness.py | 30 +++++++++++ 6 files changed, 142 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index 6fe34c4..18c0b5b 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ Native-like Hermes dashboard plugin for local Ollama model management and chat. - Streamed Ollama responses with a real Stop action that cancels the active request - Minimized-by-default expandable thinking/progress details with live stage, elapsed time, event, and character counters - Validation harness mode: choose one primary model and one or more independent validator models; validators review the primary draft and the primary model compiles one final answer -- Server-owned chat jobs continue after the browser closes and persist final answers for later resume +- Server-owned chat jobs continue after the browser closes and persist final answers for later resume. A newly opened dashboard discovers queued/running jobs from the shared server store and resumes observing them automatically. - SQLite is the default chat store for new users - Optional native PostgreSQL storage can be installed and linked explicitly from the plugin @@ -26,7 +26,7 @@ The chat supports two modes. With one selected model, it sends a normal direct r ## Chat storage and durability -Chat jobs are persisted before model execution. The browser observes job status through `/chat/status/{request_id}` but does not own generation. Closing the browser no longer cancels a queued or running job, and the dashboard recovers queued/running jobs after restart. Each conversation records immutable message versions, job attempts, stage timing, model metrics, and append-only operational events. +Chat jobs are persisted before model execution. The browser observes job status through `/chat/status/{request_id}` and discovers active jobs through `/chat/jobs`, but does not own generation. Closing the browser no longer cancels a queued or running job, and a fresh browser session automatically reconnects to the latest active job. The server worker supervisor continuously re-queues jobs whose worker disappeared and persists a heartbeat while Ollama is thinking, including during idle streaming periods. Stop requests remain responsive because the Ollama stream is checked on a short read interval. The dashboard service is configured with `Restart=always`, and queued/running jobs are recovered after a service restart. Each conversation records immutable message versions, job attempts, stage timing, model metrics, and append-only operational events. SQLite remains the default and requires no service installation. The Chat storage panel can detect native PostgreSQL, install it only after explicit confirmation, and link it as the active backend. PostgreSQL storage uses the local service environment and remains disabled until the user explicitly selects **Link PostgreSQL**. diff --git a/dashboard/dist/index.js b/dashboard/dist/index.js index 6c5c044..0b4014f 100644 --- a/dashboard/dist/index.js +++ b/dashboard/dist/index.js @@ -319,7 +319,7 @@ }).catch(function () { return []; }); } function newConversation() { setConversationId(""); setHistory([]); setMetrics([]); setValidationReports(null); setNotice({ ok: "New conversation ready." }); } - React.useEffect(function () { refreshConversations(); refreshStorage(); }, []); + React.useEffect(function () { refreshConversations(); refreshStorage(); resumeActiveJobs(); }, []); React.useEffect(function () { if (conversationId) saveChat(model, selectedModels, history); }, [model, selectedModels, history, conversationId]); React.useEffect(function () { if (!model && (loadedModels[0] || models[0])) setModel((loadedModels[0] || models[0]).name); }, [models, loadedModels, model]); React.useEffect(function () { @@ -429,6 +429,37 @@ return new Promise(function (resolve) { setTimeout(resolve, 1200); }).then(function () { return pollChatJob(requestId, current); }); }); } + function applyChatResult(result) { + var answer = result.message && result.message.content ? result.message.content : "(No response text returned.)"; + var resolvedConversationId = result.conversation_id || conversationId; + setConversationId(resolvedConversationId); + setHistory(function (old) { + var last = old.length ? old[old.length - 1] : null; + return last && last.role === "assistant" && last.content === answer ? old : old.concat([{ role: "assistant", content: answer }]); + }); + setMetrics(result.metrics || []); + setValidationReports(result.mode === "harness" ? (result.validation_reports || []) : null); + setAttachments([]); + setRuntime(result.runtime || runtime); + setNotice({ ok: result.mode === "harness" ? "One final answer compiled by " + result.primary_model + " after validation by " + (result.validator_models || []).join(", ") + "." : "Response complete. Shared conversation and performance metrics saved." }); + pollRuntime(); refreshConversations(); + return result; + } + function resumeActiveJobs() { + return fetchJSON(API + "/chat/jobs?active=true&limit=20").then(function (value) { + var jobs = value.jobs || []; + var active = jobs.find(function (job) { return !job.done && (job.status === "queued" || job.status === "running" || job.status === "stopping"); }); + if (!active) return null; + var current = { id: active.request_id, controller: null, stopped: false }; + setConversationId(active.conversation_id || ""); + setActiveRequest(current); + setBusy("send"); + setThinking({ request_id: active.request_id, startedAt: (Number(active.started_at || active.updated_at || Date.now() / 1000) * 1000), stage: active.stage || "Resuming server-side chat job" }); + setThinkingDetails(active); + if (active.conversation_id) openConversation(active.conversation_id); + return pollChatJob(active.request_id, current).then(applyChatResult).catch(function (err) { if (!current.stopped) setNotice({ error: err.message || String(err) }); }).finally(function () { if (!current.stopped) { setThinking(null); setActiveRequest(null); } setBusy(""); }); + }).catch(function () { return null; }); + } function send() { if (busy === "send" || busy === "stop" || !selectedModels.length || (!message.trim() && !attachments.length)) return; var requestId = makeRequestId(); @@ -438,7 +469,7 @@ setHistory(function (old) { return old.concat([outgoing]); }); setMessage(""); setBusy("send"); setActiveRequest(current); setThinking({ request_id: requestId, startedAt: Date.now(), stage: attachments.length ? "Preparing attachments and sending request to Ollama" : "Sending request to Ollama" }); setThinkingDetails(null); setThinkingOpen(false); setNotice(null); var requestOptions = { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(body) }; if (controller) requestOptions.signal = controller.signal; - fetchJSON(API + "/chat", requestOptions).then(function (result) { return result.done ? result : pollChatJob(requestId, current); }).then(function (result) { var answer = result.message && result.message.content ? result.message.content : "(No response text returned.)"; setConversationId(result.conversation_id || conversationId); setHistory(function (old) { return old.concat([{ role: "assistant", content: answer }]); }); setMetrics(result.metrics || []); setValidationReports(result.mode === "harness" ? (result.validation_reports || []) : null); setAttachments([]); setRuntime(result.runtime || runtime); setNotice({ ok: result.mode === "harness" ? "One final answer compiled by " + result.primary_model + " after validation by " + (result.validator_models || []).join(", ") + "." : "Response complete. Shared conversation and performance metrics saved." }); pollRuntime(); refreshConversations(); }).catch(function (err) { if (!current.stopped) setNotice({ error: err.message || String(err) }); }).finally(function () { if (!current.stopped) { setThinking(null); setActiveRequest(null); } setBusy(""); }); + fetchJSON(API + "/chat", requestOptions).then(function (result) { return result.done ? result : pollChatJob(requestId, current); }).then(applyChatResult).catch(function (err) { if (!current.stopped) setNotice({ error: err.message || String(err) }); }).finally(function () { if (!current.stopped) { setThinking(null); setActiveRequest(null); } setBusy(""); }); } return h("section", { className: "ollama-chat" }, h("div", { className: "ollama-chat-header" }, h("div", null, h("div", { className: "ollama-eyebrow" }, "LOCAL OLLAMA CHAT"), h("h2", null, "Chat with a validated model harness"), h("p", null, "Choose one primary model and at least one loaded validator model. The primary produces one final answer after reviewing the independent validation reports.")), h(Button, { className: "secondary", disabled: !history.length || busy === "send", onClick: clearChat }, "Clear chat")), diff --git a/dashboard/manifest.json b/dashboard/manifest.json index 4a2150f..cf2a8a8 100644 --- a/dashboard/manifest.json +++ b/dashboard/manifest.json @@ -3,7 +3,7 @@ "label": "Ollama Models", "description": "Inspect, manage, and chat with local Ollama models, including shared persistent conversations, performance metrics, images, PDFs, URLs, and live memory telemetry.", "icon": "Cpu", - "version": "1.7.1", + "version": "1.7.2", "tab": {"path": "/ollama-manager", "position": "after:models"}, "entry": "dist/index.js", "css": "dist/style.css", diff --git a/dashboard/plugin_api.py b/dashboard/plugin_api.py index da00dd8..459d59e 100644 --- a/dashboard/plugin_api.py +++ b/dashboard/plugin_api.py @@ -10,6 +10,7 @@ import json import mimetypes import os import re +import select import shutil import socket import sqlite3 @@ -76,6 +77,8 @@ MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024 MAX_ATTACHMENT_TEXT = 80_000 MAX_URL_BYTES = 15 * 1024 * 1024 CHAT_KEEP_ALIVE = -1 +CHAT_READ_TIMEOUT = 5.0 +CHAT_HEARTBEAT_INTERVAL = 5.0 HARNESS_MIN_VALIDATORS = 1 HARNESS_MAX_DRAFT_CHARS = 24_000 HARNESS_MAX_VALIDATION_CHARS = 8_000 @@ -93,6 +96,7 @@ _chat_db_init_lock = threading.Lock() _chat_db_ready = False _chat_job_futures: dict[str, Any] = {} _chat_job_futures_lock = threading.Lock() +_chat_job_heartbeat_at: dict[str, float] = {} _chat_job_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="ollama-chat-job") _DATABASE_URL = os.environ.get("OLLAMA_MANAGER_DATABASE_URL", "").strip() @@ -501,6 +505,16 @@ def _update_chat_job(request_id: str, **values: Any) -> None: db.close() +def _touch_chat_job(request_id: str, *, force: bool = False) -> None: + now = time.time() + with _chat_job_futures_lock: + previous = _chat_job_heartbeat_at.get(request_id, 0.0) + if not force and now - previous < CHAT_HEARTBEAT_INTERVAL: + return + _chat_job_heartbeat_at[request_id] = now + _update_chat_job(request_id, heartbeat_at=now) + + def _job_status_response(job: dict[str, Any]) -> dict[str, Any]: result = json.loads(job.get("result_json") or "{}") status = str(job.get("status") or "unknown") @@ -515,12 +529,37 @@ def _job_status_response(job: dict[str, Any]) -> dict[str, Any]: "validator_models": json.loads(job.get("validator_models_json") or "[]"), "error": job.get("error") or "", "done": status in {"completed", "failed", "stopped", "canceled"}, + "attempt": job.get("attempt"), + "started_at": job.get("started_at"), + "heartbeat_at": job.get("heartbeat_at"), + "finished_at": job.get("finished_at"), "updated_at": job.get("updated_at"), } + if job.get("heartbeat_at"): + response["heartbeat_age"] = round(max(0.0, time.time() - float(job["heartbeat_at"])), 1) response.update(result) return response +def _list_chat_jobs(*, active_only: bool = True, conversation_id: str | None = None, limit: int = 50) -> list[dict[str, Any]]: + limit = max(1, min(int(limit), 100)) + db = _chat_db() + try: + values: list[Any] = [] + clauses: list[str] = [] + if active_only: + clauses.append("status IN (?,?,?)") + values.extend(["queued", "running", "stopping"]) + if conversation_id: + clauses.append("conversation_id=?") + values.append(conversation_id) + where = " WHERE " + " AND ".join(clauses) if clauses else "" + rows = db.execute(f"SELECT * FROM chat_jobs{where} ORDER BY updated_at DESC LIMIT ?", tuple(values) + (limit,)).fetchall() + return [_job_status_response(dict(row)) for row in rows] + finally: + db.close() + + def _chat_request_from_job(job: dict[str, Any]) -> ChatRequest: return ChatRequest.model_validate(json.loads(job["payload_json"])) @@ -1843,8 +1882,11 @@ def _stream_chat_request(payload: dict[str, Any], request_id: str, cancel_event: response_text: list[str] = [] thinking_text: list[str] = [] _chat_state(request_id, state="connecting", stage="Connecting to Ollama") + _touch_chat_job(parent_id or request_id, force=True) try: with urlopen(request, timeout=1800) as response: + socket_file = getattr(getattr(response, "fp", None), "raw", None) + socket_obj = getattr(socket_file, "_sock", None) _chat_state(request_id, response=response, state="generating", stage="Ollama is generating") while True: with _chat_requests_lock: @@ -1852,9 +1894,14 @@ def _stream_chat_request(payload: dict[str, Any], request_id: str, cancel_event: if cancelled: response.close() raise _ChatStopped() + if socket_obj is not None: + readable, _, _ = select.select([socket_obj], [], [], CHAT_READ_TIMEOUT) + if not readable: + _touch_chat_job(parent_id or request_id) + continue try: raw_line = response.readline() - except (socket.timeout, TimeoutError, OSError, ValueError) as exc: + except (TimeoutError, OSError, ValueError) as exc: with _chat_requests_lock: cancelled = bool(_chat_requests.get(request_id, {}).get("cancel", threading.Event()).is_set()) if cancelled: @@ -1891,6 +1938,7 @@ def _stream_chat_request(payload: dict[str, Any], request_id: str, cancel_event: ) if event.get("done"): break + _touch_chat_job(parent_id or request_id) except _ChatStopped: _chat_state(request_id, state="stopped", stage="Stopped by user", finished_at=time.time()) raise @@ -2079,20 +2127,26 @@ def _submit_chat_job(request_id: str) -> None: def _recover_chat_jobs() -> None: - try: - db = _chat_db() + while True: try: - rows = db.execute("SELECT request_id,status,conversation_id FROM chat_jobs WHERE status IN ('queued','running') ORDER BY created_at").fetchall() - finally: - db.close() - for row in rows: - request_id = str(row["request_id"]) - if str(row["status"]) == "running": - _update_chat_job(request_id, status="queued", heartbeat_at=time.time()) - _persist_chat_event(request_id, row["conversation_id"], "job-recovered", level="warning", stage="Recovered after dashboard restart") - _submit_chat_job(request_id) - except Exception: - return + db = _chat_db() + try: + rows = db.execute("SELECT request_id,status,conversation_id FROM chat_jobs WHERE status IN ('queued','running') ORDER BY created_at").fetchall() + finally: + db.close() + for row in rows: + request_id = str(row["request_id"]) + with _chat_job_futures_lock: + future = _chat_job_futures.get(request_id) + future_missing = future is None or future.done() + if str(row["status"]) == "running" and future_missing: + _update_chat_job(request_id, status="queued", heartbeat_at=time.time()) + _persist_chat_event(request_id, row["conversation_id"], "job-recovered", level="warning", stage="Recovered by server worker supervisor") + if str(row["status"]) == "queued" or future_missing: + _submit_chat_job(request_id) + except Exception: + pass + time.sleep(10) @router.get("/chat/status/{request_id}") @@ -2113,6 +2167,13 @@ def chat_status(request_id: str) -> dict[str, Any]: raise HTTPException(404, "Chat request not found") +@router.get("/chat/jobs") +def chat_jobs(active: bool = True, conversation_id: str | None = None, limit: int = 50) -> dict[str, Any]: + if conversation_id: + conversation_id = _conversation_id(conversation_id) + return {"jobs": _list_chat_jobs(active_only=bool(active), conversation_id=conversation_id, limit=limit)} + + @router.post("/chat/stop") def chat_stop(body: ChatStopRequest) -> dict[str, Any]: request_id = _valid_chat_request_id(body.request_id) diff --git a/plugin.yaml b/plugin.yaml index cd1ea53..27552b1 100644 --- a/plugin.yaml +++ b/plugin.yaml @@ -1,5 +1,5 @@ name: ollama-manager -version: 1.7.1 +version: 1.7.2 description: Native dashboard manager and chat interface for local Ollama models, attachments, URLs, shared persistent conversations, performance metrics, and live runtime telemetry. auto_install_dependencies: true python_dependencies: diff --git a/tests/test_validation_harness.py b/tests/test_validation_harness.py index fff21b8..5c3580f 100644 --- a/tests/test_validation_harness.py +++ b/tests/test_validation_harness.py @@ -59,6 +59,36 @@ class ValidationHarnessTests(unittest.TestCase): self.assertFalse(response["done"]) self.assertEqual(response["mode"], "direct") + def test_job_status_includes_durable_heartbeat_fields(self): + job = { + "request_id": "request", + "conversation_id": "conversation", + "status": "running", + "mode": "direct", + "primary_model": "primary", + "validator_models_json": "[]", + "result_json": "{}", + "error": "", + "attempt": 2, + "started_at": 100.0, + "heartbeat_at": 110.0, + "finished_at": None, + "updated_at": 110.0, + } + with patch.object(api.time, "time", return_value=112.5): + response = api._job_status_response(job) + self.assertFalse(response["done"]) + self.assertEqual(response["attempt"], 2) + self.assertEqual(response["heartbeat_at"], 110.0) + self.assertEqual(response["heartbeat_age"], 2.5) + + def test_active_jobs_route_returns_server_owned_jobs(self): + active = [{"request_id": "request", "status": "running"}] + with patch.object(api, "_list_chat_jobs", return_value=active) as listed: + response = api.chat_jobs(active=True, conversation_id="conversation", limit=20) + self.assertEqual(response, {"jobs": active}) + listed.assert_called_once_with(active_only=True, conversation_id="conversation", limit=20) + def test_primary_draft_validators_and_primary_compilation_produce_one_answer(self): body = api.ChatRequest( primary_model="primary",