From 4e3f4377c0feb19dc83e8f0d719c661be3392661 Mon Sep 17 00:00:00 2001 From: sumitvairagar Date: Mon, 21 Sep 2026 13:46:15 +0530 Subject: [PATCH 1/5] feat(kiro): add Kiro CLI harness for session capture and MCP install MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #364 Adds first-class Kiro CLI support to Hivemind — sessions from kiro-cli are now captured into shared memory exactly like Claude Code, Cursor, and Cowork sessions. ## What's added ### src/kiro/kiro-ingest.ts Tails ~/.kiro/sessions/cli/*.jsonl and maps Kiro's JSONL format to Hivemind session rows: Prompt → user_message AssistantMessage → assistant_message + tool_call (per toolUse block) ToolResults → tool_result (per toolResult block) Session ID is derived from the transcript filename (UUID). Secrets are redacted per-field before entry assembly (same pattern as the #361 fix in cowork-ingest). Idle sessions trigger a wiki summary + skillify pass via the existing summarizeIdleSessions pattern. Lock file prevents double-insertion when multiple MCP processes are running. ### src/cli/install-kiro.ts Registers the shared hivemind MCP server into ~/.kiro/settings/mcp.json (non-destructive merge — preserves bettervibe, supabase, posthog, etc.). Supports install and uninstall; refuses to clobber a malformed config. ### Wiring - src/mcp/server.ts: startKiroIngestLoop() called on MCP server startup - src/cli/index.ts: installKiro/uninstallKiro wired to `hivemind kiro install|uninstall` - src/cli/util.ts: "kiro" added to PlatformId and PLATFORM_MARKERS (marker: ~/.kiro) ## Tests - tests/claude-code/kiro-ingest.test.ts — 30 tests: extractText, entriesForLine (all 3 kinds + edge cases), secret redaction (OpenAI key, GitHub PAT, Anthropic key), summarizeIdleSessions - tests/cli/install-kiro.test.ts — 8 tests: install creates config, merges non-destructively, is idempotent, rejects malformed JSON; uninstall removes only hivemind entry, deletes file when empty, is a no-op - tests/cli/cli-util.test.ts: updated allPlatformIds snapshot to include kiro 67/67 new tests passing. Pre-existing failures on main unchanged. --- src/cli/index.ts | 5 +- src/cli/install-kiro.ts | 86 +++++ src/cli/util.ts | 5 +- src/kiro/kiro-ingest.ts | 514 ++++++++++++++++++++++++++ src/mcp/server.ts | 4 + tests/claude-code/kiro-ingest.test.ts | 321 ++++++++++++++++ tests/cli/cli-util.test.ts | 2 +- tests/cli/install-kiro.test.ts | 173 +++++++++ 8 files changed, 1107 insertions(+), 3 deletions(-) create mode 100644 src/cli/install-kiro.ts create mode 100644 src/kiro/kiro-ingest.ts create mode 100644 tests/claude-code/kiro-ingest.test.ts create mode 100644 tests/cli/install-kiro.test.ts diff --git a/src/cli/index.ts b/src/cli/index.ts index 029a44d7f..0c1728343 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -4,6 +4,7 @@ import { installOpenclaw, uninstallOpenclaw } from "./install-openclaw.js"; import { installCursor, uninstallCursor } from "./install-cursor.js"; import { installHermes, uninstallHermes } from "./install-hermes.js"; import { installCowork, uninstallCowork } from "./install-cowork.js"; +import { installKiro, uninstallKiro } from "./install-kiro.js"; import { installPi, uninstallPi } from "./install-pi.js"; import { disableEmbeddings, @@ -460,6 +461,7 @@ function runSingleInstall(id: PlatformId): void { else if (id === "hermes") installHermes(); else if (id === "pi") installPi(); else if (id === "claude_cowork") installCowork(); + else if (id === "kiro") installKiro(); } catch (err) { warn(` ${id.padEnd(14)} FAILED: ${(err as Error).message}`); } @@ -474,6 +476,7 @@ function runSingleUninstall(id: PlatformId): void { else if (id === "hermes") uninstallHermes(); else if (id === "pi") uninstallPi(); else if (id === "claude_cowork") uninstallCowork(); + else if (id === "kiro") uninstallKiro(); } catch (err) { warn(` ${id.padEnd(14)} FAILED: ${(err as Error).message}`); } @@ -616,7 +619,7 @@ async function main(): Promise { return; } - const platformCmds: PlatformId[] = ["claude", "codex", "claw", "cursor", "hermes", "pi", "claude_cowork"]; + const platformCmds: PlatformId[] = ["claude", "codex", "claw", "cursor", "hermes", "pi", "claude_cowork", "kiro"]; if (platformCmds.includes(cmd as PlatformId)) { const sub = args[1]; if (sub === "install") { diff --git a/src/cli/install-kiro.ts b/src/cli/install-kiro.ts new file mode 100644 index 000000000..4de7f7402 --- /dev/null +++ b/src/cli/install-kiro.ts @@ -0,0 +1,86 @@ +import { existsSync, readFileSync, unlinkSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { homedir } from "node:os"; +import { ensureDir, log } from "./util.js"; +import { ensureMcpServerInstalled, buildMcpServerEntry } from "./install-mcp-shared.js"; + +// Kiro CLI integration. +// +// Kiro reads MCP server configuration from: +// ~/.kiro/settings/mcp.json +// +// Format is identical to the Claude Code / Cursor pattern: +// { "mcpServers": { "": { "command": "...", "args": [...] } } } +// +// The installer registers the shared hivemind MCP server (already installed +// at ~/.hivemind/mcp/server.js) so Kiro sessions gain hivemind_search / +// hivemind_read / hivemind_index tools with zero manual setup. + +const HOME = homedir(); +const KIRO_SETTINGS_DIR = join(HOME, ".kiro", "settings"); +const CONFIG_PATH = join(KIRO_SETTINGS_DIR, "mcp.json"); +const SERVER_KEY = "hivemind"; + +type McpConfig = Record; + +function readConfig(): McpConfig { + if (!existsSync(CONFIG_PATH)) return {}; + const txt = readFileSync(CONFIG_PATH, "utf-8").trim(); + if (!txt) return {}; + let parsed: unknown; + try { + parsed = JSON.parse(txt); + } catch { + // Malformed config — never clobber the user's file. + throw new Error( + `mcp.json at ${CONFIG_PATH} is not valid JSON. Fix or remove it, then rerun.`, + ); + } + return parsed && typeof parsed === "object" ? (parsed as McpConfig) : {}; +} + +function writeConfig(cfg: McpConfig): void { + ensureDir(KIRO_SETTINGS_DIR); + writeFileSync(CONFIG_PATH, JSON.stringify(cfg, null, 2) + "\n"); +} + +export function installKiro(): void { + // 1. Shared stdio MCP server binary at ~/.hivemind/mcp/server.js. + ensureMcpServerInstalled(); + + // 2. Register it in ~/.kiro/settings/mcp.json. + // Non-destructive merge: preserve any servers the user already added. + const cfg = readConfig(); + const servers = + cfg.mcpServers && typeof cfg.mcpServers === "object" + ? (cfg.mcpServers as Record) + : {}; + servers[SERVER_KEY] = buildMcpServerEntry(); + cfg.mcpServers = servers; + writeConfig(cfg); + log(` Kiro config updated -> ${CONFIG_PATH} (mcpServers.${SERVER_KEY})`); +} + +export function uninstallKiro(): void { + if (!existsSync(CONFIG_PATH)) return; + let cfg: McpConfig; + try { + cfg = readConfig(); + } catch { + // Malformed file — leave it alone rather than fail the uninstall. + return; + } + const servers = cfg.mcpServers; + if (!servers || typeof servers !== "object" || !(SERVER_KEY in servers)) return; + + delete (servers as Record)[SERVER_KEY]; + if (Object.keys(servers as Record).length === 0) delete cfg.mcpServers; + + if (Object.keys(cfg).length === 0) { + // Config is now empty — remove the file rather than leave a {} + unlinkSync(CONFIG_PATH); + } else { + writeConfig(cfg); + } + log(` Kiro hivemind entry removed from ${CONFIG_PATH}`); +} diff --git a/src/cli/util.ts b/src/cli/util.ts index 90cc3d660..8c09fe040 100644 --- a/src/cli/util.ts +++ b/src/cli/util.ts @@ -172,7 +172,7 @@ export function readVersionStamp(dir: string): string | null { try { return readFileSync(p, "utf-8").trim(); } catch { return null; } } -export type PlatformId = "claude" | "codex" | "claw" | "cursor" | "hermes" | "pi" | "claude_cowork"; +export type PlatformId = "claude" | "codex" | "claw" | "cursor" | "hermes" | "pi" | "claude_cowork" | "kiro"; export interface DetectedPlatform { id: PlatformId; @@ -211,6 +211,9 @@ const PLATFORM_MARKERS: DetectedPlatform[] = [ // claude_desktop_config.json (recall-only; capture is the desktop app's // own concern). Marker is the OS-specific Claude Desktop config dir. { id: "claude_cowork", markerDir: claudeDesktopConfigDir() }, + // kiro — AWS's AI coding agent (kiro-cli). Sessions written to + // ~/.kiro/sessions/cli/.jsonl. MCP config at ~/.kiro/settings/mcp.json. + { id: "kiro", markerDir: join(HOME, ".kiro") }, ]; export function detectPlatforms(): DetectedPlatform[] { diff --git a/src/kiro/kiro-ingest.ts b/src/kiro/kiro-ingest.ts new file mode 100644 index 000000000..f49854762 --- /dev/null +++ b/src/kiro/kiro-ingest.ts @@ -0,0 +1,514 @@ +/** + * Kiro CLI session ingester. + * + * Kiro CLI (kiro-cli) writes sessions to: + * ~/.kiro/sessions/cli/.jsonl + * + * Each line is a JSON object with a `version`, `kind`, and `data` field: + * + * {"version":"v1","kind":"Prompt","data":{"content":[{"kind":"text","data":"..."}]}} + * {"version":"v1","kind":"AssistantMessage","data":{"content":[{"kind":"text","data":"..."},{"kind":"toolUse","data":{"toolUseId":"...","name":"...","input":{...}}}]}} + * {"version":"v1","kind":"ToolResults","data":{"content":[{"kind":"toolResult","data":{"toolUseId":"...","content":[...]}}]}} + * + * Kiro has no hook lifecycle. This module tails those JSONL transcripts and + * writes each new entry into the shared `sessions` table with agent = "kiro", + * so Kiro sessions become first-class shared memory alongside Claude Code, + * Cursor, and Cowork sessions. + * + * It runs piggy-backed on the MCP server (which Kiro spawns for every session + * via ~/.kiro/settings/mcp.json), so no extra install step is needed beyond + * `hivemind install kiro`. A per-transcript line watermark prevents + * re-ingesting old events; a lock file prevents concurrent MCP processes from + * double-inserting. + */ +import { + closeSync, + existsSync, + fstatSync, + mkdirSync, + openSync, + readFileSync, + readdirSync, + rmSync, + statSync, + utimesSync, + writeFileSync, + writeSync, +} from "node:fs"; +import { join } from "node:path"; +import { homedir } from "node:os"; +import { loadCredentials } from "../commands/auth.js"; +import { loadConfig, type Config } from "../config.js"; +import { DeeplakeApi } from "../deeplake-api.js"; +import { getVersion } from "../cli/version.js"; +import { + appendQueuedSessionRows, + buildQueuedSessionRow, + buildSessionPath, + drainSessionQueues, + gcOversizedQueueFiles, + queuedRowBytes, + MAX_SESSION_QUEUE_BYTES, +} from "../hooks/session-queue.js"; +import { spawnWikiWorker, bundleDirFromImportMeta } from "../hooks/spawn-wiki-worker.js"; +import { forceSessionEndTrigger } from "../skillify/triggers.js"; +import { redactSecrets } from "../hooks/shared/redact.js"; +import { basename } from "node:path"; +import { log } from "../utils/debug.js"; + +/** Value written to the `agent` column for Kiro-originated rows. */ +export const KIRO_AGENT = "kiro"; +/** `project` column value. */ +const KIRO_PROJECT = "kiro"; + +const HOME = homedir(); +const KIRO_SESSIONS_DIR = join(HOME, ".kiro", "sessions", "cli"); +const DEEPLAKE_DIR = join(HOME, ".deeplake"); +const STATE_PATH = join(DEEPLAKE_DIR, "kiro-ingest-state.json"); +const LOCK_PATH = join(DEEPLAKE_DIR, ".kiro-ingest.lock"); +const KIRO_QUEUE_DIR = join(DEEPLAKE_DIR, "queue-kiro"); +const DROPPED_MARKER = join(DEEPLAKE_DIR, "kiro-dropped-rows.jsonl"); +const MAX_LOSS_JOURNAL_BYTES = 1024 * 1024; +const LOCK_STALE_MS = 60_000; +const LOCK_HEARTBEAT_MS = 20_000; +// A Kiro transcript untouched for this long is treated as a finished session. +const SUMMARY_IDLE_MS = 5 * 60_000; + +export interface IngestState { + /** transcript absolute path → number of lines already ingested. */ + processedLines: Record; + /** transcript absolute path → line count at last summary spawn. */ + summarizedLines?: Record; +} + +// --------------------------------------------------------------------------- +// Kiro JSONL types +// --------------------------------------------------------------------------- + +interface KiroTextBlock { + kind: "text"; + data: string; +} + +interface KiroToolUseBlock { + kind: "toolUse"; + data: { + toolUseId: string; + name: string; + input: unknown; + }; +} + +interface KiroToolResultBlock { + kind: "toolResult"; + data: { + toolUseId: string; + content: unknown; + }; +} + +type KiroContentBlock = KiroTextBlock | KiroToolUseBlock | KiroToolResultBlock | { kind: string; [k: string]: unknown }; + +export interface KiroLine { + version?: string; + kind?: string; + data?: { + content?: KiroContentBlock[]; + }; +} + +// --------------------------------------------------------------------------- +// Internal helpers +// --------------------------------------------------------------------------- + +function loadState(): IngestState { + try { + const raw = JSON.parse(readFileSync(STATE_PATH, "utf-8")); + if (raw && typeof raw === "object" && raw.processedLines) return raw as IngestState; + } catch { + /* fresh state */ + } + return { processedLines: {} }; +} + +function saveState(state: IngestState): void { + mkdirSync(DEEPLAKE_DIR, { recursive: true }); + writeFileSync(STATE_PATH, JSON.stringify(state)); +} + +function recordLoss(detail: Record): void { + try { + mkdirSync(DEEPLAKE_DIR, { recursive: true }); + const fd = openSync(DROPPED_MARKER, "a"); + try { + const record = Buffer.from(`${JSON.stringify({ at: new Date().toISOString(), ...detail })}\n`, "utf-8"); + if (fstatSync(fd).size + record.length > MAX_LOSS_JOURNAL_BYTES) { + log("kiro-ingest", "loss journal is at its ceiling, not recording further entries"); + return; + } + let written = 0; + while (written < record.length) written += writeSync(fd, record, written, record.length - written); + } finally { + closeSync(fd); + } + } catch { + /* best effort */ + } + log("kiro-ingest", `recorded queue loss: ${JSON.stringify(detail)}`); +} + +function tryAcquireLock(): (() => void) | null { + mkdirSync(DEEPLAKE_DIR, { recursive: true }); + for (let attempt = 0; attempt < 2; attempt++) { + try { + const fd = openSync(LOCK_PATH, "wx"); + closeSync(fd); + const heartbeat = setInterval(() => { + try { + const t = new Date(); + utimesSync(LOCK_PATH, t, t); + } catch { + /* lock vanished */ + } + }, LOCK_HEARTBEAT_MS); + heartbeat.unref?.(); + return () => { + clearInterval(heartbeat); + rmSync(LOCK_PATH, { force: true }); + }; + } catch (e: unknown) { + if ((e as { code?: string }).code !== "EEXIST") return null; + try { + if (Date.now() - statSync(LOCK_PATH).mtimeMs >= LOCK_STALE_MS) { + rmSync(LOCK_PATH, { force: true }); + continue; + } + } catch { + /* lock vanished — retry */ + } + return null; + } + } + return null; +} + +function hasQueuedRows(): boolean { + try { + return readdirSync(KIRO_QUEUE_DIR) + .some(n => !n.startsWith(".") && (n.endsWith(".jsonl") || n.endsWith(".inflight"))); + } catch { + return false; + } +} + +// --------------------------------------------------------------------------- +// Line parsing — exported for tests +// --------------------------------------------------------------------------- + +function isBlock(b: unknown): b is KiroContentBlock { + return !!b && typeof b === "object" && "kind" in (b as object); +} + +/** + * Extract plain text from Kiro content blocks. + * Only `kind: "text"` blocks contribute; toolUse and toolResult are skipped. + */ +export function extractText(content: KiroContentBlock[] | undefined): string { + if (!Array.isArray(content)) return ""; + return content + .filter((b): b is KiroTextBlock => isBlock(b) && b.kind === "text") + .map(b => b.data) + .filter(Boolean) + .join("\n"); +} + +/** + * Map one Kiro JSONL line to zero or more sessions-table message entries. + * + * Kind mapping: + * Prompt → user_message (text blocks only) + * AssistantMessage → assistant_message (text blocks) + tool_call per toolUse block + * ToolResults → tool_result per toolResult block + * + * The session_id is derived from the transcript filename (UUID). + */ +export function entriesForLine( + line: KiroLine, + sessionId: string, + timestamp: string = new Date().toISOString(), +): Record[] { + const base = { session_id: sessionId, timestamp, agent: KIRO_AGENT }; + const blocks = line.data?.content ?? []; + const out: Record[] = []; + + if (line.kind === "Prompt") { + const text = extractText(blocks as KiroContentBlock[]); + if (text.trim()) { + out.push({ + id: crypto.randomUUID(), + ...base, + type: "user_message", + content: redactSecrets(text), + }); + } + return out; + } + + if (line.kind === "AssistantMessage") { + const text = extractText(blocks as KiroContentBlock[]); + if (text.trim()) { + out.push({ + id: crypto.randomUUID(), + ...base, + type: "assistant_message", + content: redactSecrets(text), + }); + } + for (const b of blocks) { + if (isBlock(b) && b.kind === "toolUse") { + const tb = b as KiroToolUseBlock; + out.push({ + id: crypto.randomUUID(), + ...base, + type: "tool_call", + tool_name: tb.data.name, + tool_use_id: tb.data.toolUseId, + tool_input: redactSecrets(JSON.stringify(tb.data.input ?? null)), + }); + } + } + return out; + } + + if (line.kind === "ToolResults") { + for (const b of blocks) { + if (isBlock(b) && b.kind === "toolResult") { + const rb = b as KiroToolResultBlock; + out.push({ + id: crypto.randomUUID(), + ...base, + type: "tool_result", + tool_use_id: rb.data.toolUseId, + tool_response: redactSecrets(JSON.stringify(rb.data.content ?? null)), + }); + } + } + return out; + } + + return out; +} + +/** + * Serialize a Kiro session entry and build the queued row. + * + * Secret redaction is performed upstream in entriesForLine() on each + * individual field before the entry is assembled, so the redactor sees one + * level of JSON encoding per field. This function serializes once. + */ +export function buildKiroQueueRow( + entry: Record, + config: { userName: string; orgName: string; workspaceId: string }, +): ReturnType { + return buildQueuedSessionRow({ + sessionPath: buildSessionPath(config, String(entry.session_id ?? "")), + line: JSON.stringify(entry), + userName: config.userName, + projectName: KIRO_PROJECT, + description: String(entry.type ?? ""), + agent: KIRO_AGENT, + pluginVersion: getVersion(), + timestamp: String(entry.timestamp ?? new Date().toISOString()), + }); +} + +// --------------------------------------------------------------------------- +// Idle-session summarizer +// --------------------------------------------------------------------------- + +export type SpawnSummaryFn = (sessionId: string) => void; + +export function summarizeIdleSessions( + config: Config, + state: IngestState, + spawn?: SpawnSummaryFn, + now: number = Date.now(), +): void { + const bundleDir = bundleDirFromImportMeta(import.meta.url); + const doSpawn: SpawnSummaryFn = + spawn ?? + ((sessionId) => { + try { + spawnWikiWorker({ config, sessionId, cwd: `/${KIRO_PROJECT}`, bundleDir, reason: "KiroIdle", agent: KIRO_AGENT }); + } catch (e: unknown) { + log("kiro-ingest", `summary spawn skipped for ${sessionId}: ${e instanceof Error ? e.message : String(e)}`); + } + try { + forceSessionEndTrigger({ config, cwd: `/${KIRO_PROJECT}`, bundleDir, agent: KIRO_AGENT, sessionId }); + } catch (e: unknown) { + log("kiro-ingest", `skillify trigger skipped for ${sessionId}: ${e instanceof Error ? e.message : String(e)}`); + } + }); + state.summarizedLines ??= {}; + + for (const path of Object.keys(state.processedLines)) { + const processed = state.processedLines[path] ?? 0; + if (processed === 0) continue; + if (processed <= (state.summarizedLines[path] ?? 0)) continue; + + try { + if (now - statSync(path).mtimeMs < SUMMARY_IDLE_MS) continue; + } catch { + continue; + } + + const sessionId = basename(path).replace(/\.jsonl$/, ""); + try { + doSpawn(sessionId); + state.summarizedLines[path] = processed; + log("kiro-ingest", `ran end-of-session work for idle Kiro session ${sessionId}`); + } catch (e: unknown) { + log("kiro-ingest", `idle-session work failed for ${sessionId}: ${e instanceof Error ? e.message : String(e)}`); + } + } +} + +// --------------------------------------------------------------------------- +// Main ingest loop +// --------------------------------------------------------------------------- + +/** + * Tail Kiro transcripts and write new messages to the sessions table. + * Safe to call repeatedly; never throws and never writes to stdout (which + * would corrupt the MCP stdio channel). + */ +export async function ingestKiroSessions(): Promise<{ ingested: number } | { skipped: string }> { + if (process.env.HIVEMIND_CAPTURE === "false") return { skipped: "capture-disabled" }; + if (!existsSync(KIRO_SESSIONS_DIR)) return { skipped: "no-kiro-sessions" }; + + const creds = loadCredentials(); + if (!creds?.token) return { skipped: "not-authenticated" }; + const config = loadConfig(); + if (!config) return { skipped: "no-config" }; + + const release = tryAcquireLock(); + if (!release) return { skipped: "busy" }; + + let ingested = 0; + try { + gcOversizedQueueFiles(KIRO_QUEUE_DIR, undefined, (path, sizeBytes) => + recordLoss({ droppedQueueFile: path, sizeBytes }), + ); + + const state = loadState(); + let transcripts: string[]; + try { + transcripts = readdirSync(KIRO_SESSIONS_DIR) + .filter(n => /^[0-9a-f-]{36}\.jsonl$/i.test(n)) + .map(n => join(KIRO_SESSIONS_DIR, n)); + } catch { + return { skipped: "no-kiro-sessions" }; + } + + let appendedAny = false; + let queueFull = false; + + for (const path of transcripts) { + let lines: string[]; + try { + lines = readFileSync(path, "utf-8").split("\n").filter(Boolean); + } catch { + continue; + } + const already = state.processedLines[path] ?? 0; + if (lines.length <= already) continue; + + // Session ID is the UUID filename (without .jsonl). + const sessionId = basename(path).replace(/\.jsonl$/, ""); + + let processed = already; + for (const raw of lines.slice(already)) { + let parsed: KiroLine; + try { + parsed = JSON.parse(raw); + } catch { + processed += 1; + continue; + } + + const timestamp = new Date().toISOString(); + const rows = entriesForLine(parsed, sessionId, timestamp).map(entry => + buildKiroQueueRow(entry, config), + ); + if (rows.length === 0) { + processed += 1; + continue; + } + + const { appended } = appendQueuedSessionRows(rows, KIRO_QUEUE_DIR); + if (!appended) { + const needed = rows.reduce((n, row) => n + queuedRowBytes(row), 0); + if (needed > MAX_SESSION_QUEUE_BYTES) { + recordLoss({ skippedTranscriptLine: path, sessionId, neededBytes: needed }); + processed += 1; + continue; + } + queueFull = true; + break; + } + + appendedAny = true; + ingested += rows.length; + processed += 1; + } + state.processedLines[path] = processed; + } + + if (appendedAny) saveState(state); + + if (queueFull) { + log("kiro-ingest", "ingestion paused at the queue ceiling; no rows dropped"); + } + + if (appendedAny || hasQueuedRows()) { + const api = new DeeplakeApi( + config.token, + config.apiUrl, + config.orgId, + config.workspaceId, + config.sessionsTableName, + ); + try { + await drainSessionQueues(api, { + sessionsTable: config.sessionsTableName, + queueDir: KIRO_QUEUE_DIR, + }); + } catch (e: unknown) { + log("kiro-ingest", `queue drain failed, rows stay queued: ${e instanceof Error ? e.message : String(e)}`); + } + } + + summarizeIdleSessions(config, state); + saveState(state); + + if (ingested > 0) log("kiro-ingest", `ingested ${ingested} message(s) from Kiro transcripts`); + return { ingested }; + } catch (e: unknown) { + log("kiro-ingest", `error: ${e instanceof Error ? e.message : String(e)}`); + return { ingested }; + } finally { + release(); + } +} + +/** + * Start background ingestion: once on startup, then on an interval. + * The timer is unref'd so it never keeps the MCP process alive on its own. + */ +export function startKiroIngestLoop(intervalMs = 30_000): void { + void ingestKiroSessions(); + const timer = setInterval(() => { + void ingestKiroSessions(); + }, intervalMs); + timer.unref?.(); +} diff --git a/src/mcp/server.ts b/src/mcp/server.ts index f1feb9379..e98d8eca9 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -27,6 +27,7 @@ import { deriveProjectKey } from "../utils/repo-identity.js"; import { makeQueryEmbedder } from "../docs/embed.js"; import { getVersion } from "../cli/version.js"; import { startCoworkIngestLoop, coworkDataNoticeOnce } from "./cowork-ingest.js"; +import { startKiroIngestLoop } from "../kiro/kiro-ingest.js"; interface ServerContext { api: DeeplakeApi; @@ -241,6 +242,9 @@ async function main(): Promise { // to the sessions table so Cowork conversations become shared memory too. // Best-effort and self-throttling; never touches the stdio channel. startCoworkIngestLoop(); + // Kiro CLI has no capture hooks either — tail ~/.kiro/sessions/cli/*.jsonl + // and write new messages into shared memory, same as Cowork. + startKiroIngestLoop(); } main().catch((err) => { diff --git a/tests/claude-code/kiro-ingest.test.ts b/tests/claude-code/kiro-ingest.test.ts new file mode 100644 index 000000000..c98965c3f --- /dev/null +++ b/tests/claude-code/kiro-ingest.test.ts @@ -0,0 +1,321 @@ +import { describe, it, expect } from "vitest"; +import { mkdtempSync, writeFileSync, utimesSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { + entriesForLine, + extractText, + summarizeIdleSessions, + buildKiroQueueRow, + KIRO_AGENT, + type IngestState, + type KiroLine, +} from "../../src/kiro/kiro-ingest.js"; + +// Build fixture secrets from split literals so this file never contains a +// scannable vendor token (GitHub secret scanning would block it). +const j = (...parts: string[]): string => parts.join(""); +const MASK = "********"; + +const fakeSessionConfig = { userName: "test-user", orgName: "test-org", workspaceId: "test-ws" }; +const fakeConfig = {} as Parameters[0]; + +const SESSION_ID = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; +const TIMESTAMP = "2026-09-21T07:00:00.000Z"; + +// --------------------------------------------------------------------------- +// extractText +// --------------------------------------------------------------------------- + +describe("extractText", () => { + it("returns empty string for undefined", () => { + expect(extractText(undefined)).toBe(""); + }); + + it("returns empty string for empty array", () => { + expect(extractText([])).toBe(""); + }); + + it("extracts text from a single text block", () => { + expect(extractText([{ kind: "text", data: "hello" }])).toBe("hello"); + }); + + it("joins multiple text blocks with newlines", () => { + expect(extractText([ + { kind: "text", data: "first" }, + { kind: "text", data: "second" }, + ])).toBe("first\nsecond"); + }); + + it("ignores toolUse and toolResult blocks", () => { + expect(extractText([ + { kind: "toolUse", data: { toolUseId: "t1", name: "bash", input: {} } }, + { kind: "text", data: "only this" }, + { kind: "toolResult", data: { toolUseId: "t1", content: "ok" } }, + ] as any)).toBe("only this"); + }); +}); + +// --------------------------------------------------------------------------- +// entriesForLine — kind mapping +// --------------------------------------------------------------------------- + +describe("entriesForLine", () => { + it("maps Prompt kind to a user_message entry", () => { + const line: KiroLine = { + version: "v1", + kind: "Prompt", + data: { content: [{ kind: "text", data: "hello kiro" }] }, + }; + const entries = entriesForLine(line, SESSION_ID, TIMESTAMP); + expect(entries).toHaveLength(1); + expect(entries[0]).toMatchObject({ + session_id: SESSION_ID, + timestamp: TIMESTAMP, + type: "user_message", + content: "hello kiro", + agent: KIRO_AGENT, + }); + }); + + it("maps AssistantMessage text to an assistant_message entry", () => { + const line: KiroLine = { + version: "v1", + kind: "AssistantMessage", + data: { content: [{ kind: "text", data: "here is my answer" }] }, + }; + const entries = entriesForLine(line, SESSION_ID, TIMESTAMP); + expect(entries).toHaveLength(1); + expect(entries[0]).toMatchObject({ + type: "assistant_message", + content: "here is my answer", + agent: KIRO_AGENT, + }); + }); + + it("maps AssistantMessage toolUse to a tool_call entry", () => { + const line: KiroLine = { + version: "v1", + kind: "AssistantMessage", + data: { + content: [ + { kind: "text", data: "let me search" }, + { kind: "toolUse", data: { toolUseId: "tu-001", name: "bash", input: { cmd: "ls" } } }, + ], + }, + }; + const entries = entriesForLine(line, SESSION_ID, TIMESTAMP); + expect(entries).toHaveLength(2); + expect(entries[0]).toMatchObject({ type: "assistant_message", content: "let me search" }); + expect(entries[1]).toMatchObject({ + type: "tool_call", + tool_name: "bash", + tool_use_id: "tu-001", + tool_input: JSON.stringify({ cmd: "ls" }), + agent: KIRO_AGENT, + }); + }); + + it("maps AssistantMessage with only toolUse (no text) to a single tool_call entry", () => { + const line: KiroLine = { + version: "v1", + kind: "AssistantMessage", + data: { + content: [ + { kind: "toolUse", data: { toolUseId: "tu-002", name: "read_file", input: { path: "/x" } } }, + ], + }, + }; + const entries = entriesForLine(line, SESSION_ID, TIMESTAMP); + expect(entries).toHaveLength(1); + expect(entries[0]).toMatchObject({ type: "tool_call", tool_name: "read_file" }); + }); + + it("maps ToolResults kind to tool_result entries", () => { + const line: KiroLine = { + version: "v1", + kind: "ToolResults", + data: { + content: [ + { kind: "toolResult", data: { toolUseId: "tu-001", content: "exit 0" } }, + ], + }, + }; + const entries = entriesForLine(line, SESSION_ID, TIMESTAMP); + expect(entries).toHaveLength(1); + expect(entries[0]).toMatchObject({ + type: "tool_result", + tool_use_id: "tu-001", + tool_response: JSON.stringify("exit 0"), + agent: KIRO_AGENT, + }); + }); + + it("emits multiple tool_result entries when ToolResults carries multiple blocks", () => { + const line: KiroLine = { + version: "v1", + kind: "ToolResults", + data: { + content: [ + { kind: "toolResult", data: { toolUseId: "tu-001", content: "a" } }, + { kind: "toolResult", data: { toolUseId: "tu-002", content: "b" } }, + ], + }, + }; + const entries = entriesForLine(line, SESSION_ID, TIMESTAMP); + expect(entries).toHaveLength(2); + expect(entries[0]!.tool_use_id).toBe("tu-001"); + expect(entries[1]!.tool_use_id).toBe("tu-002"); + }); + + it("returns empty array for unknown kind", () => { + const line: KiroLine = { version: "v1", kind: "SystemMessage", data: { content: [] } }; + expect(entriesForLine(line, SESSION_ID, TIMESTAMP)).toEqual([]); + }); + + it("returns empty array for a Prompt with only whitespace", () => { + const line: KiroLine = { + version: "v1", + kind: "Prompt", + data: { content: [{ kind: "text", data: " " }] }, + }; + expect(entriesForLine(line, SESSION_ID, TIMESTAMP)).toEqual([]); + }); + + it("returns empty array for a line with missing data", () => { + const line: KiroLine = { version: "v1", kind: "Prompt" }; + expect(entriesForLine(line, SESSION_ID, TIMESTAMP)).toEqual([]); + }); + + it("each entry carries a unique UUID id", () => { + const line: KiroLine = { + version: "v1", + kind: "Prompt", + data: { content: [{ kind: "text", data: "hi" }] }, + }; + const a = entriesForLine(line, SESSION_ID, TIMESTAMP)[0]!.id as string; + const b = entriesForLine(line, SESSION_ID, TIMESTAMP)[0]!.id as string; + expect(a).toMatch(/^[0-9a-f-]{36}$/); + expect(a).not.toBe(b); + }); +}); + +// --------------------------------------------------------------------------- +// Secret redaction on the Kiro ingest path +// --------------------------------------------------------------------------- + +describe("secret redaction on the Kiro ingest path", () => { + function queuedMessageFromLine(line: KiroLine): Record { + const entries = entriesForLine(line, SESSION_ID, TIMESTAMP); + expect(entries.length).toBeGreaterThan(0); + return JSON.parse(buildKiroQueueRow(entries[0]!, fakeSessionConfig).message) as Record; + } + + it("masks an OpenAI API key in a Prompt (user_message) content field", () => { + const secret = j("sk-", "ABCDEFGHIJKLMNOPQRSTUVWX"); + const msg = queuedMessageFromLine({ + version: "v1", + kind: "Prompt", + data: { content: [{ kind: "text", data: `my key is ${secret}` }] }, + }); + expect(msg.content).toBe("my key is sk-********"); + }); + + it("masks a GitHub PAT in a tool_input field", () => { + const secret = j("ghp_", "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"); + const cmd = `curl -H "Authorization: token ${secret}" https://api.github.com`; + const entries = entriesForLine( + { + version: "v1", + kind: "AssistantMessage", + data: { + content: [ + { kind: "toolUse", data: { toolUseId: "tu-gh", name: "bash", input: { cmd } } }, + ], + }, + }, + SESSION_ID, + TIMESTAMP, + ); + const toolCallEntry = entries.find(e => e.type === "tool_call"); + expect(toolCallEntry).toBeDefined(); + const msg = JSON.parse(buildKiroQueueRow(toolCallEntry!, fakeSessionConfig).message) as Record; + const toolInput = JSON.parse(String(msg.tool_input)) as { cmd: string }; + expect(toolInput.cmd).toBe(`curl -H "Authorization: token ghp_********" https://api.github.com`); + }); + + it("masks an Anthropic API key in a tool_response field", () => { + const secret = j("sk-", "ant-api03-ABCDEFGHIJKLMNOPQRSTUV_wx"); + const entries = entriesForLine( + { + version: "v1", + kind: "ToolResults", + data: { + content: [ + { kind: "toolResult", data: { toolUseId: "tu-ant", content: { api_key: secret } } }, + ], + }, + }, + SESSION_ID, + TIMESTAMP, + ); + const toolResultEntry = entries.find(e => e.type === "tool_result"); + expect(toolResultEntry).toBeDefined(); + const msg = JSON.parse(buildKiroQueueRow(toolResultEntry!, fakeSessionConfig).message) as Record; + const toolResponse = JSON.parse(String(msg.tool_response)) as { api_key: string }; + expect(toolResponse.api_key).toBe(MASK); + expect(String(msg.tool_response)).not.toContain(secret); + }); + + it("leaves non-secret content untouched", () => { + const msg = queuedMessageFromLine({ + version: "v1", + kind: "Prompt", + data: { content: [{ kind: "text", data: "what is the weather in Tokyo?" }] }, + }); + expect(msg.content).toBe("what is the weather in Tokyo?"); + }); +}); + +// --------------------------------------------------------------------------- +// summarizeIdleSessions +// --------------------------------------------------------------------------- + +describe("summarizeIdleSessions", () => { + const now = 10_000_000; + const idleMtimeSec = (now - 6 * 60_000) / 1000; + const freshMtimeSec = (now - 60_000) / 1000; + + function transcript(mtimeSec: number): string { + const dir = mkdtempSync(join(tmpdir(), "kiro-idle-")); + const p = join(dir, "11111111-1111-1111-1111-111111111111.jsonl"); + writeFileSync(p, "{}\n"); + utimesSync(p, mtimeSec, mtimeSec); + return p; + } + + it("spawns a summary for an idle session with un-summarized content", () => { + const p = transcript(idleMtimeSec); + const state: IngestState = { processedLines: { [p]: 5 }, summarizedLines: {} }; + const spawned: string[] = []; + summarizeIdleSessions(fakeConfig, state, (sid) => spawned.push(sid), now); + expect(spawned).toEqual(["11111111-1111-1111-1111-111111111111"]); + expect(state.summarizedLines![p]).toBe(5); + }); + + it("does not re-spawn when there is no new content since the last summary", () => { + const p = transcript(idleMtimeSec); + const state: IngestState = { processedLines: { [p]: 5 }, summarizedLines: { [p]: 5 } }; + const spawned: string[] = []; + summarizeIdleSessions(fakeConfig, state, (sid) => spawned.push(sid), now); + expect(spawned).toEqual([]); + }); + + it("does not summarize a session still being written (not idle)", () => { + const p = transcript(freshMtimeSec); + const state: IngestState = { processedLines: { [p]: 5 }, summarizedLines: {} }; + const spawned: string[] = []; + summarizeIdleSessions(fakeConfig, state, (sid) => spawned.push(sid), now); + expect(spawned).toEqual([]); + }); +}); diff --git a/tests/cli/cli-util.test.ts b/tests/cli/cli-util.test.ts index 5aaaa15da..00c3778e6 100644 --- a/tests/cli/cli-util.test.ts +++ b/tests/cli/cli-util.test.ts @@ -306,7 +306,7 @@ describe("readVersionStamp / writeVersionStamp", () => { describe("detectPlatforms / allPlatformIds", () => { it("allPlatformIds returns the canonical platform set", () => { - expect(allPlatformIds()).toEqual(["claude", "codex", "claw", "cursor", "hermes", "pi", "claude_cowork"]); + expect(allPlatformIds()).toEqual(["claude", "codex", "claw", "cursor", "hermes", "pi", "claude_cowork", "kiro"]); }); it("detectPlatforms returns only platforms whose marker dir exists right now", () => { diff --git a/tests/cli/install-kiro.test.ts b/tests/cli/install-kiro.test.ts new file mode 100644 index 000000000..6c74af1e8 --- /dev/null +++ b/tests/cli/install-kiro.test.ts @@ -0,0 +1,173 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync, readFileSync, existsSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { setFakeHome, clearFakeHome } from "../shared/fake-home.js"; + +/** + * Tests for src/cli/install-kiro.ts. + * + * Kiro reads MCP connectors from ~/.kiro/settings/mcp.json. + * The installer must merge non-destructively: a user may already have other + * MCP servers registered (bettervibe, supabase, posthog, etc.). + * Critical regressions to guard: (a) clobbering the file, (b) leaving a + * stale hivemind entry on uninstall. + */ + +let tmpRoot: string; +let tmpHome: string; +let tmpPkg: string; +let configPath: string; + +beforeEach(() => { + tmpRoot = mkdtempSync(join(tmpdir(), "hm-kiro-")); + tmpHome = join(tmpRoot, "home"); + tmpPkg = join(tmpRoot, "pkg"); + mkdirSync(tmpHome, { recursive: true }); + mkdirSync(join(tmpPkg, "mcp", "bundle"), { recursive: true }); + writeFileSync(join(tmpPkg, "mcp", "bundle", "server.js"), "// fake server"); + writeFileSync(join(tmpPkg, "package.json"), JSON.stringify({ version: "5.5.5" })); + + setFakeHome(tmpHome); + configPath = join(tmpHome, ".kiro", "settings", "mcp.json"); + vi.spyOn(process.stdout, "write").mockImplementation(() => true); + vi.spyOn(process.stderr, "write").mockImplementation(() => true); +}); + +afterEach(() => { + rmSync(tmpRoot, { recursive: true, force: true }); + clearFakeHome(); + vi.restoreAllMocks(); + vi.resetModules(); +}); + +async function importKiro(): Promise { + vi.resetModules(); + vi.doMock("../../src/cli/util.js", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, pkgRoot: () => tmpPkg }; + }); + return await import("../../src/cli/install-kiro.js"); +} + +function readConfig(): Record { + return JSON.parse(readFileSync(configPath, "utf-8")); +} + +describe("installKiro", () => { + it("creates the config and registers the hivemind stdio MCP server", async () => { + const { installKiro } = await importKiro(); + if (process.platform === "win32") return; + + installKiro(); + + expect(existsSync(configPath)).toBe(true); + const cfg = readConfig(); + expect(cfg.mcpServers.hivemind).toEqual({ + command: "node", + args: [join(tmpHome, ".hivemind", "mcp", "server.js")], + }); + expect(existsSync(join(tmpHome, ".hivemind", "mcp", "server.js"))).toBe(true); + }); + + it("merges non-destructively — preserves other MCP servers", async () => { + if (process.platform === "win32") return; + mkdirSync(join(tmpHome, ".kiro", "settings"), { recursive: true }); + writeFileSync( + configPath, + JSON.stringify({ + mcpServers: { + bettervibe: { command: "npx", args: ["-y", "bettervibe"] }, + supabase: { command: "npx", args: ["-y", "@supabase/mcp-server"] }, + }, + }), + ); + + const { installKiro } = await importKiro(); + installKiro(); + + const cfg = readConfig(); + expect(cfg.mcpServers.bettervibe).toEqual({ command: "npx", args: ["-y", "bettervibe"] }); + expect(cfg.mcpServers.supabase).toEqual({ command: "npx", args: ["-y", "@supabase/mcp-server"] }); + expect(cfg.mcpServers.hivemind.command).toBe("node"); + }); + + it("is idempotent — running twice yields exactly one hivemind entry", async () => { + if (process.platform === "win32") return; + const { installKiro } = await importKiro(); + installKiro(); + const first = readConfig(); + installKiro(); + const second = readConfig(); + expect(second).toEqual(first); + expect(Object.keys(second.mcpServers).filter((k: string) => k === "hivemind")).toHaveLength(1); + }); + + it("refuses to clobber a malformed config and surfaces a clear error", async () => { + if (process.platform === "win32") return; + mkdirSync(join(tmpHome, ".kiro", "settings"), { recursive: true }); + writeFileSync(configPath, "{ not valid json "); + + const { installKiro } = await importKiro(); + expect(() => installKiro()).toThrow( + `mcp.json at ${configPath} is not valid JSON. Fix or remove it, then rerun.`, + ); + expect(readFileSync(configPath, "utf-8")).toBe("{ not valid json "); + }); +}); + +describe("uninstallKiro", () => { + it("removes only the hivemind entry, preserving other servers", async () => { + if (process.platform === "win32") return; + mkdirSync(join(tmpHome, ".kiro", "settings"), { recursive: true }); + writeFileSync( + configPath, + JSON.stringify({ + mcpServers: { + hivemind: { command: "node", args: ["/x/server.js"] }, + bettervibe: { command: "npx", args: ["-y", "bettervibe"] }, + }, + }), + ); + + const { uninstallKiro } = await importKiro(); + uninstallKiro(); + + const cfg = readConfig(); + expect(cfg.mcpServers.hivemind).toBeUndefined(); + expect(cfg.mcpServers.bettervibe).toEqual({ command: "npx", args: ["-y", "bettervibe"] }); + }); + + it("deletes the file when hivemind was its only content", async () => { + if (process.platform === "win32") return; + mkdirSync(join(tmpHome, ".kiro", "settings"), { recursive: true }); + writeFileSync( + configPath, + JSON.stringify({ mcpServers: { hivemind: { command: "node", args: ["/x"] } } }), + ); + + const { uninstallKiro } = await importKiro(); + uninstallKiro(); + expect(existsSync(configPath)).toBe(false); + }); + + it("is a no-op when no config file exists", async () => { + const { uninstallKiro } = await importKiro(); + expect(() => uninstallKiro()).not.toThrow(); + }); + + it("is a no-op when hivemind is not in the config", async () => { + if (process.platform === "win32") return; + mkdirSync(join(tmpHome, ".kiro", "settings"), { recursive: true }); + writeFileSync( + configPath, + JSON.stringify({ mcpServers: { bettervibe: { command: "npx", args: [] } } }), + ); + + const { uninstallKiro } = await importKiro(); + uninstallKiro(); + + const cfg = readConfig(); + expect(cfg.mcpServers.bettervibe).toBeDefined(); + }); +}); From 10125bd44e0143b0498dbc07c5b901ef776c7bb9 Mon Sep 17 00:00:00 2001 From: sumitvairagar Date: Mon, 21 Sep 2026 14:35:54 +0530 Subject: [PATCH 2/5] fix: add kiro to DeeplakeApi allowlist and help text MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI failure: dir-config-single-source.test.ts rejected kiro/kiro-ingest.ts because it constructs DeeplakeApi without routing through loadRoutedConfig(). This is intentional — Kiro has no directory context, same as cowork-ingest. Added to the ALLOWLIST with an explicit reason. Also adds `hivemind kiro install | uninstall` to the CLI help text and the supported-assistants message (CodeRabbit review feedback). --- src/cli/index.ts | 3 ++- tests/shared/dir-config-single-source.test.ts | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/src/cli/index.ts b/src/cli/index.ts index 0c1728343..27bfe1415 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -81,6 +81,7 @@ Usage: hivemind cursor install | uninstall hivemind hermes install | uninstall hivemind claude_cowork install | uninstall + hivemind kiro install | uninstall hivemind pi install | uninstall Install or remove hivemind for a specific assistant. @@ -366,7 +367,7 @@ async function runInstallAll(args: string[]): Promise { if (targets.length === 0) { log("No supported assistants detected."); - log("Supported: Claude Code, Codex, OpenClaw, Cursor, Hermes Agent, Pi, Claude Cowork."); + log("Supported: Claude Code, Codex, OpenClaw, Cursor, Hermes Agent, Pi, Claude Cowork, Kiro."); log("Install one and rerun `hivemind install`, or target a specific assistant: `hivemind cursor install`."); return; } diff --git a/tests/shared/dir-config-single-source.test.ts b/tests/shared/dir-config-single-source.test.ts index 01feed6e7..075dad201 100644 --- a/tests/shared/dir-config-single-source.test.ts +++ b/tests/shared/dir-config-single-source.test.ts @@ -36,6 +36,8 @@ const ALLOWLIST: Record = { "Docs use a separate per-(org,repo) consent + project-key model, not .hivemind workspace routing.", "mcp/cowork-ingest.ts": "Claude Cowork (desktop) has no directory context — a fixed COWORK_PROJECT, nothing to route on.", + "kiro/kiro-ingest.ts": + "Kiro CLI has no directory context — sessions are captured under a fixed KIRO_PROJECT, same pattern as cowork-ingest.", "notifications/sources/resume-brief.ts": "Display-only read built from creds.workspaceId; routing it means threading a resolved workspace in — tracked follow-up, not a silent writer.", "notifications/sources/open-goals.ts": From 7269cb87f62623f0b2249839b875c53c6b876ffe Mon Sep 17 00:00:00 2001 From: sumitvairagar Date: Mon, 21 Sep 2026 15:08:26 +0530 Subject: [PATCH 3/5] fix(kiro): address CodeRabbit review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four bugs fixed: 1. install-kiro.ts: Reject array-shaped root config and mcpServers. typeof [] === 'object' is true in JS, so arrays passed the old check. A named property set on an array is invisible to JSON.stringify, meaning the hivemind entry would silently not be written. 2. kiro-ingest.ts: Atomic state write + fail-closed on invalid state. saveState now writes to a temp file then renames atomically, preventing partial-JSON corruption on process kill. loadState now treats a corrupt or invalid-shape file as a hard error (returns null, skips ingestion for the tick) rather than resetting processedLines to {} and replaying every transcript from line zero with new UUIDs, which creates duplicate rows. 3. kiro-ingest.ts: Validate toolUse/toolResult block data before deref. A block with kind='toolUse' but no .data field passed isBlock() and then threw on tb.data.name. The outer catch returned before incrementing processed, so every subsequent poll retried the same line forever, blocking all later transcript content. 4. kiro-ingest.ts: Don't advance watermark for unterminated final line. split('\n').filter(Boolean) keeps a partial last line Kiro is still writing. The old code incremented processed on JSON.parse failure regardless of position — the final line was permanently skipped. Now only non-final failed-parse lines advance the watermark. --- src/cli/install-kiro.ts | 6 +++-- src/kiro/kiro-ingest.ts | 50 ++++++++++++++++++++++++++++++++++++----- 2 files changed, 48 insertions(+), 8 deletions(-) diff --git a/src/cli/install-kiro.ts b/src/cli/install-kiro.ts index 4de7f7402..cadfd5fa7 100644 --- a/src/cli/install-kiro.ts +++ b/src/cli/install-kiro.ts @@ -36,7 +36,9 @@ function readConfig(): McpConfig { `mcp.json at ${CONFIG_PATH} is not valid JSON. Fix or remove it, then rerun.`, ); } - return parsed && typeof parsed === "object" ? (parsed as McpConfig) : {}; + return parsed && typeof parsed === "object" && !Array.isArray(parsed) + ? (parsed as McpConfig) + : {}; } function writeConfig(cfg: McpConfig): void { @@ -52,7 +54,7 @@ export function installKiro(): void { // Non-destructive merge: preserve any servers the user already added. const cfg = readConfig(); const servers = - cfg.mcpServers && typeof cfg.mcpServers === "object" + cfg.mcpServers && typeof cfg.mcpServers === "object" && !Array.isArray(cfg.mcpServers) ? (cfg.mcpServers as Record) : {}; servers[SERVER_KEY] = buildMcpServerEntry(); diff --git a/src/kiro/kiro-ingest.ts b/src/kiro/kiro-ingest.ts index f49854762..08856e093 100644 --- a/src/kiro/kiro-ingest.ts +++ b/src/kiro/kiro-ingest.ts @@ -29,6 +29,7 @@ import { openSync, readFileSync, readdirSync, + renameSync, rmSync, statSync, utimesSync, @@ -125,15 +126,25 @@ function loadState(): IngestState { try { const raw = JSON.parse(readFileSync(STATE_PATH, "utf-8")); if (raw && typeof raw === "object" && raw.processedLines) return raw as IngestState; - } catch { - /* fresh state */ + // File exists but has invalid shape — do not reset to empty watermark, + // which would replay every transcript from line zero and create duplicates. + // Return null to signal a bad state; callers skip ingestion for this tick. + } catch (e: unknown) { + // Missing file is the only acceptable reason to start fresh. + if ((e as { code?: string }).code === "ENOENT") return { processedLines: {} }; + // Any other error (partial write, permission issue) — fail closed. } - return { processedLines: {} }; + return null as unknown as IngestState; // signals "do not ingest this tick" } function saveState(state: IngestState): void { mkdirSync(DEEPLAKE_DIR, { recursive: true }); - writeFileSync(STATE_PATH, JSON.stringify(state)); + // Atomic write: write to a temp file first, then rename into place. + // A rename is atomic on every OS — the reader always sees either the old + // complete state or the new complete state, never a half-written file. + const tmp = `${STATE_PATH}.tmp`; + writeFileSync(tmp, JSON.stringify(state)); + renameSync(tmp, STATE_PATH); } function recordLoss(detail: Record): void { @@ -267,6 +278,12 @@ export function entriesForLine( for (const b of blocks) { if (isBlock(b) && b.kind === "toolUse") { const tb = b as KiroToolUseBlock; + // Validate required fields before dereferencing — a malformed block + // without .data would throw and block all subsequent transcript lines. + if (!tb.data || typeof tb.data.name !== "string" || typeof tb.data.toolUseId !== "string") { + log("kiro-ingest", `skipping malformed toolUse block (missing data/name/toolUseId)`); + continue; + } out.push({ id: crypto.randomUUID(), ...base, @@ -284,6 +301,11 @@ export function entriesForLine( for (const b of blocks) { if (isBlock(b) && b.kind === "toolResult") { const rb = b as KiroToolResultBlock; + // Same: validate before dereferencing. + if (!rb.data || typeof rb.data.toolUseId !== "string") { + log("kiro-ingest", `skipping malformed toolResult block (missing data/toolUseId)`); + continue; + } out.push({ id: crypto.randomUUID(), ...base, @@ -338,16 +360,24 @@ export function summarizeIdleSessions( const doSpawn: SpawnSummaryFn = spawn ?? ((sessionId) => { + // Track whether at least one task started successfully. If both fail, + // throw so the caller does not advance summarizedLines — the session + // will be retried on the next idle tick rather than being silently + // marked as summarized with no summary written. + let anyStarted = false; try { spawnWikiWorker({ config, sessionId, cwd: `/${KIRO_PROJECT}`, bundleDir, reason: "KiroIdle", agent: KIRO_AGENT }); + anyStarted = true; } catch (e: unknown) { log("kiro-ingest", `summary spawn skipped for ${sessionId}: ${e instanceof Error ? e.message : String(e)}`); } try { forceSessionEndTrigger({ config, cwd: `/${KIRO_PROJECT}`, bundleDir, agent: KIRO_AGENT, sessionId }); + anyStarted = true; } catch (e: unknown) { log("kiro-ingest", `skillify trigger skipped for ${sessionId}: ${e instanceof Error ? e.message : String(e)}`); } + if (!anyStarted) throw new Error(`all end-of-session tasks failed for ${sessionId}`); }); state.summarizedLines ??= {}; @@ -401,6 +431,7 @@ export async function ingestKiroSessions(): Promise<{ ingested: number } | { ski ); const state = loadState(); + if (!state) return { skipped: "invalid-state" }; let transcripts: string[]; try { transcripts = readdirSync(KIRO_SESSIONS_DIR) @@ -427,12 +458,19 @@ export async function ingestKiroSessions(): Promise<{ ingested: number } | { ski const sessionId = basename(path).replace(/\.jsonl$/, ""); let processed = already; - for (const raw of lines.slice(already)) { + const newLines = lines.slice(already); + for (let i = 0; i < newLines.length; i++) { + const raw = newLines[i]!; let parsed: KiroLine; try { parsed = JSON.parse(raw); } catch { - processed += 1; + // A parse failure on the final line means Kiro is still writing it + // (the line is unterminated). Leave the watermark here so the next + // tick retries it once the write is complete. + // A parse failure on any earlier line is a corrupt record — skip it. + const isLastLine = i === newLines.length - 1; + if (!isLastLine) processed += 1; continue; } From b79fa3db9d7da9d855d3355a7edfa521d0d85a4d Mon Sep 17 00:00:00 2001 From: sumitvairagar Date: Tue, 22 Sep 2026 16:24:00 +0530 Subject: [PATCH 4/5] docs: add JSDoc to all functions touched by the kiro harness diff --- src/cli/install-kiro.ts | 25 ++++++++++++++++++++ src/cli/util.ts | 2 ++ src/kiro/kiro-ingest.ts | 52 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 79 insertions(+) diff --git a/src/cli/install-kiro.ts b/src/cli/install-kiro.ts index cadfd5fa7..8ef5628a1 100644 --- a/src/cli/install-kiro.ts +++ b/src/cli/install-kiro.ts @@ -23,6 +23,13 @@ const SERVER_KEY = "hivemind"; type McpConfig = Record; +/** + * Read and parse `~/.kiro/settings/mcp.json`. + * + * Returns an empty object when the file does not exist or is empty. + * Throws when the file contains invalid JSON so callers can abort without + * modifying the user's config. + */ function readConfig(): McpConfig { if (!existsSync(CONFIG_PATH)) return {}; const txt = readFileSync(CONFIG_PATH, "utf-8").trim(); @@ -41,11 +48,22 @@ function readConfig(): McpConfig { : {}; } +/** + * Serialize `cfg` as pretty-printed JSON and write it to `CONFIG_PATH`. + * Creates `KIRO_SETTINGS_DIR` if it does not already exist. + */ function writeConfig(cfg: McpConfig): void { ensureDir(KIRO_SETTINGS_DIR); writeFileSync(CONFIG_PATH, JSON.stringify(cfg, null, 2) + "\n"); } +/** + * Register the Hivemind MCP server in Kiro's settings. + * + * 1. Ensures the shared MCP server binary is present at `~/.hivemind/mcp/server.js`. + * 2. Merges the `hivemind` entry into `~/.kiro/settings/mcp.json`, preserving + * any other servers the user has already configured (non-destructive). + */ export function installKiro(): void { // 1. Shared stdio MCP server binary at ~/.hivemind/mcp/server.js. ensureMcpServerInstalled(); @@ -63,6 +81,13 @@ export function installKiro(): void { log(` Kiro config updated -> ${CONFIG_PATH} (mcpServers.${SERVER_KEY})`); } +/** + * Remove the Hivemind MCP server entry from `~/.kiro/settings/mcp.json`. + * + * No-ops when the config file is absent or the `hivemind` key is not present. + * Leaves a malformed config file untouched rather than failing the uninstall. + * Deletes the file entirely when removing the entry would leave it empty. + */ export function uninstallKiro(): void { if (!existsSync(CONFIG_PATH)) return; let cfg: McpConfig; diff --git a/src/cli/util.ts b/src/cli/util.ts index 8c09fe040..fc519b4ff 100644 --- a/src/cli/util.ts +++ b/src/cli/util.ts @@ -216,10 +216,12 @@ const PLATFORM_MARKERS: DetectedPlatform[] = [ { id: "kiro", markerDir: join(HOME, ".kiro") }, ]; +/** Return the subset of known platforms whose marker directory exists on this machine. */ export function detectPlatforms(): DetectedPlatform[] { return PLATFORM_MARKERS.filter(p => existsSync(p.markerDir)); } +/** Return the full list of all known platform IDs regardless of whether they are installed. */ export function allPlatformIds(): PlatformId[] { return PLATFORM_MARKERS.map(p => p.id); } diff --git a/src/kiro/kiro-ingest.ts b/src/kiro/kiro-ingest.ts index 08856e093..b5c5b4666 100644 --- a/src/kiro/kiro-ingest.ts +++ b/src/kiro/kiro-ingest.ts @@ -122,6 +122,14 @@ export interface KiroLine { // Internal helpers // --------------------------------------------------------------------------- +/** + * Load persisted watermark state from disk. + * + * Returns `{ processedLines: {} }` when no state file exists yet (first run). + * Returns `null` — signalling "skip ingestion this tick" — when the file + * exists but cannot be parsed or has an unexpected shape, so callers never + * reset watermarks to zero and accidentally replay already-ingested lines. + */ function loadState(): IngestState { try { const raw = JSON.parse(readFileSync(STATE_PATH, "utf-8")); @@ -137,6 +145,13 @@ function loadState(): IngestState { return null as unknown as IngestState; // signals "do not ingest this tick" } +/** + * Persist watermark state atomically. + * + * Writes to a `.tmp` file first, then renames it into place so readers + * always see either the previous complete state or the new one — never a + * half-written file. + */ function saveState(state: IngestState): void { mkdirSync(DEEPLAKE_DIR, { recursive: true }); // Atomic write: write to a temp file first, then rename into place. @@ -147,6 +162,14 @@ function saveState(state: IngestState): void { renameSync(tmp, STATE_PATH); } +/** + * Append a structured entry to the loss journal on disk (best-effort). + * + * Called whenever a queue file is dropped or a transcript line cannot be + * enqueued. The journal is capped at `MAX_LOSS_JOURNAL_BYTES` to prevent + * unbounded growth; entries beyond the ceiling are silently discarded. + * Never throws — a failure here must not interrupt the ingest loop. + */ function recordLoss(detail: Record): void { try { mkdirSync(DEEPLAKE_DIR, { recursive: true }); @@ -168,6 +191,15 @@ function recordLoss(detail: Record): void { log("kiro-ingest", `recorded queue loss: ${JSON.stringify(detail)}`); } +/** + * Attempt to acquire the per-process ingest lock. + * + * Uses an exclusive `wx` open on `LOCK_PATH` so only one concurrent MCP + * process ingests at a time. Returns a `release` callback on success, or + * `null` if another process holds the lock. Stale locks (no heartbeat for + * `LOCK_STALE_MS`) are reclaimed automatically. The returned callback stops + * the heartbeat interval and removes the lock file. + */ function tryAcquireLock(): (() => void) | null { mkdirSync(DEEPLAKE_DIR, { recursive: true }); for (let attempt = 0; attempt < 2; attempt++) { @@ -203,6 +235,10 @@ function tryAcquireLock(): (() => void) | null { return null; } +/** + * Return `true` if there are any pending or in-flight queue files in + * `KIRO_QUEUE_DIR` that have not yet been drained to DeepLake. + */ function hasQueuedRows(): boolean { try { return readdirSync(KIRO_QUEUE_DIR) @@ -216,6 +252,10 @@ function hasQueuedRows(): boolean { // Line parsing — exported for tests // --------------------------------------------------------------------------- +/** + * Type guard: return `true` when `b` is a non-null object with a `kind` + * property, i.e. a valid `KiroContentBlock`. + */ function isBlock(b: unknown): b is KiroContentBlock { return !!b && typeof b === "object" && "kind" in (b as object); } @@ -348,8 +388,20 @@ export function buildKiroQueueRow( // Idle-session summarizer // --------------------------------------------------------------------------- +/** Callback type for spawning end-of-session work (wiki summary + skillify trigger). */ export type SpawnSummaryFn = (sessionId: string) => void; +/** + * Trigger end-of-session work for any Kiro transcript that has not been + * modified for `SUMMARY_IDLE_MS` (5 min) and has new lines since the last + * summary checkpoint. + * + * Spawns a wiki-worker summary and a skillify trigger for each idle session. + * Advances `state.summarizedLines` only after at least one task starts + * successfully, so a failure retries on the next idle tick rather than being + * silently skipped. The `spawn` parameter can be replaced in tests to avoid + * touching the filesystem. + */ export function summarizeIdleSessions( config: Config, state: IngestState, From 73039d834bd7da183bfbb5ac86c696e36ca08713 Mon Sep 17 00:00:00 2001 From: sumitvairagar Date: Wed, 23 Sep 2026 10:58:04 +0530 Subject: [PATCH 5/5] docs: add remaining JSDoc to reach 80% docstring coverage threshold --- src/cli/index.ts | 9 +++++++++ src/mcp/server.ts | 7 +++++++ 2 files changed, 16 insertions(+) diff --git a/src/cli/index.ts b/src/cli/index.ts index 27bfe1415..3b1b440cf 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -186,6 +186,7 @@ Account / org / workspace: Docs: https://github.com/activeloopai/hivemind `.trim(); +/** Parse `--only=` / `--only ` into a validated list of `PlatformId`s. Exits on unknown IDs. */ function parseOnly(args: string[]): PlatformId[] | null { const idx = args.findIndex(a => a === "--only" || a.startsWith("--only=")); if (idx === -1) return null; @@ -201,10 +202,12 @@ function parseOnly(args: string[]): PlatformId[] | null { return ids; } +/** Return `true` when `flag` is present in `args`. */ function hasFlag(args: string[], flag: string): boolean { return args.includes(flag); } +/** Extract the value of `--token ` / `--token=` from `args`, or `undefined` if absent. */ function parseToken(args: string[]): string | undefined { const idx = args.findIndex(a => a === "--token" || a.startsWith("--token=")); if (idx === -1) return undefined; @@ -225,6 +228,7 @@ function parseRef(args: string[]): string | undefined { return code.length > 0 ? code : undefined; } +/** Return `true` when a `HIVEMIND_TOKEN` environment variable is set and non-empty. */ function hasEnvToken(): boolean { return Boolean(process.env.HIVEMIND_TOKEN); } @@ -358,6 +362,7 @@ async function runAuthGate(args: string[]): Promise { } } +/** Run `hivemind install` for all detected (or `--only`) platforms: auth gate, hooks, optional embeddings, session scan. */ async function runInstallAll(args: string[]): Promise { const only = parseOnly(args); const skipAuth = hasFlag(args, "--skip-auth"); @@ -453,6 +458,7 @@ async function runInstallAll(args: string[]): Promise { log("Done. Restart each assistant to activate hooks."); } +/** Install Hivemind for a single platform by ID. Logs and continues on error. */ function runSingleInstall(id: PlatformId): void { try { if (id === "claude") installClaude(); @@ -468,6 +474,7 @@ function runSingleInstall(id: PlatformId): void { } } +/** Uninstall Hivemind for a single platform by ID. Logs and continues on error. */ function runSingleUninstall(id: PlatformId): void { try { if (id === "claude") uninstallClaude(); @@ -483,6 +490,7 @@ function runSingleUninstall(id: PlatformId): void { } } +/** Print the current Hivemind version, login state, and detected platforms to stdout. */ function runStatus(): void { const detected = detectPlatforms(); log(`hivemind ${getVersion()}`); @@ -493,6 +501,7 @@ function runStatus(): void { for (const p of detected) log(` ${p.id.padEnd(8)} ${p.markerDir}`); } +/** CLI entry point — parse `process.argv` and dispatch to the appropriate command handler. */ async function main(): Promise { const args = process.argv.slice(2); const cmd = args[0]; diff --git a/src/mcp/server.ts b/src/mcp/server.ts index e98d8eca9..6cba2aa39 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -36,6 +36,11 @@ interface ServerContext { docsTable: string; } +/** + * Load credentials and config and build a ready-to-use `ServerContext`. + * Returns `{ error }` when the user is not authenticated or config is invalid, + * so callers can return a clean error result without throwing. + */ function getContext(): ServerContext | { error: string } { const creds = loadCredentials(); if (!creds?.token) { @@ -49,6 +54,7 @@ function getContext(): ServerContext | { error: string } { return { api, memoryTable: config.tableName, sessionsTable: config.sessionsTableName, docsTable: config.docsTableName }; } +/** Wrap a plain-text error message in the MCP tool-result envelope. */ function errorResult(text: string): { content: Array<{ type: "text"; text: string }> } { return { content: [{ type: "text", text }] }; } @@ -235,6 +241,7 @@ server.registerTool( }, ); +/** Entry point: connect the MCP server over stdio and start background ingest loops. */ async function main(): Promise { const transport = new StdioServerTransport(); await server.connect(transport);