feat(integrations): shared-layer enrollment + target-collection plumbing (#505, #506)

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<ProposeOutcome> {
   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);

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,
   );

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) {

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<EnrollmentRecord, "enrolledAt" | "enrolledBy">;
+
 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<Result<VerifiedWebhook, Error>>;
+  // 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<Result<EnrollmentCandidate[], Error>>;
   discover(state: ProviderState): Promise<Result<RemoteSource[], Error>>;
   fetch(source: RemoteSource, state: ProviderState): Promise<Result<NormalizedRemoteSource, Error>>;
 }
@@ -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<EngineDeps, "reconcileLimits">): 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<void, Error> {
+  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<string, unknown> | undefined,
+): Record<string, unknown> | 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<RemoteSource[], Error>;
       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<DistillationRun, Error>;
         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;

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<string, unknown>;
   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) &&

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<Record<ProviderName, ProviderAdapter>>,
+): 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<boolean> {
   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;