Improve Notion ingest UX

packages/cli/src/commands/connection-notion.test.ts

@@ -355,6 +355,53 @@ describe('runKtxConnectionNotion', () => {
     expect(io.stdout()).toContain('rootPageIds: 1');
   });
 
+  it('uses inline Notion auth_token for interactive discovery', async () => {
+    const projectDir = join(tempDir, 'project');
+    const initialized = await initKtxProject({ projectDir, projectName: 'warehouse' });
+    await writeProjectConfig(projectDir, {
+      ...initialized.config,
+      connections: {
+        'notion-main': {
+          driver: 'notion',
+          auth_token: 'ntn_inline_token',
+          crawl_mode: 'selected_roots',
+          root_page_ids: [PAGE_IDS.engineering],
+          root_database_ids: [],
+          root_data_source_ids: [],
+          max_pages_per_run: 12,
+          max_knowledge_creates_per_run: 2,
+          max_knowledge_updates_per_run: 7,
+          last_successful_cursor: null,
+        },
+      },
+    });
+    const api = fakeNotionApi([notionPage(PAGE_IDS.engineering, 'Engineering')]);
+    const createNotionApi = vi.fn((authToken: string) => {
+      expect(authToken).toBe('ntn_inline_token');
+      return api;
+    });
+    const io = makeIo();
+
+    await expect(
+      runKtxConnectionNotion(
+        {
+          command: 'pick',
+          projectDir,
+          connectionId: 'notion-main',
+          mode: 'interactive',
+        },
+        io.io,
+        {
+          createNotionApi,
+          renderPicker: vi.fn(async (): Promise<PickerRenderResult> => ({ kind: 'quit' })),
+        },
+      ),
+    ).resolves.toBe(0);
+
+    expect(createNotionApi).toHaveBeenCalledOnce();
+    expect(io.stdout()).toContain('No changes saved.');
+  });
+
   it('passes partial-discovery warnings into the TUI banner state', async () => {
     const projectDir = join(tempDir, 'project');
     const initialized = await initKtxProject({ projectDir, projectName: 'warehouse' });

packages/cli/src/commands/connection-notion.ts

@@ -1,4 +1,4 @@
-import { parseNotionConnectionConfig, resolveNotionAuthToken } from '@ktx/context/connections';
+import { parseNotionConnectionConfig, resolveNotionConnectionAuthToken } from '@ktx/context/connections';
 import { type NotionApi, type NotionBotInfo, NotionClient } from '@ktx/context/ingest';
 import {
   type KtxLocalProject,
@@ -223,7 +223,7 @@ export async function runKtxConnectionNotion(
       const project = await loadProject({ projectDir: args.projectDir });
       const rawConnection = notionConnection(project, args.connectionId);
       const notion = parseNotionConnectionConfig(rawConnection);
-      const authToken = await resolveNotionAuthToken(notion.auth_token_ref, { env: deps.env });
+      const authToken = await resolveNotionConnectionAuthToken(notion, { env: deps.env });
       const api = deps.createNotionApi ? deps.createNotionApi(authToken) : new NotionClient(authToken);
       const discovery = await discoverNotionPickerPages(api);
       const tree = buildPickerTree(discovery.pages);

packages/cli/src/ingest-viz.test.ts

@@ -186,6 +186,91 @@ describe('runKtxIngest viz and replay', () => {
     expect(io.stdout()).toContain('Connection: warehouse');
   });
 
+  it('prints live viz final summaries as errors when the report has failed work units', async () => {
+    const projectDir = join(tempDir, 'project');
+    await writeWarehouseConfig(projectDir);
+    const io = makeIo({ isTTY: true, stdinIsTTY: true, columns: 120 });
+    const liveSession = {
+      update: vi.fn(),
+      close: vi.fn(),
+      isClosed: vi.fn(() => false),
+    };
+    const startLiveMemoryFlow = vi.fn(async (_input: MemoryFlowReplayInput, _io: unknown) => liveSession);
+    const runLocal = vi.fn(async (input: RunLocalIngestOptions): Promise<LocalIngestResult> => {
+      input.memoryFlow?.emit({ type: 'source_acquired', adapter: 'notion', trigger: 'manual_resync', fileCount: 37 });
+      input.memoryFlow?.update({
+        syncId: 'sync-notion',
+        plannedWorkUnits: [
+          {
+            unitKey: 'notion-cluster-1',
+            rawFiles: ['pages/a.md'],
+            peerFileCount: 0,
+            dependencyCount: 0,
+          },
+        ],
+      });
+      input.memoryFlow?.emit({ type: 'chunks_planned', chunkCount: 1, workUnitCount: 1, evictionCount: 0 });
+      input.memoryFlow?.emit({
+        type: 'work_unit_finished',
+        unitKey: 'notion-cluster-1',
+        status: 'failed',
+        reason: 'notion-cluster-1 failed: {"error":"invalid_grant","error_description":"reauth related error (invalid_rapt)"}',
+      });
+      input.memoryFlow?.emit({ type: 'report_created', runId: 'live-failed' });
+      input.memoryFlow?.finish('done');
+
+      const failedWorkUnit = {
+        ...localFakeBundleReport('live-failed').body.workUnits[0],
+        unitKey: 'notion-cluster-1',
+        rawFiles: ['pages/a.md'],
+        status: 'failed' as const,
+        reason: 'notion-cluster-1 failed: {"error":"invalid_grant","error_description":"reauth related error (invalid_rapt)"}',
+        actions: [],
+        touchedSlSources: [],
+      };
+      const report = localFakeBundleReport('live-failed', {
+        id: 'report-live-failed',
+        runId: 'run-live-failed',
+        connectionId: input.connectionId,
+        sourceKey: input.adapter,
+        body: {
+          workUnits: [failedWorkUnit],
+          failedWorkUnits: [failedWorkUnit.unitKey],
+        },
+      });
+      return {
+        result: {
+          jobId: 'live-failed',
+          runId: report.runId,
+          syncId: report.body.syncId,
+          diffSummary: report.body.diffSummary,
+          workUnitCount: report.body.workUnits.length,
+          failedWorkUnits: report.body.failedWorkUnits,
+          artifactsWritten: report.body.provenanceRows.length,
+          commitSha: report.body.commitSha,
+        },
+        report,
+      };
+    });
+
+    await expect(
+      runKtxIngest(
+        {
+          command: 'run',
+          projectDir,
+          connectionId: 'notion-main',
+          adapter: 'notion',
+          outputMode: 'viz',
+        },
+        io.io,
+        { runLocalIngest: runLocal, startLiveMemoryFlow },
+      ),
+    ).resolves.toBe(1);
+
+    expect(io.stdout()).toContain('Memory-flow summary: error');
+    expect(io.stdout()).toContain('Notion authorization expired');
+  });
+
   it('falls back to text live rendering when the TUI live session is unavailable', async () => {
     const projectDir = join(tempDir, 'project');
     await writeWarehouseConfig(projectDir);

packages/cli/src/ingest.test.ts

@@ -1076,17 +1076,17 @@ describe('runKtxIngest', () => {
       ),
     ).resolves.toBe(0);
 
-    const stdout = io.stdout();
-    expect(stdout).toContain('[5%] Fetching source files for warehouse/historic-sql');
-    expect(stdout).toContain('[15%] Fetched 3 source files from historic-sql');
-    expect(stdout).toContain('[45%] Planned 1 work unit');
-    expect(stdout).toContain('[80%] Processed 1/1 work units');
-    expect(stdout).toContain('[100%] Ingest completed');
-    expect(stdout).toContain('Report: report-live-1');
-    expect(io.stderr()).toBe('');
+    const stderr = io.stderr();
+    expect(stderr).toContain('[5%] Fetching source files for warehouse/historic-sql');
+    expect(stderr).toContain('[15%] Fetched 3 source files from historic-sql');
+    expect(stderr).toContain('[45%] Planned 1 work unit');
+    expect(stderr).toContain('[80%] Processed 1/1 work units');
+    expect(stderr).toContain('[100%] Ingest completed');
+    expect(io.stdout()).toContain('Report: report-live-1');
+    expect(io.stdout()).not.toContain('[5%]');
   });
 
-  it('writes plain TTY ingest progress and final report to stdout', async () => {
+  it('writes plain TTY ingest progress to stderr and final report to stdout', async () => {
     const projectDir = join(tempDir, 'project');
     await writeWarehouseConfig(projectDir);
     const sourceDir = join(tempDir, 'source');
@@ -1113,9 +1113,9 @@ describe('runKtxIngest', () => {
       ),
     ).resolves.toBe(0);
 
-    expect(io.stdout()).toContain('[5%] Fetching source files for warehouse/fake');
+    expect(io.stderr()).toContain('[5%] Fetching source files for warehouse/fake');
     expect(io.stdout()).toContain('Report: report-live-1');
-    expect(io.stderr()).toBe('');
+    expect(io.stdout()).not.toContain('[5%]');
   });
 
   it('prints plain WorkUnit step progress during long-running local ingest', async () => {
@@ -1202,13 +1202,13 @@ describe('runKtxIngest', () => {
       ),
     ).resolves.toBe(0);
 
-    const stdout = io.stdout();
-    expect(stdout).toContain('[45%] Planned 2 work units');
-    expect(stdout).toContain('[55%] Processing 1/2 work units: historic-sql-table-public-orders');
-    expect(stdout).toContain(
+    const stderr = io.stderr();
+    expect(stderr).toContain('[45%] Planned 2 work units');
+    expect(stderr).toContain('[55%] Processing 1/2 work units: historic-sql-table-public-orders');
+    expect(stderr).toContain(
       '\r[58%] Processing work units: 0/2 complete, 1 active; latest historic-sql-table-public-orders step 7/40\u001b[K',
     );
-    expect(stdout).toContain('[68%] Processed 1/2 work units');
+    expect(stderr).toContain('[68%] Processed 1/2 work units');
   });
 
   it('renders concurrent WorkUnit step progress as transient aggregate status', async () => {
@@ -1294,14 +1294,14 @@ describe('runKtxIngest', () => {
       ),
     ).resolves.toBe(0);
 
-    const stdout = io.stdout();
-    expect(stdout).toContain(
+    const stderr = io.stderr();
+    expect(stderr).toContain(
       '\r[56%] Processing work units: 0/6 complete, 6 active; latest historic-sql-table-public-suppliers step 1/40\u001b[K',
     );
-    expect(stdout).not.toContain(
+    expect(stderr).not.toContain(
       '\n[56%] Processing 6/6 work units: historic-sql-table-public-suppliers step 1/40\n',
     );
-    expect(stdout).toContain('\n[100%] Ingest completed\n');
+    expect(stderr).toContain('\n[100%] Ingest completed\n');
   });
 
   it('passes local Looker pull-config options and agent runner into scheduled ingest for Looker scheduled ingest', async () => {

packages/cli/src/ingest.ts

@@ -306,7 +306,7 @@ function createPlainIngestProgressRenderer(
     if (!hasPendingTransient) {
       return;
     }
-    io.stdout.write('\n');
+    io.stderr.write('\n');
     hasPendingTransient = false;
   };
 
@@ -315,12 +315,12 @@ function createPlainIngestProgressRenderer(
     lastPercent = nextPercent;
     const line = `[${nextPercent}%] ${message}`;
     if (options?.transient === true) {
-      io.stdout.write(`\r${line}\u001b[K`);
+      io.stderr.write(`\r${line}\u001b[K`);
       hasPendingTransient = true;
       return;
     }
     flush();
-    io.stdout.write(`${line}\n`);
+    io.stderr.write(`${line}\n`);
   };
 
   return {
@@ -437,6 +437,21 @@ function initialRunMemoryFlowInput(
   };
 }
 
+function finalRunMemoryFlowInput(snapshot: MemoryFlowReplayInput, report: IngestReportSnapshot): MemoryFlowReplayInput {
+  const status = reportStatus(report);
+  return {
+    ...snapshot,
+    runId: report.runId,
+    connectionId: report.connectionId,
+    adapter: report.sourceKey,
+    status,
+    syncId: report.body.syncId,
+    reportId: report.id,
+    reportPath: report.id,
+    errors: status === 'error' ? report.body.failedWorkUnits : snapshot.errors,
+  };
+}
+
 function managedDaemonOptionsForIngestRun(
   args: Extract<KtxIngestArgs, { command: 'run' }>,
   io: KtxIngestIo,
@@ -592,7 +607,7 @@ export async function runKtxIngest(
           ...(memoryFlow ? { memoryFlow } : {}),
         });
         if (shouldUseLiveViz && memoryFlow) {
-          latestMemoryFlowSnapshot = memoryFlow.snapshot();
+          latestMemoryFlowSnapshot = finalRunMemoryFlowInput(memoryFlow.snapshot(), result.report);
           liveTui?.close();
           liveTui = null;
           io.stdout.write(formatMemoryFlowFinalSummary(latestMemoryFlowSnapshot));