diff --git a/mcp-server/src/api-client.ts b/mcp-server/src/api-client.ts index af0783305..b94abe33d 100644 --- a/mcp-server/src/api-client.ts +++ b/mcp-server/src/api-client.ts @@ -296,6 +296,56 @@ export class LlmWikiApiClient { }) } + async ingestStatus(projectId = "current"): Promise { + const json = await this.request(`/projects/${encodeURIComponent(projectId)}/ingest/status`) + const result = requireObject(json.result, "ingest status result") + const summary = requireObject(result.summary, "ingest summary") + const tasks = Array.isArray(result.tasks) ? result.tasks : [] + return { + summary: parseIngestSummary(summary), + tasks: tasks.map(parseIngestTask), + } + } + + async ingestPause(projectId = "current"): Promise { + return this.parseIngestControlResponse( + await this.request(`/projects/${encodeURIComponent(projectId)}/ingest/pause`, { + method: "POST", + }), + "pause", + ) + } + + async ingestResume(projectId = "current"): Promise { + return this.parseIngestControlResponse( + await this.request(`/projects/${encodeURIComponent(projectId)}/ingest/resume`, { + method: "POST", + }), + "resume", + ) + } + + async ingestRetryFailed(projectId = "current"): Promise { + return this.parseIngestControlResponse( + await this.request(`/projects/${encodeURIComponent(projectId)}/ingest/retry-failed`, { + method: "POST", + }), + "retry-failed", + ) + } + + private parseIngestControlResponse( + json: Record, + action: string, + ): ApiIngestControlResponse { + const result = requireObject(json.result, `ingest ${action} result`) + return { + action: typeof json.action === "string" ? json.action : action, + projectId: typeof json.projectId === "string" ? json.projectId : "", + result, + } + } + private async request(path: string, options: { method?: "GET" | "POST"; body?: unknown; auth?: boolean } = {}): Promise> { const url = `${this.baseUrl}${apiPath(path)}` const headers: Record = { Accept: "application/json" } @@ -456,3 +506,70 @@ function parseGraphEdge(value: unknown): ApiGraphEdge { weight: numberOrUndefined(obj.weight), } } + +// ── Ingest control ─────────────────────────────────────────────────────── + +export interface ApiIngestSummary { + pending: number + processing: number + failed: number + cancelled: number + completed: number + total: number + paused: boolean + userPaused: boolean + restoredBacklogWaiting: boolean +} + +export interface ApiIngestTask { + id: string + sourcePath: string + folderContext: string + status: "pending" | "processing" | "done" | "failed" | "cancelled" + error: string | null + retryCount: number + addedAt: number +} + +export interface ApiIngestStatusResponse { + summary: ApiIngestSummary + tasks: ApiIngestTask[] +} + +export interface ApiIngestControlResponse { + action: string + projectId: string + result: Record +} + +function parseIngestSummary(value: unknown): ApiIngestSummary { + const obj = requireObject(value, "ingest summary") + return { + pending: numberOrUndefined(obj.pending) ?? 0, + processing: numberOrUndefined(obj.processing) ?? 0, + failed: numberOrUndefined(obj.failed) ?? 0, + cancelled: numberOrUndefined(obj.cancelled) ?? 0, + completed: numberOrUndefined(obj.completed) ?? 0, + total: numberOrUndefined(obj.total) ?? 0, + paused: obj.paused === true, + userPaused: obj.userPaused === true, + restoredBacklogWaiting: obj.restoredBacklogWaiting === true, + } +} + +function parseIngestTask(value: unknown): ApiIngestTask { + const obj = requireObject(value, "ingest task") + const status = obj.status + return { + id: String(obj.id ?? ""), + sourcePath: String(obj.sourcePath ?? ""), + folderContext: typeof obj.folderContext === "string" ? obj.folderContext : "", + status: + status === "pending" || status === "processing" || status === "done" || status === "failed" || status === "cancelled" + ? status + : "pending", + error: typeof obj.error === "string" ? obj.error : null, + retryCount: numberOrUndefined(obj.retryCount) ?? 0, + addedAt: numberOrUndefined(obj.addedAt) ?? 0, + } +} diff --git a/mcp-server/src/index.ts b/mcp-server/src/index.ts index 49cc31890..065f5d020 100644 --- a/mcp-server/src/index.ts +++ b/mcp-server/src/index.ts @@ -16,6 +16,8 @@ import { type ApiChatResponse, type ApiSearchResult, type ApiProject, + type ApiIngestStatusResponse, + type ApiIngestControlResponse, } from "./api-client.js" import { VERSION } from "./version.js" import { McpProjectBinding, withActiveProject } from "./project-binding.js" @@ -169,6 +171,50 @@ server.setRequestHandler(ListToolsRequestSchema, async () => ({ additionalProperties: false, }, }, + { + name: "llm_wiki_ingest_status", + description: "Get the current ingest pipeline status for a project: queue counts (pending/processing/failed/cancelled/completed), paused flag, and task list. Useful for monitoring automated ingest progress via MCP.", + inputSchema: { + type: "object", + properties: { + project_id: { type: "string", description: "Project UUID, project path, or 'current'. Defaults to current." }, + }, + additionalProperties: false, + }, + }, + { + name: "llm_wiki_ingest_pause", + description: "Pause the ingest pipeline for a project. Aborts any in-flight LLM ingest task and stops new pending tasks from starting. Token spend stops immediately. The queue is preserved.", + inputSchema: { + type: "object", + properties: { + project_id: { type: "string", description: "Project UUID, project path, or 'current'. Defaults to current." }, + }, + additionalProperties: false, + }, + }, + { + name: "llm_wiki_ingest_resume", + description: "Resume the ingest pipeline for a project after a pause. Pending tasks immediately start processing. No-op if already running.", + inputSchema: { + type: "object", + properties: { + project_id: { type: "string", description: "Project UUID, project path, or 'current'. Defaults to current." }, + }, + additionalProperties: false, + }, + }, + { + name: "llm_wiki_ingest_retry_failed", + description: "Retry all failed ingest tasks for a project. Requeues every failed task back to pending and kicks off processing. Returns the number of tasks requeued and the updated queue summary.", + inputSchema: { + type: "object", + properties: { + project_id: { type: "string", description: "Project UUID, project path, or 'current'. Defaults to current." }, + }, + additionalProperties: false, + }, + }, ], })) @@ -268,6 +314,30 @@ server.setRequestHandler(CallToolRequestSchema, async (request) => { const scope = await resolveProjectScope(args) return textResult(withActiveProject(JSON.stringify(await client.rescan(scope.id), null, 2), scope.project, scope.id)) } + case "llm_wiki_ingest_status": { + await assertMcpEnabled() + const scope = await resolveProjectScope(args) + const status = await client.ingestStatus(scope.id) + return textResult(withActiveProject(formatIngestStatus(status), scope.project, scope.id)) + } + case "llm_wiki_ingest_pause": { + await assertMcpEnabled() + const scope = await resolveProjectScope(args) + const response = await client.ingestPause(scope.id) + return textResult(withActiveProject(formatIngestControl(response), scope.project, scope.id)) + } + case "llm_wiki_ingest_resume": { + await assertMcpEnabled() + const scope = await resolveProjectScope(args) + const response = await client.ingestResume(scope.id) + return textResult(withActiveProject(formatIngestControl(response), scope.project, scope.id)) + } + case "llm_wiki_ingest_retry_failed": { + await assertMcpEnabled() + const scope = await resolveProjectScope(args) + const response = await client.ingestRetryFailed(scope.id) + return textResult(withActiveProject(formatIngestControl(response), scope.project, scope.id)) + } default: throw new McpError(ErrorCode.MethodNotFound, `Unknown tool: ${request.params.name}`) } @@ -503,6 +573,63 @@ function formatGraph(nodes: ApiGraphNode[], edges: Array<{ source: string; targe return lines.join("\n") } +function formatIngestStatus(status: ApiIngestStatusResponse): string { + const { summary, tasks } = status + const lines = [ + "# Ingest pipeline status", + "", + `Paused: ${summary.paused ? "yes" : "no"}${summary.userPaused ? " (user-paused)" : ""}${summary.restoredBacklogWaiting ? " | restored backlog waiting" : ""}`, + "", + "## Queue summary", + `- Pending: ${summary.pending}`, + `- Processing: ${summary.processing}`, + `- Failed: ${summary.failed}`, + `- Cancelled: ${summary.cancelled}`, + `- Completed (this session): ${summary.completed}`, + `- Total: ${summary.total}`, + "", + ] + if (tasks.length > 0) { + lines.push("## Tasks") + tasks.forEach((task, index) => { + lines.push(`${index + 1}. [${task.status}] ${task.sourcePath}`) + if (task.folderContext) lines.push(` Folder: ${task.folderContext}`) + if (task.error) lines.push(` Error: ${task.error}`) + if (task.retryCount > 0) lines.push(` Retries: ${task.retryCount}`) + }) + lines.push("") + } + return lines.join("\n") +} + +function formatIngestControl(response: ApiIngestControlResponse): string { + const result = response.result + const summary = (result.summary ?? {}) as Record + const lines = [ + `# Ingest ${response.action}`, + "", + ] + if (typeof result.paused === "boolean") { + lines.push(`Paused: ${result.paused ? "yes" : "no"}`) + } + if (typeof result.resumed === "boolean") { + lines.push(`Resumed: ${result.resumed ? "yes" : "no"}`) + } + if (typeof result.requeued === "number") { + lines.push(`Requeued: ${result.requeued} task(s)`) + } + if (typeof summary.pending === "number") { + lines.push("") + lines.push("## Queue summary") + lines.push(`- Pending: ${summary.pending}`) + lines.push(`- Processing: ${summary.processing ?? 0}`) + lines.push(`- Failed: ${summary.failed ?? 0}`) + lines.push(`- Cancelled: ${summary.cancelled ?? 0}`) + lines.push(`- Paused: ${summary.paused ? "yes" : "no"}`) + } + return lines.join("\n") +} + async function main(): Promise { const transport = new StdioServerTransport() await server.connect(transport) diff --git a/mcp-server/test/api-client.test.ts b/mcp-server/test/api-client.test.ts index 678a852a8..5bb627d6a 100644 --- a/mcp-server/test/api-client.test.ts +++ b/mcp-server/test/api-client.test.ts @@ -231,3 +231,117 @@ test("API errors include status and server message", async () => { const client = new LlmWikiApiClient({ fetchImpl }) await assert.rejects(() => client.projects(), /LLM Wiki API 401: Unauthorized/) }) + +// ── Ingest control tests ───────────────────────────────────────────────── + +test("ingestStatus sends GET and parses summary + tasks", async () => { + const calls: Array<{ url: string; method?: string }> = [] + const fetchImpl = async (url: string | URL | Request, init?: RequestInit): Promise => { + calls.push({ url: String(url), method: init?.method }) + return new Response(JSON.stringify({ + ok: true, + action: "status", + projectId: "p1", + result: { + summary: { + pending: 2, + processing: 1, + failed: 3, + cancelled: 0, + completed: 5, + total: 11, + paused: false, + userPaused: false, + restoredBacklogWaiting: false, + }, + tasks: [ + { id: "t1", sourcePath: "raw/sources/a.pdf", folderContext: "papers", status: "processing", error: null, retryCount: 0, addedAt: 1700000000 }, + { id: "t2", sourcePath: "raw/sources/b.md", folderContext: "", status: "failed", error: "LLM error", retryCount: 3, addedAt: 1700000001 }, + ], + }, + }), { status: 200 }) + } + + const client = new LlmWikiApiClient({ fetchImpl }) + const status = await client.ingestStatus("p1") + + assert.equal(calls[0]?.url, "http://127.0.0.1:19828/api/v1/projects/p1/ingest/status") + assert.equal(calls[0]?.method, "GET") // GET requests + assert.equal(status.summary.pending, 2) + assert.equal(status.summary.processing, 1) + assert.equal(status.summary.failed, 3) + assert.equal(status.summary.paused, false) + assert.equal(status.tasks.length, 2) + assert.equal(status.tasks[0]?.status, "processing") + assert.equal(status.tasks[1]?.error, "LLM error") + assert.equal(status.tasks[1]?.retryCount, 3) +}) + +test("ingestPause sends POST and parses control response", async () => { + const calls: Array<{ url: string; method?: string }> = [] + const fetchImpl = async (url: string | URL | Request, init?: RequestInit): Promise => { + calls.push({ url: String(url), method: init?.method }) + return new Response(JSON.stringify({ + ok: true, + action: "pause", + projectId: "p1", + result: { + paused: true, + summary: { pending: 2, processing: 0, failed: 0, cancelled: 0, completed: 3, total: 5, paused: true, userPaused: true, restoredBacklogWaiting: false }, + }, + }), { status: 200 }) + } + + const client = new LlmWikiApiClient({ fetchImpl }) + const response = await client.ingestPause("p1") + + assert.equal(calls[0]?.url, "http://127.0.0.1:19828/api/v1/projects/p1/ingest/pause") + assert.equal(calls[0]?.method, "POST") + assert.equal(response.action, "pause") + assert.equal(response.projectId, "p1") + assert.equal(response.result.paused, true) +}) + +test("ingestResume sends POST and parses control response", async () => { + let method = "" + const fetchImpl = async (_url: string | URL | Request, init?: RequestInit): Promise => { + method = init?.method ?? "GET" + return new Response(JSON.stringify({ + ok: true, + action: "resume", + projectId: "p1", + result: { + resumed: true, + summary: { pending: 2, processing: 0, failed: 0, cancelled: 0, completed: 3, total: 5, paused: false, userPaused: false, restoredBacklogWaiting: false }, + }, + }), { status: 200 }) + } + + const client = new LlmWikiApiClient({ fetchImpl }) + const response = await client.ingestResume("p1") + + assert.equal(method, "POST") + assert.equal(response.result.resumed, true) +}) + +test("ingestRetryFailed sends POST and parses requeued count", async () => { + let method = "" + const fetchImpl = async (_url: string | URL | Request, init?: RequestInit): Promise => { + method = init?.method ?? "GET" + return new Response(JSON.stringify({ + ok: true, + action: "retry-failed", + projectId: "p1", + result: { + requeued: 3, + summary: { pending: 5, processing: 0, failed: 0, cancelled: 0, completed: 3, total: 8, paused: false, userPaused: false, restoredBacklogWaiting: false }, + }, + }), { status: 200 }) + } + + const client = new LlmWikiApiClient({ fetchImpl }) + const response = await client.ingestRetryFailed("p1") + + assert.equal(method, "POST") + assert.equal(response.result.requeued, 3) +}) diff --git a/src-tauri/src/api_server.rs b/src-tauri/src/api_server.rs index 0d8ebb836..9386580bf 100644 --- a/src-tauri/src/api_server.rs +++ b/src-tauri/src/api_server.rs @@ -1,15 +1,16 @@ -use std::collections::{BTreeMap, BTreeSet, VecDeque}; +use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque}; use std::fs; use std::io::Read; use std::path::{Component, Path, PathBuf}; use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering}; +use std::sync::mpsc; use std::sync::{Mutex, OnceLock}; use std::thread; use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; use serde_json::{json, Map, Value}; -use tauri::{AppHandle, Manager}; +use tauri::{AppHandle, Emitter, Manager}; use tiny_http::{Header, Method, Response, Server, StatusCode}; use uuid::Uuid; use walkdir::WalkDir; @@ -33,12 +34,22 @@ const APP_STATE_CACHE_TTL: Duration = Duration::from_secs(5); const RATE_LIMIT_WINDOW: Duration = Duration::from_secs(1); const RATE_LIMIT_MAX_REQUESTS: usize = 120; const MAX_IN_FLIGHT_REQUESTS: usize = 64; +/// How long an ingest control request waits for the frontend to respond +/// before timing out. The frontend listener calls ingest-queue functions +/// which may be async (e.g. retryAllFailedTasks), so this needs to be +/// generous enough for a queue save + processNext kickoff. +const INGEST_API_TIMEOUT: Duration = Duration::from_secs(15); /// API status: 0=starting, 1=running, 2=port_conflict, 3=error static API_STATUS: AtomicU8 = AtomicU8::new(0); static IN_FLIGHT_REQUESTS: AtomicUsize = AtomicUsize::new(0); static APP_STATE_CACHE: OnceLock>> = OnceLock::new(); static RATE_LIMIT: OnceLock>> = OnceLock::new(); +/// Pending ingest API requests waiting for a frontend response. The HTTP +/// handler emits a Tauri event, then blocks on the channel receiver until +/// the frontend calls the `ingest_api_response` command (which calls +/// `complete_ingest_api_request`). Keyed by request UUID. +static INGEST_API_PENDING: OnceLock>>> = OnceLock::new(); #[derive(Clone)] struct CachedAppState { @@ -290,6 +301,18 @@ fn handle_request( (&Method::Post, ["projects", project_id, "chat", session_id, "cancel"]) => { handle_cancel_chat(app, project_id, session_id) } + (&Method::Get, ["projects", project_id, "ingest", "status"]) => { + handle_ingest_control(app, project_id, "status") + } + (&Method::Post, ["projects", project_id, "ingest", "pause"]) => { + handle_ingest_control(app, project_id, "pause") + } + (&Method::Post, ["projects", project_id, "ingest", "resume"]) => { + handle_ingest_control(app, project_id, "resume") + } + (&Method::Post, ["projects", project_id, "ingest", "retry-failed"]) => { + handle_ingest_control(app, project_id, "retry-failed") + } _ => err(404, "Not found"), } } @@ -2077,6 +2100,104 @@ fn resolve_link(raw: &str, ids: &BTreeSet) -> Option { .cloned() } +/// Completes a pending ingest API request by sending the frontend's +/// result through the channel. Called by the `ingest_api_response` +/// Tauri command when the frontend finishes processing an +/// `ingest-api://request` event. Returns true if a matching request +/// was found (so the command can report success to the frontend). +pub fn complete_ingest_api_request(request_id: &str, result: Value) -> bool { + let Some(lock) = INGEST_API_PENDING.get() else { + return false; + }; + let Ok(mut map) = lock.lock() else { + return false; + }; + if let Some(sender) = map.remove(request_id) { + let _ = sender.send(result); + true + } else { + false + } +} + +/// Bridge between the HTTP API and the frontend ingest queue. The ingest +/// queue lives entirely in frontend TypeScript (src/lib/ingest-queue.ts), +/// so the HTTP handler emits a Tauri event and blocks until the frontend +/// responds via the `ingest_api_response` command. +/// +/// `action` is one of "status", "pause", "resume", "retry-failed". +fn handle_ingest_control( + app: &AppHandle, + project_id: &str, + action: &str, +) -> ApiResponse { + let project = match resolve_project(app, project_id) { + Ok(project) => project, + Err(e) => return err(404, e), + }; + + let request_id = Uuid::new_v4().to_string(); + let (tx, rx) = mpsc::channel::(); + + // Register the pending request so the frontend's response can find us. + { + let map = INGEST_API_PENDING.get_or_init(|| Mutex::new(HashMap::new())); + if let Ok(mut guard) = map.lock() { + guard.insert(request_id.clone(), tx); + } else { + return err(500, "Failed to register ingest API request"); + } + } + + // Emit the event to the frontend webview. The frontend listener + // (src/lib/ingest-api-bridge.ts) picks this up, calls the + // corresponding ingest-queue function, and responds via the + // `ingest_api_response` Tauri command. + let payload = json!({ + "requestId": request_id, + "action": action, + "projectId": project.id, + }); + + if app.emit("ingest-api://request", payload).is_err() { + // Clean up the pending entry to avoid a stale channel. + if let Some(lock) = INGEST_API_PENDING.get() { + if let Ok(mut map) = lock.lock() { + map.remove(&request_id); + } + } + return err(503, "Failed to dispatch ingest control request to the desktop UI"); + } + + // Block until the frontend responds or the timeout expires. + match rx.recv_timeout(INGEST_API_TIMEOUT) { + Ok(result) => { + // The frontend may return { "error": "..." } for internal failures. + if let Some(err_msg) = result.get("error").and_then(Value::as_str) { + return err(500, err_msg); + } + ok(json!({ + "ok": true, + "action": action, + "projectId": project.id, + "result": result, + })) + } + Err(_) => { + // Timeout — remove the stale entry so it doesn't linger. + if let Some(lock) = INGEST_API_PENDING.get() { + if let Ok(mut map) = lock.lock() { + map.remove(&request_id); + } + } + err( + 504, + "Ingest control request timed out — the desktop UI may not be visible or no project is open", + ) + } + } +} + fn handle_rescan(app: &AppHandle, project_id: &str) -> ApiResponse { let project = match resolve_project(app, project_id) { Ok(project) => project, diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index ab2d11e3b..d5bab1c3b 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -61,6 +61,15 @@ fn api_server_reload_config() -> String { .unwrap_or_else(|e| format!("error: {e}")) } +/// Frontend → backend response channel for ingest API control requests. +/// The HTTP API handler emits an `ingest-api://request` event and blocks +/// until the frontend calls this command with the result. See +/// `handle_ingest_control` in api_server.rs. +#[tauri::command] +fn ingest_api_response(request_id: String, result: serde_json::Value) -> bool { + api_server::complete_ingest_api_request(&request_id, result) +} + #[tauri::command] async fn agent_start_turn( app: tauri::AppHandle, @@ -654,6 +663,7 @@ pub fn run() { clip_server_status, api_server_status, api_server_reload_config, + ingest_api_response, agent_start_turn, agent_start_turn_stream, agent_cancel_turn, diff --git a/src/App.tsx b/src/App.tsx index 709f3fa45..78873fa5e 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -123,6 +123,15 @@ function App() { useEffect(() => { setupAutoSave() startClipWatcher() + // Start the ingest API event bridge so MCP/API clients can control + // the ingest queue (pause/resume/status/retry-failed) without the UI. + import("@/lib/ingest-api-bridge") + .then(({ startIngestApiBridge }) => { + startIngestApiBridge().catch((err) => + console.error("Failed to start ingest API bridge:", err), + ) + }) + .catch(() => {}) }, []) useEffect(() => { diff --git a/src/lib/ingest-api-bridge.ts b/src/lib/ingest-api-bridge.ts new file mode 100644 index 000000000..81ddaf86e --- /dev/null +++ b/src/lib/ingest-api-bridge.ts @@ -0,0 +1,92 @@ +import { listen, type UnlistenFn } from "@tauri-apps/api/event" +import { invoke } from "@tauri-apps/api/core" +import { + getQueue, + getQueueSummary, + pauseProcessing, + resumeProcessing, + retryAllFailedTasks, + type IngestTask, +} from "@/lib/ingest-queue" + +interface IngestApiRequest { + requestId: string + action: "status" | "pause" | "resume" | "retry-failed" + projectId: string +} + +let unlisten: UnlistenFn | null = null + +/** + * Start listening for ingest control requests emitted by the HTTP API + * server. The Rust backend emits `ingest-api://request` events when an + * MCP/API client calls the ingest control endpoints. This bridge calls + * the corresponding ingest-queue function and sends the result back via + * the `ingest_api_response` Tauri command. + * + * Safe to call multiple times — subsequent calls are no-ops. + */ +export async function startIngestApiBridge(): Promise { + if (unlisten) return + unlisten = await listen( + "ingest-api://request", + async (event) => { + const { requestId, action, projectId } = event.payload + let result: Record + try { + switch (action) { + case "status": { + const summary = getQueueSummary() + const tasks = getQueue() as readonly IngestTask[] + result = { + summary, + tasks: tasks.map((task) => ({ + id: task.id, + sourcePath: task.sourcePath, + folderContext: task.folderContext, + status: task.status, + error: task.error, + retryCount: task.retryCount, + addedAt: task.addedAt, + })), + } + break + } + case "pause": { + pauseProcessing() + result = { paused: true, summary: getQueueSummary() } + break + } + case "resume": { + resumeProcessing() + result = { resumed: true, summary: getQueueSummary() } + break + } + case "retry-failed": { + const requeued = await retryAllFailedTasks() + result = { requeued, summary: getQueueSummary() } + break + } + default: + result = { error: `Unknown ingest action: ${action}` } + } + } catch (err) { + result = { error: err instanceof Error ? err.message : String(err) } + } + try { + await invoke("ingest_api_response", { requestId, result }) + } catch (err) { + console.error("[Ingest API Bridge] Failed to send response:", err) + } + }, + ) + console.log("[Ingest API Bridge] Listening for ingest-api://request events") +} + +/** Stop listening and clean up. */ +export async function stopIngestApiBridge(): Promise { + if (unlisten) { + unlisten() + unlisten = null + } +}