diff --git a/src/distill/propose.ts b/src/distill/propose.ts index b4566020..503f9b66 100644 --- a/src/distill/propose.ts +++ b/src/distill/propose.ts @@ -41,6 +41,18 @@ export const DISTILL_COLLECTION = "distill"; /** The proposing agent identity, recorded on every proposal. */ export const DISTILL_AGENT = "agent:distill"; +// A collection name is a single physical top-level directory AND the exact +// string RBAC's canWrite/canRead match against (src/access/rbac.ts). Those two +// checks must never diverge, so a collection may not contain a path separator +// or traversal segment — otherwise a value that passes an RBAC check for one +// string could resolve to a different directory (or escape the vault root). +const COLLECTION_NAME_PATTERN = /^[A-Za-z0-9_-]+$/; + +/** True if `value` is safe to use as both an RBAC-checked collection name and a physical path segment. */ +export function isValidCollectionName(value: string): boolean { + return COLLECTION_NAME_PATTERN.test(value); +} + /** * Maximum number of overlap paths attached to a proposal rationale (U8). * Small and bounded: the hint is advisory context for the ratifier, not a @@ -132,6 +144,14 @@ export interface DistillIds { * is date-stable across runs. */ asOf?: string; + /** + * Optional target collection override (M365 ingestion design, #506). + * Defaults to DISTILL_COLLECTION. Set by a selected-source connector whose + * enrollment names a target collection other than the default; every other + * caller is unaffected. The raw-tier fence (refuseRawDistillOutput) runs + * against the resulting path regardless of which collection produced it. + */ + collection?: string; } /** Per-claim staging outcome (the StageOutcome from the queue, or an error). */ @@ -174,10 +194,12 @@ function hash8FromClaimKey(claimKey: string): string { // co-located AND stable across runs — U5's re-distill join relies on it); // falls back to "claims" if the source-id is empty or non-slug-friendly. // -// Path-traversal safety: slugifyKey strips everything except [a-z0-9-], so -// none of the join components can contain ".." or path separators — the -// sanitizer is the invariant; don't remove it in a future refactor. -function derivePath(claim: ExtractedClaim, sourceId: string): string { +// Path-traversal safety: slugifyKey strips everything except [a-z0-9-] from +// sourceGroup/titleSlug, and proposeAllClaims rejects the batch before this +// runs if `collection` fails isValidCollectionName — none of the three join +// components can contain ".." or a path separator. Don't remove either +// sanitizer in a future refactor. +function derivePath(claim: ExtractedClaim, sourceId: string, collection: string): string { const title = claim.proposed_frontmatter.title; const hash8 = hash8FromClaimKey(claim.claim_key); const sourceGroup = slugifyKey(sourceId) || "claims"; @@ -188,7 +210,7 @@ function derivePath(claim: ExtractedClaim, sourceId: string): string { // "memory", which makes U5's targetPath-based upsert join harder to // reason about and produces semantically useless names. const titleSlug = title.trim() ? slugifyKey(title) : slugifyKey(claim.claim_key); - return join(DISTILL_COLLECTION, sourceGroup, `${titleSlug}--${hash8}.md`); + return join(collection, sourceGroup, `${titleSlug}--${hash8}.md`); } // --------------------------------------------------------------------------- @@ -393,9 +415,30 @@ export async function proposeAllClaims( ): Promise { const results: ClaimProposalResult[] = []; const errors: Array<{ claim_key: string; error: string }> = []; + const collection = ids.collection ?? DISTILL_COLLECTION; + + // collection is shared across the whole batch (see isValidCollectionName) — + // an invalid value fails every claim rather than being silently sanitized, + // since sanitizing here could make the written path diverge from the + // string an RBAC check upstream (e.g. requireCollectionWriteAccess) saw. + if (!isValidCollectionName(collection)) { + const error = `invalid collection name ${JSON.stringify(collection)}: must match ${COLLECTION_NAME_PATTERN}`; + for (const claim of claims) errors.push({ claim_key: claim.claim_key, error }); + return { proposed: 0, results, errors }; + } for (const claim of claims) { - const targetPath = pathOverrides?.[claim.claim_key] ?? derivePath(claim, ids.sourceId); + const targetPath = + pathOverrides?.[claim.claim_key] ?? derivePath(claim, ids.sourceId, collection); + const isUpdate = pathOverrides?.[claim.claim_key] !== undefined; + // U5: an update-in-place proposal's targetPath is pinned to wherever the + // claim landed on a PRIOR run (see joinClaims in state.ts) — under + // whatever collection was in effect then, which can differ from the + // current run's `collection` if the enrollment's targetCollection was + // since changed. frontmatter.collection drives RBAC/collection-scoped + // logic downstream, so it must describe where the file actually lives, + // not the current run's batch collection. + const landedCollection = isUpdate ? (targetPath.split("/")[0] ?? collection) : collection; // R3: frontmatter is hardcoded to draft/low/synthesized. No caller can // override these — the emitter owns the invariant. @@ -407,7 +450,7 @@ export async function proposeAllClaims( // missing `created` cannot be approved). created: ids.asOf ?? new Date().toISOString().slice(0, 10), domain: "accumulation", - collection: DISTILL_COLLECTION, + collection: landedCollection, status: "draft", confidence: "low", provenance: "synthesized", @@ -444,7 +487,7 @@ export async function proposeAllClaims( // 6mf.4: the op is "update" iff this claim has a path override (meaning it is // an update-in-place re-distillation of an existing landed belief), else "ingest". // The land-time union (Task 2) merges the incoming lineage with the existing one. - const isUpdate = pathOverrides?.[claim.claim_key] !== undefined; + // (isUpdate computed above, alongside landedCollection.) const lineageOp: LineageOp = isUpdate ? "update" : "ingest"; const reader = claim.run_meta ? buildReaderFrontmatter(claim.run_meta, lineageOp) : null; if (reader) Object.assign(frontmatter, reader); diff --git a/src/distill/state.ts b/src/distill/state.ts index b37497a0..c2621d91 100644 --- a/src/distill/state.ts +++ b/src/distill/state.ts @@ -204,6 +204,12 @@ export interface DistillUpsertInput { * callers remain valid — this field is optional. */ overlapSearch?: OverlapSearchFn; + /** + * Optional target collection override (#506). Defaults to + * propose.ts's DISTILL_COLLECTION when absent — every existing caller is + * unaffected. Set by a selected-source connector's enrolled collection. + */ + collection?: string; /** Injectable proposal writer used to verify atomic retry behavior. */ proposeClaims?: typeof proposeAllClaims; } @@ -288,7 +294,7 @@ export async function distillUpsert( const attempted = await (input.proposeClaims ?? proposeAllClaims)( vaultRoot, toPropose, - { sourceId: input.sourceId, runId: input.runId }, + { sourceId: input.sourceId, runId: input.runId, collection: input.collection }, pathOverrides, input.overlapSearch, ); diff --git a/src/integrations/distill.ts b/src/integrations/distill.ts index 9a6ae5c0..70a2b8a3 100644 --- a/src/integrations/distill.ts +++ b/src/integrations/distill.ts @@ -170,6 +170,7 @@ function preparedIntegrationDistill( claims: extracted.claims, runId: id, overlapSearch: makeOverlapHinter(vaultRoot), + collection: input.targetCollection, }); if (!upserted.ok) return upserted; if ((upserted.value.propose?.errors.length ?? 0) > 0) { diff --git a/src/integrations/engine.ts b/src/integrations/engine.ts index 8821433b..33d54e51 100644 --- a/src/integrations/engine.ts +++ b/src/integrations/engine.ts @@ -2,7 +2,9 @@ // and normalization; this module owns encrypted metadata and the change gate. import { randomBytes, timingSafeEqual } from "node:crypto"; +import { canWrite } from "../access/rbac.js"; import { err, ok, type Result } from "../frontmatter/types.js"; +import type { RoleConfig } from "../utils/config.js"; import { sha256Hex } from "../utils/hash.js"; import { readIntegrationState, @@ -11,6 +13,7 @@ import { writeIntegrationState, } from "./state.js"; import type { + EnrollmentRecord, IntegrationConfig, IntegrationProviderConfig, ProviderName, @@ -74,9 +77,20 @@ export type VerifiedWebhook = | { kind: "verification"; channel: WebhookChannel } | { kind: "event"; eventId: string; hint: RefreshHint }; +/** Untrusted, operator-picked enrollment input before server-side validation. */ +export type EnrollmentCandidate = Omit; + export interface RemoteSource { id: string; revision: string; + /** + * Ref of the EnrollmentRecord that owns this source (#506). Defaults to + * `id` when absent — correct for a directly-enrolled file, whose own ref + * IS its id. A folder-enrolled provider whose discovered descendant ids + * differ from their owning folder's ref must set this explicitly; the + * engine never infers folder membership itself (adapterData is opaque). + */ + enrolledRef?: string; } export interface NormalizedRemoteSource extends RemoteSource { @@ -101,6 +115,13 @@ export interface ProviderAdapter { input: WebhookRequest, state: ProviderState, ): Promise>; + // Selected-source providers re-validate operator-picked candidates with the + // connected account before they become enrollment records; a provider whose + // discover() enumerates everything the token can see omits this. + resolveEnrollment?( + candidates: EnrollmentCandidate[], + state: ProviderState, + ): Promise>; discover(state: ProviderState): Promise>; fetch(source: RemoteSource, state: ProviderState): Promise>; } @@ -109,16 +130,27 @@ export interface DistillationInput { providerSourceId: string; revision: string; text: string; + /** The owning EnrollmentRecord's collection (#506); absent for google/notion. */ + targetCollection?: string; } export interface DistillationRun { runId: string; } +// The review queue distinguishes an operator's un-enrollment from the remote +// side taking a source away; the engine's availability sweep only ever emits +// "no_longer_discovered", the other reasons come from adapters and routes. +export type UnavailableSourceReason = + | "no_longer_discovered" + | "access_denied" + | "deleted" + | "unenrolled"; + export interface UnavailableSourceEvent { idempotencyKey: string; providerSourceId: string; - reason: "no_longer_discovered"; + reason: UnavailableSourceReason; revision: string; occurredAt: string; } @@ -138,6 +170,8 @@ export interface ReconcileLimits { maxSources: number; maxSourceTextBytes: number; maxCycleTextBytes: number; + /** Soft wall-time cap: stop starting new fetches, commit what is done. */ + maxCycleMs: number; } export interface ReconcileOutcome { @@ -157,6 +191,7 @@ const DEFAULT_RECONCILE_LIMITS: ReconcileLimits = { maxSources: 10_000, maxSourceTextBytes: 8 * 1024 * 1024, maxCycleTextBytes: 64 * 1024 * 1024, + maxCycleMs: Number.POSITIVE_INFINITY, }; function reconcileLimits(deps: Pick): ReconcileLimits { @@ -210,6 +245,34 @@ export function sourceIdentity(provider: ProviderName, sourceId: string): string return `${provider}:${sourceId}`; } +// Stage-time write gate for enrollment (#506) — the same rationale as +// vault_stage_action's own gate (docs/architecture.md "Stage-time write +// gate"): manage_integrations alone must not let an operator aim distilled +// proposals at a collection the serve process cannot write to. #509's +// enroll route calls this before persisting an EnrollmentRecord. +export function requireCollectionWriteAccess( + role: RoleConfig | null, + targetCollection: string, +): Result { + if (!canWrite(role, targetCollection)) { + return err(new Error(`the serve process cannot write to collection "${targetCollection}"`)); + } + return ok(undefined); +} + +// The EnrollmentRecord that owns a discovered source: an exact ref match for +// a directly-enrolled file, or the adapter-attested owner (RemoteSource +// .enrolledRef) for a folder descendant. Absent enrollment (google/notion) +// or no match ⇒ undefined, so targetCollection is never set for them. +function owningEnrollment( + enrollment: EnrollmentRecord[] | undefined, + remote: RemoteSource, +): EnrollmentRecord | undefined { + if (enrollment === undefined) return undefined; + const ownerRef = remote.enrolledRef ?? remote.id; + return enrollment.find((record) => record.ref === ownerRef); +} + export function unavailableEventKey( provider: ProviderName, sourceId: string, @@ -230,6 +293,14 @@ function reconciliationKey(vaultRoot: string, provider: ProviderName): string { return `${vaultRoot}\u0000${provider}`; } +// Adapters may mutate adapterData in place during discovery (delta links, +// subscription bookkeeping), so a replayable snapshot must be a deep copy. +function snapshotAdapterData( + value: Record | undefined, +): Record | undefined { + return value === undefined ? undefined : structuredClone(value); +} + function sourceState( source: NormalizedRemoteSource, previous: SourceState | undefined, @@ -502,8 +573,10 @@ export async function reconcileProvider( providerState = refreshed.value; const limits = reconcileLimits(deps); + const cycleStart = currentTime(deps).getTime(); const shouldDiscover = hint.kind === "reconcile" || hint.rediscover; const previousCursor = providerState.cursor; + const previousAdapterData = snapshotAdapterData(providerState.adapterData); let discovered: Result; if (shouldDiscover) { try { @@ -525,10 +598,13 @@ export async function reconcileProvider( return err(new Error(`integration provider ${adapter.name} discovery failed`)); } const discoveredCursor = providerState.cursor; - // Discovery adapters may advance a remote change cursor. Keep that - // cursor provisional until every source in this page has been handled; - // all intermediate state writes must retain the replayable cursor. + const discoveredAdapterData = providerState.adapterData; + // Discovery adapters may advance a remote change cursor or their opaque + // adapterData (per-drive delta links behave exactly like a cursor). Keep + // both provisional until every source in this page has been handled; + // all intermediate state writes must retain the replayable values. providerState.cursor = previousCursor; + providerState.adapterData = previousAdapterData; if (!discovered.value.every(validRemoteSource)) { return err(new Error(`integration provider ${adapter.name} returned an invalid source`)); } @@ -579,6 +655,13 @@ export async function reconcileProvider( const scopedSources = [...currentSources.values()]; for (const [index, remote] of scopedSources.entries()) { const providerSourceId = sourceIdentity(adapter.name, remote.id); + if (currentTime(deps).getTime() - cycleStart > limits.maxCycleMs) { + outcome.failedSourceIds.push(providerSourceId); + for (const remaining of scopedSources.slice(index + 1)) { + outcome.failedSourceIds.push(sourceIdentity(adapter.name, remaining.id)); + } + break; + } const previous = providerState.sources[remote.id]; const targetedWithoutDiscovery = hint.kind === "sources" && !hint.rediscover; if ( @@ -643,12 +726,14 @@ export async function reconcileProvider( const beforeDistill = writeState(vaultRoot, key.value, persisted.value, deps); if (!beforeDistill.ok) return beforeDistill; + const owner = owningEnrollment(providerState.enrollment, remote); let distilled: Result; try { distilled = await deps.distill({ providerSourceId, revision: fetched.value.revision, text: fetched.value.text, + ...(owner === undefined ? {} : { targetCollection: owner.targetCollection }), }); } catch { outcome.failedSourceIds.push(providerSourceId); @@ -670,6 +755,7 @@ export async function reconcileProvider( if (shouldDiscover && outcome.failedSourceIds.length === 0) { providerState.cursor = discoveredCursor; + providerState.adapterData = discoveredAdapterData; } const finalStateWritten = writeState(vaultRoot, key.value, persisted.value, deps); if (!finalStateWritten.ok) return finalStateWritten; diff --git a/src/integrations/queue.ts b/src/integrations/queue.ts index 4de6080f..ee25b048 100644 --- a/src/integrations/queue.ts +++ b/src/integrations/queue.ts @@ -91,7 +91,7 @@ function validQueueItem(value: unknown): value is IntegrationQueueItem { if (typeof value !== "object" || value === null) return false; const item = value as Record; return ( - (item.provider === "google" || item.provider === "notion") && + (item.provider === "google" || item.provider === "notion" || item.provider === "m365") && typeof item.eventId === "string" && item.eventId.length > 0 && validHint(item.hint) && diff --git a/src/integrations/routes.ts b/src/integrations/routes.ts index a924430d..9a0c859a 100644 --- a/src/integrations/routes.ts +++ b/src/integrations/routes.ts @@ -47,9 +47,16 @@ function writeJson(response: ServerResponse, status: number, body: unknown): voi response.end(JSON.stringify(body)); } -function providerFrom(pathname: string): ProviderName | null { - const matched = /^\/integrations\/(google|notion)(?:\/|$)/.exec(pathname); - return matched === null ? null : (matched[1] as ProviderName); +// Provider names come from the registered adapters, never a hardcoded list — +// serve stays provider-neutral and a new adapter needs no route change. +function providerFrom( + pathname: string, + adapters: Partial>, +): ProviderName | null { + const matched = /^\/integrations\/([a-z0-9-]+)(?:\/|$)/.exec(pathname); + if (matched === null) return null; + const name = matched[1] as ProviderName; + return adapters[name] === undefined ? null : name; } function nodeHeaders(request: IncomingMessage): WebhookRequest["headers"] { @@ -134,7 +141,7 @@ export async function handleIntegrationRoute( deps: IntegrationRouteDependencies, ): Promise { if (!url.pathname.startsWith("/integrations/")) return false; - const provider = providerFrom(url.pathname); + const provider = providerFrom(url.pathname, deps.adapters); if (provider === null) { writeJson(response, 404, { error: "not_found" }); return true; diff --git a/src/integrations/runtime.ts b/src/integrations/runtime.ts index eeb2e78f..40f02394 100644 --- a/src/integrations/runtime.ts +++ b/src/integrations/runtime.ts @@ -57,13 +57,16 @@ export interface ConfiguredIntegrationRuntimeOptions { waitForShutdown?: (cycle: Promise, timeoutMilliseconds: number) => Promise; } -const DEFAULT_FACTORIES: Record = { +// Partial: a provider name may exist in the type before its adapter ships. +const DEFAULT_FACTORIES: Partial> = { google: (redirectUri) => createGoogleDocsAdapter({ redirectUri }), notion: (redirectUri) => createNotionAdapter({ redirectUri }), }; function configuredProviders(config: IntegrationConfig): ProviderName[] { - return (["google", "notion"] as const).filter((provider) => config[provider] !== undefined); + return (["google", "notion", "m365"] as const).filter( + (provider) => config[provider] !== undefined, + ); } function callbackUrl(baseUrl: string, provider: ProviderName): string { @@ -266,12 +269,17 @@ export function createConfiguredIntegrationRuntime( const resolvedRouteBaseUrl = options.publicBaseUrl ?? fallback.value; routeBaseUrl = resolvedRouteBaseUrl; integrationRoutePrefix = routePrefix(resolvedRouteBaseUrl); - try { - adapters = providers.map((provider) => - factories[provider](callbackUrl(resolvedRouteBaseUrl, provider)), - ); - } catch { - return err(new Error("integration adapter construction failed")); + adapters = []; + for (const provider of providers) { + const factory = factories[provider]; + if (factory === undefined) { + return err(new Error(`integration provider ${provider} has no adapter factory`)); + } + try { + adapters.push(factory(callbackUrl(resolvedRouteBaseUrl, provider))); + } catch { + return err(new Error("integration adapter construction failed")); + } } for (const adapter of adapters) { const capability = validateContinuousAdapterCapabilities(adapter, { diff --git a/src/integrations/state.ts b/src/integrations/state.ts index def58742..01ecac65 100644 --- a/src/integrations/state.ts +++ b/src/integrations/state.ts @@ -5,6 +5,7 @@ import { createCipheriv, createDecipheriv, randomBytes } from "node:crypto"; import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs"; import { dirname, join } from "node:path"; +import { isValidCollectionName } from "../distill/propose.js"; import { err, ok, type Result } from "../frontmatter/types.js"; import type { IntegrationState, @@ -108,7 +109,7 @@ function asBase64( } function isProviderName(value: unknown): value is ProviderName { - return value === "google" || value === "notion"; + return value === "google" || value === "notion" || value === "m365"; } function isStringRecord(value: unknown): value is Record { @@ -131,6 +132,23 @@ function validSourceState(value: unknown): value is SourceState { ); } +function validEnrollmentRecord(value: unknown): boolean { + return ( + isStringRecord(value) && + typeof value.ref === "string" && + (value.kind === "file" || value.kind === "folder") && + typeof value.label === "string" && + // targetCollection is later joined into a staged file path (see + // src/distill/propose.ts derivePath) and separately checked against RBAC + // as an exact string — it must be a single safe path segment, not just a + // string, or the two checks could diverge (confused-deputy escalation). + typeof value.targetCollection === "string" && + isValidCollectionName(value.targetCollection) && + typeof value.enrolledAt === "string" && + typeof value.enrolledBy === "string" + ); +} + function validProviderState(value: unknown): value is ProviderState { if ( !isStringRecord(value) || @@ -146,6 +164,12 @@ function validProviderState(value: unknown): value is ProviderState { (value.webhookSetupToken !== undefined && value.webhookSetupToken.length < 16) ) return false; + if ( + value.enrollment !== undefined && + (!Array.isArray(value.enrollment) || !value.enrollment.every(validEnrollmentRecord)) + ) + return false; + if (value.adapterData !== undefined && !isStringRecord(value.adapterData)) return false; if (!isStringRecord(value.sources) || !Object.values(value.sources).every(validSourceState)) return false; if (value.webhook === undefined) return true; diff --git a/src/integrations/types.ts b/src/integrations/types.ts index 6908ba22..6478211f 100644 --- a/src/integrations/types.ts +++ b/src/integrations/types.ts @@ -1,7 +1,7 @@ // Shared, provider-neutral integration state. Source text never appears in // these types: connector state records only credentials and change metadata. -export type ProviderName = "google" | "notion"; +export type ProviderName = "google" | "notion" | "m365"; export interface IntegrationProviderConfig { clientIdEnv: string; @@ -13,6 +13,22 @@ export interface IntegrationConfig { pollingIntervalMinutes: number; google?: IntegrationProviderConfig; notion?: IntegrationProviderConfig; + m365?: IntegrationProviderConfig; +} + +// An operator's selected-source grant. Enrollment — not the provider token — +// is the privilege boundary for selected-source providers: discover() expands +// exactly this set and nothing else is ever fetched. +export interface EnrollmentRecord { + /** Provider-scoped source reference, e.g. "drive::". */ + ref: string; + kind: "file" | "folder"; + /** Display metadata for the operator UI only — never used for dispatch. */ + label: string; + targetCollection: string; + enrolledAt: string; + /** Authenticated principal who made the enrollment. */ + enrolledBy: string; } export interface SourceState { @@ -37,6 +53,14 @@ export interface ProviderState { expiresAt?: string; verificationRequired?: boolean; }; + /** Selected-source providers only; absent = discover() enumerates everything. */ + enrollment?: EnrollmentRecord[]; + /** + * Opaque adapter-owned change metadata (delta links, subscription ids). + * Held provisionally with the change cursor: only committed once every + * source in a discovery page has been handled. + */ + adapterData?: Record; sources: Record; } diff --git a/src/utils/config.ts b/src/utils/config.ts index 7e1c15d0..2c1d99b2 100644 --- a/src/utils/config.ts +++ b/src/utils/config.ts @@ -539,12 +539,13 @@ const RECOGNISED_INTEGRATIONS_KEYS = [ "polling_interval_minutes", "google", "notion", + "m365", ] as const; const RECOGNISED_INTEGRATION_PROVIDER_KEYS = ["client_id_env", "client_secret_env"] as const; const DEFAULT_INTEGRATION_POLLING_INTERVAL_MINUTES = 15; function validateIntegrationProvider( - provider: "google" | "notion", + provider: "google" | "notion" | "m365", raw: unknown, ): Result { const mapping = requireMapping(raw, `'integrations.${provider}'`); @@ -592,7 +593,7 @@ function validateIntegrations(raw: unknown): Result { expect(listed.value).toHaveLength(0); }); }); + +describe("refuseRawDistillOutput against an overridden collection (#506)", () => { + it("allows a normal synthesized proposal under a non-default collection", () => { + const r = refuseRawDistillOutput("sensitive-reports/m365-item/x--aabbccdd.md", { + collection: "sensitive-reports", + provenance: "synthesized", + status: "draft", + }); + expect(r.ok).toBe(true); + }); + + it("still refuses a top-level raw/ landing even when it came from a collection override", () => { + // The fence is purely structural (path + frontmatter.tier); it must fire + // identically regardless of which collection produced the path. + const r = refuseRawDistillOutput("raw/m365-item/x.md", { + collection: "raw", + provenance: "synthesized", + status: "draft", + }); + expect(r.ok).toBe(false); + }); +}); diff --git a/test/distill/idempotency.test.ts b/test/distill/idempotency.test.ts index a8cd5d81..f009573e 100644 --- a/test/distill/idempotency.test.ts +++ b/test/distill/idempotency.test.ts @@ -455,3 +455,50 @@ describe("distillUpsert (U5 idempotency)", () => { expect(action.rationale).toMatch(/Possible overlaps:/); }); }); + +describe("distillUpsert — target collection override (#506)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-distill-collection-")); + }); + + afterEach(() => { + rmSync(vault, { recursive: true, force: true }); + }); + + it("lands the new claim under the overridden collection", async () => { + const outcome = await distillUpsert(vault, { + sourceId: SOURCE_ID, + sourceContent: CONTENT_V1, + claims: [CLAIM_A], + runId: "run-1", + collection: "sensitive-reports", + }); + expect(outcome.ok).toBe(true); + if (!outcome.ok) return; + expect(outcome.value.created).toEqual([CLAIM_A.claim_key]); + + const listed = await listStagedActions(vault, "pending"); + expect(listed.ok).toBe(true); + if (!listed.ok) return; + const staged = listed.value.find((a) => a.runId === "run-1"); + expect(staged?.targetPath).toMatch(/^sensitive-reports\//); + }); + + it("lands under the default DISTILL_COLLECTION when no override is given", async () => { + const outcome = await distillUpsert(vault, { + sourceId: SOURCE_ID, + sourceContent: CONTENT_V1, + claims: [CLAIM_A], + runId: "run-2", + }); + expect(outcome.ok).toBe(true); + if (!outcome.ok) return; + const listed = await listStagedActions(vault, "pending"); + expect(listed.ok).toBe(true); + if (!listed.ok) return; + const staged = listed.value.find((a) => a.runId === "run-2"); + expect(staged?.targetPath).toMatch(/^distill\//); + }); +}); diff --git a/test/distill/propose.test.ts b/test/distill/propose.test.ts index 7e9d695c..6d7e5689 100644 --- a/test/distill/propose.test.ts +++ b/test/distill/propose.test.ts @@ -20,6 +20,7 @@ import { listStagedActions } from "../../src/curation/staged-actions.js"; import type { ClaimRunMeta, ExtractedClaim } from "../../src/distill/extract.js"; import { DISTILL_COLLECTION, + isValidCollectionName, type OverlapHint, type ProposeOutcome, proposeAllClaims, @@ -581,3 +582,145 @@ describe("proposeAllClaims (U4)", () => { expect(body).toContain("### Reader"); }); }); + +// ----------------------------------------------------------------------------- +// #506: an overridden target collection lands proposals and paths there +// ----------------------------------------------------------------------------- + +describe("proposeAllClaims — target collection override (#506)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-propose-collection-")); + }); + + afterEach(() => { + rmSync(vault, { recursive: true, force: true }); + }); + + it("lands proposals under an overridden collection when ids.collection is set", async () => { + const claim = makeClaim(); + + const outcome = await proposeAllClaims(vault, [claim], { + sourceId: "m365:drive:d1:item-1", + runId: "run-collection-override", + collection: "sensitive-reports", + }); + + expect(outcome.proposed).toBe(1); + const listed = await listStagedActions(vault, "pending"); + expect(listed.ok).toBe(true); + if (!listed.ok) return; + const action = listed.value[0]; + if (!action) throw new Error("expected a staged action"); + const diff = action.proposedDiff as Record; + const fm = diff.frontmatter as Record; + + expect(fm.collection).toBe("sensitive-reports"); + expect(action.targetPath).toMatch(/^sensitive-reports\//); + // Every other default is unaffected by the override. + expect(fm.status).toBe("draft"); + expect(fm.confidence).toBe("low"); + expect(fm.provenance).toBe("synthesized"); + }); + + it("defaults to DISTILL_COLLECTION when ids.collection is absent (unchanged callers)", async () => { + const claim = makeClaim(); + const outcome = await proposeAllClaims(vault, [claim], { + sourceId: "chat-export-2", + runId: "run-no-override", + }); + expect(outcome.proposed).toBe(1); + const listed = await listStagedActions(vault, "pending"); + expect(listed.ok && listed.value[0]?.targetPath).toMatch(new RegExp(`^${DISTILL_COLLECTION}/`)); + }); +}); + +describe("isValidCollectionName", () => { + it("accepts realistic collection names", () => { + for (const name of ["distill", "competitive-intel", "pricing", "moonshot", "_drafts"]) { + expect(isValidCollectionName(name)).toBe(true); + } + }); + + it("rejects path separators, traversal, and empty strings", () => { + for (const name of ["../secrets", "a/b", "a\\b", "", ".", ".."]) { + expect(isValidCollectionName(name)).toBe(false); + } + }); +}); + +describe("proposeAllClaims — update-in-place preserves the landed collection (security)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-propose-landed-collection-")); + }); + + afterEach(() => { + rmSync(vault, { recursive: true, force: true }); + }); + + it("stamps frontmatter.collection from the override path, not the current run's collection", async () => { + const claim = makeClaim(); + const landedPath = "old-collection/source-group/title--abcd1234.md"; + + const outcome = await proposeAllClaims( + vault, + [claim], + { + sourceId: "m365:drive:d1:item-1", + runId: "run-reenrolled", + // The enrollment's targetCollection changed since this claim landed. + collection: "new-collection", + }, + { [claim.claim_key]: landedPath }, + ); + + expect(outcome.proposed).toBe(1); + const listed = await listStagedActions(vault, "pending"); + expect(listed.ok).toBe(true); + if (!listed.ok) return; + const action = listed.value[0]; + if (!action) throw new Error("expected a staged action"); + + // Physical path is unchanged (still the prior landing spot). + expect(action.targetPath).toBe(landedPath); + // frontmatter.collection must describe where the file actually lives, + // not the batch's current collection — a mismatch here would make + // downstream RBAC/collection-scoped logic reason about the wrong grant. + const diff = action.proposedDiff as Record; + const fm = diff.frontmatter as Record; + expect(fm.collection).toBe("old-collection"); + }); +}); + +describe("proposeAllClaims — invalid collection is rejected (security)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-propose-bad-collection-")); + }); + + afterEach(() => { + rmSync(vault, { recursive: true, force: true }); + }); + + it("fails every claim in the batch instead of joining an unsafe collection into a path", async () => { + const claim = makeClaim(); + + const outcome = await proposeAllClaims(vault, [claim], { + sourceId: "m365:drive:d1:item-1", + runId: "run-bad-collection", + collection: "../../etc", + }); + + expect(outcome.proposed).toBe(0); + expect(outcome.results).toHaveLength(0); + expect(outcome.errors).toEqual([ + { claim_key: claim.claim_key, error: expect.stringContaining("invalid collection name") }, + ]); + const listed = await listStagedActions(vault, "pending"); + expect(listed.ok && listed.value).toHaveLength(0); + }); +}); diff --git a/test/integrations/distill.test.ts b/test/integrations/distill.test.ts index 819b987a..963c3d1d 100644 --- a/test/integrations/distill.test.ts +++ b/test/integrations/distill.test.ts @@ -350,3 +350,106 @@ describe("integration distillation", () => { expect(resolve).toHaveBeenCalledTimes(1); }); }); + +describe("integration distillation — target collection threading (#506)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-integration-distill-collection-")); + }); + + afterEach(() => rmSync(vault, { recursive: true, force: true })); + + it("forwards DistillationInput.targetCollection to the upsert collection field", async () => { + const upsert = vi.fn(async () => + ok({ + noop: false, + skipped: [], + updated: [], + created: ["anchor:claim-1"], + propose: null, + stateWritten: true, + }), + ); + const distill = createIntegrationDistill(vault, { + resolve: () => + ok({ + client: { complete: vi.fn(), completeJson: vi.fn(), completeWithTools: vi.fn() }, + config: { + model: "test-model", + maxLlmCalls: 2, + maxClaims: 3, + maxVerbatimChars: 100, + inCallInputCap: 64, + corroborationThreshold: 0.8, + }, + transport: "anthropic", + }), + extract: async () => ({ + claims: [ + { + claim_key: "anchor:claim-1", + statement: "Enrolled source claim.", + proposed_frontmatter: { title: "Enrolled source claim." }, + }, + ], + budget_exhausted: false, + llmCalls: 1, + chunkErrors: [], + }), + upsert, + now: () => new Date("2026-09-04T00:00:00.000Z"), + runNonce: () => "nonce", + }); + + const result = await distill({ + providerSourceId: "m365:drive:d1:item-1", + revision: "1", + text: "enrolled text", + targetCollection: "sensitive-reports", + }); + + expect(result.ok).toBe(true); + expect(upsert).toHaveBeenCalledWith( + vault, + expect.objectContaining({ collection: "sensitive-reports" }), + ); + }); + + it("forwards undefined collection when targetCollection is absent (google/notion unaffected)", async () => { + const upsert = vi.fn(async () => + ok({ + noop: false, + skipped: [], + updated: [], + created: [], + propose: null, + stateWritten: true, + }), + ); + const distill = createIntegrationDistill(vault, { + resolve: () => + ok({ + client: { complete: vi.fn(), completeJson: vi.fn(), completeWithTools: vi.fn() }, + config: { + model: "test-model", + maxLlmCalls: 2, + maxClaims: 3, + maxVerbatimChars: 100, + inCallInputCap: 64, + corroborationThreshold: 0.8, + }, + transport: "anthropic", + }), + extract: async () => ({ claims: [], budget_exhausted: false, llmCalls: 0, chunkErrors: [] }), + upsert, + now: () => new Date("2026-09-04T00:00:00.000Z"), + runNonce: () => "nonce", + }); + + await distill({ providerSourceId: "google:doc-1", revision: "1", text: "plain" }); + + const call = upsert.mock.calls[0]?.[1] as { collection?: string }; + expect(call.collection).toBeUndefined(); + }); +}); diff --git a/test/integrations/engine.test.ts b/test/integrations/engine.test.ts index 5387181d..d078ed0e 100644 --- a/test/integrations/engine.test.ts +++ b/test/integrations/engine.test.ts @@ -12,12 +12,14 @@ import { type ProviderAdapter, readProviderWebhookVerificationToken, reconcileProvider, + requireCollectionWriteAccess, startPeriodicIntegrationSync, validateContinuousAdapterCapabilities, verifyProviderWebhook, } from "../../src/integrations/engine.js"; import { readIntegrationState, writeIntegrationState } from "../../src/integrations/state.js"; import type { IntegrationConfig, ProviderState } from "../../src/integrations/types.js"; +import type { RoleConfig } from "../../src/utils/config.js"; import { sha256Hex } from "../../src/utils/hash.js"; const KEY = Buffer.alloc(32, 7); @@ -1320,3 +1322,375 @@ describe("provider reconciliation", () => { vi.useRealTimers(); }); }); + +describe("enrollment-scoped providers (#505)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-integration-enroll-")); + }); + + afterEach(() => { + rmSync(vault, { recursive: true, force: true }); + }); + + const m365Config: IntegrationConfig = { + encryptionKeyEnv: "DAFTARI_INTEGRATIONS_KEY", + pollingIntervalMinutes: 15, + m365: { clientIdEnv: "M365_CLIENT_ID", clientSecretEnv: "M365_CLIENT_SECRET" }, + }; + const m365Environment = { + DAFTARI_INTEGRATIONS_KEY: KEY.toString("base64"), + M365_CLIENT_ID: "m365-client-id", + M365_CLIENT_SECRET: "m365-client-secret", + }; + + function enrolledState(extra: Partial = {}): ProviderState { + return { + accessToken: "access", + refreshToken: "refresh", + enrollment: [ + { + ref: "file:f1", + kind: "file", + label: "F1.docx", + targetCollection: "distill", + enrolledAt: "2026-09-04T00:00:00.000Z", + enrolledBy: "user:test", + }, + { + ref: "folder:dir", + kind: "folder", + label: "Dir", + targetCollection: "distill", + enrolledAt: "2026-09-04T00:00:00.000Z", + enrolledBy: "user:test", + }, + ], + sources: {}, + ...extra, + }; + } + + // The fake selected-source adapter: discover() is DEFINED as expanding the + // enrollment set — enrolled files plus current folder descendants — so the + // engine's reconciliation machinery is exercised unchanged. + function enrolledAdapter( + folderChildren: string[], + overrides: Partial = {}, + ): ProviderAdapter { + return { + name: "m365", + authorizationUrl: () => "https://login.example/authorize", + exchangeCode: async () => ok({ accessToken: "access", refreshToken: "refresh" }), + resolveEnrollment: async (candidates) => ok(candidates), + discover: async (state) => { + const sources: { id: string; revision: string }[] = []; + for (const record of state.enrollment ?? []) { + if (record.kind === "file") { + sources.push({ id: record.ref, revision: "1" }); + } else { + for (const child of folderChildren) sources.push({ id: child, revision: "1" }); + } + } + return ok(sources); + }, + fetch: async (source) => + ok({ id: source.id, revision: source.revision, text: `text of ${source.id}` }), + ...overrides, + }; + } + + function m365Deps(overrides: Partial = {}): EngineDeps { + return { + config: m365Config, + environment: m365Environment, + adapters: {}, + now, + distill: async () => ok({ runId: "run-1" }), + ...overrides, + }; + } + + it("discover() expands the enrollment set and rides the unchanged engine", async () => { + expect( + writeIntegrationState(vault, { providers: { m365: enrolledState() }, oauthStates: {} }, KEY), + ).toEqual(ok(undefined)); + + const first = await reconcileProvider(vault, enrolledAdapter(["folder:dir/d1"]), m365Deps()); + expect(first.ok && first.value.distilledSourceIds).toEqual([ + "m365:file:f1", + "m365:folder:dir/d1", + ]); + + // A file leaving an enrolled folder stops being discovered and takes the + // existing unavailable path — no new engine mechanism. + const recorded: unknown[] = []; + const second = await reconcileProvider( + vault, + enrolledAdapter([]), + m365Deps({ + recordUnavailable: (event) => { + recorded.push(event); + return ok(undefined); + }, + }), + ); + expect(second.ok && second.value.unavailableSourceIds).toEqual(["m365:folder:dir/d1"]); + expect(second.ok && second.value.unchangedSourceIds).toEqual(["m365:file:f1"]); + expect(recorded).toMatchObject([ + { reason: "no_longer_discovered", providerSourceId: "m365:folder:dir/d1" }, + ]); + }); + + it("keeps adapterData replayable when a source in the discovery page fails", async () => { + expect( + writeIntegrationState( + vault, + { + providers: { + m365: enrolledState({ + cursor: "old-cursor", + adapterData: { deltaLinks: { drive: "old" } }, + }), + }, + oauthStates: {}, + }, + KEY, + ), + ).toEqual(ok(undefined)); + + const result = await reconcileProvider( + vault, + enrolledAdapter([], { + discover: async (state) => { + state.cursor = "new-cursor"; + state.adapterData = { deltaLinks: { drive: "new" } }; + return ok([ + { id: "file:f1", revision: "1" }, + { id: "folder:dir/d1", revision: "1" }, + ]); + }, + fetch: async (source) => + source.id === "file:f1" + ? ok({ id: source.id, revision: source.revision, text: "fine" }) + : err(new Error("transient fetch failure")), + }), + m365Deps(), + ); + + expect(result.ok && result.value.failedSourceIds).toEqual(["m365:folder:dir/d1"]); + const persisted = readIntegrationState(vault, KEY); + expect(persisted.value.providers.m365?.cursor).toBe("old-cursor"); + expect(persisted.value.providers.m365?.adapterData).toEqual({ deltaLinks: { drive: "old" } }); + }); + + it("commits adapterData with the cursor once every source in the page is handled", async () => { + expect( + writeIntegrationState( + vault, + { + providers: { + m365: enrolledState({ + cursor: "old-cursor", + adapterData: { deltaLinks: { drive: "old" } }, + }), + }, + oauthStates: {}, + }, + KEY, + ), + ).toEqual(ok(undefined)); + + const result = await reconcileProvider( + vault, + enrolledAdapter([], { + discover: async (state) => { + state.cursor = "new-cursor"; + state.adapterData = { deltaLinks: { drive: "new" } }; + return ok([{ id: "file:f1", revision: "1" }]); + }, + }), + m365Deps(), + ); + + expect(result.ok && result.value.distilledSourceIds).toEqual(["m365:file:f1"]); + const persisted = readIntegrationState(vault, KEY); + expect(persisted.value.providers.m365?.cursor).toBe("new-cursor"); + expect(persisted.value.providers.m365?.adapterData).toEqual({ deltaLinks: { drive: "new" } }); + }); + + it("maxCycleMs stops starting new fetches and leaves the remainder retryable", async () => { + expect( + writeIntegrationState( + vault, + { providers: { m365: enrolledState({ cursor: "old-cursor" }) }, oauthStates: {} }, + KEY, + ), + ).toEqual(ok(undefined)); + + let tick = 0; + const clock = () => new Date(Date.parse("2026-08-24T12:00:00.000Z") + 1000 * tick++); + const fetch = vi.fn(async () => err(new Error("must not fetch"))); + const result = await reconcileProvider( + vault, + enrolledAdapter([], { + discover: async () => + ok([ + { id: "file:f1", revision: "1" }, + { id: "folder:dir/d1", revision: "1" }, + ]), + fetch, + }), + m365Deps({ now: clock, reconcileLimits: { maxCycleMs: 500 } }), + ); + + expect(fetch).not.toHaveBeenCalled(); + expect(result.ok && result.value.failedSourceIds).toEqual([ + "m365:file:f1", + "m365:folder:dir/d1", + ]); + expect(readIntegrationState(vault, KEY).value.providers.m365?.cursor).toBe("old-cursor"); + }); +}); + +describe("target-collection plumbing (#506)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-integration-collection-")); + }); + + afterEach(() => { + rmSync(vault, { recursive: true, force: true }); + }); + + it("threads the owning EnrollmentRecord's targetCollection into distill()", async () => { + const state: ProviderState = { + accessToken: "access", + refreshToken: "refresh", + enrollment: [ + { + ref: "file:f1", + kind: "file", + label: "F1", + targetCollection: "sensitive-reports", + enrolledAt: "2026-09-04T00:00:00.000Z", + enrolledBy: "user:test", + }, + ], + sources: {}, + }; + expect( + writeIntegrationState(vault, { providers: { google: state }, oauthStates: {} }, KEY), + ).toEqual(ok(undefined)); + + const distill = vi.fn(async () => ok({ runId: "run-1" })); + const result = await reconcileProvider( + vault, + adapter({ + discover: async () => ok([{ id: "file:f1", revision: "1" }]), + fetch: async () => ok({ id: "file:f1", revision: "1", text: "enrolled text" }), + }), + deps({ distill }), + ); + + expect(result.ok && result.value.distilledSourceIds).toEqual(["google:file:f1"]); + expect(distill).toHaveBeenCalledWith( + expect.objectContaining({ targetCollection: "sensitive-reports" }), + ); + }); + + it("passes no targetCollection when the source has no enrollment (google/notion unaffected)", async () => { + expect( + writeIntegrationState( + vault, + { providers: { google: providerState() }, oauthStates: {} }, + KEY, + ), + ).toEqual(ok(undefined)); + + const distill = vi.fn(async () => ok({ runId: "run-1" })); + const result = await reconcileProvider( + vault, + adapter({ + discover: async () => ok([{ id: "doc-1", revision: "1" }]), + fetch: async () => ok({ id: "doc-1", revision: "1", text: "plain text" }), + }), + deps({ distill }), + ); + + expect(result.ok && result.value.distilledSourceIds).toEqual(["google:doc-1"]); + const call = distill.mock.calls[0]?.[0] as { targetCollection?: string }; + expect(call.targetCollection).toBeUndefined(); + }); + + it("resolves a folder descendant's owning collection via RemoteSource.enrolledRef", async () => { + const state: ProviderState = { + accessToken: "access", + refreshToken: "refresh", + enrollment: [ + { + ref: "folder:dir", + kind: "folder", + label: "Dir", + targetCollection: "team-notes", + enrolledAt: "2026-09-04T00:00:00.000Z", + enrolledBy: "user:test", + }, + ], + sources: {}, + }; + expect( + writeIntegrationState(vault, { providers: { google: state }, oauthStates: {} }, KEY), + ).toEqual(ok(undefined)); + + const distill = vi.fn(async () => ok({ runId: "run-1" })); + const result = await reconcileProvider( + vault, + adapter({ + discover: async () => + ok([{ id: "folder:dir/child-1", revision: "1", enrolledRef: "folder:dir" }]), + fetch: async (source) => ok({ id: source.id, revision: "1", text: "child text" }), + }), + deps({ distill }), + ); + + expect(result.ok && result.value.distilledSourceIds).toEqual(["google:folder:dir/child-1"]); + expect(distill).toHaveBeenCalledWith( + expect.objectContaining({ targetCollection: "team-notes" }), + ); + }); +}); + +describe("requireCollectionWriteAccess (#506)", () => { + it("allows a role whose write list includes the target collection", () => { + const result = requireCollectionWriteAccess( + { read: ["*"], write: ["sensitive-reports"] } as RoleConfig, + "sensitive-reports", + ); + expect(result.ok).toBe(true); + }); + + it("allows a role with a wildcard write grant", () => { + const result = requireCollectionWriteAccess( + { read: ["*"], write: ["*"] } as RoleConfig, + "anything", + ); + expect(result.ok).toBe(true); + }); + + it("refuses by name when the role cannot write the target collection", () => { + const result = requireCollectionWriteAccess( + { read: ["*"], write: ["distill"] } as RoleConfig, + "sensitive-reports", + ); + expect(result.ok).toBe(false); + expect(result.ok || result.error.message).toContain('"sensitive-reports"'); + }); + + it("refuses a null role (deny-all fallback)", () => { + const result = requireCollectionWriteAccess(null, "distill"); + expect(result.ok).toBe(false); + }); +}); diff --git a/test/integrations/queue.test.ts b/test/integrations/queue.test.ts index 8e70e55b..862403d5 100644 --- a/test/integrations/queue.test.ts +++ b/test/integrations/queue.test.ts @@ -198,4 +198,16 @@ describe("durable integration queue", () => { const queue = createIntegrationQueue(vault); expect(queue.pending().ok).toBe(false); }); + + it("accepts an m365 event without corrupting the rest of the queue (security)", () => { + const first = createIntegrationQueue(vault); + first.enqueue({ provider: "notion", eventId: "notion-evt", hint: { kind: "reconcile" } }); + first.enqueue({ provider: "m365", eventId: "m365-evt", hint: { kind: "reconcile" } }); + + const recovered = createIntegrationQueue(vault); + const pending = recovered.pending(); + expect(pending.ok).toBe(true); + if (!pending.ok) return; + expect(pending.value.map((item) => item.provider).sort()).toEqual(["m365", "notion"]); + }); }); diff --git a/test/integrations/review.test.ts b/test/integrations/review.test.ts index 5d011961..dc504ead 100644 --- a/test/integrations/review.test.ts +++ b/test/integrations/review.test.ts @@ -33,3 +33,37 @@ describe("integration unavailable review", () => { expect(readFileSync(derived, "utf8")).toBe("canonical derived knowledge\n"); }); }); + +describe("integration unavailable review reasons (#505)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-integration-review-reasons-")); + }); + + afterEach(() => rmSync(vault, { recursive: true, force: true })); + + it("accepts the widened reason vocabulary alongside existing entries", () => { + const legacy = { + idempotencyKey: "google:doc-1:rev-1", + providerSourceId: "google:doc-1", + reason: "no_longer_discovered" as const, + revision: "rev-1", + occurredAt: "2026-09-04T00:00:00.000Z", + }; + const widened = { + idempotencyKey: "m365:doc-2:rev-1", + providerSourceId: "m365:doc-2", + reason: "unenrolled" as const, + revision: "rev-1", + occurredAt: "2026-09-04T00:00:00.000Z", + }; + + expect(appendUnavailableReview(vault, legacy).ok).toBe(true); + expect(appendUnavailableReview(vault, widened).ok).toBe(true); + + const lines = readFileSync(integrationReviewPath(vault), "utf8").trim().split("\n"); + expect(lines).toHaveLength(2); + expect(JSON.parse(lines.at(-1) ?? "")).toMatchObject({ reason: "unenrolled" }); + }); +}); diff --git a/test/integrations/routes.test.ts b/test/integrations/routes.test.ts index f88c4558..ecff30b1 100644 --- a/test/integrations/routes.test.ts +++ b/test/integrations/routes.test.ts @@ -335,3 +335,63 @@ describe("integration routes", () => { } }); }); + +describe("provider-neutral route matching (#505)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-integration-routes-m365-")); + }); + + afterEach(() => rmSync(vault, { recursive: true, force: true })); + + it("routes any registered adapter and 404s unregistered provider names", async () => { + const m365 = adapter({ name: "m365" }); + const queue = createIntegrationQueue(vault, () => new Date("2026-08-24T12:00:00.000Z")); + const engineDeps: EngineDeps = { + config, + environment, + adapters: { m365 }, + distill: async () => ok({ runId: "run" }), + }; + const server = createServer((req, res) => { + const url = new URL(req.url ?? "/", "http://localhost"); + void handleIntegrationRoute(req, res, url, { + vaultRoot: vault, + config, + environment, + adapters: { m365 }, + engineDeps, + queue, + publicBaseUrl: "https://vault.example/daftari", + authorize: async () => ({ cookieAuthenticated: false, canManageIntegrations: true }), + admitPublic: () => () => undefined, + checkCsrf: () => null, + }).then((handled) => { + if (!handled) { + res.statusCode = 404; + res.end(); + } + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (typeof address !== "object" || address === null) throw new Error("missing address"); + const base = `http://127.0.0.1:${address.port}`; + try { + // The registered m365 adapter resolves; it fails later (no m365 OAuth + // config here), proving the matcher — not a hardcoded name list — routed. + const registered = await fetch(`${base}/integrations/m365/connect`, { + method: "POST", + redirect: "manual", + }); + expect(registered.status).toBe(503); + + // google is in the old hardcoded list but not registered on this server. + const unregistered = await fetch(`${base}/integrations/google/connect`, { method: "POST" }); + expect(unregistered.status).toBe(404); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + } + }); +}); diff --git a/test/integrations/runtime.test.ts b/test/integrations/runtime.test.ts index fa844278..98962d8d 100644 --- a/test/integrations/runtime.test.ts +++ b/test/integrations/runtime.test.ts @@ -160,6 +160,72 @@ describe("configured integration runtime", () => { expect(factorySpy).not.toHaveBeenCalled(); }); + it("activates m365 when configured with a matching adapter factory (#505/#506 follow-up)", async () => { + const spy = { discover: 0, ensure: 0 }; + const m365Factory = (redirectUri: string): ProviderAdapter => { + spy.redirect = redirectUri; + return { + name: "m365", + authorizationUrl: () => "https://login.example/authorize", + exchangeCode: async () => ok({ accessToken: "access", refreshToken: "refresh" }), + refreshTokens: async () => ok({ accessToken: "access", refreshToken: "refresh" }), + ensureWebhook: async () => { + spy.ensure += 1; + return ok({ id: "channel", secret: "secret" }); + }, + verifyWebhook: async () => + ok({ kind: "event", eventId: "evt", hint: { kind: "reconcile" } }), + discover: async (state) => { + spy.discover += 1; + state.cursor = "cursor"; + return ok([]); + }, + fetch: async () => err(new Error("not used")), + }; + }; + writeIntegrationState( + vault, + { + providers: { m365: { accessToken: "access", refreshToken: "refresh", sources: {} } }, + oauthStates: {}, + }, + KEY, + ); + const created = createConfiguredIntegrationRuntime({ + vaultRoot: vault, + config: { + ...config, + m365: { clientIdEnv: "M365_ID", clientSecretEnv: "M365_SECRET" }, + }, + environment: { ...environment, M365_ID: "id", M365_SECRET: "secret" }, + distill, + adapterFactories: { google: factory({ discover: 0, ensure: 0 }), m365: m365Factory }, + }); + expect(created.ok).toBe(true); + if (!created.ok) return; + expect(await created.value.start("http://127.0.0.1:8787")).toEqual(ok(undefined)); + await created.value.runOnce(); + expect(spy.discover).toBeGreaterThan(0); + await created.value.close(); + }); + + it("surfaces a clear error when m365 is configured without a matching factory", async () => { + const created = createConfiguredIntegrationRuntime({ + vaultRoot: vault, + config: { + ...config, + m365: { clientIdEnv: "M365_ID", clientSecretEnv: "M365_SECRET" }, + }, + environment: { ...environment, M365_ID: "id", M365_SECRET: "secret" }, + distill, + adapterFactories: { google: factory({ discover: 0, ensure: 0 }) }, + }); + expect(created.ok).toBe(true); + if (!created.ok) return; + const started = await created.value.start("http://127.0.0.1:8787"); + expect(started).toEqual(err(new Error("integration provider m365 has no adapter factory"))); + }); + it("surfaces safe lifecycle failures without provider response data", async () => { const messages: string[] = []; const created = createConfiguredIntegrationRuntime({ diff --git a/test/integrations/state.test.ts b/test/integrations/state.test.ts index 45b1a639..ba1edfde 100644 --- a/test/integrations/state.test.ts +++ b/test/integrations/state.test.ts @@ -120,3 +120,80 @@ describe("encrypted integration state", () => { expect(writeIntegrationState(vault, both, KEY).ok).toBe(false); }); }); + +describe("enrollment and adapter data state (#505)", () => { + let vault: string; + + beforeEach(() => { + vault = mkdtempSync(join(tmpdir(), "daftari-integration-state-enroll-")); + }); + + afterEach(() => { + rmSync(vault, { recursive: true, force: true }); + }); + + it("round-trips enrollment records and opaque adapter data", () => { + const input: IntegrationState = { + providers: { + m365: { + accessToken: "access-token", + refreshToken: "refresh-token", + enrollment: [ + { + ref: "drive:d1:item-1", + kind: "folder", + label: "Reports", + targetCollection: "distill", + enrolledAt: "2026-09-04T00:00:00.000Z", + enrolledBy: "user:me", + }, + ], + adapterData: { deltaLinks: { d1: "delta-token" } }, + sources: {}, + }, + }, + oauthStates: {}, + }; + expect(writeIntegrationState(vault, input, KEY)).toEqual(ok(undefined)); + expect(readIntegrationState(vault, KEY)).toEqual(ok(input)); + }); + + it("rejects a malformed enrollment record at the validation gate", () => { + const malformed = { + providers: { + m365: { + accessToken: "access-token", + refreshToken: "refresh-token", + enrollment: [{ ref: "drive:d1:item-1", kind: "everything" }], + sources: {}, + }, + }, + oauthStates: {}, + } as unknown as IntegrationState; + expect(writeIntegrationState(vault, malformed, KEY).ok).toBe(false); + }); + + it("rejects an enrollment record whose targetCollection is not a safe path segment (security)", () => { + const unsafe = { + providers: { + m365: { + accessToken: "access-token", + refreshToken: "refresh-token", + enrollment: [ + { + ref: "drive:d1:item-1", + kind: "folder", + label: "Reports", + targetCollection: "../../etc", + enrolledAt: "2026-09-04T00:00:00.000Z", + enrolledBy: "user:me", + }, + ], + sources: {}, + }, + }, + oauthStates: {}, + } as unknown as IntegrationState; + expect(writeIntegrationState(vault, unsafe, KEY).ok).toBe(false); + }); +}); diff --git a/test/utils/config.test.ts b/test/utils/config.test.ts index ed931140..5c58dea2 100644 --- a/test/utils/config.test.ts +++ b/test/utils/config.test.ts @@ -1190,6 +1190,23 @@ describe("loadConfig — integrations", () => { }); }); + it("accepts an m365 provider block (#505/#506 follow-up)", () => { + writeConfig( + "integrations:\n" + + " encryption_key_env: DAFTARI_INTEGRATIONS_KEY\n" + + " m365:\n" + + " client_id_env: M365_CLIENT_ID\n" + + " client_secret_env: M365_CLIENT_SECRET\n", + ); + const result = loadConfig(dir); + expect(result.ok).toBe(true); + if (!result.ok) return; + expect(result.value.integrations?.m365).toEqual({ + clientIdEnv: "M365_CLIENT_ID", + clientSecretEnv: "M365_CLIENT_SECRET", + }); + }); + it("rejects a client secret declared directly in YAML", () => { writeConfig( "integrations:\n" +