Method produceEventsFromDiff

server/src/crawler.ts:685–771

(
    preCrawlAgents: Map<string, { domains: Set<string> }>,
    profiles: Map<string, ProfileUpsertInput>
  )

  }

  private async produceEventsFromDiff(
    preCrawlAgents: Map<string, { domains: Set<string> }>,
    profiles: Map<string, ProfileUpsertInput>
  ): Promise<void> {
    if (!this.eventsDb) return;

    const events: WriteEventInput[] = [];
    const postCrawlAgents = await this.snapshotAgentState();

    // Detect new agents — include full profile in the event so RegistrySync
    // clients get a complete agent without waiting for next bootstrap
    for (const [url] of postCrawlAgents) {
      if (!preCrawlAgents.has(url)) {
        const profile = profiles.get(url);
        events.push({
          event_type: 'agent.discovered',
          entity_type: 'agent',
          entity_id: url,
          payload: {
            agent_url: url,
            ...(profile ? {
              channels: profile.channels,
              property_types: profile.property_types,
              markets: profile.markets,
              categories: profile.categories,
              tags: profile.tags,
              delivery_types: profile.delivery_types,
              property_count: profile.property_count,
              publisher_count: profile.publisher_count,
              has_tmp: profile.has_tmp,
            } : {}),
          },
          actor: 'pipeline:crawler',
        });
      }
    }

    // Detect removed agents (global aggregation: gone from ALL publishers)
    for (const [url] of preCrawlAgents) {
      if (!postCrawlAgents.has(url)) {
        events.push({
          event_type: 'agent.removed',
          entity_type: 'agent',
          entity_id: url,
          payload: { agent_url: url },
          actor: 'pipeline:crawler',
        });
      }
    }

    // Detect authorization changes per agent
    for (const [url, postState] of postCrawlAgents) {
      const preState = preCrawlAgents.get(url);
      const preDomains = preState?.domains ?? new Set<string>();

      // New authorizations
      for (const domain of postState.domains) {
        if (!preDomains.has(domain)) {