From 2558f2d6a3cff1457418896b86deeaa63baa18f9 Mon Sep 17 00:00:00 2001 From: Shah Hussain <95882307+shhahhussain@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:22:42 +0500 Subject: [PATCH 1/7] test: define scoped resilient source adapters --- src/adapters/file-adapters.test.ts | 41 +++++++++++++++++++++++++ src/adapters/supermemory.test.ts | 49 ++++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+) create mode 100644 src/adapters/supermemory.test.ts diff --git a/src/adapters/file-adapters.test.ts b/src/adapters/file-adapters.test.ts index 5bf2eb1..7b40f2a 100644 --- a/src/adapters/file-adapters.test.ts +++ b/src/adapters/file-adapters.test.ts @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { DirectoryAdapter } from "./directory.js"; import { RawCsvAdapter } from "./raw-csv.js"; import { RawJsonAdapter } from "./raw-json.js"; +import { RawJsonlAdapter } from "./raw-jsonl.js"; async function collect(adapter: { iterate(): AsyncIterable }): Promise { const records: unknown[] = []; @@ -96,4 +97,44 @@ describe("file adapters", () => { Status: "Customer", }); }); + + it("maps each record to its own destination container", async () => { + const file = join(directory, "customers.csv"); + await fs.writeFile(file, "id,notes,customer\n1,Alpha note,customer:alpha\n"); + const adapter = new RawCsvAdapter({ + file, + map: "content=notes,id=id,container=customer", + }); + + await expect(collect(adapter)).resolves.toEqual([ + { + sourceId: "1", + content: "Alpha note", + contentType: undefined, + containerTag: "customer:alpha", + metadata: { source: "raw-csv", file }, + }, + ]); + }); + + it("reports malformed JSONL records and continues with valid records", async () => { + const file = join(directory, "mixed.jsonl"); + await fs.writeFile( + file, + '{"id":"1","content":"first"}\nnot-json\n{"id":"3","content":"third"}\n', + ); + const adapter = new RawJsonlAdapter({ file }); + const issues: Array<{ sourceId: string; error: string }> = []; + adapter.setIssueHandler((issue) => issues.push(issue)); + + const records = await collect(adapter) as Array<{ sourceId?: string }>; + + expect(records.map((record) => record.sourceId)).toEqual(["1", "3"]); + expect(issues).toEqual([ + { + sourceId: "line:2", + error: expect.stringContaining("Invalid JSON on line 2"), + }, + ]); + }); }); diff --git a/src/adapters/supermemory.test.ts b/src/adapters/supermemory.test.ts new file mode 100644 index 0000000..31a97a4 --- /dev/null +++ b/src/adapters/supermemory.test.ts @@ -0,0 +1,49 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { SupermemoryAdapter } from "./supermemory.js"; + +afterEach(() => { + delete process.env.SUPERMEMORY_API_KEY; + vi.unstubAllGlobals(); +}); + +describe("SupermemoryAdapter", () => { + it("reads current v4 memory pages for an explicitly scoped source container", async () => { + process.env.SUPERMEMORY_API_KEY = "source-key"; + const requests: Array<{ url: string; body: Record }> = []; + vi.stubGlobal( + "fetch", + vi.fn(async (url: string | URL | Request, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as Record; + requests.push({ url: String(url), body }); + const page = Number(body.page); + return new Response( + JSON.stringify({ + memoryEntries: [ + { + id: `memory-${page}`, + memory: page === 1 ? "first memory" : "second memory", + createdAt: "2026-07-22T00:00:00.000Z", + }, + ], + pagination: { currentPage: page, totalPages: 2, totalItems: 2, limit: 1 }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + }), + ); + + const adapter = new SupermemoryAdapter({ + sourceContainers: ["customer:alpha"], + pageSize: 1, + }); + const records = []; + for await (const record of adapter.iterate()) records.push(record); + + expect(requests).toHaveLength(2); + expect(requests[0]).toEqual({ + url: "https://api.supermemory.ai/v4/memories/list", + body: { containerTags: ["customer:alpha"], limit: 1, page: 1 }, + }); + expect(records.map((record) => record.content)).toEqual(["first memory", "second memory"]); + }); +}); From 01efce94295f43edd0308fc184766817932eecc6 Mon Sep 17 00:00:00 2001 From: Shah Hussain <95882307+shhahhussain@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:25:05 +0500 Subject: [PATCH 2/7] feat: support scoped resilient source adapters --- src/adapters/base.ts | 21 +++++++++++++ src/adapters/raw-jsonl.ts | 6 +++- src/adapters/record-mapping.ts | 14 ++++++++- src/adapters/supermemory.ts | 56 +++++++++++++++++++++++----------- 4 files changed, 77 insertions(+), 20 deletions(-) diff --git a/src/adapters/base.ts b/src/adapters/base.ts index a443923..9030c1a 100644 --- a/src/adapters/base.ts +++ b/src/adapters/base.ts @@ -9,8 +9,17 @@ export interface SourceRecord { metadata?: Record; /** Optional source-provided content classification. */ contentType?: string; + /** Optional per-record destination container resolved from source data. */ + containerTag?: string; } +export interface AdapterIssue { + sourceId: string; + error: string; +} + +export type AdapterIssueHandler = (issue: AdapterIssue) => void | Promise; + export interface AdapterConfig { /** Optional path to a JSONL file (used by raw-jsonl adapter). */ file?: string; @@ -22,14 +31,26 @@ export interface AdapterConfig { baseUrl?: string; /** Page size hint for paginated providers. */ pageSize?: number; + /** Source-side containers to read from when the provider requires scoping. */ + sourceContainers?: string[]; } export abstract class Adapter { abstract readonly name: string; abstract readonly defaultLane: ImportLane; + private issueHandler?: AdapterIssueHandler; + constructor(protected readonly cfg: AdapterConfig = {}) {} + setIssueHandler(handler: AdapterIssueHandler): void { + this.issueHandler = handler; + } + + protected async reportIssue(issue: AdapterIssue): Promise { + await this.issueHandler?.(issue); + } + /** Best-effort total record count. Return null if unknown. */ abstract count(): Promise; diff --git a/src/adapters/raw-jsonl.ts b/src/adapters/raw-jsonl.ts index 3589705..fa4e29f 100644 --- a/src/adapters/raw-jsonl.ts +++ b/src/adapters/raw-jsonl.ts @@ -40,7 +40,11 @@ export class RawJsonlAdapter extends Adapter { try { parsed = JSON.parse(trimmed) as Record; } catch (err) { - throw new Error(`Invalid JSON on line ${lineNo} of ${this.filePath}: ${(err as Error).message}`); + await this.reportIssue({ + sourceId: `line:${lineNo}`, + error: `Invalid JSON on line ${lineNo} of ${this.filePath}: ${(err as Error).message}`, + }); + continue; } const record = recordFromObject(parsed, this.map, { source: "raw-jsonl", diff --git a/src/adapters/record-mapping.ts b/src/adapters/record-mapping.ts index 2680c81..3ba9a1e 100644 --- a/src/adapters/record-mapping.ts +++ b/src/adapters/record-mapping.ts @@ -44,9 +44,17 @@ export function recordFromObject( const id = pluck(input, map.id!); const contentType = map.contentType ? pluck(input, map.contentType) : undefined; + const container = map.container ? pluck(input, map.container) : undefined; const metadata: Record = { ...baseMetadata }; for (const [target, source] of Object.entries(map)) { - if (target === "content" || target === "id" || target === "contentType") continue; + if ( + target === "content" || + target === "id" || + target === "contentType" || + target === "container" + ) { + continue; + } const value = pluck(input, source); if (value !== undefined) metadata[target] = value; } @@ -56,5 +64,9 @@ export function recordFromObject( content, metadata, contentType: typeof contentType === "string" ? contentType : undefined, + containerTag: + typeof container === "string" || typeof container === "number" + ? String(container) + : undefined, }; } diff --git a/src/adapters/supermemory.ts b/src/adapters/supermemory.ts index 394e1b3..d24ee50 100644 --- a/src/adapters/supermemory.ts +++ b/src/adapters/supermemory.ts @@ -2,17 +2,23 @@ import { Adapter, AdapterConfig, SourceRecord } from "./base.js"; interface SuperMemoryItem { id?: string; - content?: string; - text?: string; - type?: string; + memory?: string; metadata?: Record; createdAt?: string; + updatedAt?: string; + version?: number; + isLatest?: boolean; + isForgotten?: boolean; } interface SuperMemoryPage { - results?: SuperMemoryItem[]; - total?: number; - nextCursor?: string | null; + memoryEntries?: SuperMemoryItem[]; + pagination?: { + currentPage?: number; + totalPages?: number; + totalItems?: number; + limit?: number; + }; } const DEFAULT_BASE_URL = "https://api.supermemory.ai"; @@ -30,16 +36,26 @@ export class SupermemoryAdapter extends Adapter { if (!key) { throw new Error("SUPERMEMORY_API_KEY is required for the supermemory adapter"); } + if (!cfg.sourceContainers?.length) { + throw new Error("--source-container is required for the supermemory adapter"); + } this.apiKey = key; this.baseUrl = (cfg.baseUrl ?? DEFAULT_BASE_URL).replace(/\/$/, ""); } - private async request(path: string): Promise { - const res = await fetch(new URL(path, this.baseUrl + "/"), { + private async request(page: number, limit: number): Promise { + const res = await fetch(new URL("/v4/memories/list", this.baseUrl + "/"), { + method: "POST", headers: { Authorization: `Bearer ${this.apiKey}`, Accept: "application/json", + "Content-Type": "application/json", }, + body: JSON.stringify({ + containerTags: this.cfg.sourceContainers, + limit, + page, + }), }); if (!res.ok) { throw new Error(`supermemory request failed: ${res.status} ${res.statusText}`); @@ -48,31 +64,35 @@ export class SupermemoryAdapter extends Adapter { } async count(): Promise { - const page = await this.request("/v3/memories?limit=1"); - return typeof page.total === "number" ? page.total : null; + const page = await this.request(1, 1); + return page.pagination?.totalItems ?? null; } async *iterate(): AsyncIterable { const pageSize = this.cfg.pageSize ?? 100; - let cursor: string | null = null; + let pageNumber = 1; + let totalPages = 1; do { - const path = `/v3/memories?limit=${pageSize}${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ""}`; - const page = await this.request(path); - for (const item of page.results ?? []) { - const content = item.content ?? item.text; + const page = await this.request(pageNumber, pageSize); + for (const item of page.memoryEntries ?? []) { + const content = item.memory; if (!content) continue; yield { sourceId: item.id, content, metadata: { source: "supermemory", - ...(item.type ? { type: item.type } : {}), ...(item.createdAt ? { sourceCreatedAt: item.createdAt } : {}), + ...(item.updatedAt ? { sourceUpdatedAt: item.updatedAt } : {}), + ...(item.version !== undefined ? { sourceVersion: item.version } : {}), + ...(item.isLatest !== undefined ? { sourceIsLatest: item.isLatest } : {}), + ...(item.isForgotten !== undefined ? { sourceIsForgotten: item.isForgotten } : {}), ...(item.metadata ?? {}), }, }; } - cursor = page.nextCursor ?? null; - } while (cursor); + totalPages = page.pagination?.totalPages ?? pageNumber; + pageNumber += 1; + } while (pageNumber <= totalPages); } } From 581df565b30bd87fa467eb40a2869444e6128b2c Mon Sep 17 00:00:00 2001 From: Shah Hussain <95882307+shhahhussain@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:26:05 +0500 Subject: [PATCH 3/7] test: define bulk and isolated destination writes --- src/lib/mnemo-writer.test.ts | 73 ++++++++++++++++++++++++++++++++++++ src/migrator.test.ts | 15 ++++++++ 2 files changed, 88 insertions(+) diff --git a/src/lib/mnemo-writer.test.ts b/src/lib/mnemo-writer.test.ts index cbdf0d0..529f6d1 100644 --- a/src/lib/mnemo-writer.test.ts +++ b/src/lib/mnemo-writer.test.ts @@ -51,6 +51,79 @@ describe("MnemoWriter", () => { expect(result.remoteJobIds).toEqual(["job_1"]); }); + it("writes small documents through the batch endpoint", async () => { + const requests: Array<{ url: string; init?: RequestInit }> = []; + vi.stubGlobal( + "fetch", + vi.fn(async (url: string | URL | Request, init?: RequestInit) => { + requests.push({ url: String(url), init }); + return new Response( + JSON.stringify({ + success: 2, + failed: 0, + results: [ + { index: 0, status: "accepted", jobId: "job_1", reused: false }, + { index: 1, status: "accepted", jobId: "job_2", reused: true }, + ], + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + }), + ); + const writer = new MnemoWriter({ apiKey: "key", workspaceId: "workspace", apiUrl: "https://api.test" }); + + const results = await writer.writeBatch([ + { + record: { sourceId: "1", content: "First" }, + options: { lane: "document", source: "raw-jsonl", containerTag: "customer:one" }, + }, + { + record: { sourceId: "2", content: "Second" }, + options: { lane: "document", source: "raw-jsonl", containerTag: "customer:two" }, + }, + ]); + const body = JSON.parse(String(requests[0]?.init?.body)) as { documents: unknown[] }; + + expect(requests).toHaveLength(1); + expect(requests[0]?.url).toBe("https://api.test/v1/documents/batch"); + expect(body.documents).toHaveLength(2); + expect(results).toEqual([ + { reused: false, remoteJobIds: ["job_1"] }, + { reused: true, remoteJobIds: ["job_2"] }, + ]); + }); + + it("falls back to individual writes when the batch endpoint is unavailable", async () => { + const paths: string[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (url: string | URL | Request) => { + const path = new URL(String(url)).pathname; + paths.push(path); + if (path === "/v1/documents/batch") return new Response("not found", { status: 404 }); + return new Response(JSON.stringify({ jobId: `job_${paths.length}`, reused: false }), { + status: 202, + headers: { "content-type": "application/json" }, + }); + }), + ); + const writer = new MnemoWriter({ apiKey: "key", workspaceId: "workspace", apiUrl: "https://api.test" }); + + const results = await writer.writeBatch([ + { + record: { sourceId: "1", content: "First" }, + options: { lane: "document", source: "raw-jsonl", containerTag: "customer:one" }, + }, + { + record: { sourceId: "2", content: "Second" }, + options: { lane: "document", source: "raw-jsonl", containerTag: "customer:two" }, + }, + ]); + + expect(paths).toEqual(["/v1/documents/batch", "/v1/documents", "/v1/documents"]); + expect(results.every((result) => result instanceof Error)).toBe(false); + }); + it("retries rate limits and preserves the same request body", async () => { const bodies: string[] = []; const fetchMock = vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { diff --git a/src/migrator.test.ts b/src/migrator.test.ts index caa6c2c..c511ca6 100644 --- a/src/migrator.test.ts +++ b/src/migrator.test.ts @@ -109,6 +109,21 @@ describe("migrate", () => { expect(job.remotePending).toBe(3); }); + it("routes mapped records to their own destination containers", async () => { + const writer = new MockWriter(); + const records: SourceRecord[] = [ + { sourceId: "1", content: "First", containerTag: "customer:one" }, + { sourceId: "2", content: "Second", containerTag: "customer:two" }, + ]; + + await migrate(new MockAdapter(records), writer, options("memory")); + + expect(writer.calls.map((call) => call.options.containerTag)).toEqual([ + "customer:one", + "customer:two", + ]); + }); + it("resumes by retrying only unresolved records", async () => { const firstWriter = new MockWriter(); firstWriter.failIds.add("2"); From 887385737bb0c5c2a1fa99f59baee5177fa755a8 Mon Sep 17 00:00:00 2001 From: Shah Hussain <95882307+shhahhussain@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:26:54 +0500 Subject: [PATCH 4/7] feat: batch isolated document imports --- src/lib/migrator.ts | 20 +++++++- src/lib/mnemo-writer.ts | 104 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 122 insertions(+), 2 deletions(-) diff --git a/src/lib/migrator.ts b/src/lib/migrator.ts index 029652d..4a85878 100644 --- a/src/lib/migrator.ts +++ b/src/lib/migrator.ts @@ -50,6 +50,24 @@ async function processBatch( writer: DestinationWriter, options: MigrateOptions, ): Promise { + if (writer.writeBatch) { + const inputs = records.map((record) => ({ + record, + options: { + lane: options.lane, + source: adapter.name, + containerTag: record.containerTag ?? options.containerTag, + }, + })); + const results = await writer.writeBatch(inputs); + return records.map((record, index): RecordOutcome => { + const sourceId = sourceIdentity(record); + const result = results[index]; + if (!result) return { kind: "failure", sourceId, error: "batch writer omitted a result" }; + if (result instanceof Error) return { kind: "failure", sourceId, error: result.message }; + return { kind: "success", sourceId, ...result }; + }); + } return Promise.all( records.map(async (record): Promise => { const sourceId = sourceIdentity(record); @@ -57,7 +75,7 @@ async function processBatch( const result = await writer.write(record, { lane: options.lane, source: adapter.name, - containerTag: options.containerTag, + containerTag: record.containerTag ?? options.containerTag, }); return { kind: "success", sourceId, ...result }; } catch (error) { diff --git a/src/lib/mnemo-writer.ts b/src/lib/mnemo-writer.ts index b9c6479..f6eae7d 100644 --- a/src/lib/mnemo-writer.ts +++ b/src/lib/mnemo-writer.ts @@ -4,6 +4,8 @@ import type { ImportLane, SourceRecord } from "../adapters/index.js"; const MAX_DOCUMENT_CHARS = 450_000; const MAX_MEMORY_CHARS = 3_900; const MAX_ATTEMPTS = 5; +const MAX_BATCH_DOCUMENTS = 50; +const MAX_BATCH_BODY_BYTES = 64_000; export interface WriteOptions { lane: ImportLane; @@ -16,6 +18,13 @@ export interface WriteResult { remoteJobIds: string[]; } +export interface BatchWriteInput { + record: SourceRecord; + options: WriteOptions; +} + +export type BatchWriteResult = WriteResult | Error; + export interface RemoteJob { id: string; status: string; @@ -24,6 +33,7 @@ export interface RemoteJob { export interface DestinationWriter { write(record: SourceRecord, options: WriteOptions): Promise; + writeBatch?(inputs: BatchWriteInput[]): Promise; getJobs(ids: string[]): Promise; } @@ -37,6 +47,13 @@ function sha256(value: string): string { return createHash("sha256").update(value).digest("hex"); } +class MnemoHttpError extends Error { + constructor(readonly status: number, message: string) { + super(message); + this.name = "MnemoHttpError"; + } +} + export function sourceIdentity(record: SourceRecord): string { return record.sourceId ?? `sha256:${sha256(record.content)}`; } @@ -100,7 +117,10 @@ export class MnemoWriter implements DestinationWriter { if (response.ok) return (await response.json()) as T; const body = await response.text(); - lastError = new Error(`Mnemo request failed: ${response.status} ${body.slice(0, 500)}`); + lastError = new MnemoHttpError( + response.status, + `Mnemo request failed: ${response.status} ${body.slice(0, 500)}`, + ); if (response.status !== 429 && response.status < 500) throw lastError; } catch (error) { lastError = error instanceof Error ? error : new Error(String(error)); @@ -117,6 +137,77 @@ export class MnemoWriter implements DestinationWriter { : this.writeMemory(record, options); } + async writeBatch(inputs: BatchWriteInput[]): Promise { + if (inputs.length === 0) return []; + if (inputs.some(({ record, options }) => + options.lane !== "document" || splitDocument(record.content).length !== 1)) { + return Promise.all(inputs.map(({ record, options }) => + this.write(record, options).catch((error: unknown) => + error instanceof Error ? error : new Error(String(error))))); + } + + const outputs: BatchWriteResult[] = []; + let offset = 0; + while (offset < inputs.length) { + let end = Math.min(offset + MAX_BATCH_DOCUMENTS, inputs.length); + let documents = inputs.slice(offset, end).map(({ record, options }) => + this.documentPayload(record, options)); + while (documents.length > 1 && Buffer.byteLength(JSON.stringify({ documents })) > MAX_BATCH_BODY_BYTES) { + end -= 1; + documents = inputs.slice(offset, end).map(({ record, options }) => + this.documentPayload(record, options)); + } + + if (Buffer.byteLength(JSON.stringify({ documents })) > MAX_BATCH_BODY_BYTES) { + const input = inputs[offset]!; + outputs.push(await this.write(input.record, input.options).catch((error: unknown) => + error instanceof Error ? error : new Error(String(error)))); + offset += 1; + continue; + } + + try { + const response = await this.request<{ + results: Array<{ + index: number; + status: "accepted" | "failed"; + jobId?: string; + reused?: boolean; + error?: string; + }>; + }>("/v1/documents/batch", { + method: "POST", + body: JSON.stringify({ documents }), + }); + const byIndex = new Map(response.results.map((result) => [result.index, result])); + for (let index = 0; index < documents.length; index += 1) { + const result = byIndex.get(index); + if (!result) { + outputs.push(new Error(`Mnemo batch response omitted item ${index}`)); + } else if (result.status === "failed") { + outputs.push(new Error(result.error ?? `Mnemo batch item ${index} failed`)); + } else if (!result.jobId) { + outputs.push(new Error(`Mnemo batch item ${index} did not return a job id`)); + } else { + outputs.push({ reused: result.reused === true, remoteJobIds: [result.jobId] }); + } + } + } catch (error) { + if (error instanceof MnemoHttpError && [404, 405, 501].includes(error.status)) { + const fallback = await Promise.all(inputs.slice(offset, end).map(({ record, options }) => + this.write(record, options).catch((writeError: unknown) => + writeError instanceof Error ? writeError : new Error(String(writeError))))); + outputs.push(...fallback); + } else { + const normalized = error instanceof Error ? error : new Error(String(error)); + outputs.push(...documents.map(() => normalized)); + } + } + offset = end; + } + return outputs; + } + private importMetadata(record: SourceRecord, source: string): Record { return { ...(record.metadata ?? {}), @@ -125,6 +216,17 @@ export class MnemoWriter implements DestinationWriter { }; } + private documentPayload(record: SourceRecord, options: WriteOptions): Record { + const identity = sourceIdentity(record); + return { + containerTag: options.containerTag, + content: record.content, + contentType: record.contentType ?? "record", + customId: `imp_${sha256(`${options.source}:${identity}`)}`, + metadata: this.importMetadata(record, options.source), + }; + } + private async writeMemory(record: SourceRecord, options: WriteOptions): Promise { const identity = sourceIdentity(record); const chunks = splitContent(record.content, MAX_MEMORY_CHARS); From 948e52a7cabd128c4aced742acbd6d832f3e5d72 Mon Sep 17 00:00:00 2001 From: Shah Hussain <95882307+shhahhussain@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:27:44 +0500 Subject: [PATCH 5/7] test: define retry and retrieval verification reports --- src/lib/verification.test.ts | 24 ++++++++++++++++++++++++ src/migrator.test.ts | 34 +++++++++++++++++++++++++++++++++- 2 files changed, 57 insertions(+), 1 deletion(-) create mode 100644 src/lib/verification.test.ts diff --git a/src/lib/verification.test.ts b/src/lib/verification.test.ts new file mode 100644 index 0000000..1b65865 --- /dev/null +++ b/src/lib/verification.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from "vitest"; +import { verifyRetrieval, type RetrievalClient } from "./verification.js"; + +describe("verifyRetrieval", () => { + it("reports missing expectations without hiding successful questions", async () => { + const client: RetrievalClient = { + search: async (query) => query.includes("support") + ? { results: [{ content: "Support is available within two business hours." }] } + : { results: [{ content: "The plan is Basic." }] }, + }; + + const report = await verifyRetrieval(client, "pilot:customer", [ + { question: "What is the support SLA?", expect: ["two business hours"] }, + { question: "What is the plan?", expect: ["Enterprise"] }, + ]); + + expect(report.passed).toBe(1); + expect(report.failed).toBe(1); + expect(report.results).toEqual([ + expect.objectContaining({ passed: true, missing: [] }), + expect.objectContaining({ passed: false, missing: ["Enterprise"] }), + ]); + }); +}); diff --git a/src/migrator.test.ts b/src/migrator.test.ts index c511ca6..ddaecd9 100644 --- a/src/migrator.test.ts +++ b/src/migrator.test.ts @@ -3,7 +3,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { Adapter, type SourceRecord } from "./adapters/base.js"; -import { listJobs, requestCancellation } from "./lib/jobs.js"; +import { listJobs, loadFailures, requestCancellation } from "./lib/jobs.js"; import { migrate, planMigration, reconcileMigration } from "./lib/migrator.js"; import type { DestinationWriter, @@ -29,6 +29,14 @@ class MockAdapter extends Adapter { } } +class IssueAdapter extends MockAdapter { + async *iterate(): AsyncIterable { + yield { sourceId: "1", content: "valid" }; + await this.reportIssue({ sourceId: "line:2", error: "invalid JSON" }); + yield { sourceId: "3", content: "also valid" }; + } +} + class MockWriter implements DestinationWriter { readonly calls: Array<{ record: SourceRecord; options: WriteOptions }> = []; readonly remoteJobs = new Map(); @@ -124,6 +132,30 @@ describe("migrate", () => { ]); }); + it("journals malformed source records without stopping valid writes", async () => { + const writer = new MockWriter(); + const job = await migrate(new IssueAdapter([]), writer, options("memory")); + + expect(job.written).toBe(2); + expect(job.failed).toBe(1); + expect(job.status).toBe("completed_with_errors"); + await expect(loadFailures(job.id)).resolves.toEqual([ + expect.objectContaining({ sourceId: "line:2", kind: "source", error: "invalid JSON" }), + ]); + }); + + it("can create a new job that writes only selected failed source ids", async () => { + const writer = new MockWriter(); + const job = await migrate(new MockAdapter(RECORDS), writer, { + ...options("memory"), + includeSourceIds: new Set(["2"]), + retryOf: "mig_original", + }); + + expect(writer.calls.map((call) => call.record.sourceId)).toEqual(["2"]); + expect(job.retryOf).toBe("mig_original"); + }); + it("resumes by retrying only unresolved records", async () => { const firstWriter = new MockWriter(); firstWriter.failIds.add("2"); From 4491e062d052785cd781be972ec68084d4ca92c3 Mon Sep 17 00:00:00 2001 From: Shah Hussain <95882307+shhahhussain@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:29:23 +0500 Subject: [PATCH 6/7] feat: journal failures and verify retrieval --- src/lib/jobs.ts | 15 ++++++++++++ src/lib/migrator.ts | 21 +++++++++++++++++ src/lib/mnemo-writer.ts | 17 ++++++++++++++ src/lib/verification.ts | 52 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 105 insertions(+) create mode 100644 src/lib/verification.ts diff --git a/src/lib/jobs.ts b/src/lib/jobs.ts index 37ad8d1..9662e20 100644 --- a/src/lib/jobs.ts +++ b/src/lib/jobs.ts @@ -32,12 +32,14 @@ export interface JobState { /** Backward compatibility with jobs written by 0.1.0. */ processedIds?: string[]; lastError?: string; + retryOf?: string; } export interface FailedRecord { sourceId: string; error: string; occurredAt: string; + kind?: "source" | "write"; } function resolveJobsDir(): string { @@ -111,6 +113,19 @@ export async function appendFailure(jobId: string, failure: FailedRecord): Promi await appendLine(artifactPath(jobId, "failures.jsonl"), JSON.stringify(failure)); } +export async function loadFailures(jobId: string): Promise { + try { + const raw = await fs.readFile(artifactPath(jobId, "failures.jsonl"), "utf8"); + return raw + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line) as FailedRecord); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return []; + throw error; + } +} + export async function requestCancellation(jobId: string): Promise { await fs.writeFile(artifactPath(jobId, "cancel"), `${new Date().toISOString()}\n`, { mode: 0o600 }); } diff --git a/src/lib/migrator.ts b/src/lib/migrator.ts index 4a85878..365993a 100644 --- a/src/lib/migrator.ts +++ b/src/lib/migrator.ts @@ -22,6 +22,8 @@ export interface MigrateOptions { lane: ImportLane; containerTag: string; adapterConfig: AdapterConfig; + includeSourceIds?: Set; + retryOf?: string; onProgress?: (state: { written: number; reused: number; failed: number; total: number | null }) => void; } @@ -106,6 +108,7 @@ function newState(adapter: Adapter, options: MigrateOptions, total: number | nul remoteCompleted: 0, remoteFailed: 0, remotePending: 0, + ...(options.retryOf ? { retryOf: options.retryOf } : {}), }; } @@ -139,6 +142,22 @@ export async function migrate( await clearCancellation(state.id); await writeJob(state); + adapter.setIssueHandler(async (issue) => { + state.failed += 1; + state.lastError = issue.error; + await appendFailure(state.id, { + ...issue, + kind: "source", + occurredAt: new Date().toISOString(), + }); + options.onProgress?.({ + written: state.written, + reused: state.reused, + failed: state.failed, + total: state.total ?? null, + }); + }); + const seen = await loadProcessedIds(state.id, state.processedIds ?? []); let batch: SourceRecord[] = []; @@ -162,6 +181,7 @@ export async function migrate( await appendFailure(state.id, { sourceId: outcome.sourceId, error: outcome.error, + kind: "write", occurredAt: new Date().toISOString(), }); } @@ -184,6 +204,7 @@ export async function migrate( return state; } const id = sourceIdentity(record); + if (options.includeSourceIds && !options.includeSourceIds.has(id)) continue; if (seen.has(id)) continue; batch.push(record); if (batch.length >= concurrency) await commitBatch(); diff --git a/src/lib/mnemo-writer.ts b/src/lib/mnemo-writer.ts index f6eae7d..15807fa 100644 --- a/src/lib/mnemo-writer.ts +++ b/src/lib/mnemo-writer.ts @@ -289,6 +289,23 @@ export class MnemoWriter implements DestinationWriter { ); return response.items; } + + async search( + question: string, + containerTag: string, + options: { limit?: number; searchMode?: "hybrid" | "memories" | "documents" } = {}, + ): Promise { + return this.request("/v1/search", { + method: "POST", + body: JSON.stringify({ + q: question, + containerTags: [containerTag], + searchMode: options.searchMode ?? "hybrid", + limit: options.limit ?? 10, + includeSources: true, + }), + }); + } } export { splitDocument }; diff --git a/src/lib/verification.ts b/src/lib/verification.ts new file mode 100644 index 0000000..4f1820d --- /dev/null +++ b/src/lib/verification.ts @@ -0,0 +1,52 @@ +export interface VerificationQuestion { + question: string; + expect: string[]; + limit?: number; + searchMode?: "hybrid" | "memories" | "documents"; +} + +export interface RetrievalClient { + search( + question: string, + containerTag: string, + options?: { limit?: number; searchMode?: VerificationQuestion["searchMode"] }, + ): Promise; +} + +export interface VerificationResult { + question: string; + passed: boolean; + missing: string[]; +} + +export interface VerificationReport { + schemaVersion: "getmnemo-migrate.verification.v1"; + containerTag: string; + passed: number; + failed: number; + results: VerificationResult[]; +} + +export async function verifyRetrieval( + client: RetrievalClient, + containerTag: string, + questions: VerificationQuestion[], +): Promise { + const results: VerificationResult[] = []; + for (const item of questions) { + const response = await client.search(item.question, containerTag, { + limit: item.limit, + searchMode: item.searchMode, + }); + const haystack = JSON.stringify(response).toLocaleLowerCase(); + const missing = item.expect.filter((value) => !haystack.includes(value.toLocaleLowerCase())); + results.push({ question: item.question, passed: missing.length === 0, missing }); + } + return { + schemaVersion: "getmnemo-migrate.verification.v1", + containerTag, + passed: results.filter((result) => result.passed).length, + failed: results.filter((result) => !result.passed).length, + results, + }; +} From 4574e7f4e7e5d7822492f964a32a71b3b547e364 Mon Sep 17 00:00:00 2001 From: Shah Hussain <95882307+shhahhussain@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:31:18 +0500 Subject: [PATCH 7/7] feat: release migration operations and reports --- README.md | 36 +++++++++++- docs/pilot-integration.md | 12 +++- package-lock.json | 4 +- package.json | 2 +- src/cli.ts | 120 ++++++++++++++++++++++++++++++++------ 5 files changed, 148 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index a2417dc..82918fa 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,8 @@ getmnemo-migrate from raw-csv \ --container pilot:customer-name ``` +Small documents are sent through Mnemo's bulk endpoint in bounded batches. If the destination is running an older API without that endpoint, the CLI falls back to the existing single-document calls automatically. + The `content=*` mapping serializes each complete CSV or JSON record. For cleaner source data, map a specific field instead, such as `content=notes,id=record_id,company=company`. ## Import Lanes @@ -56,13 +58,15 @@ getmnemo-migrate from raw-json \ --container pilot:customer-name ``` +Map `container=` when one approved export contains multiple customers. Each record is then written to its own isolated container; `--container` remains the fallback. + ## Provider Sources | Source | Source credentials | Useful options | | --- | --- | --- | | `mem0` | `MEM0_API_KEY` | `--user`, `--base-url` | | `zep` | `ZEP_API_KEY` | `--user` is required, `--base-url` | -| `supermemory` | `SUPERMEMORY_API_KEY` | `--base-url` | +| `supermemory` | `SUPERMEMORY_API_KEY` | `--source-container` is required, `--base-url` | | `letta` | `LETTA_API_KEY`, `LETTA_BASE_URL` | Imports core and archival memory | ## Safe Operating Flow @@ -72,15 +76,40 @@ getmnemo-migrate from raw-json \ 3. Import into a customer-specific container with conservative concurrency. 4. Keep the printed migration job ID. 5. Run `status ` or `reconcile ` to obtain the final completion report. -6. Validate retrieval against a small set of customer-approved questions before connecting an agent. +6. Retry isolated write failures with `retry-failed `. +7. Validate retrieval against a small set of customer-approved questions before connecting an agent. ```bash getmnemo-migrate status getmnemo-migrate cancel mig_... getmnemo-migrate resume mig_... getmnemo-migrate reconcile mig_... +getmnemo-migrate retry-failed mig_... +``` + +Use `--json` with planning, imports, resume, retry, reconciliation, status, and cancellation when another system needs to consume the report. + +Create a verification file after agreeing expected answers with the customer: + +```json +[ + { + "question": "What is the current support response time?", + "expect": ["two business hours"], + "searchMode": "hybrid", + "limit": 10 + } +] ``` +Then run retrieval acceptance against the same isolated container: + +```bash +getmnemo-migrate verify mig_... --questions ./acceptance-questions.json --json +``` + +The command exits non-zero when an expected phrase is missing. For an import that mapped records to several containers, pass `--container` to verify one customer boundary at a time. + Cancellation pauses after the current bounded batch. Resume skips successful source records and retries unresolved ones. Mnemo custom IDs and idempotency keys also protect against duplicate writes if local state is lost. Job state, processed IDs, remote job IDs, and failure journals are stored under `~/.getmnemo/migrations` with owner-only permissions. Set `GETMNEMO_MIGRATION_DIR` to use another state directory. @@ -89,6 +118,9 @@ Job state, processed IDs, remote job IDs, and failure journals are stored under - Concurrency is capped at 25 even if a larger value is requested. - HTTP 429 and server/network failures retry up to five times with backoff. +- Malformed JSONL rows are journaled while valid rows continue. +- Failed writes can be retried in a new job without replaying successful records. +- Bulk requests contain at most 50 documents and stay below a conservative request-size ceiling. - Documents larger than the API limit are split deterministically near line boundaries. - Atomic memories longer than the memory endpoint limit are split with part metadata. - JSON arrays are capped at 50 MB to avoid loading unbounded files into memory; JSONL and CSV stream. diff --git a/docs/pilot-integration.md b/docs/pilot-integration.md index 542c6aa..d048053 100644 --- a/docs/pilot-integration.md +++ b/docs/pilot-integration.md @@ -43,9 +43,15 @@ getmnemo-migrate from raw-csv \ --map 'content=*,id=Record ID' \ --container pilot:customer-name \ --concurrency 5 + +getmnemo-migrate reconcile mig_... --json +getmnemo-migrate retry-failed mig_... --json +getmnemo-migrate verify mig_... \ + --questions ./acceptance-questions.json \ + --json ``` -Record the migration job ID and keep the local migration directory until acceptance is complete. A paused or interrupted import is resumed with `getmnemo-migrate resume `; it does not restart successful records. +Record the migration job ID and keep the local migration directory until acceptance is complete. A paused or interrupted import is resumed with `getmnemo-migrate resume `; it does not restart successful records. Malformed rows and write failures are isolated in the job journal, and `retry-failed` creates a separate auditable retry job for unresolved writes. ## Runtime API Contract @@ -53,7 +59,8 @@ The integrated system needs three Mnemo operations: | Operation | Endpoint | Purpose | | --- | --- | --- | -| Add source material | `POST /v1/documents` | Asynchronous ingestion with provenance and deterministic `customId`. | +| Add source material | `POST /v1/documents/batch` | Bounded asynchronous ingestion with per-record results, provenance, and deterministic `customId`. | +| Single-write fallback | `POST /v1/documents` | Used automatically for oversized records or older API deployments. | | Check ingestion | `GET /v1/jobs?ids=...` | Completion and failure reconciliation in groups of at most 50 job IDs. | | Retrieve context | `POST /v1/search` | Scoped retrieval for the governance layer before an agent acts. | @@ -69,6 +76,7 @@ The pilot is ready for agent integration when all of the following are true: - the final report has no pending jobs and any failed records are explained or retried; - rerunning or resuming does not create duplicate source records; - customer-approved test questions retrieve the expected current context and provenance; +- the machine-readable verification report has zero failed questions; - searches cannot cross the pilot customer’s workspace and container boundary; - removing Mnemo access from the governance service prevents further reads and writes. diff --git a/package-lock.json b/package-lock.json index da41cd6..ff53b71 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "getmnemo-migrate", - "version": "0.1.1", + "version": "0.2.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "getmnemo-migrate", - "version": "0.1.1", + "version": "0.2.0", "license": "MIT", "dependencies": { "cli-progress": "^3.12.0", diff --git a/package.json b/package.json index 763fdc4..bfff183 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "getmnemo-migrate", - "version": "0.1.1", + "version": "0.2.0", "description": "Resumable CLI for importing historical documents and memories into Mnemo.", "type": "module", "bin": { diff --git a/src/cli.ts b/src/cli.ts index 7b2d283..0ea7360 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -1,6 +1,7 @@ #!/usr/bin/env node import { Command, Option } from "commander"; import { realpathSync } from "node:fs"; +import { promises as fs } from "node:fs"; import { fileURLToPath } from "node:url"; import cliProgress from "cli-progress"; import kleur from "kleur"; @@ -13,6 +14,7 @@ import { } from "./adapters/index.js"; import { listJobs, + loadFailures, readJob, requestCancellation, writeJob, @@ -20,8 +22,9 @@ import { } from "./lib/jobs.js"; import { migrate, planMigration, reconcileMigration } from "./lib/migrator.js"; import { MnemoWriter } from "./lib/mnemo-writer.js"; +import { verifyRetrieval, type VerificationQuestion } from "./lib/verification.js"; -const VERSION = "0.1.1"; +const VERSION = "0.2.0"; interface CommonAdapterOpts { file?: string; @@ -29,6 +32,8 @@ interface CommonAdapterOpts { user?: string; baseUrl?: string; pageSize?: string; + sourceContainer?: string; + json?: boolean; } interface RunOpts extends CommonAdapterOpts { @@ -55,6 +60,9 @@ function buildAdapterConfig(opts: CommonAdapterOpts): AdapterConfig { ...(opts.pageSize !== undefined ? { pageSize: parsePositiveInt(opts.pageSize, 100, "--page-size") } : {}), + ...(opts.sourceContainer !== undefined + ? { sourceContainers: opts.sourceContainer.split(",").map((value) => value.trim()).filter(Boolean) } + : {}), }; } @@ -95,11 +103,13 @@ function addAdapterOptions(command: Command, includeDefaults: boolean): Command .option("--map ", "field mapping such as content=body,id=record_id") .option("--user ", "source user id (required by some providers)") .option("--base-url ", "override provider base URL") + .option("--source-container ", "comma-separated source containers") .option( "--page-size ", "source page-size hint", includeDefaults ? "100" : undefined, - ); + ) + .option("--json", "print a machine-readable JSON report"); return command; } @@ -125,7 +135,11 @@ function startProgress(source: string, total: number | null): cliProgress.Single return bar; } -function printSummary(state: JobState): void { +function printSummary(state: JobState, json = false): void { + if (json) { + process.stdout.write(`${JSON.stringify({ schemaVersion: "getmnemo-migrate.report.v1", job: state }, null, 2)}\n`); + return; + } process.stdout.write( `\n${kleur.green("✓")} job ${kleur.bold(state.id)}: ${state.status}\n` + ` records written: ${state.written}\n` + @@ -153,18 +167,19 @@ export function buildCli(): Command { const containerTag = requireContainer(opts.container); const writer = getWriter(); const total = await adapter.count(); - const bar = startProgress(source, total); + const bar = opts.json ? null : startProgress(source, total); let state = await migrate(adapter, writer, { concurrency: parsePositiveInt(opts.concurrency, 5, "--concurrency"), lane, containerTag, adapterConfig, onProgress: (progress) => { + if (!bar) return; if (progress.total !== null && bar.getTotal() !== progress.total) bar.setTotal(progress.total); bar.update(progress.written + progress.failed, { failed: progress.failed }); }, }); - bar.stop(); + bar?.stop(); if (lane === "document" && opts.wait && state.status !== "paused") { state = await reconcileMigration( writer, @@ -172,7 +187,7 @@ export function buildCli(): Command { parsePositiveInt(opts.waitTimeout, 86_400, "--wait-timeout") * 1000, ); } - printSummary(state); + printSummary(state, opts.json); }); const plan = program.command("plan").description("preview an import without writing"); @@ -181,9 +196,13 @@ export function buildCli(): Command { .option("--sample ", "sample size", "5") .action(async (source: string, opts: CommonAdapterOpts & { sample?: string }) => { const adapter = createAdapter(source, buildAdapterConfig(opts)); - const spinner = ora(`Inspecting ${source}...`).start(); + const spinner = opts.json ? null : ora(`Inspecting ${source}...`).start(); const result = await planMigration(adapter, parsePositiveInt(opts.sample, 5, "--sample")); - spinner.stop(); + spinner?.stop(); + if (opts.json) { + process.stdout.write(`${JSON.stringify({ schemaVersion: "getmnemo-migrate.plan.v1", source, lane: adapter.defaultLane, ...result }, null, 2)}\n`); + return; + } process.stdout.write(`${kleur.cyan("source:")} ${source}\n`); process.stdout.write(`${kleur.cyan("lane:")} ${adapter.defaultLane}\n`); process.stdout.write(`${kleur.cyan("total:")} ${result.total ?? "unknown"}\n`); @@ -202,8 +221,8 @@ export function buildCli(): Command { const lane = parseLane(opts.lane, existing.lane); const containerTag = opts.container ?? existing.containerTag; const writer = getWriter(); - const bar = startProgress(existing.source, existing.total ?? null); - bar.update(existing.written + existing.failed, { failed: existing.failed }); + const bar = opts.json ? null : startProgress(existing.source, existing.total ?? null); + bar?.update(existing.written + existing.failed, { failed: existing.failed }); let state = await migrate(adapter, writer, { concurrency: parsePositiveInt(opts.concurrency, 5, "--concurrency"), resumeJobId: jobId, @@ -211,9 +230,9 @@ export function buildCli(): Command { containerTag, adapterConfig, onProgress: (progress) => - bar.update(progress.written + progress.failed, { failed: progress.failed }), + bar?.update(progress.written + progress.failed, { failed: progress.failed }), }); - bar.stop(); + bar?.stop(); if (lane === "document" && opts.wait && state.status !== "paused") { state = await reconcileMigration( writer, @@ -221,31 +240,38 @@ export function buildCli(): Command { parsePositiveInt(opts.waitTimeout, 86_400, "--wait-timeout") * 1000, ); } - printSummary(state); + printSummary(state, opts.json); }); program .command("reconcile ") .description("refresh all remote ingestion jobs and print the completion report") .option("--wait-timeout ", "maximum wait", "86400") - .action(async (jobId: string, opts: { waitTimeout: string }) => { + .option("--json", "print a machine-readable JSON report") + .action(async (jobId: string, opts: { waitTimeout: string; json?: boolean }) => { const state = await reconcileMigration( getWriter(), await readJob(jobId), parsePositiveInt(opts.waitTimeout, 86_400, "--wait-timeout") * 1000, ); - printSummary(state); + printSummary(state, opts.json); }); program .command("status [jobId]") .description("show one import report or list recent imports") - .action(async (jobId?: string) => { + .option("--json", "print a machine-readable JSON report") + .action(async (jobId: string | undefined, opts: { json?: boolean }) => { if (jobId) { - process.stdout.write(`${JSON.stringify(await readJob(jobId), null, 2)}\n`); + const job = await readJob(jobId); + process.stdout.write(`${JSON.stringify(opts.json ? { schemaVersion: "getmnemo-migrate.report.v1", job } : job, null, 2)}\n`); return; } const jobs = await listJobs(); + if (opts.json) { + process.stdout.write(`${JSON.stringify({ schemaVersion: "getmnemo-migrate.jobs.v1", jobs: jobs.slice(0, 20) }, null, 2)}\n`); + return; + } if (jobs.length === 0) process.stdout.write("No migration jobs yet.\n"); for (const job of jobs.slice(0, 20)) { process.stdout.write( @@ -254,10 +280,64 @@ export function buildCli(): Command { } }); + const retry = program.command("retry-failed ").description("retry only failed writes as a new job"); + addRunOptions(retry, false).action(async (jobId: string, opts: RunOpts) => { + const original = await readJob(jobId); + const failures = await loadFailures(jobId); + const sourceIds = new Set( + failures.filter((failure) => failure.kind !== "source").map((failure) => failure.sourceId), + ); + if (sourceIds.size === 0) throw new Error(`job ${jobId} has no failed writes to retry`); + const adapterConfig = mergeAdapterConfig(original.adapterConfig, opts); + const adapter = createAdapter(original.source, adapterConfig); + const writer = getWriter(); + let state = await migrate(adapter, writer, { + concurrency: parsePositiveInt(opts.concurrency, 5, "--concurrency"), + lane: original.lane, + containerTag: opts.container ?? original.containerTag, + adapterConfig, + includeSourceIds: sourceIds, + retryOf: original.id, + }); + if (original.lane === "document" && opts.wait) { + state = await reconcileMigration( + writer, + state, + parsePositiveInt(opts.waitTimeout, 86_400, "--wait-timeout") * 1000, + ); + } + printSummary(state, opts.json); + }); + + program + .command("verify ") + .description("run retrieval questions against an imported container") + .requiredOption("--questions ", "JSON verification questions") + .option("--container ", "override the job container") + .option("--json", "print a machine-readable JSON report") + .action(async (jobId: string, opts: { questions: string; container?: string; json?: boolean }) => { + const job = await readJob(jobId); + const parsed: unknown = JSON.parse(await fs.readFile(opts.questions, "utf8")); + if (!Array.isArray(parsed) || parsed.length === 0) { + throw new Error("--questions must contain a non-empty JSON array"); + } + const questions = parsed as VerificationQuestion[]; + for (const [index, item] of questions.entries()) { + if (!item || typeof item.question !== "string" || !Array.isArray(item.expect) || + !item.expect.every((value) => typeof value === "string")) { + throw new Error(`invalid verification question at index ${index}`); + } + } + const report = await verifyRetrieval(getWriter(), opts.container ?? job.containerTag, questions); + process.stdout.write(`${JSON.stringify(report, null, 2)}\n`); + if (report.failed > 0) process.exitCode = 2; + }); + program .command("cancel ") .description("request a running import to pause after its current bounded batch") - .action(async (jobId: string) => { + .option("--json", "print a machine-readable JSON report") + .action(async (jobId: string, opts: { json?: boolean }) => { const state = await readJob(jobId); if (["completed", "completed_with_errors", "failed"].includes(state.status)) { throw new Error(`job ${jobId} is already ${state.status}`); @@ -265,7 +345,9 @@ export function buildCli(): Command { await requestCancellation(jobId); state.status = "paused"; await writeJob(state); - process.stdout.write(`${kleur.yellow("!")} pause requested for ${jobId}\n`); + process.stdout.write(opts.json + ? `${JSON.stringify({ schemaVersion: "getmnemo-migrate.cancel.v1", jobId, status: "paused" })}\n` + : `${kleur.yellow("!")} pause requested for ${jobId}\n`); }); return program;