feat: add MCP tools for ingest pipeline control (status/pause/resume/retry-failed)

mcp-server/src/api-client.ts

@@ -296,6 +296,56 @@ export class LlmWikiApiClient {
     })
   }
 
+  async ingestStatus(projectId = "current"): Promise<ApiIngestStatusResponse> {
+    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<ApiIngestControlResponse> {
+    return this.parseIngestControlResponse(
+      await this.request(`/projects/${encodeURIComponent(projectId)}/ingest/pause`, {
+        method: "POST",
+      }),
+      "pause",
+    )
+  }
+
+  async ingestResume(projectId = "current"): Promise<ApiIngestControlResponse> {
+    return this.parseIngestControlResponse(
+      await this.request(`/projects/${encodeURIComponent(projectId)}/ingest/resume`, {
+        method: "POST",
+      }),
+      "resume",
+    )
+  }
+
+  async ingestRetryFailed(projectId = "current"): Promise<ApiIngestControlResponse> {
+    return this.parseIngestControlResponse(
+      await this.request(`/projects/${encodeURIComponent(projectId)}/ingest/retry-failed`, {
+        method: "POST",
+      }),
+      "retry-failed",
+    )
+  }
+
+  private parseIngestControlResponse(
+    json: Record<string, unknown>,
+    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<Record<string, unknown>> {
     const url = `${this.baseUrl}${apiPath(path)}`
     const headers: Record<string, string> = { 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<string, unknown>
+}
+
+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,
+  }
+}

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<string, unknown>
+  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<void> {
   const transport = new StdioServerTransport()
   await server.connect(transport)