From 58799d209b109e5c4e4d73718ec801657f4521f9 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 02:03:57 +0000 Subject: [PATCH 1/4] feat(integrations): provider-neutral enrollment in the shared connector layer (#505) Slice 1 of the accepted M365 ingestion design (#492): - EnrollmentRecord + optional enrollment/adapterData on ProviderState; "m365" joins ProviderName with an optional config block - optional ProviderAdapter.resolveEnrollment capability for selected-source providers - adapterData held provisionally with the change cursor: snapshotted before discovery, restored for intermediate writes, committed only when no source in the page failed - UnavailableSourceEvent.reason widens to no_longer_discovered | access_denied | deleted | unenrolled (backward-compatible JSONL) - optional maxCycleMs soft wall-time cap in ReconcileLimits: stop starting new fetches, leave the remainder retryable, cursor uncommitted - integration route matcher derives provider names from registered adapters instead of a hardcoded (google|notion) list - state validation covers the new fields; google/notion adapters unchanged Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01BgaRyc5Nji2GpMYwXEYpmk --- src/integrations/engine.ts | 52 ++++++- src/integrations/routes.ts | 15 +- src/integrations/runtime.ts | 20 ++- src/integrations/state.ts | 20 ++- src/integrations/types.ts | 26 +++- test/integrations/engine.test.ts | 231 +++++++++++++++++++++++++++++++ test/integrations/review.test.ts | 34 +++++ test/integrations/routes.test.ts | 60 ++++++++ test/integrations/state.test.ts | 53 +++++++ 9 files changed, 494 insertions(+), 17 deletions(-) diff --git a/src/integrations/engine.ts b/src/integrations/engine.ts index 8821433b..e49a8307 100644 --- a/src/integrations/engine.ts +++ b/src/integrations/engine.ts @@ -11,6 +11,7 @@ import { writeIntegrationState, } from "./state.js"; import type { + EnrollmentRecord, IntegrationConfig, IntegrationProviderConfig, ProviderName, @@ -74,6 +75,9 @@ 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; @@ -101,6 +105,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>; } @@ -115,10 +126,19 @@ 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 +158,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 +179,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 { @@ -230,6 +253,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 +533,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 +558,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 +615,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 ( @@ -670,6 +713,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/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..36039787 100644 --- a/src/integrations/runtime.ts +++ b/src/integrations/runtime.ts @@ -57,7 +57,8 @@ 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 }), }; @@ -266,12 +267,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..20d20f12 100644 --- a/src/integrations/state.ts +++ b/src/integrations/state.ts @@ -108,7 +108,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 +131,18 @@ 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" && + typeof value.targetCollection === "string" && + typeof value.enrolledAt === "string" && + typeof value.enrolledBy === "string" + ); +} + function validProviderState(value: unknown): value is ProviderState { if ( !isStringRecord(value) || @@ -146,6 +158,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/test/integrations/engine.test.ts b/test/integrations/engine.test.ts index 5387181d..abf2530a 100644 --- a/test/integrations/engine.test.ts +++ b/test/integrations/engine.test.ts @@ -1320,3 +1320,234 @@ 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"); + }); +}); 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/state.test.ts b/test/integrations/state.test.ts index 45b1a639..15081977 100644 --- a/test/integrations/state.test.ts +++ b/test/integrations/state.test.ts @@ -120,3 +120,56 @@ 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); + }); +}); From 4178890fb1fc890dabb2a69b27f016c0f0488b1e Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 02:17:39 +0000 Subject: [PATCH 2/4] feat(distill): thread targetCollection through the integration distill path (#506) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Slice 2 of the accepted M365 ingestion design (#492). Depends on #505's EnrollmentRecord/adapterData (already on this branch, PR #511). - DistillIds/DistillUpsertInput gain an optional collection override, defaulting to DISTILL_COLLECTION; propose.ts's derivePath and the frontmatter.collection field use it instead of the hardcoded constant. Every existing caller (no override passed) is unaffected. - DistillationInput gains targetCollection?: string. RemoteSource gains an optional enrolledRef so a folder-enrolled provider can attest which enrollment record owns a discovered descendant; the reconcile loop looks up the owning EnrollmentRecord (exact ref match, or enrolledRef for a folder descendant) and threads its targetCollection into deps.distill(). google/notion carry no enrollment, so they see no behavior change. - src/integrations/distill.ts forwards targetCollection into the distillUpsert collection field. - refuseRawDistillOutput needed no change: it is purely structural (path + frontmatter.tier), so it already fires identically against an overridden collection — added tests proving it. - New requireCollectionWriteAccess(role, targetCollection) reuses canWrite (the same stage-time write-gate rationale as vault_stage_action, docs/architecture.md): manage_integrations alone must not let an enrollment aim proposals at a collection the serve process cannot write. #509's enroll route will call this before persisting an EnrollmentRecord. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01BgaRyc5Nji2GpMYwXEYpmk --- src/distill/propose.ts | 18 +++- src/distill/state.ts | 8 +- src/integrations/distill.ts | 1 + src/integrations/engine.ts | 42 +++++++++ test/distill/fence.test.ts | 22 +++++ test/distill/idempotency.test.ts | 47 ++++++++++ test/distill/propose.test.ts | 53 +++++++++++ test/integrations/distill.test.ts | 103 +++++++++++++++++++++ test/integrations/engine.test.ts | 143 ++++++++++++++++++++++++++++++ 9 files changed, 432 insertions(+), 5 deletions(-) diff --git a/src/distill/propose.ts b/src/distill/propose.ts index b4566020..cca3dca2 100644 --- a/src/distill/propose.ts +++ b/src/distill/propose.ts @@ -132,6 +132,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). */ @@ -177,7 +185,7 @@ function hash8FromClaimKey(claimKey: string): string { // 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 { +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 +196,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 +401,11 @@ export async function proposeAllClaims( ): Promise { const results: ClaimProposalResult[] = []; const errors: Array<{ claim_key: string; error: string }> = []; + const collection = ids.collection ?? DISTILL_COLLECTION; 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); // R3: frontmatter is hardcoded to draft/low/synthesized. No caller can // override these — the emitter owns the invariant. @@ -407,7 +417,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, status: "draft", confidence: "low", provenance: "synthesized", 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 e49a8307..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, @@ -81,6 +83,14 @@ export type EnrollmentCandidate = Omit { + 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, @@ -686,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); diff --git a/test/distill/fence.test.ts b/test/distill/fence.test.ts index 2d799853..4bb250e6 100644 --- a/test/distill/fence.test.ts +++ b/test/distill/fence.test.ts @@ -114,3 +114,25 @@ describe("proposeAllClaims enforces the distill output fence (U11)", () => { 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..6631ef35 100644 --- a/test/distill/propose.test.ts +++ b/test/distill/propose.test.ts @@ -581,3 +581,56 @@ 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}/`)); + }); +}); 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 abf2530a..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); @@ -1551,3 +1553,144 @@ describe("enrollment-scoped providers (#505)", () => { 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); + }); +}); From 2ef4a30ccf25d857515872134ea103a12bb332c1 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 6 Sep 2026 00:03:07 +0000 Subject: [PATCH 3/4] fix(distill): reject unsafe collection names before path join MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Security Review bot on PR #511 flagged that derivePath() joins the caller-supplied `collection` string into the staged file path without sanitizing it, even though the comment above it claims all join components are traversal-safe. Before #506, `collection` was always the hardcoded DISTILL_COLLECTION constant; now it can come from EnrollmentRecord.targetCollection, which state.ts only checked with typeof === "string" — no charset restriction. Add isValidCollectionName() (single path segment: [A-Za-z0-9_-]+, no separators or traversal) and enforce it in two places: proposeAllClaims rejects the whole batch with a clear error rather than silently sanitizing (sanitizing could make the written path diverge from the string requireCollectionWriteAccess checks against RBAC), and validEnrollmentRecord rejects an unsafe targetCollection at the persistence boundary so bad state can never be written or read back. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01BgaRyc5Nji2GpMYwXEYpmk --- src/distill/propose.ts | 30 +++++++++++++++++++--- src/integrations/state.ts | 6 +++++ test/distill/propose.test.ts | 45 +++++++++++++++++++++++++++++++++ test/integrations/state.test.ts | 24 ++++++++++++++++++ 4 files changed, 102 insertions(+), 3 deletions(-) diff --git a/src/distill/propose.ts b/src/distill/propose.ts index cca3dca2..2fbce0b6 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 @@ -182,9 +194,11 @@ 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. +// 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); @@ -403,6 +417,16 @@ export async function proposeAllClaims( 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, collection); diff --git a/src/integrations/state.ts b/src/integrations/state.ts index 20d20f12..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, @@ -137,7 +138,12 @@ function validEnrollmentRecord(value: unknown): boolean { 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" ); diff --git a/test/distill/propose.test.ts b/test/distill/propose.test.ts index 6631ef35..7678d771 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, @@ -634,3 +635,47 @@ describe("proposeAllClaims — target collection override (#506)", () => { 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 — 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/state.test.ts b/test/integrations/state.test.ts index 15081977..ba1edfde 100644 --- a/test/integrations/state.test.ts +++ b/test/integrations/state.test.ts @@ -172,4 +172,28 @@ describe("enrollment and adapter data state (#505)", () => { } 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); + }); }); From 897a4b0eb81fa977505ffeeec4f7f6ca9612957f Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 6 Sep 2026 00:11:19 +0000 Subject: [PATCH 4/4] fix(integrations): wire m365 into config/queue gates, fix collection drift MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three more findings from the Claude Code Review pass on PR #511, all confirmed against the actual code path: - runtime.ts's configuredProviders() and config.ts's RECOGNISED_INTEGRATIONS_KEYS / per-provider parse loop still hardcoded ["google", "notion"], even though this PR widened ProviderName to include "m365" specifically so an operator could configure it ahead of #508's adapter landing. An integrations.m365 block in config.yaml was rejected as an unknown key before ever reaching the "missing factory" error this PR added for exactly that case. - queue.ts's validQueueItem still only accepted provider "google" | "notion". Once an m365 webhook event reached the durable queue, the next read of the whole pending array would fail as malformed, taking every other provider's queued events down with it. Mirrors the isProviderName fix already in state.ts. - proposeAllClaims stamped the current run's `collection` into every claim's frontmatter uniformly, including update-in-place (pathOverrides) claims whose targetPath is pinned to wherever a PRIOR run landed them. If an enrollment's targetCollection changes between runs, the physical path stays under the old collection while frontmatter.collection would claim the new one — a metadata/location mismatch that matters because collection drives RBAC downstream. Now derived from the landed path's own leading segment for update claims. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01BgaRyc5Nji2GpMYwXEYpmk --- src/distill/propose.ts | 13 +++++- src/integrations/queue.ts | 2 +- src/integrations/runtime.ts | 4 +- src/utils/config.ts | 5 ++- test/distill/propose.test.ts | 45 +++++++++++++++++++++ test/integrations/queue.test.ts | 12 ++++++ test/integrations/runtime.test.ts | 66 +++++++++++++++++++++++++++++++ test/utils/config.test.ts | 17 ++++++++ 8 files changed, 158 insertions(+), 6 deletions(-) diff --git a/src/distill/propose.ts b/src/distill/propose.ts index 2fbce0b6..503f9b66 100644 --- a/src/distill/propose.ts +++ b/src/distill/propose.ts @@ -430,6 +430,15 @@ export async function proposeAllClaims( for (const claim of claims) { 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. @@ -441,7 +450,7 @@ export async function proposeAllClaims( // missing `created` cannot be approved). created: ids.asOf ?? new Date().toISOString().slice(0, 10), domain: "accumulation", - collection, + collection: landedCollection, status: "draft", confidence: "low", provenance: "synthesized", @@ -478,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/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/runtime.ts b/src/integrations/runtime.ts index 36039787..40f02394 100644 --- a/src/integrations/runtime.ts +++ b/src/integrations/runtime.ts @@ -64,7 +64,9 @@ const DEFAULT_FACTORIES: Partial }; 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 { 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 { }); }); +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; 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/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/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" +