diff --git a/docs-site/content/docs/configuration/ktx-yaml.mdx b/docs-site/content/docs/configuration/ktx-yaml.mdx index 0f67877d..91940cb7 100644 --- a/docs-site/content/docs/configuration/ktx-yaml.mdx +++ b/docs-site/content/docs/configuration/ktx-yaml.mdx @@ -120,6 +120,7 @@ context-source drivers share the map. | `dbt` | Context source | `driver`, one of `source_dir` or `repo_url` | `branch`, `path`, `profiles_path`, `target`, `project_name` | | `metricflow` | Context source | `driver`, `metricflow.repoUrl` | `metricflow.branch`, `metricflow.path`, `metricflow.auth_token_ref` | | `notion` | Context source | `driver`, `auth_token_ref` | `crawl_mode`, `root_*_ids`, `max_*_per_run` | +| `confluence` | Context source | `driver`, `base_url`, `email`, one of `api_token` or `api_token_ref` | `space_keys` | | `sigma` | Context source | `driver`, `client_id`, `client_secret_ref` | `api_url` | ### Warehouse drivers @@ -347,6 +348,27 @@ connections: | `max_knowledge_creates_per_run` | Max new wiki pages created per run (0-25). | | `max_knowledge_updates_per_run` | Max existing wiki pages updated per run (0-100). | +### Confluence + +```yaml +connections: + my-confluence: + driver: confluence + base_url: https://yourorg.atlassian.net + email: you@yourorg.com + api_token_ref: env:CONFLUENCE_API_TOKEN + space_keys: + - ENG + - PROD +``` + +| Field | Purpose | +|-------|---------| +| `base_url` | Confluence Cloud base URL (e.g. `https://yourorg.atlassian.net`). Required. | +| `email` | Atlassian account email used for Basic auth. Required. | +| `api_token` / `api_token_ref` | Literal API token or reference. Prefer the `_ref`. | +| `space_keys` | List of space keys to ingest. Omit to ingest all accessible global spaces. | + ### Sigma ```yaml diff --git a/docs-site/content/docs/integrations/context-sources.mdx b/docs-site/content/docs/integrations/context-sources.mdx index 3bfc3ff3..c672c90f 100644 --- a/docs-site/content/docs/integrations/context-sources.mdx +++ b/docs-site/content/docs/integrations/context-sources.mdx @@ -1,6 +1,6 @@ --- title: Context Sources -description: Ingest semantic context from dbt, MetricFlow, LookML, Metabase, Looker, Notion, Sigma, and Google Drive. +description: Ingest semantic context from dbt, MetricFlow, LookML, Metabase, Looker, Notion, Confluence, Sigma, and Google Drive. --- Context sources feed your existing analytics tooling into **ktx**. During ingestion, **ktx** extracts metadata from each source and uses a reconciliation agent to reconcile it with your existing semantic layer and knowledge base - preserving accepted edits rather than overwriting. @@ -27,7 +27,7 @@ LookML uses top-level `repoUrl`, and MetricFlow uses nested | Field | Required | Description | |-------|----------|-------------| -| `driver` | Yes | Source connector: `dbt`, `metricflow`, `lookml`, `metabase`, `looker`, `notion`, `sigma`, or `gdrive` | +| `driver` | Yes | Source connector: `dbt`, `metricflow`, `lookml`, `metabase`, `looker`, `notion`, `confluence`, `sigma`, or `gdrive` | | `source_dir` | For local file sources | Absolute or project-relative source directory | | `repo_url` | For Git-hosted dbt sources | Git repository URL | | `repoUrl` | For Git-hosted LookML sources | Git repository URL | @@ -378,6 +378,51 @@ Create an integration at [notion.so/my-integrations](https://www.notion.so/my-in --- +## Confluence + +Ingests pages from Confluence Cloud spaces as wiki context. Uses the Confluence REST API v2 to fetch page content and metadata, then synthesizes durable wiki items from the storage XML. + +### What it provides + +- Page titles, breadcrumbs, and hierarchy (parent–child relationships) +- Full page body in Confluence storage XML format +- Space keys, space names, and page URLs +- Version metadata for incremental sync + +### Connection config + +```yaml title="ktx.yaml" +connections: + my-confluence: + driver: confluence + base_url: https://yourorg.atlassian.net + email: you@yourorg.com + api_token_ref: env:CONFLUENCE_API_TOKEN # or api_token: "" + space_keys: [ENG, PROD] # optional — omit to ingest all global spaces +``` + +### Authentication + +| Method | Config | +|--------|--------| +| API token | `email` + `api_token_ref: env:CONFLUENCE_API_TOKEN` | + +Generate an API token at [id.atlassian.net/manage-profile/security/api-tokens](https://id.atlassian.net/manage-profile/security/api-tokens). + +### What gets ingested + +- All `current`-status pages in the configured spaces (or all accessible global spaces when `space_keys` is omitted) +- Each page's full content in Confluence storage XML — the `confluence_synthesize` skill extracts durable wiki candidates from it +- Ingest is incremental: pages whose version number and timestamp are unchanged since the last run are skipped + +### Notes + +- `space_keys` is case-insensitive; `ENG` and `eng` both match the same space +- Large Confluence instances with thousands of pages may take several minutes on the first run; subsequent runs are fast due to incremental sync +- Raw storage XML is kept in the staged directory; only synthesized wiki items are written to the ktx knowledge base + +--- + ## Sigma Ingests data model definitions and workbook metadata from a Sigma workspace as semantic context. Uses the Sigma REST API to fetch data model specs and workbook summaries. @@ -537,6 +582,8 @@ connections: - Only Google Docs are ingested in v1; other file types (Sheets, Slides, PDFs) in the folder are skipped and recorded in the staged manifest - The service account must be granted access to the target folder explicitly +--- + ## Common errors | Error or symptom | Likely cause | Recovery | diff --git a/packages/cli/src/context/ingest/adapters/confluence/chunk.ts b/packages/cli/src/context/ingest/adapters/confluence/chunk.ts new file mode 100644 index 00000000..5f00757e --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/confluence/chunk.ts @@ -0,0 +1,131 @@ +import { readdir, readFile } from 'node:fs/promises'; +import { join, relative } from 'node:path'; +import type { ChunkResult, DiffSet, WorkUnit } from '../../types.js'; +import { + type ConfluenceManifest, + type StagedPageFile, + confluenceManifestSchema, + stagedPageFileSchema, + STAGED_FILES, +} from './types.js'; + +interface LoadedBundle { + manifest: ConfluenceManifest | null; + pagesByPath: Map; + allPaths: string[]; +} + +async function walkStagedDir(stagedDir: string): Promise { + let entries; + try { + entries = await readdir(stagedDir, { withFileTypes: true, recursive: true }); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') return []; + throw err; + } + const paths: string[] = []; + for (const entry of entries) { + if (!entry.isFile()) continue; + const abs = join(entry.parentPath, entry.name); + paths.push(relative(stagedDir, abs).replace(/\\/g, '/')); + } + paths.sort(); + return paths; +} + +async function loadBundle(stagedDir: string): Promise { + const allPaths = await walkStagedDir(stagedDir); + let manifest: ConfluenceManifest | null = null; + try { + const body = await readFile(join(stagedDir, STAGED_FILES.manifest), 'utf-8'); + manifest = confluenceManifestSchema.parse(JSON.parse(body)); + } catch { + manifest = null; + } + + const pagesByPath = new Map(); + const pagesPrefix = `${STAGED_FILES.pagesDir}/`; + for (const path of allPaths) { + if (!path.startsWith(pagesPrefix) || !path.endsWith('.json')) continue; + try { + const body = await readFile(join(stagedDir, path), 'utf-8'); + const parsed = stagedPageFileSchema.parse(JSON.parse(body)); + pagesByPath.set(path, parsed); + } catch { + // Malformed file — skip. + } + } + + return { manifest, pagesByPath, allPaths }; +} + +/** @internal */ +export const PAGES_PER_UNIT = 30; + +function emitBatches( + paths: string[], + pages: Map, + perUnit: number, + allPaths: string[], +): WorkUnit[] { + const batches = Math.ceil(paths.length / perUnit) || 0; + const units: WorkUnit[] = []; + for (let i = 0; i < batches; i++) { + const batch = paths.slice(i * perUnit, (i + 1) * perUnit); + const spaceKeys = [ + ...new Set(batch.map((p) => pages.get(p)?.spaceKey).filter((k): k is string => Boolean(k))), + ].sort(); + const spaceLabel = spaceKeys.length > 0 ? spaceKeys.join(', ') : 'pages'; + const rawFiles = [...batch].sort(); + const rawFilesSet = new Set(rawFiles); + const suffix = batches > 1 ? `-${i}` : ''; + const labelBase = + batches > 1 + ? `Confluence: ${spaceLabel} (${i + 1}/${batches})` + : `Confluence: ${spaceLabel}`; + units.push({ + unitKey: `confluence-pages${suffix}`, + displayLabel: labelBase, + rawFiles, + peerFileIndex: allPaths.filter((p) => !rawFilesSet.has(p)).sort(), + dependencyPaths: [], + notes: `${batch.length} Confluence page${batch.length === 1 ? '' : 's'} from space${spaceKeys.length === 1 ? '' : 's'} ${spaceLabel}`, + }); + } + return units; +} + +interface ChunkOptions { + diffSet?: DiffSet; +} + +export async function chunkConfluenceStagedDir( + stagedDir: string, + opts: ChunkOptions = {}, +): Promise { + const bundle = await loadBundle(stagedDir); + if (!bundle.manifest) { + return { workUnits: [] }; + } + + const pagePaths = [...bundle.pagesByPath.keys()].sort(); + const firstRunUnits = emitBatches(pagePaths, bundle.pagesByPath, PAGES_PER_UNIT, bundle.allPaths); + + if (!opts.diffSet) { + return { workUnits: firstRunUnits }; + } + + const touched = new Set([...opts.diffSet.added, ...opts.diffSet.modified]); + const kept: WorkUnit[] = []; + for (const wu of firstRunUnits) { + const anyTouched = wu.rawFiles.some((p) => touched.has(p)); + if (!anyTouched) continue; + const changedFiles = wu.rawFiles.filter((p) => touched.has(p)); + const unchangedFiles = wu.rawFiles.filter((p) => !touched.has(p)); + const deps = new Set([...wu.dependencyPaths, ...unchangedFiles]); + kept.push({ ...wu, rawFiles: changedFiles.sort(), dependencyPaths: [...deps].sort() }); + } + const eviction = + opts.diffSet.deleted.length > 0 ? { deletedRawPaths: [...opts.diffSet.deleted].sort() } : undefined; + return { workUnits: kept, eviction }; +} diff --git a/packages/cli/src/context/ingest/adapters/confluence/client-port.ts b/packages/cli/src/context/ingest/adapters/confluence/client-port.ts new file mode 100644 index 00000000..d7996d38 --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/confluence/client-port.ts @@ -0,0 +1,47 @@ +import type { FetchContext } from '../../types.js'; +import type { ConfluencePullConfig } from './types.js'; + +export interface ConfluenceSpaceSummary { + id: string; + key: string; + name: string; + type: string; + status: string; +} + +export interface ConfluencePageSummary { + id: string; + title: string; + spaceId: string; + parentId: string | null; + status: string; + version: { number: number; createdAt: string }; + _links: { webui: string }; +} + +export interface ConfluencePageBody { + id: string; + title: string; + spaceId: string; + parentId: string | null; + status: string; + version: { number: number; createdAt: string }; + body: { storage: { value: string } }; + _links: { webui: string }; +} + +export interface ConfluenceRuntimeClient { + readonly baseUrl: string; + listSpaces(opts?: { spaceKeys?: string[] }): Promise; + listPagesInSpace(spaceId: string, opts?: { limit?: number }): Promise; + getPageWithBody(pageId: string): Promise; + testConnection(): Promise<{ success: boolean; message?: string; error?: string }>; + cleanup(): Promise; +} + +export interface ConfluenceClientFactory { + createClient( + config: ConfluencePullConfig, + ctx: FetchContext, + ): Promise | ConfluenceRuntimeClient; +} diff --git a/packages/cli/src/context/ingest/adapters/confluence/client.ts b/packages/cli/src/context/ingest/adapters/confluence/client.ts new file mode 100644 index 00000000..627d4d94 --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/confluence/client.ts @@ -0,0 +1,142 @@ +import type { + ConfluencePageBody, + ConfluencePageSummary, + ConfluenceRuntimeClient, + ConfluenceSpaceSummary, +} from './client-port.js'; + +export interface ConfluenceClientRuntimeConfig { + baseUrl: string; + email: string; + apiToken: string; +} + +export interface ConfluenceClientConfig { + maxRetries: number; + baseDelayMs: number; + maxDelayMs: number; + timeoutMs: number; +} + +export const DEFAULT_CONFLUENCE_CLIENT_CONFIG: ConfluenceClientConfig = { + maxRetries: 3, + baseDelayMs: 500, + maxDelayMs: 10_000, + timeoutMs: 30_000, +}; + +type PaginatedV2Response = { + results: T[]; + _links: { next?: string; base?: string }; +}; + +export class DefaultConfluenceClient implements ConfluenceRuntimeClient { + constructor( + private readonly runtimeConfig: ConfluenceClientRuntimeConfig, + private readonly clientConfig: ConfluenceClientConfig = DEFAULT_CONFLUENCE_CLIENT_CONFIG, + ) {} + + get baseUrl(): string { + return this.runtimeConfig.baseUrl.replace(/\/$/, ''); + } + + private get apiBase(): string { + return `${this.baseUrl}/wiki/api/v2`; + } + + private authHeader(): string { + const credentials = Buffer.from( + `${this.runtimeConfig.email}:${this.runtimeConfig.apiToken}`, + ).toString('base64'); + return `Basic ${credentials}`; + } + + private async fetchWithRetry(url: string): Promise { + let lastErr: unknown; + for (let attempt = 0; attempt <= this.clientConfig.maxRetries; attempt++) { + if (attempt > 0) { + const delay = Math.min( + this.clientConfig.baseDelayMs * 2 ** (attempt - 1), + this.clientConfig.maxDelayMs, + ); + await new Promise((resolve) => setTimeout(resolve, delay)); + } + try { + const res = await fetch(url, { + headers: { + Authorization: this.authHeader(), + Accept: 'application/json', + }, + signal: AbortSignal.timeout(this.clientConfig.timeoutMs), + }); + if (res.status === 429 || res.status >= 500) { + const text = await res.text().catch(() => ''); + lastErr = new Error(`Confluence API error (${res.status}): ${text}`); + continue; + } + return res; + } catch (err) { + lastErr = err; + } + } + throw lastErr; + } + + private async getJson(path: string): Promise { + const url = path.startsWith('http') ? path : `${this.apiBase}${path}`; + const res = await this.fetchWithRetry(url); + if (!res.ok) { + const text = await res.text().catch(() => ''); + throw new Error(`Confluence API error (${res.status}) at ${url}: ${text}`); + } + return res.json() as Promise; + } + + private async paginateAll(initialPath: string): Promise { + const results: T[] = []; + let nextUrl: string | null = initialPath; + while (nextUrl) { + const page: PaginatedV2Response = await this.getJson>(nextUrl); + results.push(...page.results); + const nextLink: string | undefined = page._links?.next; + if (!nextLink) break; + // next is a relative path like /wiki/api/v2/spaces?cursor=... + const base: string = page._links?.base ?? this.runtimeConfig.baseUrl.replace(/\/$/, ''); + nextUrl = nextLink.startsWith('http') ? nextLink : `${base}${nextLink}`; + } + return results; + } + + async listSpaces(opts?: { spaceKeys?: string[] }): Promise { + const all = await this.paginateAll('/spaces?limit=250&status=current&type=global'); + if (!opts?.spaceKeys?.length) return all; + const filter = new Set(opts.spaceKeys.map((k) => k.toUpperCase())); + return all.filter((s) => filter.has(s.key.toUpperCase())); + } + + async listPagesInSpace(spaceId: string, opts?: { limit?: number }): Promise { + const limit = Math.min(opts?.limit ?? 250, 250); + return this.paginateAll(`/spaces/${spaceId}/pages?limit=${limit}&status=current`); + } + + async getPageWithBody(pageId: string): Promise { + return this.getJson(`/pages/${pageId}?body-format=storage`); + } + + async testConnection(): Promise<{ success: boolean; message?: string; error?: string }> { + try { + const res = await this.fetchWithRetry(`${this.apiBase}/spaces?limit=1`); + if (res.ok) { + return { success: true, message: 'Connected to Confluence Cloud' }; + } + const text = await res.text().catch(() => ''); + return { success: false, error: `HTTP ${res.status}: ${text.slice(0, 200)}` }; + } catch (err) { + return { success: false, error: err instanceof Error ? err.message : String(err) }; + } + } + + async cleanup(): Promise { + // No persistent connections to close. + } +} diff --git a/packages/cli/src/context/ingest/adapters/confluence/confluence.adapter.ts b/packages/cli/src/context/ingest/adapters/confluence/confluence.adapter.ts new file mode 100644 index 00000000..cb8b161b --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/confluence/confluence.adapter.ts @@ -0,0 +1,36 @@ +import type { ChunkResult, DiffSet, FetchContext, SourceAdapter } from '../../types.js'; +import { chunkConfluenceStagedDir } from './chunk.js'; +import type { ConfluenceClientFactory } from './client-port.js'; +import { detectConfluenceStagedDir } from './detect.js'; +import { fetchConfluenceBundle, type ConfluenceFetchLogger } from './fetch.js'; +import { CONFLUENCE_SOURCE_KEY } from './types.js'; + +export interface ConfluenceSourceAdapterDeps { + clientFactory: ConfluenceClientFactory; + logger?: ConfluenceFetchLogger; +} + +export class ConfluenceSourceAdapter implements SourceAdapter { + readonly source = CONFLUENCE_SOURCE_KEY; + readonly skillNames: string[] = ['confluence_synthesize']; + + constructor(private readonly deps: ConfluenceSourceAdapterDeps) {} + + detect(stagedDir: string): Promise { + return detectConfluenceStagedDir(stagedDir); + } + + async fetch(pullConfig: unknown, stagedDir: string, ctx: FetchContext): Promise { + await fetchConfluenceBundle({ + pullConfig, + stagedDir, + ctx, + clientFactory: this.deps.clientFactory, + ...(this.deps.logger ? { logger: this.deps.logger } : {}), + }); + } + + chunk(stagedDir: string, diffSet?: DiffSet): Promise { + return chunkConfluenceStagedDir(stagedDir, { diffSet }); + } +} diff --git a/packages/cli/src/context/ingest/adapters/confluence/detect.ts b/packages/cli/src/context/ingest/adapters/confluence/detect.ts new file mode 100644 index 00000000..02bd2cfc --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/confluence/detect.ts @@ -0,0 +1,12 @@ +import { stat } from 'node:fs/promises'; +import { join } from 'node:path'; +import { STAGED_FILES } from './types.js'; + +export async function detectConfluenceStagedDir(stagedDir: string): Promise { + try { + await stat(join(stagedDir, STAGED_FILES.manifest)); + return true; + } catch { + return false; + } +} diff --git a/packages/cli/src/context/ingest/adapters/confluence/fetch.ts b/packages/cli/src/context/ingest/adapters/confluence/fetch.ts new file mode 100644 index 00000000..59279c53 --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/confluence/fetch.ts @@ -0,0 +1,157 @@ +import { mkdir, readdir, readFile, rm, writeFile } from 'node:fs/promises'; +import { join } from 'node:path'; +import type { FetchContext } from '../../types.js'; +import type { ConfluenceClientFactory } from './client-port.js'; +import { + type ConfluenceManifest, + type StagedPageFile, + parseConfluencePullConfig, + stagedPageFileSchema, + STAGED_FILES, +} from './types.js'; + +export interface ConfluenceFetchLogger { + log(message: string): void; + warn(message: string): void; +} + +const noopLogger: ConfluenceFetchLogger = { log: () => undefined, warn: () => undefined }; + +export interface FetchConfluenceBundleParams { + pullConfig: unknown; + stagedDir: string; + ctx: FetchContext; + clientFactory: ConfluenceClientFactory; + logger?: ConfluenceFetchLogger; +} + +async function loadExistingStagedPages(stagedDir: string): Promise> { + const existing = new Map(); + const pagesDir = join(stagedDir, STAGED_FILES.pagesDir); + let entries: string[]; + try { + entries = await readdir(pagesDir); + } catch { + return existing; + } + for (const entry of entries) { + if (!entry.endsWith('.json')) continue; + try { + const body = await readFile(join(pagesDir, entry), 'utf-8'); + const parsed = stagedPageFileSchema.parse(JSON.parse(body)); + existing.set(parsed.pageId, parsed); + } catch { + // Skip malformed files. + } + } + return existing; +} + +function buildBreadcrumb(spaceName: string, title: string): string { + return `${spaceName} > ${title}`; +} + +export async function fetchConfluenceBundle({ + pullConfig, + stagedDir, + ctx, + clientFactory, + logger = noopLogger, +}: FetchConfluenceBundleParams): Promise { + const config = parseConfluencePullConfig(pullConfig); + const client = await clientFactory.createClient(config, ctx); + + try { + await mkdir(join(stagedDir, STAGED_FILES.pagesDir), { recursive: true }); + + const existingByPageId = await loadExistingStagedPages(stagedDir); + + logger.log('Listing Confluence spaces...'); + const spaces = await client.listSpaces( + config.spaceKeys?.length ? { spaceKeys: config.spaceKeys } : undefined, + ); + logger.log(`Found ${spaces.length} space(s).`); + + const seenPageIds = new Set(); + let totalFetched = 0; + let totalSkipped = 0; + + for (const space of spaces) { + logger.log(`Listing pages in space "${space.name}" (${space.key})...`); + let summaries; + try { + summaries = await client.listPagesInSpace(space.id); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + logger.warn(`Failed to list pages in space "${space.key}": ${msg}`); + continue; + } + logger.log(`Found ${summaries.length} page(s) in "${space.key}".`); + + for (const summary of summaries) { + seenPageIds.add(summary.id); + const existing = existingByPageId.get(summary.id); + + if (existing && existing.lastEditedAt === summary.version.createdAt && existing.version === summary.version.number) { + totalSkipped++; + continue; + } + + let pageWithBody; + try { + pageWithBody = await client.getPageWithBody(summary.id); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + logger.warn(`Failed to fetch body for page "${summary.title}" (${summary.id}): ${msg}`); + continue; + } + + const staged: StagedPageFile = { + pageId: summary.id, + spaceId: space.id, + spaceKey: space.key, + spaceName: space.name, + title: summary.title, + parentId: summary.parentId ?? null, + breadcrumb: buildBreadcrumb(space.name, summary.title), + url: `${client.baseUrl}/wiki${summary._links.webui}`, + lastEditedAt: summary.version.createdAt, + version: summary.version.number, + status: summary.status, + contentStorage: pageWithBody.body.storage.value, + }; + + await writeFile( + join(stagedDir, STAGED_FILES.pagesDir, `${summary.id}.json`), + JSON.stringify(staged, null, 2), + 'utf-8', + ); + totalFetched++; + } + } + + // Remove staged files for pages that no longer exist. + for (const [pageId] of existingByPageId) { + if (seenPageIds.has(pageId)) continue; + try { + await rm(join(stagedDir, STAGED_FILES.pagesDir, `${pageId}.json`)); + logger.log(`Removed stale staged file for page ${pageId}.`); + } catch { + // Best-effort removal. + } + } + + const manifest: ConfluenceManifest = { + confluenceConnectionId: config.confluenceConnectionId, + baseUrl: client.baseUrl, + fetchedAt: new Date().toISOString(), + spaceCount: spaces.length, + pageCount: totalFetched + totalSkipped, + ...(config.spaceKeys?.length ? { spaceKeys: config.spaceKeys } : {}), + }; + await writeFile(join(stagedDir, STAGED_FILES.manifest), JSON.stringify(manifest, null, 2), 'utf-8'); + logger.log(`Confluence fetch complete. Pages: ${totalFetched} fetched, ${totalSkipped} unchanged.`); + } finally { + await client.cleanup(); + } +} diff --git a/packages/cli/src/context/ingest/adapters/confluence/local-confluence.adapter.ts b/packages/cli/src/context/ingest/adapters/confluence/local-confluence.adapter.ts new file mode 100644 index 00000000..18d890c7 --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/confluence/local-confluence.adapter.ts @@ -0,0 +1,78 @@ +import type { KtxProjectConnectionConfig } from '../../../../context/project/config.js'; +import type { KtxLocalProject } from '../../../../context/project/project.js'; +import { resolveKtxConfigReference } from '../../../core/config-reference.js'; +import { + DEFAULT_CONFLUENCE_CLIENT_CONFIG, + DefaultConfluenceClient, + type ConfluenceClientConfig, + type ConfluenceClientRuntimeConfig, +} from './client.js'; +import type { ConfluenceClientFactory, ConfluenceRuntimeClient } from './client-port.js'; +import type { ConfluenceFetchLogger } from './fetch.js'; +import type { ConfluencePullConfig } from './types.js'; +import { ConfluenceSourceAdapter } from './confluence.adapter.js'; +import type { FetchContext } from '../../types.js'; + +function stringField(value: unknown): string | null { + return typeof value === 'string' && value.trim().length > 0 ? value.trim() : null; +} + +/** @internal */ +export function confluenceRuntimeConfigFromLocalConnection( + connectionId: string, + connection: KtxProjectConnectionConfig | undefined, + env: NodeJS.ProcessEnv = process.env, +): ConfluenceClientRuntimeConfig { + if (!connection || String(connection.driver).toLowerCase() !== 'confluence') { + throw new Error(`Connection "${connectionId}" is not a Confluence connection`); + } + + const baseUrl = stringField(connection.base_url); + const email = stringField(connection.email); + const literalToken = stringField(connection.api_token); + const tokenRef = stringField(connection.api_token_ref); + const apiToken = literalToken ?? (tokenRef ? (resolveKtxConfigReference(tokenRef, env) ?? null) : null); + + if (!baseUrl) throw new Error(`Connection "${connectionId}" is missing Confluence base_url`); + if (!email) throw new Error(`Connection "${connectionId}" is missing Confluence email`); + if (!apiToken) { + throw new Error(`Connection "${connectionId}" is missing Confluence api_token or api_token_ref`); + } + + return { baseUrl, email, apiToken }; +} + +interface CreateLocalConfluenceSourceAdapterOptions { + env?: NodeJS.ProcessEnv; + defaultClientConfig?: ConfluenceClientConfig; + logger?: ConfluenceFetchLogger; +} + +class LocalConfluenceClientFactory implements ConfluenceClientFactory { + constructor( + private readonly project: KtxLocalProject, + private readonly options: CreateLocalConfluenceSourceAdapterOptions, + ) {} + + createClient(config: ConfluencePullConfig, _ctx: FetchContext): ConfluenceRuntimeClient { + const runtimeConfig = confluenceRuntimeConfigFromLocalConnection( + config.confluenceConnectionId, + this.project.config.connections[config.confluenceConnectionId], + this.options.env, + ); + return new DefaultConfluenceClient( + runtimeConfig, + this.options.defaultClientConfig ?? DEFAULT_CONFLUENCE_CLIENT_CONFIG, + ); + } +} + +export function createLocalConfluenceSourceAdapter( + project: KtxLocalProject, + options: CreateLocalConfluenceSourceAdapterOptions = {}, +): ConfluenceSourceAdapter { + return new ConfluenceSourceAdapter({ + clientFactory: new LocalConfluenceClientFactory(project, options), + ...(options.logger ? { logger: options.logger } : {}), + }); +} diff --git a/packages/cli/src/context/ingest/adapters/confluence/types.ts b/packages/cli/src/context/ingest/adapters/confluence/types.ts new file mode 100644 index 00000000..7cf57adc --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/confluence/types.ts @@ -0,0 +1,56 @@ +import { z } from 'zod'; + +/** @internal */ +export const CONFLUENCE_SOURCE_KEY = 'confluence'; + +const confluenceLocalConnectionIdSchema = z.string().regex(/^[a-zA-Z0-9][a-zA-Z0-9_-]*$/); + +/** @internal */ +export const confluencePullConfigSchema = z.object({ + confluenceConnectionId: confluenceLocalConnectionIdSchema, + /** Limit crawl to specific space keys (e.g. ["ENG", "PROD"]). Defaults to all accessible spaces. */ + spaceKeys: z.array(z.string().min(1)).optional(), +}); + +export type ConfluencePullConfig = z.infer; + +export function parseConfluencePullConfig(raw: unknown): ConfluencePullConfig { + return confluencePullConfigSchema.parse(raw); +} + +/** Written to stagedDir/pages/{pageId}.json during fetch(). */ +export const stagedPageFileSchema = z.object({ + pageId: z.string(), + spaceId: z.string(), + spaceKey: z.string(), + spaceName: z.string(), + title: z.string(), + parentId: z.string().nullable().default(null), + breadcrumb: z.string(), + url: z.string(), + lastEditedAt: z.string().nullable().default(null), + version: z.number().int(), + status: z.string(), + /** Raw Confluence storage-format XML. */ + contentStorage: z.string(), +}); + +export type StagedPageFile = z.infer; + +/** Written once per fetch() to stagedDir/confluence-manifest.json. Presence acts as detect() sentinel. */ +export const confluenceManifestSchema = z.object({ + confluenceConnectionId: confluenceLocalConnectionIdSchema, + baseUrl: z.string(), + fetchedAt: z.string(), + spaceCount: z.number().int(), + pageCount: z.number().int(), + spaceKeys: z.array(z.string()).optional(), +}); + +export type ConfluenceManifest = z.infer; + +/** Filenames inside stagedDir. Centralized so chunk() + fetch() + detect() all agree. */ +export const STAGED_FILES = { + manifest: 'confluence-manifest.json', + pagesDir: 'pages', +} as const; diff --git a/packages/cli/src/context/ingest/local-adapters.ts b/packages/cli/src/context/ingest/local-adapters.ts index 75e54a33..182fa68b 100644 --- a/packages/cli/src/context/ingest/local-adapters.ts +++ b/packages/cli/src/context/ingest/local-adapters.ts @@ -47,6 +47,8 @@ import { pullConfigFromMetricflowIntegration } from './adapters/metricflow/pull- import { LocalNotionRuntimeStore } from './adapters/notion/local-state-store.js'; import { NotionSourceAdapter } from './adapters/notion/notion.adapter.js'; import type { NotionFetchLogger } from './adapters/notion/fetch.js'; +import { createLocalConfluenceSourceAdapter } from './adapters/confluence/local-confluence.adapter.js'; +import type { ConfluenceFetchLogger } from './adapters/confluence/fetch.js'; import { seedLocalMappingStateFromKtxYaml } from './local-mapping-reconcile.js'; import type { SourceAdapter } from './types.js'; @@ -75,6 +77,7 @@ type LocalIngestOperationalLogger = MetabaseClientLogger & MetabaseFetchLogger & LookerClientLogger & NotionFetchLogger & + ConfluenceFetchLogger & SigmaFetchLogger; export function createDefaultLocalIngestAdapters( @@ -133,6 +136,9 @@ export function createDefaultLocalIngestAdapters( }, ...(options.logger ? { logger: options.logger } : {}), }), + createLocalConfluenceSourceAdapter(project, { + ...(options.logger ? { logger: options.logger } : {}), + }), ]; if (options.historicSql) { @@ -360,6 +366,13 @@ export async function localPullConfigForAdapter( lastSuccessfulCursor: await localNotionRuntimeStore(project).readCursor(connectionId), }; } + if (adapter.source === 'confluence') { + const spaceKeys = Array.isArray(connection?.space_keys) ? connection.space_keys : undefined; + return { + confluenceConnectionId: connectionId, + ...(spaceKeys ? { spaceKeys } : {}), + }; + } if (adapter.source === 'gdrive') { return await gdriveConnectionToPullConfig(parseGdriveConnectionConfig(connection)); } diff --git a/packages/cli/src/context/ingest/repo-fetch.ts b/packages/cli/src/context/ingest/repo-fetch.ts index 1fecb422..021f4d4a 100644 --- a/packages/cli/src/context/ingest/repo-fetch.ts +++ b/packages/cli/src/context/ingest/repo-fetch.ts @@ -154,7 +154,10 @@ async function tryPull(cacheDir: string, authUrl: string, branch: string): Promi await git.remote(['set-url', 'origin', authUrl]); await git.fetch(['origin', branch]); await git.checkout(branch); - await git.pull('origin', branch); + // --ff-only makes divergence fail deterministically instead of silently rebasing or + // merging per the caller's ambient pull.rebase/pull.ff config, which would let a + // locally-diverged cache survive instead of falling back to cloneFresh below. + await git.pull('origin', branch, ['--ff-only']); return true; } catch { return false; diff --git a/packages/cli/src/context/project/driver-schemas.ts b/packages/cli/src/context/project/driver-schemas.ts index 9f808d00..a24e6c43 100644 --- a/packages/cli/src/context/project/driver-schemas.ts +++ b/packages/cli/src/context/project/driver-schemas.ts @@ -253,6 +253,27 @@ const metricflowConnectionSchema = z }) .describe('MetricFlow / SL context-source connection.'); +const confluenceConnectionSchema = z + .looseObject({ + driver: z.literal('confluence'), + base_url: z + .string() + .url() + .describe('Confluence Cloud base URL (e.g. https://yourorg.atlassian.net).'), + email: z.string().min(1).describe('Atlassian account email used for Basic auth.'), + api_token: z.string().min(1).optional().describe('Literal Atlassian API token. Prefer api_token_ref.'), + api_token_ref: z + .string() + .min(1) + .optional() + .describe('Reference to Atlassian API token (e.g. env:CONFLUENCE_API_TOKEN or file:/path).'), + space_keys: z + .array(z.string().min(1)) + .optional() + .describe('Limit ingest to specific space keys (e.g. ["ENG", "PROD"]). Defaults to all accessible global spaces.'), + }) + .describe('Confluence Cloud context-source connection.'); + const sigmaConnectionSchema = z .looseObject({ driver: z.literal('sigma'), @@ -304,5 +325,6 @@ export const connectionConfigSchema = z.discriminatedUnion('driver', [ gdriveConnectionSchema, dbtConnectionSchema, metricflowConnectionSchema, + confluenceConnectionSchema, sigmaConnectionSchema, ]); diff --git a/packages/cli/src/public-ingest.ts b/packages/cli/src/public-ingest.ts index beab6dca..a9f39235 100644 --- a/packages/cli/src/public-ingest.ts +++ b/packages/cli/src/public-ingest.ts @@ -135,6 +135,7 @@ const sourceAdapterByDriver = new Map([ ['local_metabase', 'metabase'], ['looker', 'looker'], ['notion', 'notion'], + ['confluence', 'confluence'], ['gdrive', 'gdrive'], ['metricflow', 'metricflow'], ['dbt', 'dbt'], diff --git a/packages/cli/src/skills/confluence_synthesize/SKILL.md b/packages/cli/src/skills/confluence_synthesize/SKILL.md new file mode 100644 index 00000000..1944fd60 --- /dev/null +++ b/packages/cli/src/skills/confluence_synthesize/SKILL.md @@ -0,0 +1,106 @@ +--- +name: confluence_synthesize +description: Extract durable ktx wiki knowledge from staged Confluence Cloud pages. Load for WorkUnits with unitKey starting with confluence-pages. +callers: [memory_agent] +--- + +# Confluence Synthesize + +Confluence synthesize turns staged Confluence page snapshots into durable **ktx** wiki knowledge. Pages contain business processes, technical documentation, metric definitions, team conventions, and domain knowledge. + +## Work unit structure + +Confluence produces one or more work units per ingest run: + +- `confluence-pages` — all staged pages when ≤ 30 pages total. +- `confluence-pages-N` — batched when more than 30 pages are staged, e.g. `confluence-pages-0`, `confluence-pages-1`. The `displayLabel` includes `(N/total)`. + +Each work unit's `rawFiles` is a subset of `pages/{pageId}.json` files. `confluence-manifest.json` is never in `rawFiles`; it always appears in `peerFileIndex` for context. + +## Staged file shapes + +**`pages/{pageId}.json`** — one per Confluence page: +```json +{ + "pageId": "393358", + "spaceId": "393220", + "spaceKey": "SEC", + "spaceName": "Global Information Security", + "title": "Global Information Security Office (GIS)", + "parentId": null, + "breadcrumb": "Global Information Security > Global Information Security Office (GIS)", + "url": "https://yourorg.atlassian.net/wiki/spaces/SEC/overview", + "lastEditedAt": "2026-04-27T17:04:03.456Z", + "version": 113, + "status": "current", + "contentStorage": "

Welcome to the Security Wiki...

" +} +``` + +`contentStorage` is Confluence storage-format XML. Key constructs: +- `

`, `

` through `

` — paragraph and heading content +- ``, `` — emphasis +- `