mirror of
https://github.com/Kaelio/ktx.git
synced 2026-06-22 08:38:08 +02:00
feat(cli): stream plain ktx ingest progress to stderr (KLO-726) (#251)
* 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
GitHub
parent
d01abe6f3c
commit
13774bfcef
8 changed files with 445 additions and 49 deletions
35
packages/cli/test/progress-port-adapter.test.ts
Normal file
35
packages/cli/test/progress-port-adapter.test.ts
Normal 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' },
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
|
@ -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) }),
|
||||
);
|
||||
});
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue