{
  "markdown": "# configurable-prospect-scanner\n\nThe single, config-driven prospect scanner behind every skill the AEO platform\nbuilds. Registered in conqrse-queue as **one** catalog entry, `taskRef:\nconfigurable-prospect-scanner` — decision C-1: many skills, one runtime. Per-skill\nbehaviour arrives as **config in the org's runtime context**, never as a separate\nimage, so this repo must contain no vertical logic and no customer data.\n\nThe repo name, the queue `taskRef`, and `skills.runtime_slug` in aeo-backend are\ndeliberately the **same string**. The Conversational Skill Builder feature produced\nfive distinct name-collision defects; this removes a sixth.\n\n## What is ours and what is not\n\n| Path | Provenance | Rule |\n|---|---|---|\n| `av_lead_scanner.py` | **vendored** from `skill-av-lead-scanner` @ `3791a71` | **do not edit** — see [UPSTREAM.md](UPSTREAM.md) |\n| `tests/` | vendored | upstream's tests for the engine |\n| `aeo/` | **ours** | all AEO-specific behaviour lives here |\n| `reference/` | vendored, for reading only | not imported, not shipped |\n\nThat split is not our invention — upstream drew it themselves. Their integration\nexample states the core tool \"stays generic — it knows nothing about AEO\" and that\nAEO-specific behaviour belongs in a wrapper. `aeo/` is that wrapper, promoted from\nexample to product.\n\n## How a run works\n\n```\nqueue task  →  aeo/runner.py\n                 1. GET  {AEO_BACKEND_URL}/runtime/organizations/{ORGANIZATION_ID}/context\n                 2. aeo/config_mapping.py   AEO context ─→ engine context\n                 3. av_lead_scanner.discover(...)  →  score_prospects(...)\n                 4. POST {AEO_BACKEND_URL}/runtime/scans/{SCAN_RUN_ID}/events   (per event)\n```\n\n## The mapping is the point of this repo\n\nThe engine reads a flat context: `organization`, `product_description`, `gemini`,\n`output`, `sources`, `scoring`. AEO's runtime context returns something else — org\ncolumns, resolved geography, personas, products, and the authored skill recipe\nnested under `skill.config` in the Skill Builder's schema.\n\n**Those two shapes were never reconciled.** Upstream's example asserted AEO returns\nthe engine's own shape; it does not — four of six top-level keys are absent. Left\nunmapped, a conversationally-authored config reaches the engine and is **silently\nignored**: the scan runs on defaults, returns plausible prospects, and nothing\nerrors. `aeo/config_mapping.py` exists to make that impossible, and it **refuses\nrather than defaults** whenever the authored recipe cannot drive a real scan —\nbecause a default discovery strategy is not a smaller feature, it is a wrong answer\nwearing a confident face.\n\n## Events out: `aeo/event_mapping.py`\n\nThe mirror of the config mapping. Verified field-by-field against aeo-backend's\n`scan-event.dto.ts`, and the good news is that the hard parts already agree: the\nengine's prospect `id` is a `uuid5` of `(scan_run_id, normalised name)`, which is\nexactly AEO's \"skill-generated, stable on retry\"; `discovery_data` is already an\nobject, matching migration 071; and `prospect_id` / `contact_name` / `score` /\n`rank` / `score_factors` map 1:1 on the scored event.\n\nThree things a pass-through gets wrong, all of them total failures rather than\npartial ones:\n\n- **`phase` / `phase_name` are per-ITEM in AEO, per-EVENT in the engine.** Unmapped,\n  every prospect in the sweep fails validation.\n- **AEO caps an event at 1000 items**; the engine emits one event per sweep. Over the\n  cap the whole callback 400s and the sweep is lost, not truncated.\n- **`pipeline_status` is a name collision** — see below.\n\nUndeclared fields are folded into `scoring_payload` rather than dropped: the\nengine's vertical-shaped extras (`denomination`, `campaign_goal`, `project_type`)\nare real signal, they just are not columns.\n\n### ⚠️ `pipeline_status` — do not connect these two fields\n\nThe engine emits `pipeline_status` from `calculate_pipeline()`: a **construction\nproject** stage inferred from timeline arithmetic (\"in campaign\", \"breaking\nground\"). AEO's `prospects.pipeline_status` is the **sales** pipeline workflow an\noperator drives by hand via `PATCH /prospects/:id/pipeline-status`.\n\nSame name, unrelated meanings. Mapping one onto the other silently overwrites\noperator sales state with construction strings on every scan, with nothing erroring.\nIt stays inside `scoring_payload`, and a test asserts it never travels top-level.\n\nThis is the sixth name collision this feature has produced. The other five each cost\na defect.\n\n### Known gaps, stated rather than discovered later\n\n- **`discovery.sources` is a proposal.** The Skill Builder's config schema leaves\n  the `discovery` section's internals open (`additionalProperties: true`) because\n  they were never ratified. This repo proposes the engine's own source shape as\n  those internals — settled by something that executes rather than by discussion.\n- **`validation` and `contacts` do not run.** The PRD's scanner has five phases;\n  this engine exposes discovery and scoring. A config authoring those sections gets\n  a loud warning at start-up, not silence and a short result set.\n- The corrected context path: upstream fetched `/api/runtime/...`; aeo-backend sets\n  **no global prefix**, so that 404s. Fixed here, verified against its controller.\n\n## Local run (no Docker, no queue)\n\nThe cheapest useful test, and it needs neither of the platform's blocked\ndependencies — the ordinary scan path is live, only the builder's chat and\ntest-run endpoints are stubs.\n\n```bash\npython -m venv .venv && . .venv/bin/activate     # Windows: .venv\\Scripts\\activate\npip install -r requirements.txt\n\nexport AEO_BACKEND_URL=http://localhost:3000\nexport ORGANIZATION_ID=<an org in your local aeo_platform>\nexport SCAN_RUN_ID=<a row in scan_runs, so events have something to bind to>\nexport CI_USER=... CI_PASSWORD=...\nexport AV_SCANNER_MOCK=1        # offline provider — no model key needed\n\npython -m aeo.runner\n```\n\nExpect one of three informative outcomes: a mapping refusal naming exactly which\nconfig sections are unusable, an engine rejection of the mapped context, or a real\n(mock-sourced) run whose events land on the scan run.\n\n## Running under conqrse-queue (the real path)\n\n**The queue injects exactly one variable: `TASK_RECORD_ID`.** The business payload\nstays in the portal's database and is delivered *by reference* — both repos say so\nindependently (`k8s-job.executor.ts`: \"Only the reference is injected\";\naeo-backend's enqueue site: \"the opaque business envelope the scan task reads back by\nreference\"). So a container expecting `ORGANIZATION_ID`/`SCAN_RUN_ID` in its\nenvironment gets **none of them**. `aeo/bootstrap.py` fetches\n`GET {QUEUE_API_URL}/api/tasks/{TASK_RECORD_ID}` and maps the payload:\n\n| Payload field | Runner value |\n|---|---|\n| `organization_id` *(not `org_id`)* | `ORGANIZATION_ID` |\n| `tenant_id` | `TENANT_ID` |\n| `scan_run_id` | `SCAN_RUN_ID` |\n| `skill.slug` | `SKILL_SLUG` |\n| `phases[]` *(absent = all)* | `PHASES` — this is R6's reduced-phase run |\n\nIn queue mode the **payload wins** over any ambient env: the task record is the\nauthoritative statement of what the run is, and a stale env var silently scanning the\nwrong org is worse than a missing one.\n\nEverything else — credentials, `AEO_BACKEND_URL`, `QUEUE_API_URL` — arrives through\nthe catalog entry's `envFrom` (Secrets/ConfigMaps), which the executor wires normally.\nOnly the business payload is by-reference.\n\n### Verified locally, end to end\n\nDocker Desktop's Kubernetes, the queue on `:3200`, the gateway on `:3000`:\n\n```bash\ndocker build -t configurable-prospect-scanner:local .\n\nkubectl create secret generic scanner-secrets \\\n  --from-literal=GEMINI_API_KEY=... --from-literal=CI_USER=... --from-literal=CI_PASSWORD=...\nkubectl create configmap scanner-config \\\n  --from-literal=AEO_BACKEND_URL=http://host.docker.internal:3000 \\\n  --from-literal=QUEUE_API_URL=http://host.docker.internal:3200 \\\n  --from-literal=AV_SCANNER_PROVIDER=gemini\n\ncurl -X POST localhost:3200/api/catalog/entries -H 'Content-Type: application/json' -d '{\n  \"taskRef\": \"configurable-prospect-scanner\",\n  \"image\": \"configurable-prospect-scanner:local\",\n  \"resources\": {\"cpu\":\"500m\",\"memory\":\"512Mi\"},\n  \"retry\": {\"attempts\":1,\"backoff\":\"fixed\",\"delaySec\":10},\n  \"timeoutSec\": 900, \"namespace\": \"default\",\n  \"envFrom\": [\"secret/scanner-secrets\",\"configmap/scanner-config\"]\n}'\n\ncurl -X POST localhost:3200/api/tasks -H 'Content-Type: application/json' \\\n  -d '{\"taskRef\":\"configurable-prospect-scanner\",\"idempotencyKey\":\"<runId>\",\"payload\":{...}}'\n```\n\nTwo local-only details that matter: **`host.docker.internal`** is how a pod reaches a\nservice on the host, and the catalog image must be resolvable without a registry —\nthe queue's `K8S_IMAGE_PULL_POLICY=IfNotPresent` plus Docker Desktop's shared image\nstore makes a locally-built tag work as-is.\n\n## Container\n\n```bash\ndocker build -t configurable-prospect-scanner:dev .\n```\n\nThe entrypoint is `aeo.runner`, never the raw engine — invoking the tool directly\nskips the mapping and quietly scans on defaults. Register the image against\n`taskRef: configurable-prospect-scanner` in conqrse-queue's catalog (DB-backed CRUD,\nno code change needed there).\n\n## Environment\n\n| Var | Required | Purpose |\n|---|---|---|\n| `ORGANIZATION_ID` | yes | whose runtime context to fetch |\n| `SCAN_RUN_ID` | yes | binds every emitted event to a scan run |\n| `AEO_BACKEND_URL` | yes | e.g. `http://aeo-backend` |\n| `CI_USER` / `CI_PASSWORD` | for real runs | HTTP Basic on both AEO calls |\n| `AV_SCANNER_MOCK` | no | `1` → offline provider, no model key |\n| `AV_SCANNER_PROVIDER` | no | `gemini` \\| `claude` when not mocking |\n| `GEMINI_API_KEY` | when using gemini | grounded search |\n| `SCANNER_TOP_N` | no | default 50. Fallback for `discovery.target_prospects` and `contacts.max_prospects` when the config omits them. **Not a ranking cut on this path** — the cut lives in `FileSink.close()` and the AEO path uses `AeoEventSink`, which has no `close()`. **Not** a fallback for `discovery.max_prospects` either: at the production value of `1` that would cap every run at one prospect. |\n| `SCANNER_PHASE_CONCURRENCY` | no | width of the per-prospect phases, default **2**. The biggest single lever on run duration — location verification and validation are one grounded call per prospect each, so wall-clock ≈ `2 × prospects ÷ this × call latency` (~47 s). Raise only as far as the model key's rate limit allows; a 429's backoff costs more than the parallelism wins. |\n\n### Bounding a run\n\nTwo knobs decide how long a scan takes, and they multiply:\n\n- **`discovery.max_prospects`** (skill config) — a hard ceiling on prospects per run,\n  cumulative across discovery rounds and applied *before* any prospect is persisted or\n  verified. Absent means no ceiling. This is the only real cap; see\n  `aeo/phases/prospect_budget.py` for why the three things that look like one are not.\n- **`SCANNER_PHASE_CONCURRENCY`** (env) — how many of the per-prospect calls run at once.\n\nMeasured, on the run that motivated the ceiling: 262 discovered prospects at concurrency\n2 needs roughly 3.5 hours of grounded calls. The same set at concurrency 6, capped to\n100, is about 40 minutes.\n",
  "bytes": 11207,
  "sha": "c90ece431956ab1ac4606e6e03081ab481739f3082100deb7ebfbed30c67b09b",
  "repo_slug": "conexustech/configurable-prospect-scanner",
  "fonte": "repo",
  "truncated": false,
  "api": "https://agentalog.com/api/listings/okf_conexustech_configurable_prospect_scanne_92c652e8/readme"
}