all repos — sz-chrome-remote-cdp @ master

pi-browser-harness, patched to drive a Chrome on another host over CDP, with castle_cdp_search web search

castle-search/src/search.ts (view raw)

 1
 2
 3
 4
 5
 6
 7
 8
 9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
 100
 101
 102
 103
 104
 105
 106
 107
 108
 109
 110
 111
 112
 113
 114
 115
 116
 117
 118
 119
 120
 121
 122
 123
 124
 125
 126
 127
 128
 129
 130
 131
 132
 133
 134
 135
 136
 137
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
/**
 * Search orchestration: resolve a browser, borrow a page, navigate, extract,
 * hand the page back.
 *
 * The connection is process-wide and lazy. It is *not* opened by the extension
 * factory — pi runs factories in invocations that never start a session, and
 * the docs are explicit that sockets must not start there. It is opened on the
 * first search and closed by an idempotent `session_shutdown`.
 */

import { isGoogleGoto } from "./google-links.ts";
import { GoogleVerificationError } from "./errors.ts";
import { BrowserConnection, PageSession, raceAbort } from "./cdp.ts";
import {
  type CdpTarget,
  describeCdpTarget,
  describeResolutionFailure,
  resolveCdpTarget,
} from "./endpoint.ts";
import { describeEngineFailure, describeEngineSource, resolveEngine } from "./engine-target.ts";
import { readStoredEngine } from "./engine-store.ts";
import {
  ENGINE_IDS as allEngineIdList,
  type EngineDefinition,
  type EngineId,
  type Recency,
  describeRecencySupport,
  enginesSupportingRecency,
  getEngine,
} from "./engines.ts";

const allEngineIds = (): readonly EngineId[] => allEngineIdList;
import {
  BrowserUnavailableError,
  NoResultsError,
  RecencyUnsupportedError,
  SearchChallengeError,
} from "./errors.ts";
import { buildProbeScript, type PageProbe, type SearchResult } from "./extract.ts";
import { readStoredTarget } from "./target-store.ts";

/** Per-search wall clock, covering connect, navigate and extract together. */
export const SEARCH_TIMEOUT_MS = 20_000;
/** How many searches may drive the shared browser at once. */
export const MAX_CONCURRENT_SEARCHES = 2;
/** Results sometimes land a beat after the load event; re-read a couple of times. */
const EXTRACT_RETRIES = 2;
const EXTRACT_RETRY_DELAY_MS = 700;

export type { Recency } from "./engines.ts";

export type SearchRequest = {
  query: string;
  numResults: number;
  recency?: Recency | undefined;
  /** Per-call engine override; falls back to the configured default. */
  engine?: string | null | undefined;
};

export type SearchOutcome = {
  results: SearchResult[];
  searchUrl: string;
  /** The URL the page settled on, which differs from searchUrl after a redirect. */
  finalUrl: string;
  endpoint: string;
  engine: EngineDefinition;
  /** Where the engine choice came from, for the "via" line in the output. */
  engineSource: string;
};

/** Build the SERP URL for a request. Exported for tests. */
export const buildSearchUrl = (engine: EngineDefinition, request: SearchRequest): string =>
  engine.buildUrl(request.query, request.numResults, request.recency);

/**
 * Resolve the engine, refusing rather than silently ignoring a recency window
 * the chosen engine cannot express.
 *
 * Dropping an unsupported filter would be the same failure mode as Google's
 * `tbs=qdr:` — the caller asks for recent results, gets whatever the engine
 * ranked, and has no way to tell. Naming the engines that *do* support the
 * window makes the error actionable, since the model can retry with one.
 */
export const selectEngine = async (
  request: SearchRequest,
): Promise<{ engine: EngineDefinition; source: string }> => {
  const resolution = resolveEngine(request.engine, process.env, await readStoredEngine());
  if (resolution.kind === "invalid") {
    throw new BrowserUnavailableError(describeEngineFailure(resolution));
  }
  const engine = getEngine(resolution.id);

  if (request.recency && engine.recency[request.recency] === undefined) {
    const alternatives = enginesSupportingRecency(request.recency);
    throw new RecencyUnsupportedError(
      `${engine.label} cannot restrict results to the past ${request.recency}. ` +
        `It supports: ${describeRecencySupport(engine)}. ` +
        (alternatives.length > 0
          ? `Engines that support "${request.recency}": ${alternatives.join(", ")}. ` +
            `Retry with engine set to one of those, or drop the recency parameter.`
          : `No configured engine supports that window; drop the recency parameter.`),
    );
  }

  return { engine, source: describeEngineSource(resolution.source) };
};

/** For status reporting — resolves the configured engine without searching. */
export const describeConfiguredEngine = async (): Promise<{ id: EngineId; source: string; label: string }> => {
  const resolution = resolveEngine(null, process.env, await readStoredEngine());
  if (resolution.kind === "invalid") {
    return { id: "google", source: describeEngineFailure(resolution), label: "unresolved" };
  }
  return {
    id: resolution.id,
    source: describeEngineSource(resolution.source),
    label: getEngine(resolution.id).label,
  };
};

// ── connection lifecycle ───────────────────────────────────────────────────

let connection: BrowserConnection | null = null;
/** In-flight connect, so two concurrent searches share one dial rather than racing. */
let connecting: Promise<BrowserConnection> | null = null;

const resolveTarget = async (): Promise<CdpTarget> => {
  const resolution = resolveCdpTarget(process.env, await readStoredTarget());
  if (resolution.kind === "invalid") {
    throw new BrowserUnavailableError(describeResolutionFailure(resolution));
  }
  return resolution.target;
};

/**
 * Return a live connection, dialling one if there isn't one or the last one
 * died. Reconnecting transparently matters here: castle recycles Chrome hourly,
 * so a session that searched an hour ago is holding a dead socket.
 */
const ensureConnection = async (signal?: AbortSignal): Promise<BrowserConnection> => {
  if (connection && connection.isOpen) return connection;
  if (connection) {
    connection.close();
    connection = null;
  }

  if (!connecting) {
    // Assigned synchronously, before the first await inside: two searches
    // starting in the same tick must share one dial, not open two sockets and
    // leak whichever loses the assignment race.
    connecting = (async () => {
      const target = await resolveTarget();
      const conn = await BrowserConnection.connect(target);
      connection = conn;
      return conn;
    })().finally(() => {
      connecting = null;
    });
  }

  // The caller's signal is raced here rather than passed into connect(). A
  // shared dial must not be cancelled by whichever caller happened to start it
  // — one search pressing Esc would otherwise fail the other search waiting on
  // the same connection. connect() is bounded by its own timeouts regardless.
  return raceAbort(connecting, signal);
};

/** Idempotent: safe to call from `session_shutdown` however many times it fires. */
export const disconnect = (): void => {
  connection?.close();
  connection = null;
};

/** For status reporting — never dials. */
export const describeConfiguredEndpoint = async (): Promise<string> => {
  const resolution = resolveCdpTarget(process.env, await readStoredTarget());
  return resolution.kind === "invalid"
    ? describeResolutionFailure(resolution)
    : describeCdpTarget(resolution.target);
};

// ── concurrency cap ────────────────────────────────────────────────────────

let active = 0;
const waiting: Array<() => void> = [];

/** Give a held slot to the next waiter, or return it to the pool. */
const handOff = (): void => {
  const next = waiting.shift();
  // Hand the slot straight over rather than decrementing and letting the waiter
  // re-check, so a queued search cannot be overtaken by a newly arriving one.
  if (next) next();
  else active--;
};

const acquire = async (signal?: AbortSignal): Promise<() => void> => {
  if (active < MAX_CONCURRENT_SEARCHES) {
    active++;
  } else {
    let resolver!: () => void;
    const queued = new Promise<void>((resolve) => {
      resolver = resolve;
    });
    waiting.push(resolver);
    try {
      await raceAbort(queued, signal);
      // The slot was handed over by handOff(), which already counted it.
    } catch (e) {
      // An abort while queued must not strand the slot. Either we are still in
      // the queue — drop out of it — or the slot was handed to us in the same
      // tick the abort landed, in which case give it back. Without this, two
      // cancelled-while-queued searches would wedge the semaphore permanently.
      const index = waiting.indexOf(resolver);
      if (index !== -1) waiting.splice(index, 1);
      else handOff();
      throw e;
    }
  }
  let released = false;
  return () => {
    if (released) return;
    released = true;
    handOff();
  };
};

// ── the search itself ──────────────────────────────────────────────────────

export type ProgressReporter = (phase: SearchPhase, note: string) => void;
export type SearchPhase = "queued" | "connecting" | "opening" | "navigating" | "extracting";

const sleep = (ms: number, signal?: AbortSignal): Promise<void> =>
  raceAbort(new Promise<void>((resolve) => setTimeout(resolve, ms)), signal);

/**
 * The other engines, named in failure messages. A blocked engine is the case
 * where the agent most needs to know it has somewhere else to go, and telling
 * it inside the error beats hoping it remembers the guidelines.
 */
const otherEngines = (current: EngineId): readonly string[] =>
  allEngineIds().filter((id) => id !== current);

/**
 * Run one search. Throws on every failure — see errors.ts for why a returned
 * error object would be worse than useless here.
 */
export const search = async (
  request: SearchRequest,
  signal: AbortSignal,
  report: ProgressReporter,
): Promise<SearchOutcome> => {
  // Reported before acquiring, so a search that dies waiting for a slot is
  // distinguishable from one that died reaching the browser. The deadline is
  // the caller's and covers the queue wait too — with a cap of two and searches
  // that take a couple of seconds, waiting is the rare case, and a tool call
  // that silently takes twice its stated deadline would be worse.
  // Resolved before acquiring a slot: an unknown engine name or an unsupported
  // recency window is a caller error, and should not wait behind other searches
  // or touch the browser at all.
  const { engine, source: engineSource } = await selectEngine(request);
  const probeScript = buildProbeScript(engine.extraction);

  report("queued", "waiting for a search slot");
  const release = await acquire(signal);
  try {
    report("connecting", "resolving browser");
    const conn = await ensureConnection(signal);
    const endpoint = describeCdpTarget(conn.target);

    report("opening", `opening a page on ${endpoint}`);
    const page = await PageSession.open(conn, signal);
    const searchUrl = buildSearchUrl(engine, request);

    try {
      report("navigating", `searching ${engine.label} for "${request.query}"`);
      await page.navigate(searchUrl, signal);

      report("extracting", "reading results");
      let probe = await page.evaluate<PageProbe>(probeScript, signal);
      for (let attempt = 0; attempt < EXTRACT_RETRIES; attempt++) {
        if (probe?.challenge || (probe?.results?.length ?? 0) > 0) break;
        await sleep(EXTRACT_RETRY_DELAY_MS, signal);
        probe = await page.evaluate<PageProbe>(probeScript, signal);
      }

      if (!probe) {
        throw new BrowserUnavailableError("the page returned nothing — the extraction script did not run");
      }

      if (probe.challenge) {
        throw new SearchChallengeError({
          challenge: probe.challenge.kind,
          // The page the human must clear is where the browser *ended up*, which
          // after a consent redirect is not the URL we asked for.
          pageUrl: probe.url || searchUrl,
          endpoint,
          engine: engine.label,
          alternatives: otherEngines(engine.id),
          detail: probe.challenge.detail || undefined,
        });
      }

      if (probe.results.length === 0 && probe.empty) {
        // The engine itself says nothing matched. That is the query's doing, not
        // the extractor's or a rate limit's, and saying so stops the caller from
        // switching engines for a query no index will answer.
        throw new NoResultsError(
          `${engine.label} reports no matches for this query (${probe.url || searchUrl}). The page loaded ` +
            `normally and the engine says so itself, so this is not a block or a broken extractor — the query ` +
            `is too narrow. Drop the quotes or some of the terms; retrying another engine with the same ` +
            `over-specific query is unlikely to help.`,
        );
      }

      if (probe.results.length === 0) {
        // "container missing entirely" and "container present but empty" are
        // different faults, and saying which saves the next person a browser
        // session.
        const diagnosis =
          probe.container === "found"
            ? `${engine.label}'s results container was rendered but nothing could be parsed out of it, which ` +
              `suggests its markup changed and castle_cdp_search's extractor for ${engine.id} needs updating.`
            : `${engine.label} served a page with no results area at all. This is often a soft rate-limit: as a ` +
              `block decays an engine stops serving its challenge page and returns an empty result page instead, ` +
              `which is indistinguishable from a genuine zero-hit search (observed on Google lasting ~90 minutes ` +
              `after heavy querying). It can also mean the query truly has no hits.`;
        throw new NoResultsError(
          `No results could be read from ${probe.url || searchUrl} (page title: "${probe.title}"). ${diagnosis} ` +
            `Before concluding the extractor is broken, retry with a different engine (${otherEngines(engine.id).join(", ")}); ` +
            `if those also come back empty the query is the problem, and if only this one does, it is rate limiting — ` +
            `wait rather than retrying in a loop.`,
        );
      }

      return {
        results: probe.results.slice(0, request.numResults),
        searchUrl,
        finalUrl: probe.url || searchUrl,
        endpoint,
        engine,
        engineSource,
      };
    } finally {
      // Always, including the challenge path. Leaving the tab open would let the
      // operator solve the captcha in place, but it would also litter a browser
      // someone else is using with abandoned tabs on every failure; the error
      // carries the URL instead, and the cookie it sets is profile-wide.
      await page.close();
    }
  } finally {
    release();
  }
};

export type ResolveOutcome = { from: string; to: string; endpoint: string };

/**
 * Follows one Google `/goto` link in Chrome and reports where it lands.
 *
 * Lazy by design: search leaves the links wrapped, and the caller resolves only
 * the ones it means to open. One tab, one navigation, closed afterwards — the
 * pattern of a person clicking a result, which Google serves directly.
 *
 * If Google answers with its "Verifying your request" page instead, the tab is
 * still on google.com when the page loads. That is reported as such. The
 * countdown is not waited out and nothing is clicked.
 */
export const resolveLink = async (url: string, signal: AbortSignal): Promise<ResolveOutcome> => {
  if (!isGoogleGoto(url)) {
    throw new Error(
      "castle_cdp_resolve only follows Google result links of the form https://www.google.com/goto?url=… " +
        "as printed by castle_cdp_search. Results from other engines already carry their real URL.",
    );
  }
  const release = await acquire(signal);
  try {
    const conn = await ensureConnection(signal);
    const endpoint = describeCdpTarget(conn.target);
    const page = await PageSession.open(conn, signal);
    try {
      await page.navigate(url, signal);
      const landed = await page.evaluate<string>("location.href", signal);
      if (!landed || isGoogleHost(landed)) {
        throw new GoogleVerificationError(landed || url);
      }
      return { from: url, to: landed, endpoint };
    } finally {
      await page.close();
    }
  } finally {
    release();
  }
};

const isGoogleHost = (url: string): boolean => /^https?:\/\/([a-z0-9-]+\.)*google\.(?:com?\.)?[a-z]{2,3}\//.test(url);