diff --git a/packages/cli/src/context/ingest/local-ingest.ts b/packages/cli/src/context/ingest/local-ingest.ts index 2351d420..6ce40e66 100644 --- a/packages/cli/src/context/ingest/local-ingest.ts +++ b/packages/cli/src/context/ingest/local-ingest.ts @@ -13,6 +13,7 @@ import { localPullConfigForAdapter, type DefaultLocalIngestAdaptersOptions } fro import { createLocalBundleIngestRuntime } from './local-bundle-runtime.js'; import type { MemoryFlowEventSink } from './memory-flow/types.js'; import { buildSyncId } from './raw-sources-paths.js'; +import { ingestReportOutcome } from './reports.js'; import type { IngestReportBody, IngestReportSnapshot } from './reports.js'; import { SqliteBundleIngestStore } from './sqlite-bundle-ingest-store.js'; import type { IngestBundleResult, IngestJobContext, IngestJobPhase, IngestTrigger, SourceAdapter } from './types.js'; @@ -79,7 +80,7 @@ export interface LocalMetabaseFanoutProgress { metabaseDatabaseId: number; targetConnectionId: string; jobId: string; - status: 'done' | 'failed'; + status: 'done' | 'partial' | 'failed'; }): void; } @@ -232,11 +233,11 @@ export async function runLocalIngest(options: RunLocalIngestOptions): Promise child.report.body.failedWorkUnits.length === 0).length; - if (succeeded === children.length) { + const outcomes = children.map((child) => ingestReportOutcome(child.report)); + if (outcomes.every((outcome) => outcome === 'done')) { return 'all_succeeded'; } - if (succeeded === 0) { + if (outcomes.every((outcome) => outcome === 'error')) { return 'all_failed'; } return 'partial_failure'; @@ -406,7 +407,7 @@ export async function runLocalMetabaseIngest( metabaseDatabaseId: childPlan.metabaseDatabaseId, targetConnectionId, jobId: child.report.jobId, - status: child.report.body.failedWorkUnits.length > 0 ? 'failed' : 'done', + status: ingestReportOutcome(child.report) === 'error' ? 'failed' : ingestReportOutcome(child.report), }); children.push({ jobId: child.report.jobId, diff --git a/packages/cli/src/context/ingest/reports.ts b/packages/cli/src/context/ingest/reports.ts index ea02a31a..09f92170 100644 --- a/packages/cli/src/context/ingest/reports.ts +++ b/packages/cli/src/context/ingest/reports.ts @@ -146,6 +146,20 @@ export function savedMemoryCountsForReport(report: IngestReportSnapshot): Ingest }; } +/** @internal */ +export type IngestReportOutcome = 'done' | 'partial' | 'error'; + +export function ingestReportOutcome(report: IngestReportSnapshot): IngestReportOutcome { + if (report.body.status === 'failed') { + return 'error'; + } + if (report.body.failedWorkUnits.length === 0) { + return 'done'; + } + const { wikiCount, slCount } = savedMemoryCountsForReport(report); + return wikiCount + slCount > 0 ? 'partial' : 'error'; +} + export function buildStageIndexFromReportBody(jobId: string, connectionId: string, body: IngestReportBody): StageIndex { return { jobId, diff --git a/packages/cli/test/context/ingest/local-metabase-ingest.test.ts b/packages/cli/test/context/ingest/local-metabase-ingest.test.ts index 06822aa2..8fb89bd0 100644 --- a/packages/cli/test/context/ingest/local-metabase-ingest.test.ts +++ b/packages/cli/test/context/ingest/local-metabase-ingest.test.ts @@ -6,6 +6,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { initKtxProject, type KtxLocalProject } from '../../../src/context/project/project.js'; import { LocalMetabaseDiscoveryCache } from '../../../src/context/ingest/adapters/metabase/local-source-state-store.js'; import { getLocalIngestStatus, runLocalMetabaseIngest } from '../../../src/context/ingest/local-ingest.js'; +import { ingestReportOutcome } from '../../../src/context/ingest/reports.js'; import type { ChunkResult, FetchContext, SourceAdapter } from '../../../src/context/ingest/types.js'; class TestAgentRunner implements AgentRunnerPort { @@ -202,6 +203,24 @@ describe('runLocalMetabaseIngest', () => { expect(result.children[1]?.report.body.failedWorkUnits).toEqual(['metabase-db-2']); }); + it('keeps a child that saved memory out of all_failed when another child fails', async () => { + await seedMetabaseState(); + const agentRunner = new TestAgentRunner(); + const ids = ['metabase-child-1', 'metabase-child-2']; + + const result = await runLocalMetabaseIngest({ + project, + adapters: [new FakeMetabaseSourceAdapter()], + metabaseConnectionId: 'prod-metabase', + agentRunner, + jobIdFactory: () => ids.shift() ?? 'metabase-child-extra', + }); + + expect(result.status).toBe('partial_failure'); + expect(ingestReportOutcome(result.children[0].report)).toBe('done'); + expect(ingestReportOutcome(result.children[1].report)).toBe('error'); + }); + it('captures fetch-time child failures and continues later mappings', async () => { await seedMetabaseState(); project.config.connections.warehouse_c = { driver: 'postgres', url: 'postgres://localhost/c' }; diff --git a/packages/cli/test/context/ingest/reports.test.ts b/packages/cli/test/context/ingest/reports.test.ts new file mode 100644 index 00000000..5fc24f6d --- /dev/null +++ b/packages/cli/test/context/ingest/reports.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, it } from 'vitest'; +import { ingestReportOutcome } from '../../../src/context/ingest/reports.js'; +import type { IngestReportSnapshot } from '../../../src/context/ingest/reports.js'; + +function report(body: Partial): IngestReportSnapshot { + return { + id: 'r', + runId: 'run', + jobId: 'job', + connectionId: 'warehouse', + sourceKey: 'metabase', + createdAt: '2026-05-29T00:00:00.000Z', + body: { + syncId: 'sync', + diffSummary: { added: 0, modified: 0, deleted: 0, unchanged: 0 }, + commitSha: null, + workUnits: [], + failedWorkUnits: [], + reconciliationSkipped: false, + conflictsResolved: [], + evictionsApplied: [], + unmappedFallbacks: [], + evictionInputs: [], + unresolvedCards: [], + supersededBy: null, + overrideOf: null, + provenanceRows: [], + toolTranscripts: [], + ...body, + }, + }; +} + +const savingWorkUnit = { + unitKey: 'ok', + rawFiles: ['cards/1.json'], + status: 'success' as const, + actions: [{ target: 'sl' as const, type: 'updated' as const, key: 'warehouse.orders', detail: 'measure' }], + touchedSlSources: [], +}; + +const failedWorkUnit = { + unitKey: 'bad', + rawFiles: ['cards/2.json'], + status: 'failed' as const, + reason: 'tool write failed', + actions: [], + touchedSlSources: [], +}; + +describe('ingestReportOutcome', () => { + it('returns done when there are no failed work units', () => { + expect(ingestReportOutcome(report({ workUnits: [savingWorkUnit] }))).toBe('done'); + }); + + it('returns partial when failed work units coexist with saved memory', () => { + expect( + ingestReportOutcome(report({ workUnits: [savingWorkUnit, failedWorkUnit], failedWorkUnits: ['bad'] })), + ).toBe('partial'); + }); + + it('returns error when failed work units produced no saved memory', () => { + expect(ingestReportOutcome(report({ workUnits: [failedWorkUnit], failedWorkUnits: ['bad'] }))).toBe('error'); + }); + + it('returns error for a stage-level failure even if artifacts were recorded', () => { + expect(ingestReportOutcome(report({ status: 'failed', workUnits: [savingWorkUnit], failedWorkUnits: [] }))).toBe( + 'error', + ); + }); +});