apunkt/ktx
1
0
Fork 0
mirror of https://github.com/Kaelio/ktx.git synced 2026-06-07 07:55:13 +02:00
Code Issues Projects Releases Packages Wiki Activity

feat(cli): stream plain ktx ingest progress to stderr (KLO-726) (#251)

Browse source 
* feat(cli): share public ingest progress adapter

* feat(cli): stream plain public ingest progress

* test(cli): update plain ingest progress assertions

* chore(cli): satisfy plain ingest progress checks

* fix(artifacts): expect plain ingest stderr progress in installed-CLI smoke

* ci(coverage): make Codecov upload non-fatal and fix repo slug

The Coverage job failed because the Codecov upload returned
'Repository not found' while fail_ci_if_error was true, turning a
Codecov-side issue into a hard CI failure even though all tests pass.

- Set fail_ci_if_error: false on both uploads so Codecov outages or an
  unlinked repo no longer break CI (upload stays best-effort).
- Correct the stale slug Kaelio/ktx -> Kaelio/ktx-ai-data-agents-context
  to match the actual GitHub repo (aligns with main).

* fix(cli): isolate query-history failure capture from scan output

The plain public-ingest progress path passes one captured IO as the
target-level `io`. With progress deps set, both the schema scan and the
query-history ingest resolved their capture to that same shared buffer,
so a non-actionable query-history failure surfaced leftover scan report
text (e.g. "Mode: enriched") as the skipped-facet detail instead of the
real query-history message.

Give the query-history ingest a phase-local capture while preserving the
flow-to-io branch the foreground context-build view relies on.

---------

Co-authored-by: Andrey Avtomonov <7889985+andreybavt@users.noreply.github.com>
This commit is contained in:
Andrey Avtomonov 2026-06-01 23:31:31 +02:00 • committed by GitHub
parent d01abe6f3c
commit 13774bfcef
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 445 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

4
.github/workflows/ci.yml vendored
View file

@ -217,7 +217,7 @@ jobs:
flags: typescript
name: typescript
disable_search: true
fail_ci_if_error: true
fail_ci_if_error: false
- name: Warn when Codecov token is missing for TypeScript
if: env.CODECOV_TOKEN_CONFIGURED != 'true'
@ -236,7 +236,7 @@ jobs:
flags: python
name: python
disable_search: true
fail_ci_if_error: true
fail_ci_if_error: false
- name: Warn when Codecov token is missing for Python
if: env.CODECOV_TOKEN_CONFIGURED != 'true'

41
packages/cli/src/context-build-view.ts
View file

@ -1,8 +1,6 @@
import type { KtxProgressPort, KtxProgressUpdateOptions } from './context/scan/types.js';
import type { KtxCliIo } from './index.js';
import type { KtxIngestProgressUpdate } from './ingest.js';
import type { KtxManagedPythonInstallPolicy } from './managed-python-command.js';
import { publicDatabaseIngestMessage, publicQueryHistoryMessage } from './public-ingest-copy.js';
import type {
KtxPublicIngestArgs,
KtxPublicIngestDeps,
@ -10,7 +8,8 @@ import type {
KtxPublicIngestProject,
KtxPublicIngestTargetResult,
} from './public-ingest.js';
import { buildPublicIngestPlan, executePublicIngestTarget } from './public-ingest.js';
import { buildPublicIngestPlan, executePublicIngestTarget, publicProgressMessage } from './public-ingest.js';
import { createAggregateProgressPort } from './progress-port-adapter.js';
import { formatDuration } from './demo-metrics.js';
import { profileMark } from './startup-profile.js';
@ -810,17 +809,6 @@ export function initViewState(targets: KtxPublicIngestPlanTarget[]): ContextBuil
};
}
function publicProgressMessage(message: string, target: KtxPublicIngestPlanTarget): string {
let current = message;
if (target.operation === 'database-ingest') {
current = publicDatabaseIngestMessage(current);
}
if (target.steps.includes('query-history')) {
current = publicQueryHistoryMessage(current, target.connectionId);
}
return current;
}
function formatProgressDetail(
update: Pick<KtxIngestProgressUpdate, 'percent' | 'message'>,
target: KtxPublicIngestPlanTarget,
@ -829,29 +817,6 @@ function formatProgressDetail(
return `[${percent}%] ${publicProgressMessage(update.message, target)}`;
}
function createContextBuildProgressPort(
onProgress: (update: KtxIngestProgressUpdate) => void,
state: { progress: number } = { progress: 0 },
start = 0,
weight = 1,
): KtxProgressPort {
return {
async update(value: number, message?: string, options?: KtxProgressUpdateOptions): Promise<void> {
const absoluteValue = start + Math.max(0, Math.min(1, value)) * weight;
state.progress = Math.max(state.progress, Math.min(1, absoluteValue));
if (!message) return;
onProgress({
percent: Math.max(0, Math.min(100, Math.round(state.progress * 100))),
message,
...(options?.transient !== undefined ? { transient: options.transient } : {}),
});
},
startPhase(phaseWeight: number): KtxProgressPort {
return createContextBuildProgressPort(onProgress, state, state.progress, weight * phaseWeight);
},
};
}
export async function runContextBuild(
project: KtxPublicIngestProject,
args: ContextBuildArgs,
@ -1022,7 +987,7 @@ export async function runContextBuild(
};
const progressDeps: KtxPublicIngestDeps = {
scanProgress: createContextBuildProgressPort(updateSchemaPhase),
scanProgress: createAggregateProgressPort(updateSchemaPhase),
ingestProgress: updateIngestPhase,
runtimeIo: io,
onPhaseStart,

29
packages/cli/src/progress-port-adapter.ts Normal file
View file

@ -0,0 +1,29 @@
import type { KtxProgressPort, KtxProgressUpdateOptions } from './context/scan/types.js';
import type { KtxIngestProgressUpdate } from './ingest.js';
export interface AggregateProgressState {
progress: number;
}
export function createAggregateProgressPort(
onProgress: (update: KtxIngestProgressUpdate) => void,
state: AggregateProgressState = { progress: 0 },
start = 0,
weight = 1,
): KtxProgressPort {
return {
async update(value: number, message?: string, options?: KtxProgressUpdateOptions): Promise<void> {
const absoluteValue = start + Math.max(0, Math.min(1, value)) * weight;
state.progress = Math.max(state.progress, Math.min(1, absoluteValue));
if (!message) return;
onProgress({
percent: Math.max(0, Math.min(100, Math.round(state.progress * 100))),
message,
...(options?.transient !== undefined ? { transient: options.transient } : {}),
});
},
startPhase(phaseWeight: number): KtxProgressPort {
return createAggregateProgressPort(onProgress, state, state.progress, weight * phaseWeight);
},
};
}

139
packages/cli/src/public-ingest.ts
View file

@ -11,7 +11,12 @@ import {
type ManagedPythonCommandRuntime,
} from './managed-python-command.js';
import type { KtxRuntimeFeature } from './managed-python-runtime.js';
import { publicIngestOutputLine } from './public-ingest-copy.js';
import {
publicDatabaseIngestMessage,
publicIngestOutputLine,
publicQueryHistoryMessage,
} from './public-ingest-copy.js';
import { createAggregateProgressPort } from './progress-port-adapter.js';
import { resolvePublicIngestRuntimeRequirements } from './runtime-requirements.js';
import type { KtxScanArgs, KtxScanDeps } from './scan.js';
import { profileMark } from './startup-profile.js';
@ -129,6 +134,17 @@ const sourceAdapterByDriver = new Map<string, string>([
['lookml', 'lookml'],
]);
export function publicProgressMessage(message: string, target: KtxPublicIngestPlanTarget): string {
let current = message;
if (target.operation === 'database-ingest') {
current = publicDatabaseIngestMessage(current);
}
if (target.steps.includes('query-history')) {
current = publicQueryHistoryMessage(current, target.connectionId);
}
return current;
}
const queryHistoryDialectByDriver = new Map<string, HistoricSqlDialect>([
['postgres', 'postgres'],
['bigquery', 'bigquery'],
@ -729,6 +745,80 @@ function createCapturedPublicIngestIo(): CapturedPublicIngestIo {
};
}
function isCapturedPublicIngestIo(io: KtxCliIo): io is CapturedPublicIngestIo {
return typeof (io as Partial<CapturedPublicIngestIo>).capturedOutput === 'function';
}
const PLAIN_PUBLIC_INGEST_PHASE_LABELS: Record<KtxPublicIngestPhaseKey, string> = {
'database-schema': 'database schema',
'query-history': 'query history',
'source-ingest': 'source ingest',
};
interface PlainPublicIngestProgressOptions {
target: KtxPublicIngestPlanTarget;
index: number;
total: number;
}
function firstSummaryLine(summary: string | undefined): string | undefined {
if (!summary) return undefined;
return summary.split(/\r?\n/).find((line) => line.trim().length > 0)?.trim();
}
function plainPhaseHeader(options: PlainPublicIngestProgressOptions, phaseKey: KtxPublicIngestPhaseKey): string {
const prefix = options.total > 1 ? `[${options.index + 1}/${options.total}] ` : '';
return `${prefix}${options.target.connectionId} · ${PLAIN_PUBLIC_INGEST_PHASE_LABELS[phaseKey]}`;
}
function plainPhaseEndLine(status: 'done' | 'failed' | 'skipped', summary?: string): string {
const firstLine = firstSummaryLine(summary);
return firstLine ? ` ${status} · ${firstLine}` : ` ${status}`;
}
function createPlainPublicIngestProgress(io: KtxCliIo, options: PlainPublicIngestProgressOptions): Required<
Pick<KtxPublicIngestDeps, 'scanProgress' | 'ingestProgress' | 'onPhaseStart' | 'onPhaseEnd'>
> {
let currentPhase: KtxPublicIngestPhaseKey | null = null;
const startedPhases = new Set<KtxPublicIngestPhaseKey>();
const lastPercentByPhase = new Map<KtxPublicIngestPhaseKey, number>();
const startPhase = (phaseKey: KtxPublicIngestPhaseKey): void => {
currentPhase = phaseKey;
startedPhases.add(phaseKey);
lastPercentByPhase.set(phaseKey, -1);
io.stderr.write(`${plainPhaseHeader(options, phaseKey)}\n`);
};
const ensurePhaseStarted = (phaseKey: KtxPublicIngestPhaseKey): void => {
if (!startedPhases.has(phaseKey)) {
startPhase(phaseKey);
return;
}
currentPhase = phaseKey;
};
const emitProgress = (update: KtxIngestProgressUpdate): void => {
if (currentPhase === null) return;
const rounded = Math.max(0, Math.min(100, Math.round(update.percent)));
const lastPercent = lastPercentByPhase.get(currentPhase) ?? -1;
if (rounded <= lastPercent) return;
lastPercentByPhase.set(currentPhase, rounded);
io.stderr.write(` [${rounded}%] ${publicProgressMessage(update.message, options.target)}\n`);
};
return {
onPhaseStart: startPhase,
onPhaseEnd(phaseKey, status, summary) {
ensurePhaseStarted(phaseKey);
io.stderr.write(`${plainPhaseEndLine(status, summary)}\n`);
currentPhase = null;
},
scanProgress: createAggregateProgressPort(emitProgress),
ingestProgress: emitProgress,
};
}
const INTERNAL_STATUS_LINE_RE =
/^(Report|Run|Job|Status|Adapter|Connection|Sync|Diff|Tasks|Work units|Failed tasks|Saved memory|Provenance rows):\s*/;
const ACTIONABLE_FAILURE_LINE_RE =
@ -790,7 +880,7 @@ export async function executePublicIngestTarget(
? {
...step,
status: 'failed',
detail: target.preflightFailure,
detail: `${target.connectionId} failed: ${target.preflightFailure}`,
}
: step,
),
@ -810,7 +900,11 @@ export async function executePublicIngestTarget(
...(args.runtimeInstallPolicy ? { runtimeInstallPolicy: args.runtimeInstallPolicy } : {}),
};
const runScan = deps.runScan ?? runKtxScan;
const capturedScanIo = deps.scanProgress ? null : createCapturedPublicIngestIo();
const capturedScanIo = deps.scanProgress
? isCapturedPublicIngestIo(io)
? io
: null
: createCapturedPublicIngestIo();
const scanIo = capturedScanIo ?? io;
const scanDeps = {
...(deps.scanProgress ? { progress: deps.scanProgress } : {}),
@ -853,7 +947,13 @@ export async function executePublicIngestTarget(
...(target.queryHistory.windowDays !== undefined ? { windowDays: target.queryHistory.windowDays } : {}),
},
};
const capturedIngestIo = deps.ingestProgress ? null : createCapturedPublicIngestIo();
// Query history runs after the schema scan has already written its report
// into the shared target io, so it needs a phase-local capture. Reusing
// `io` here would let leftover scan text (e.g. "Mode: enriched") surface as
// the query-history failure detail. Only skip capture when progress is
// active and the caller manages its own buffer (io is not a capture).
const capturedIngestIo =
deps.ingestProgress && !isCapturedPublicIngestIo(io) ? null : createCapturedPublicIngestIo();
const ingestIo = capturedIngestIo ?? io;
const ingestDeps = {
...(deps.ingestProgress ? { progress: deps.ingestProgress } : {}),
@ -893,7 +993,11 @@ export async function executePublicIngestTarget(
allowImplicitAdapter: true,
};
const runIngest = deps.runIngest ?? runKtxIngest;
const capturedIngestIo = deps.ingestProgress ? null : createCapturedPublicIngestIo();
const capturedIngestIo = deps.ingestProgress
? isCapturedPublicIngestIo(io)
? io
: null
: createCapturedPublicIngestIo();
const ingestIo = capturedIngestIo ?? io;
const ingestDeps = {
...(deps.ingestProgress ? { progress: deps.ingestProgress } : {}),
@ -976,9 +1080,30 @@ export async function runKtxPublicIngest(
}
}
for (const target of plan.targets) {
for (const [index, target] of plan.targets.entries()) {
const startedAt = performance.now();
const result = await executePublicIngestTarget(target, args, io, deps);
if (args.json) {
const result = await executePublicIngestTarget(target, args, io, deps);
results.push(result);
await emitIngestCompleted({ args, project, target, result, startedAt, io });
continue;
}
const capture = createCapturedPublicIngestIo();
const progress = createPlainPublicIngestProgress(io, {
target,
index,
total: plan.targets.length,
});
const targetDeps: KtxPublicIngestDeps = {
...deps,
scanProgress: progress.scanProgress,
ingestProgress: progress.ingestProgress,
onPhaseStart: progress.onPhaseStart,
onPhaseEnd: progress.onPhaseEnd,
runtimeIo: deps.runtimeIo ?? io,
};
const result = await executePublicIngestTarget(target, args, capture, targetDeps);
results.push(result);
await emitIngestCompleted({ args, project, target, result, startedAt, io });
}

35
packages/cli/test/progress-port-adapter.test.ts Normal file
View file

@ -0,0 +1,35 @@
import { describe, expect, it } from 'vitest';
import { createAggregateProgressPort } from '../src/progress-port-adapter.js';
describe('createAggregateProgressPort', () => {
it('flattens nested weighted progress into absolute percent updates', async () => {
const updates: Array<{ percent: number; message: string; transient?: boolean }> = [];
const progress = createAggregateProgressPort((update) => updates.push(update));
await progress.update(0.1, 'Preparing scan');
const nested = progress.startPhase(0.5);
await nested.update(0.5, 'Generating descriptions 2/4 tables', { transient: true });
await progress.update(0.95, 'Writing schema artifacts');
expect(updates).toEqual([
{ percent: 10, message: 'Preparing scan' },
{ percent: 35, message: 'Generating descriptions 2/4 tables', transient: true },
{ percent: 95, message: 'Writing schema artifacts' },
]);
});
it('clamps updates and never moves the shared progress state backward', async () => {
const updates: Array<{ percent: number; message: string }> = [];
const progress = createAggregateProgressPort((update) => updates.push(update));
await progress.update(0.8, 'Building enriched schema context');
await progress.update(0.2, 'Older scan callback');
await progress.update(1.4, 'Scan completed');
expect(updates).toEqual([
{ percent: 80, message: 'Building enriched schema context' },
{ percent: 80, message: 'Older scan callback' },
{ percent: 100, message: 'Scan completed' },
]);
});
});

235
packages/cli/test/public-ingest.test.ts
View file

@ -8,6 +8,7 @@ import {
buildPublicIngestPlan,
type KtxPublicIngestDeps,
type KtxPublicIngestProject,
publicProgressMessage,
runKtxPublicIngest,
} from '../src/public-ingest.js';
import type { ManagedPythonCommandRuntime } from '../src/managed-python-command.js';
@ -346,6 +347,29 @@ describe('buildPublicIngestPlan', () => {
});
});
describe('publicProgressMessage', () => {
it('rewrites internal scan and historic-sql phrasing for public ingest progress', () => {
const databaseProject = deepReadyProject({
warehouse: { driver: 'postgres', context: { queryHistory: { enabled: true, dialect: 'postgres' } } },
});
const databaseTarget = buildPublicIngestPlan(databaseProject, {
projectDir: '/tmp/project',
all: false,
targetConnectionId: 'warehouse',
queryHistory: 'default',
}).targets[0];
expect(databaseTarget).toBeDefined();
expect(publicProgressMessage('Inspecting database schema', databaseTarget)).toBe('Reading database schema');
expect(publicProgressMessage('Enriching schema metadata', databaseTarget)).toBe(
'Building enriched schema context',
);
expect(publicProgressMessage('Fetching source files for warehouse/historic-sql', databaseTarget)).toBe(
'Fetching query history for warehouse',
);
});
});
describe('runKtxPublicIngest', () => {
afterEach(() => {
vi.unstubAllEnvs();
@ -371,11 +395,13 @@ describe('runKtxPublicIngest', () => {
1,
expect.objectContaining({ connectionId: 'first', mode: 'enriched', detectRelationships: true }),
expect.anything(),
expect.objectContaining({ progress: expect.any(Object) }),
);
expect(runScan).toHaveBeenNthCalledWith(
2,
expect.objectContaining({ connectionId: 'second', mode: 'enriched', detectRelationships: true }),
expect.anything(),
expect.objectContaining({ progress: expect.any(Object) }),
);
});
@ -655,7 +681,10 @@ describe('runKtxPublicIngest', () => {
expect(io.stdout()).not.toContain('Report: report-docs-1');
expect(io.stdout()).not.toContain('Adapter:');
expect(io.stdout()).not.toContain('notion\n');
expect(io.stderr()).toBe('');
expect(io.stderr()).toContain('docs · source ingest\n');
expect(io.stderr()).toContain(' done\n');
expect(io.stderr()).not.toContain('Report: report-docs-1');
expect(io.stderr()).not.toContain('Adapter:');
});
it('suppresses historic-sql report output during direct public query-history ingest', async () => {
@ -694,9 +723,168 @@ describe('runKtxPublicIngest', () => {
expect(io.stdout()).not.toContain('Report: report-query-history-1');
expect(io.stdout()).not.toContain('Adapter:');
expect(io.stdout()).not.toContain('historic-sql');
expect(io.stderr()).toContain('warehouse · database schema\n');
expect(io.stderr()).toContain('warehouse · query history\n');
expect(io.stderr()).toContain(' done\n');
expect(io.stderr()).not.toContain('Report: report-query-history-1');
expect(io.stderr()).not.toContain('Adapter:');
expect(io.stderr()).not.toContain('historic-sql');
});
it('streams plain non-json progress to stderr while keeping final results on stdout', async () => {
const io = makeIo();
const project = deepReadyProject({
warehouse: { driver: 'postgres', context: { queryHistory: { enabled: true, dialect: 'postgres' } } },
docs: { driver: 'notion' },
});
const runScan = vi.fn<NonNullable<KtxPublicIngestDeps['runScan']>>(async (_args, scanIo, deps) => {
scanIo.stdout.write('KTX scan completed\n');
scanIo.stdout.write('Report: raw-sources/warehouse/live-database/sync-1/scan-report.json\n');
await deps?.progress?.update(0.12, 'Inspecting database schema');
const enrichmentProgress = deps?.progress?.startPhase(0.5);
await enrichmentProgress?.update(0.75, 'Enriching schema metadata', { transient: true });
await deps?.progress?.update(1, 'Writing schema artifacts');
return 0;
});
const runIngest = vi.fn<NonNullable<KtxPublicIngestDeps['runIngest']>>(async (ingestArgs, ingestIo, deps) => {
if (ingestArgs.command !== 'run') {
throw new Error(`Unexpected ingest command: ${ingestArgs.command}`);
}
ingestIo.stdout.write(`Adapter: ${ingestArgs.adapter}\n`);
ingestIo.stdout.write('Report: report-progress-1\n');
if (ingestArgs.adapter === 'historic-sql') {
deps?.progress?.({ percent: 15, message: 'Fetching source files for warehouse/historic-sql' });
deps?.progress?.({ percent: 90, message: 'Saved memory: 1 wiki, 1 SL' });
return 0;
}
deps?.progress?.({ percent: 55, message: 'Processing 3/8 tasks' });
deps?.progress?.({ percent: 90, message: 'Saved memory: 6 wiki, 2 SL' });
return 0;
});
await expect(
runKtxPublicIngest(
{
command: 'run',
projectDir: '/tmp/project',
all: true,
json: false,
inputMode: 'disabled',
queryHistory: 'default',
},
io.io,
{ loadProject: vi.fn(async () => project), runScan, runIngest },
),
).resolves.toBe(0);
expect(io.stdout()).toContain('Ingest finished');
expect(io.stdout()).toContain('warehouse');
expect(io.stdout()).toContain('docs');
expect(io.stdout()).not.toContain('KTX scan completed');
expect(io.stdout()).not.toContain('Report:');
expect(io.stdout()).not.toContain('Adapter:');
expect(io.stderr()).toContain('[1/2] warehouse · database schema\n');
expect(io.stderr()).toContain(' [12%] Reading database schema\n');
expect(io.stderr()).toContain(' [50%] Building enriched schema context\n');
expect(io.stderr()).toContain('[1/2] warehouse · query history\n');
expect(io.stderr()).toContain(' [15%] Fetching query history for warehouse\n');
expect(io.stderr()).toContain('[2/2] docs · source ingest\n');
expect(io.stderr()).toContain(' [55%] Processing 3/8 tasks\n');
expect(io.stderr()).not.toContain('\r');
});
it('does not emit plain progress for json public ingest output', async () => {
const io = makeIo();
const project = deepReadyProject({
warehouse: { driver: 'postgres' },
});
const runScan = vi.fn<NonNullable<KtxPublicIngestDeps['runScan']>>(async (_args, _scanIo, deps) => {
expect(deps?.progress).toBeUndefined();
return 0;
});
await expect(
runKtxPublicIngest(
{
command: 'run',
projectDir: '/tmp/project',
targetConnectionId: 'warehouse',
all: false,
json: true,
inputMode: 'disabled',
},
io.io,
{ loadProject: vi.fn(async () => project), runScan },
),
).resolves.toBe(0);
expect(JSON.parse(io.stdout())).toMatchObject({
plan: { projectDir: '/tmp/project' },
results: [{ connectionId: 'warehouse', driver: 'postgres' }],
});
expect(io.stderr()).toBe('');
});
it('keeps captured failure details when plain progress ports are active', async () => {
const io = makeIo();
const project = deepReadyProject({ warehouse: { driver: 'postgres' } });
const runScan = vi.fn<NonNullable<KtxPublicIngestDeps['runScan']>>(async (_args, scanIo, deps) => {
await deps?.progress?.update(0.42, 'Enriching schema metadata');
scanIo.stdout.write('KTX scan enrichment failed after structural scan completed: embedding service timed out\n');
return 1;
});
await expect(
runKtxPublicIngest(
{
command: 'run',
projectDir: '/tmp/project',
targetConnectionId: 'warehouse',
all: false,
json: false,
inputMode: 'disabled',
},
io.io,
{ loadProject: vi.fn(async () => project), runScan },
),
).resolves.toBe(1);
expect(io.stderr()).toContain('warehouse · database schema\n');
expect(io.stderr()).toContain(' [42%] Building enriched schema context\n');
expect(io.stderr()).toContain(' failed\n');
expect(io.stdout()).toContain(
'warehouse failed: Database enrichment failed after schema context completed: embedding service timed out.',
);
expect(io.stdout()).not.toContain('KTX scan enrichment failed');
expect(io.stdout()).not.toContain('structural scan');
});
it('prints a failed plain phase when preflight fails before phase start', async () => {
const io = makeIo();
const project = projectWithConnections({
warehouse: { driver: 'postgres' },
});
await expect(
runKtxPublicIngest(
{
command: 'run',
projectDir: '/tmp/project',
targetConnectionId: 'warehouse',
all: false,
json: false,
inputMode: 'disabled',
},
io.io,
{ loadProject: vi.fn(async () => project) },
),
).resolves.toBe(1);
expect(io.stderr()).toContain('warehouse · database schema\n');
expect(io.stderr()).toContain(' failed · warehouse cannot be ingested: enrichment is not configured');
expect(io.stdout()).toContain('warehouse failed: warehouse cannot be ingested: enrichment is not configured');
});
it('delegates interactive TTY public ingest to the foreground context-build view', async () => {
const io = makeIo({ isTTY: true, interactive: true });
const project = projectWithConnections({ warehouse: { driver: 'postgres' } });
@ -872,6 +1060,7 @@ describe('runKtxPublicIngest', () => {
inputMode: 'disabled',
}),
expect.anything(),
expect.objectContaining({ progress: expect.any(Function) }),
);
expect(runScan).toHaveBeenCalledWith(
{
@ -883,6 +1072,7 @@ describe('runKtxPublicIngest', () => {
dryRun: false,
},
expect.anything(),
expect.objectContaining({ progress: expect.any(Object) }),
);
expect(io.stdout()).toContain('Ingest finished with partial failures');
expect(io.stdout()).toContain('warehouse failed at database-schema.');
@ -930,6 +1120,45 @@ describe('runKtxPublicIngest', () => {
expect(io.stdout()).not.toContain('historic-sql');
});
it('reports the query-history failure without leaking earlier scan report output', async () => {
const io = makeIo();
const project = deepReadyProject({
warehouse: { driver: 'postgres' },
});
const runScan = vi.fn(async (_args, scanIo) => {
scanIo.stdout.write('Run: scan-run-1\n');
scanIo.stdout.write('Mode: enriched\n');
scanIo.stdout.write('Dry run: no\n');
scanIo.stdout.write('KTX scan completed\n');
return 0;
});
const runIngest = vi.fn(async (_args, ingestIo) => {
ingestIo.stderr.write('Stopped query history before persisting any results\n');
return 1;
});
await expect(
runKtxPublicIngest(
{
command: 'run',
projectDir: '/tmp/project',
targetConnectionId: 'warehouse',
all: false,
json: false,
inputMode: 'disabled',
queryHistory: 'enabled',
},
io.io,
{ loadProject: vi.fn(async () => project), runScan, runIngest },
),
).resolves.toBe(0);
expect(io.stdout()).toContain('Skipped query history:');
expect(io.stdout()).toContain('Stopped query history before persisting any results');
expect(io.stdout()).not.toContain('Dry run: no');
expect(io.stdout()).not.toContain('Mode: enriched');
});
it('prints the runtime artifact build hint for missing query-history runtime assets', async () => {
const io = makeIo();
const project = deepReadyProject({
@ -989,6 +1218,7 @@ describe('runKtxPublicIngest', () => {
expect(runIngest).toHaveBeenCalledWith(
expect.objectContaining({ command: 'run', connectionId: 'docs', adapter: 'notion' }),
expect.anything(),
expect.objectContaining({ progress: expect.any(Function) }),
);
expect(io.stdout()).toContain('warehouse cannot be ingested: enrichment is not configured');
});
@ -1027,6 +1257,7 @@ describe('runKtxPublicIngest', () => {
dryRun: false,
},
expect.objectContaining({ capturedOutput: expect.any(Function) }),
expect.objectContaining({ progress: expect.any(Object) }),
);
});
@ -1099,6 +1330,7 @@ describe('runKtxPublicIngest', () => {
sourceDir: '/repo/dbt',
}),
expect.objectContaining({ capturedOutput: expect.any(Function) }),
expect.objectContaining({ progress: expect.any(Function) }),
);
});
@ -1135,6 +1367,7 @@ describe('runKtxPublicIngest', () => {
allowImplicitAdapter: true,
}),
expect.objectContaining({ capturedOutput: expect.any(Function) }),
expect.objectContaining({ progress: expect.any(Function) }),
);
});

10
scripts/package-artifacts.mjs
View file

@ -518,7 +518,10 @@ function requireExitCodeWithProjectStderr(label, result, projectDir, expectedCod
expectedCode,
label + ' failed with code ' + result.code + '\\nstdout:\\n' + result.stdout + '\\nstderr:\\n' + result.stderr,
);
assert.equal(result.stderr, 'Project: ' + projectDir + '\\n', label + ' wrote unexpected stderr');
assert.ok(
result.stderr.startsWith('Project: ' + projectDir + '\\n'),
label + ' did not lead stderr with the project notice\\nstderr:\\n' + result.stderr,
);
}
function requireSuccessWithStderr(label, result, stderrPattern) {
@ -534,6 +537,10 @@ function requireOutput(label, result, text) {
assert.match(result.stdout, text, label + ' output did not match ' + text);
}
function requireStderr(label, result, stderrPattern) {
assert.match(result.stderr, stderrPattern, label + ' stderr did not match ' + stderrPattern);
}
function escapeRegExp(value) {
return value.replace(/[|\\\\{}()[\\]^$+*?.]/g, '\\\\$&');
}
@ -857,6 +864,7 @@ try {
),
);
requireExitCodeWithProjectStderr('ktx ingest enrichment guard', databaseIngest, projectDir, 1);
requireStderr('ktx ingest enrichment guard', databaseIngest, /^ {2}failed /m);
requireOutput('ktx ingest enrichment guard', databaseIngest, /Ingest finished with partial failures/);
requireOutput('ktx ingest enrichment guard', databaseIngest, /enrichment is not configured/);
process.stdout.write('ktx ingest enrichment guard verified\\n');

1
scripts/package-artifacts.test.mjs
View file

@ -535,6 +535,7 @@ describe('verification snippets', () => {
assert.doesNotMatch(source, /'--enrich'/);
assert.match(source, /ktx ingest enrichment guard verified/);
assert.match(source, /enrichment is not configured/);
assert.match(source, /requireStderr\('ktx ingest enrichment guard'/);
assert.match(source, /enrichment:/);
assert.match(source, /mode: deterministic/);
assert.doesNotMatch(source, /run\('pnpm', \['exec', 'ktx', 'ingest', 'run'/);