spb/social-runtime-crawler

1import fs from "node:fs";2import path from "node:path";3import {4  createLogger,5  nowIso,6  truncate,7  type AgentDecision,8  type AppConfig,9  type CrawlJob,10  type ObservedEntity,11  type PageState,12  type Platform,13} from "@src/shared";14import { EventBus, JsonlEventLog } from "@src/events";15import { SocialBrowserSession } from "@src/browser";16import { NetworkObserver, DomObserver, snapshotDom, classifyPage, type ClassifiedResponse } from "@src/observers";17import { mergeSurfaces } from "@src/entities";18import { detectVideos, sampleVideoFrames } from "@src/media";19import { WorldModel, LoopDetector, HeuristicPlanner, LlmPlanner, createLlmClient, actionKey, type Planner, type GainContext } from "@src/agent";20import { PlatformModel, compileConnector } from "@src/platform-model";21import { openStore, type PostgresStore } from "@src/storage";22import { getAdapter, entitiesFromDom, buildActions, pageFingerprint, type SocialPlatformAdapter } from "@src/connectors";23import { ActionExecutor } from "./executor.ts";2425const log = createLogger("engine");2627export interface CrawlSummary {28  session_id: string;29  platform: Platform;30  steps: number;31  entities: number;32  videos: number;33  network_responses: number;34  schemas: number;35  patterns_learned: number;36  ended_because: string;37  world: ReturnType<WorldModel["stats"]>;38  connector: ReturnType<PlatformModel["summary"]>;39  session_dir: string;40}4142/**43 * The crawl loop (§2): observe → summarize → plan → act → learn, until the budget or the goal ends it.44 * Every subsystem is a separate package; this file only orchestrates.45 */46export class CrawlEngine {47  private readonly bus = new EventBus();48  private readonly world = new WorldModel();49  private readonly loops = new LoopDetector();50  private readonly adapter: SocialPlatformAdapter;51  private readonly model: PlatformModel;52  private session!: SocialBrowserSession;53  private net!: NetworkObserver;54  private dom!: DomObserver;55  private store?: PostgresStore;56  private eventLog!: JsonlEventLog;57  private planner!: Planner;58  private executor!: ActionExecutor;59  private sessionDir!: string;60  private step = 0;61  private stepsWithoutNew = 0;62  private recentFailures = new Map<string, number>(); // action key → step of last failure63  private counts = { entities: 0, videos: 0, profiles: 0, posts: 0, videos_seen: 0, profiles_seen: 0, posts_seen: 0, patterns: 0, schemas: 0 };64  private degraded = false;6566  constructor(private readonly cfg: AppConfig, private readonly job: CrawlJob) {67    this.adapter = getAdapter(job.platform);68    this.model = new PlatformModel(job.platform, cfg.platformModelDir);69  }7071  async run(): Promise<CrawlSummary> {72    const { job, cfg } = this;73    this.session = new SocialBrowserSession({ platform: job.platform, accountAlias: job.account_alias, profilesDir: cfg.profilesDir, headless: cfg.headless, channel: cfg.browserChannel, bus: this.bus });74    this.sessionDir = path.join(cfg.sessionsDir, this.session.id);75    this.eventLog = new JsonlEventLog(this.sessionDir);76    this.eventLog.attach(this.bus);77    this.store = await openStore(cfg.databaseUrl);78    this.store?.attach(this.bus);79    fs.writeFileSync(path.join(this.sessionDir, "job.json"), JSON.stringify({ ...job, session_id: this.session.id }, null, 2));80    await this.store?.linkJob(job, this.session.id).catch(() => {});8182    if (!this.session.hasProfile()) log.warn(`No saved browser profile for ${job.platform}/${job.account_alias}. Run: pnpm login ${job.platform} --account ${job.account_alias}`);8384    const llm = createLlmClient(cfg);85    this.planner = llm ? new LlmPlanner(llm, new HeuristicPlanner(), job.budget.max_llm_tokens) : new HeuristicPlanner();86    log.info("planner", { planner: this.planner.name, mode: job.mode, goal: job.goal });8788    await this.session.start();89    // SESSION_STARTED is emitted by the session; enrich the DB row with mode/goal.90    this.bus.emit({ event_type: "PAGE_OPENED", platform: job.platform, session_id: this.session.id, step: 0, payload: { url: "about:blank", mode: job.mode, goal: job.goal, budget: job.budget } });91    const page = this.session.getPage();92    this.net = new NetworkObserver({ page, bus: this.bus, sessionId: this.session.id, platform: job.platform, hints: this.adapter.classifierHints });93    this.net.attach();94    this.dom = new DomObserver({ page, bus: this.bus, sessionId: this.session.id, platform: job.platform });95    await this.dom.attach();96    this.executor = new ActionExecutor(page, this.adapter, { stayOnPlatform: job.stay_on_platform });9798    const started = Date.now();99    let endedBecause = "unknown";100    try {101      // Seed navigation102      this.net.beginStep(0, "seed");103      this.dom.beginStep(0);104      const seed = job.seed_url ?? (job.query && job.mode !== "observe" ? this.adapter.searchUrl(job.query) : this.adapter.homeUrl);105      await this.session.navigate(seed);106      await page.waitForLoadState("networkidle", { timeout: 6000 }).catch(() => {});107      await this.adapter.dismissOverlays?.(page).catch(() => {});108      let state = await this.observe(await this.net.collectStep(), "seed");109      if (state.classification.page_type === "LOGIN") {110        this.session.setHealth("auth_required");111        this.bus.emit({ event_type: "AUTH_REQUIRED", platform: job.platform, session_id: this.session.id, step: 0, payload: { url: state.url, hint: `pnpm login ${job.platform} --account ${job.account_alias}` } });112        endedBecause = "auth_required";113        return this.finish(endedBecause, started);114      }115116      while (true) {117        const elapsedMin = (Date.now() - started) / 60_000;118        const b = job.budget;119        if (this.step >= b.max_actions) { endedBecause = "max_actions"; break; }120        if (elapsedMin >= b.max_minutes) { endedBecause = "max_minutes"; break; }121        if (this.counts.profiles >= b.max_profiles) { endedBecause = "max_profiles"; break; }122        if (this.counts.posts >= b.max_posts) { endedBecause = "max_posts"; break; }123        if (this.counts.videos >= b.max_videos) { endedBecause = "max_videos"; break; }124        if (this.session.getInfo().health === "crashed") { endedBecause = "browser_crashed"; break; }125126        this.step++;127        const decision = await this.plan(state);128        this.bus.emit({ event_type: "ACTION_PLANNED", platform: job.platform, session_id: this.session.id, step: this.step, payload: { action: decision.chosen_action, expected_information_gain: decision.expected_information_gain, novelty: decision.novelty, relevance: decision.relevance, reason: decision.reason, planner: decision.planner, top_scores: decision.scores.slice(0, 5) } });129        log.info(`step ${this.step}: ${decision.chosen_action.type} — ${truncate(decision.chosen_action.label, 80)}`, { gain: decision.expected_information_gain.toFixed(3), planner: decision.planner, reason: truncate(decision.reason, 100) });130        if (decision.chosen_action.type === "END_SESSION") { endedBecause = "agent_ended"; break; }131132        const before = { url: state.url, page_type: state.classification.page_type, visible_entities: state.entities.length, world_nodes: this.world.nodes.size };133        this.net.beginStep(this.step, decision.chosen_action.id);134        this.dom.beginStep(this.step);135        const target = decision.chosen_action.target_ref ? state.entities.find((e) => e.ref === decision.chosen_action.target_ref) : undefined;136        const result = await this.executor.execute(decision.chosen_action, { entityUrl: target?.url });137        if (!result.ok) this.recentFailures.set(actionKey(decision.chosen_action), this.step);138        for (const [k, s] of this.recentFailures) if (this.step - s > 8) this.recentFailures.delete(k);139        this.session.recordAction(decision.chosen_action.type, result.navigated && /^OPEN_/.test(decision.chosen_action.type) ? 1 : decision.chosen_action.type === "BACK" ? -1 : 0);140        if (target) this.world.markVisited(target.fingerprint, target.url);141        if (result.navigated) this.bus.emit({ event_type: "NAVIGATION_COMPLETED", platform: job.platform, session_id: this.session.id, step: this.step, payload: { from: result.url_before, to: result.url_after, action: decision.chosen_action.type, duration_ms: result.duration_ms } });142143        const responses = await this.net.collectStep();144        const domDeltas = this.dom.collectStep();145        const prevState = state;146        // Observation is bounded too (§66): a page that never settles must not stall the loop.147        state = await Promise.race([148          this.observe(responses, decision.chosen_action.id, target?.fingerprint),149          new Promise<PageState>((_, rej) => setTimeout(() => rej(new Error("observe timeout")), 45_000)),150        ]).catch((err: Error) => {151          log.warn("observation failed", { err: err.message });152          return prevState;153        });154        const newEntities = this.world.observe(state.entities, this.step, target?.fingerprint);155        this.stepsWithoutNew = newEntities.length ? 0 : this.stepsWithoutNew + 1;156        this.tally(newEntities);157        if (result.ok && result.navigated) this.tallyVisit(decision.chosen_action.type);158159        const after = { url: state.url, page_type: state.classification.page_type, visible_entities: state.entities.length, new_entities: newEntities.length, network_responses: responses.length, dom_nodes_added: domDeltas.reduce((s, d) => s + d.added, 0), world_nodes: this.world.nodes.size };160        this.bus.emit({ event_type: result.ok ? "ACTION_EXECUTED" : "ACTION_FAILED", platform: job.platform, session_id: this.session.id, step: this.step, payload: { action: decision.chosen_action, ok: result.ok, error: result.error, before, after, duration_ms: result.duration_ms }, discovered_via: { action_id: decision.chosen_action.id } });161        await this.store?.recordAction(this.session.id, decision, before, after, result.ok, result.error, result.duration_ms).catch((e) => log.debug("recordAction failed", { err: (e as Error).message }));162        await this.persistStep(state, newEntities, responses);163164        // Learning (§30, §71)165        const outcome = this.model.learnStep({ step: this.step, action_type: decision.chosen_action.type, action_id: decision.chosen_action.id, from_page_type: prevState.classification.page_type, to_page_type: state.classification.page_type, to_url: state.url, page_confidence: state.classification.confidence, entities_total: state.entities.length, new_entities: newEntities.length, entity_types: countTypes(state.entities), responses: responses.map((r) => ({ fingerprint: r.fingerprint, entity_count: r.entities.length, kind: r.kind, url: r.response.url, fields: r.schema?.fields.filter((f) => f.semantic[0]?.kind !== "unknown").slice(0, 60).map((f) => ({ path: f.path, semantic: f.semantic[0], example: f.examples[0] })) })), failed: !result.ok }, this.session.id);166        if (outcome.newPatterns.length) {167          this.counts.patterns += outcome.newPatterns.length;168          this.model.announce(this.bus, this.session.id, this.step, outcome.newPatterns);169        }170        this.model.save();171172        // Degradation / self-healing signal (§32)173        const degradedNow = this.model.isDegraded(state.classification.page_type, state.entities.length, this.adapter.expectedEntities(state.classification.page_type));174        if (degradedNow && !this.degraded) {175          this.degraded = true;176          this.bus.emit({ event_type: "CONNECTOR_DEGRADED", platform: job.platform, session_id: this.session.id, step: this.step, payload: { page_type: state.classification.page_type, expected: this.adapter.expectedEntities(state.classification.page_type), observed: 0, url: state.url, action: "switching to exploration heuristics" } });177        } else if (!degradedNow && this.degraded && state.entities.length > 0) {178          this.degraded = false;179          this.bus.emit({ event_type: "CONNECTOR_REPAIRED", platform: job.platform, session_id: this.session.id, step: this.step, payload: { page_type: state.classification.page_type, observed: state.entities.length } });180        }181182        // Loop safety (§41)183        this.loops.record(state.fingerprint, `${decision.chosen_action.type}:${decision.chosen_action.target_url ?? ""}`);184        const loop = this.loops.detect();185        if (loop.loop) {186          this.bus.emit({ event_type: "LOOP_DETECTED", platform: job.platform, session_id: this.session.id, step: this.step, payload: { pattern: loop.pattern, url: state.url } });187          log.warn("navigation loop detected — breaking out", { pattern: loop.pattern });188          await this.executor.execute({ id: "loopbreak", type: "RETURN_TO_FEED", label: "loop break", cost: 1, target_url: job.query ? this.adapter.searchUrl(job.query) : this.adapter.homeUrl }, {});189          state = await this.observe(await this.net.collectStep(), "loopbreak");190        }191        if (state.classification.page_type === "LOGIN") {192          this.session.setHealth("auth_required");193          this.bus.emit({ event_type: "AUTH_REQUIRED", platform: job.platform, session_id: this.session.id, step: this.step, payload: { url: state.url } });194          endedBecause = "auth_required";195          break;196        }197        if (this.stepsWithoutNew >= 10) { endedBecause = "no_new_information"; break; }198      }199    } catch (err) {200      endedBecause = `error: ${(err as Error).message.split("\n")[0]}`;201      log.error("engine error", { err: (err as Error).stack?.split("\n").slice(0, 3).join(" | ") });202      this.bus.emit({ event_type: "WORKER_ERROR", platform: job.platform, session_id: this.session.id, step: this.step, payload: { error: (err as Error).message } });203    }204    return this.finish(endedBecause, started);205  }206207  /** Observe the current page across surfaces and build the compact PageState (§9, §14). */208  private async observe(responses: ClassifiedResponse[], via: string, discoveredFrom?: string): Promise<PageState> {209    const page = this.session.getPage();210    const snapshot = await snapshotDom(page).catch((err: Error) => {211      log.warn("dom snapshot failed", { err: err.message.split("\n")[0] });212      return undefined;213    });214    if (!snapshot) {215      return { url: page.url(), title: "", platform: this.job.platform, classification: { page_type: "UNKNOWN", confidence: 0.1, signals: ["snapshot failed"] }, entities: [], actions: [{ id: "A1", type: "WAIT_FOR_CONTENT", label: "Wait for content", cost: 1 }, { id: "A2", type: "RETURN_TO_FEED", label: "Return to feed", cost: 2, target_url: this.adapter.homeUrl }, { id: "A3", type: "END_SESSION", label: "End", cost: 0.5 }], media: [], summary_text: "", fingerprint: pageFingerprint(page.url(), []), captured_at: nowIso() };216    }217    const classification = classifyPage(snapshot, this.adapter.pageTypeHints);218    const domEntities = entitiesFromDom(snapshot, this.adapter);219    const netEntities = responses.flatMap((r) => r.entities);220    const merged = mergeSurfaces(netEntities, domEntities);221    // Keep the planner's list focused: DOM-visible entities first, then network-only ones that have a page URL.222    const entities = merged.merged.filter((e) => e.provenance.some((p) => p.surface === "dom") || e.url).slice(0, 80);223    const media = detectVideos({ platform: this.job.platform, snapshot, entities, responses, pageUrl: snapshot.url });224    const visited = new Set([...this.world.nodes.values()].filter((n) => n.visited).map((n) => n.fingerprint));225    const actions = buildActions(entities, snapshot, this.adapter, { query: this.job.query, visited });226    const summary = renderSummary(snapshot.title, classification.page_type, entities, actions);227    const state: PageState = { url: snapshot.url, title: snapshot.title, platform: this.job.platform, classification, entities, actions, media, summary_text: summary, fingerprint: pageFingerprint(snapshot.url, entities), captured_at: nowIso() };228229    this.bus.emit({ event_type: "PAGE_OPENED", platform: this.job.platform, session_id: this.session.id, step: this.step, payload: { url: snapshot.url, title: snapshot.title, page_type: classification.page_type, confidence: classification.confidence, signals: classification.signals, entities: entities.length, dom_entities: domEntities.length, network_entities: netEntities.length, both_surfaces: merged.both, field_agreements: merged.field_agreements, field_conflicts: merged.field_conflicts.slice(0, 5), videos: media.length, actions: actions.length, summary }, provenance: [{ surface: "dom", confidence: classification.confidence }], discovered_via: { action_id: via } });230    this.bus.emit({ event_type: "PAGE_CLASSIFIED", platform: this.job.platform, session_id: this.session.id, step: this.step, payload: { url: snapshot.url, ...classification } });231    for (const e of entities) {232      const type = e.type === "video" ? "VIDEO_DISCOVERED" : e.type === "post" ? "POST_DISCOVERED" : e.type === "comment" ? "COMMENT_DISCOVERED" : e.type === "profile" || e.type === "channel" || e.type === "person" ? "PROFILE_DISCOVERED" : "ENTITY_OBSERVED";233      const known = this.world.nodes.has(e.fingerprint);234      this.bus.emit({ event_type: known ? "ENTITY_OBSERVED" : type, platform: this.job.platform, session_id: this.session.id, step: this.step, payload: { entity: { type: e.type, platform_id: e.platform_id, url: e.url, name: e.name, text: e.text, author: e.author, metrics: e.metrics, media: e.media, context: e.context, fingerprint: e.fingerprint, canonical_id: null }, fields: e.fields }, provenance: e.provenance, discovered_via: { action_id: via, url: snapshot.url } });235    }236    for (const m of media) if (m.width || m.duration_s || m.delivery) this.bus.emit({ event_type: "MEDIA_DISCOVERED", platform: this.job.platform, session_id: this.session.id, step: this.step, payload: { ...m } as Record<string, unknown>, provenance: m.provenance, discovered_via: { action_id: via } });237    if (this.step === 0) this.world.observe(entities, 0, discoveredFrom);238    return state;239  }240241  private async plan(state: PageState): Promise<AgentDecision> {242    // Relevance is measured against the goal *and* the topic query (the query carries the domain words).243    const goal = this.job.query ? `${this.job.goal} ${this.job.query}` : this.job.goal;244    const ctx: GainContext = { goal, mode: this.job.mode, world: this.world, state, recentActionTypes: this.loops.recentActionTypes(), recentUrls: [...this.world.visitedUrls].slice(-10), stepsWithoutNewEntities: this.stepsWithoutNew, learnedYield: this.model.learnedYield(), recentFailures: new Set(this.recentFailures.keys()) };245    return this.planner.plan(ctx, this.step);246  }247248  /** Discovery counters (what we have seen). */249  private tally(fresh: ObservedEntity[]): void {250    for (const e of fresh) {251      this.counts.entities++;252      if (e.type === "video") this.counts.videos_seen++;253      else if (e.type === "post" || e.type === "comment") this.counts.posts_seen++;254      else if (e.type === "profile" || e.type === "channel" || e.type === "person" || e.type === "page") this.counts.profiles_seen++;255    }256  }257258  /** Budget counters (§40) count pages we actually opened, not entities merely seen in a feed. */259  private tallyVisit(actionType: string): void {260    if (/^OPEN_(CHANNEL|PROFILE|PAGE)$/.test(actionType)) this.counts.profiles++;261    else if (/^OPEN_(POST|COMMENTS|ENTITY)$/.test(actionType)) this.counts.posts++;262    else if (actionType === "OPEN_VIDEO") this.counts.videos++;263  }264265  private async persistStep(state: PageState, fresh: ObservedEntity[], responses: ClassifiedResponse[]): Promise<void> {266    const ts = nowIso();267    const frames: Record<string, string[]> = {};268    if (this.job.media_level >= 2) {269      const current = state.media.find((m) => m.width || m.delivery?.hostnames.length);270      if (current) {271        const files = await sampleVideoFrames(this.session.getPage(), current, this.cfg.mediaDir, this.job.media_level).catch(() => []);272        if (files.length) {273          frames[current.fingerprint] = files;274          this.bus.emit({ event_type: "VIDEO_DISCOVERED", platform: this.job.platform, session_id: this.session.id, step: this.step, payload: { fingerprint: current.fingerprint, title: current.title, frames: files, level: this.job.media_level }, provenance: [{ surface: "visual", confidence: 0.8 }] });275        }276      }277    }278    if (!this.store) return;279    try {280      await this.store.upsertEntities(state.entities, this.session.id, this.step, ts);281      await this.store.upsertMedia(state.media, frames, ts);282      await this.store.recordFeedItems(this.session.id, this.step, state, ts);283      const recentEdges = this.world.edges.filter((e) => e.step === this.step);284      await this.store.recordRelationships(this.session.id, recentEdges);285    } catch (err) {286      log.debug("persist failed", { err: (err as Error).message, fresh: fresh.length, responses: responses.length });287    }288  }289290  private async finish(endedBecause: string, started: number): Promise<CrawlSummary> {291    this.model.save();292    const connector = this.model.summary();293    if (connector.confidence >= 40) {294      const manifest = compileConnector(this.model, path.join(process.cwd(), "connectors", this.job.platform));295      await this.store?.recordConnectorVersion(this.job.platform, connector, manifest).catch(() => {});296    }297    const world = this.world.stats();298    fs.writeFileSync(path.join(this.sessionDir, "world_model.json"), JSON.stringify(this.world.toJSON(), null, 2));299    const summary: CrawlSummary = { session_id: this.session.id, platform: this.job.platform, steps: this.step, entities: this.world.nodes.size, videos: this.counts.videos_seen, network_responses: this.net?.totalResponses ?? 0, schemas: this.net?.seenShapeHashes.size ?? 0, patterns_learned: this.counts.patterns, ended_because: endedBecause, world, connector, session_dir: this.sessionDir };300    this.bus.emit({ event_type: "BUDGET_EXHAUSTED", platform: this.job.platform, session_id: this.session.id, step: this.step, payload: { reason: endedBecause, elapsed_ms: Date.now() - started } });301    await this.session.stop();302    await this.bus.flush();303    fs.writeFileSync(path.join(this.sessionDir, "summary.json"), JSON.stringify(summary, null, 2));304    await this.eventLog.close();305    await this.store?.close();306    return summary;307  }308}309310function countTypes(entities: ObservedEntity[]): Record<string, number> {311  const out: Record<string, number> = {};312  for (const e of entities) out[e.type] = (out[e.type] ?? 0) + 1;313  return out;314}315316/** Textual semantic page representation (§14) — what a planner (or a human in the dashboard) reads. */317export function renderSummary(title: string, pageType: string, entities: ObservedEntity[], actions: PageState["actions"]): string {318  const lines = [`PAGE: ${pageType} — ${truncate(title, 90)}`, "", "VISIBLE ENTITIES"];319  for (const e of entities.slice(0, 25)) {320    lines.push(`[${e.ref}] ${e.type}${e.context ? ` (${e.context})` : ""}`);321    if (e.name) lines.push(`    Name: ${truncate(e.name, 100)}`);322    if (e.author) lines.push(`    Author: ${truncate(e.author, 60)}`);323    if (e.media?.has_video) lines.push(`    Video: yes${e.media.duration_s ? ` (${Math.round(e.media.duration_s)}s)` : ""}`);324    if (e.metrics && Object.keys(e.metrics).length) lines.push(`    Metrics: ${Object.entries(e.metrics).map(([k, v]) => `${k}=${v}`).join(", ")}`);325    lines.push(`    Sources: ${[...new Set(e.provenance.map((p) => p.surface))].join("+")}`);326  }327  if (entities.length > 25) lines.push(`… ${entities.length - 25} more`);328  lines.push("", "ACTIONS");329  for (const a of actions) lines.push(`[${a.id}] ${a.label}`);330  return lines.join("\n");331}332