Close #113–#116: lead-finder fully declarative (full sweep, no code)

workers/api/src/lib/connectors/http.test.ts

@@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
 import { getRegistryTool } from "../tool-registry.js";
 import type { RegistryToolCtx } from "../tool-registry.js";
 import type { ConnectorClient } from "./client.js";
+import { getPath } from "./http.js";
 
 // The http_request tool, resolved from the registry (proves it's registered → callable via
 // runtime, MCP proxy, and POST …/tools/http_request with no bespoke route).
@@ -37,6 +38,38 @@ function safeParse(s: string): any {
 	}
 }
 
+describe("getPath — type-predicate array selection (#116)", () => {
+	// Google returns addressComponents as an array of {longText, types:[…]}.
+	const comps = [
+		{ longText: "Newtown", types: ["locality", "political"] },
+		{ longText: "New South Wales", types: ["administrative_area_level_1", "political"] },
+		{ longText: "Australia", types: ["country", "political"] },
+	];
+	const rec = { addressComponents: comps };
+
+	it("selects the element whose types[] contains the token, then continues the sub-path", () => {
+		expect(getPath(rec, "addressComponents[types~=locality].longText")).toBe("Newtown");
+		expect(getPath(rec, "addressComponents[types~=administrative_area_level_1].longText")).toBe("New South Wales");
+		expect(getPath(rec, "addressComponents[types~=country].longText")).toBe("Australia");
+	});
+
+	it("returns undefined when no element matches the predicate", () => {
+		expect(getPath(rec, "addressComponents[types~=sublocality].longText")).toBeUndefined();
+	});
+
+	it("matches a scalar field too (not only arrays)", () => {
+		const arr = { parts: [{ kind: "a", v: 1 }, { kind: "b", v: 2 }] };
+		expect(getPath(arr, "parts[kind~=b].v")).toBe(2);
+	});
+
+	it("plain dotted/index grammar still works (backward compatible)", () => {
+		expect(getPath({ a: { b: [10, 20] } }, "a.b.1")).toBe(20);
+		expect(getPath(comps, "0.longText")).toBe("Newtown");
+		// the old (broken) shape without a predicate still resolves to undefined, not a throw.
+		expect(getPath(comps, "locality")).toBeUndefined();
+	});
+});
+
 describe("http_request — registration & schema", () => {
 	it("is registered as an http-connector, read-scoped tool", () => {
 		expect(httpRequest.connector).toBe("http");

workers/api/src/lib/connectors/http.ts

@@ -42,14 +42,37 @@ function interpolateDeep(value: unknown, inputs: Record<string, unknown>): unkno
 //   • array-projection         "places[].displayName.text"      → map each element to that sub-path
 //   • projection with reshape  "places[].{id,name:displayName.text,site:websiteUri}"
 //                                → [{id, name, site}, …] pulling each field's dotted sub-path
+//   • type-predicate select    "addressComponents[types~=locality].longText"  → in an ARRAY of
+//                                typed components, pick the element whose `types` (array or
+//                                scalar) contains the token, then continue the sub-path (#116).
+//                                General for any "array of typed components" API (Google, etc.).
 // Returns undefined for a path that doesn't resolve (rather than throwing) so a partial
 // response maps to nulls, not an error.
+
+// One segment with a type-predicate filter, e.g. "addressComponents[types~=locality]".
+const PREDICATE_SEG = /^([\w-]+)\[([\w-]+)~=([^\]]+)\]$/;
+
+/** Does `field`'s value (an array, or a scalar) contain `token`? Used by the [k~=v] predicate. */
+function fieldContains(el: unknown, field: string, token: string): boolean {
+	if (el === null || typeof el !== "object") return false;
+	const v = (el as Record<string, unknown>)[field];
+	if (Array.isArray(v)) return v.some((x) => String(x) === token);
+	return v !== undefined && v !== null && String(v) === token;
+}
+
 export function getPath(obj: unknown, path: string): unknown {
 	if (!path) return obj;
 	let cur: unknown = obj;
 	for (const seg of path.split(".")) {
 		if (cur === null || cur === undefined) return undefined;
-		if (Array.isArray(cur)) {
+		const pred = PREDICATE_SEG.exec(seg);
+		if (pred) {
+			// `name[key~=token]`: read `name` off the current object (an array of components),
+			// then select the element whose `key` contains `token`.
+			const [, name, key, token] = pred;
+			const arr = typeof cur === "object" && !Array.isArray(cur) ? (cur as Record<string, unknown>)[name] : undefined;
+			cur = Array.isArray(arr) ? arr.find((el) => fieldContains(el, key, token)) : undefined;
+		} else if (Array.isArray(cur)) {
 			const idx = Number(seg);
 			cur = Number.isInteger(idx) ? cur[idx] : undefined;
 		} else if (typeof cur === "object") {

workers/api/src/lib/pipeline.test.ts

@@ -75,6 +75,21 @@ describe("resolveInputValue / resolveInputs", () => {
 	it("resolves $param:item to the fan-out item", () => {
 		expect(resolveInputValue({ $param: "item" }, { ...scope, item: { n: 1 } })).toEqual({ n: 1 });
 	});
+
+	it("resolves $param:'item.<path>' to a dotted field of the fan-out item (#114)", () => {
+		const s = { ...scope, item: { lat: -33.9, lng: 151.1, websiteUri: "https://x.example" } };
+		expect(resolveInputValue({ $param: "item.lat" }, s)).toBe(-33.9);
+		expect(resolveInputValue({ $param: "item.websiteUri" }, s)).toBe("https://x.example");
+		// whole item still works, and a nested dotted path resolves too.
+		expect(resolveInputValue({ $param: "item" }, s)).toEqual(s.item);
+		const nested = { ...scope, item: { location: { lat: 1, lng: 2 } } };
+		expect(resolveInputValue({ $param: "item.location.lng" }, nested)).toBe(2);
+	});
+
+	it("$param:'item.<path>' is undefined when there is no item scope (backward compatible)", () => {
+		// Outside a forEach body, `item.foo` is just a (missing) param name — never crashes.
+		expect(resolveInputValue({ $param: "item.lat" }, scope)).toBeUndefined();
+	});
 });
 
 describe("executePipelineStep — dispatches via runRegistryTool + threads outputs", () => {

workers/api/src/lib/pipeline.ts

@@ -176,8 +176,13 @@ export function resolveInputValue(
 	if (Array.isArray(value)) return value.map((v) => resolveInputValue(v, scope));
 	const obj = value as Record<string, unknown>;
 	if (typeof obj.$param === "string") {
-		// `$param: "item"` inside a forEach body reads the current fan-out item.
-		if (obj.$param === "item" && scope.item !== undefined) return scope.item;
+		// `$param: "item"` inside a forEach body reads the current fan-out item;
+		// `$param: "item.<path>"` reads a dotted field off it (issue #114) — so a per-item
+		// body can plug `item.lat`/`item.lng` into a request or `item.websiteUri` into a probe.
+		if (scope.item !== undefined) {
+			if (obj.$param === "item") return scope.item;
+			if (obj.$param.startsWith("item.")) return readPath(scope.item, obj.$param.slice("item.".length));
+		}
 		return scope.params[obj.$param];
 	}
 	if (typeof obj.$ref === "string") return readPath(scope.outputs, obj.$ref);

workers/api/src/lib/pipelines/README.md

@@ -10,55 +10,62 @@ core step library (`workers/api/src/lib/steps.ts`) + the generic HTTP connector
 
 The lead-finder ("find small businesses with no / dead website") expressed as pure
 configuration. It is the *proof* that the common **source → transform → sink** class can be
-built without a bespoke Worker. Proven end-to-end in `lead-finder.test.ts`: the JSON is driven
-through the real runner + real `map`/`filter`/`dedupe` handlers; only the two I/O boundaries
-(outbound HTTP, the collection sink DO) are mocked.
+built without a bespoke Worker. As of the #113–#116 fixes this is the **FULL sweep** (grid +
+geo + reachability), no longer a spine. Proven end-to-end in `lead-finder.test.ts`: the JSON is
+driven through the real runner + real `fan_out`/`flatten`/`map`/`enrich`/`filter`/`dedupe`
+handlers; only the two I/O boundaries (outbound HTTP, the collection sink DO) are mocked.
 
 ### Epic proof-bullet → concrete step
 
 | Epic bullet | Step in `lead-finder.json` | Status |
 |---|---|---|
 | **source:** HTTP connector → Google Places nearby (city→geocode, type, radius = params) | `geocode` (city→centre) → `http_request` POST `places:searchNearby`, `X-Goog-Api-Key` via vault `auth`, `X-Goog-FieldMask` header, `responseMap: "places[].{place_id:id,name:displayName.text,…}"` | ✅ composes |
-| **paginate:** grid cells around the geocoded centre | *intended:* `fan_out` (grid) → `http_request` `forEach: {$ref:"grid.cells"}` | ⚠️ **gap** — see #113 + #114. Shipped JSON uses a **single** `searchNearby` at the centre instead. |
-| **map:** addressComponents → {city, suburb, state, country} | `map` `extract: {city:"addressComponents.locality", …}` + `derive: {category,status,website_status}` | ⚠️ derive/rename/passthrough ✅; the geo `extract` targets are correct **intent** but resolve to `null` against real Google data — see #116 |
-| **enrich:** HTTP-reachability on websiteUri | *intended:* `http_reachable` per record, merged back onto the place | ⚠️ **gap** — see #115 (+ #114). Not in the shipped JSON. |
-| **filter:** keep no-website OR unreachable | `filter` `where:[{field:"websiteUri",op:"missing"}]` | ⚠️ **partial** — the "no-website" half composes; the "OR unreachable" half is blocked by #115 (nothing merges reachability onto the record) |
+| **paginate:** grid cells around the geocoded centre | `fan_out` (grid, `center:{lat,lng}` from geo) → `http_request` `forEach:{$ref:"grid.cells"}` with `{$param:"item.lat"}`/`{$param:"item.lng"}` in the body → `flatten` (`path:"data"`) collapses the per-cell envelopes to one flat list | ✅ **closed** — #113 (`flatten`) + #114 (dotted-item) |
+| **map:** addressComponents → {city, suburb, state, country} | `map` `extract:{city:"addressComponents[types~=locality].longText", state:"…[types~=administrative_area_level_1]…", country:"…[types~=country]…", suburb:"…[types~=sublocality]…"}` + `derive:{category,status}` | ✅ **closed** — #116 (type-predicate `getPath`) populates geo from Google's typed array |
+| **enrich:** HTTP-reachability on websiteUri | `enrich` `{tool:"http_reachable", input:{url:{$item:"websiteUri"}}, as:"reachable"}` — probes per record, merges `{ok,code}` back onto a copy of the place | ✅ **closed** — #115 (`enrich` merge primitive) |
+| **filter:** keep no-website OR unreachable | `filter` `any:true, where:[{field:"websiteUri",op:"missing"},{field:"reachable.ok",op:"eq",value:false}]` | ✅ **closed** — both halves compose (the "OR unreachable" half rides the enriched `reachable.ok`) |
 | **dedupe:** upsert by place_id | `dedupe_upsert` `key:"place_id"` | ✅ composes |
 | **sink:** instance collection `leads` | `sink.collection: "leads"` + `dedupe_upsert` into it | ✅ composes |
 | **audit:** per-record run log | `attachAudit(records, trail, "leads")` — a `{step,detail,at}` entry per step + a sink line | ✅ composes |
 
 ### Chaining note (envelopes)
 
-`map`/`filter` return `{items,count,…}` and `http_request` returns `{status,data,…}`. A step's
-`$ref` reads the **whole** bound output, so the JSON chains off the envelope field:
-`places.data` → `map.items`, `shaped.items` → `filter.items`, `leads.items` → `dedupe_upsert.items`.
+`map`/`filter`/`flatten`/`enrich` return `{items,count,…}` and `http_request` returns
+`{status,data,…}`. A step's `$ref` reads the **whole** bound output, so the JSON chains off the
+envelope field: `grid.cells` → `http_request` `forEach`; the per-cell `forEach` binds an array
+of `{status,data}` envelopes at `pages`, so `flatten` uses `{items:{$ref:"pages"}, path:"data"}`
+to lift + concatenate each cell's `data`; then `flat.items` → `map.items`, `shaped.items` →
+`enrich.items`, `enriched.items` → `filter.items`, `leads.items` → `dedupe_upsert.items`.
 This is the contract, not a gap.
 
 ### The enrich follow-on (#99 web_search → extract_contacts)
 
-The socials/email enrichment the epic mentions is a second enrich stage:
-`web_search` a business name + suburb → `extract_contacts` (pulls the first Instagram / Facebook
-/ email out of the results) → columns on the lead. Both tools already exist (`steps.ts`
-`extract_contacts`, the `web_search` connector, #99). Wiring them declaratively per-record
-depends on the same **enrich-merge** primitive as reachability — **#115**.
+The socials/email enrichment the epic mentions is a second `enrich` stage that reuses the SAME
+primitive as reachability: `enrich {tool:"web_search", input:{query:{$item:"name"}}, as:"hits"}`
+→ `map`/`extract_contacts` over `hits` → Instagram / Facebook / email columns on the lead. Both
+tools already exist (`steps.ts` `extract_contacts`, the `web_search` connector, #99). Now that
+**#115** (`enrich`) has landed, this is fully expressible declaratively — no new primitive needed.
 
-## Gaps found by the capstone (all children of #94)
+## Gaps found by the capstone — all CLOSED (children of #94)
 
-Each is asserted in `lead-finder.test.ts` (the "GAP #N" tests) so the green suite hides nothing.
+Each was asserted in `lead-finder.test.ts`; those "GAP #N" tests now assert the fix **works**
+(they flipped from asserting-the-limitation), so the green suite hides nothing.
 
-| # | Gap | Blocks | Issue |
+| # | Gap (now closed) | Fix | Issue |
 |---|---|---|---|
-| 1 | `forEach` over a source whose tool returns an array → **array-of-arrays**; no `flatten` step / `arr[]` `$ref` grammar to collapse it | grid fan-out source | [#113](https://github.com/ProAgentStore/platform/issues/113) |
-| 2 | `forEach` `item` has **no dotted access** — `{$param:"item.lat"}` is undefined; only the whole `{$param:"item"}` works | per-cell request body; per-record reachability url | [#114](https://github.com/ProAgentStore/platform/issues/114) |
-| 3 | **enrich-merge** missing — a `forEach` tool result is a parallel array; no `enrich`/`zip` step to join it back onto the records | "unreachable" filter; socials/email enrich (#99) | [#115](https://github.com/ProAgentStore/platform/issues/115) |
-| 4 | `map`/`responseMap` `getPath` grammar can't **type-select** Google's `addressComponents` array (`find element whose types[] contains X`) | geo fields {city,suburb,state,country} | [#116](https://github.com/ProAgentStore/platform/issues/116) |
+| 1 | `forEach` over a source whose tool returns an array → **array-of-arrays** couldn't be collapsed | `flatten` step (`{items, depth?, path?}`) — `path` also lifts a sub-array out of each result envelope | [#113](https://github.com/ProAgentStore/platform/issues/113) ✅ |
+| 2 | `forEach` `item` had **no dotted access** — `{$param:"item.lat"}` was undefined | `resolveInputValue`: `$param:"item.<path>"` → `readPath(scope.item, path)` (whole-item still works) | [#114](https://github.com/ProAgentStore/platform/issues/114) ✅ |
+| 3 | **enrich-merge** missing — a `forEach` tool result was a parallel array, unjoinable | `enrich` step (`{items, tool, input, as, concurrency?}`) — runs `tool` per item (`{$item:"path"}` templates), merges result under `as` on a copy | [#115](https://github.com/ProAgentStore/platform/issues/115) ✅ |
+| 4 | `getPath` grammar couldn't **type-select** Google's `addressComponents` array | `getPath` predicate `arr[key~=token].sub` — select the element whose `key` contains the token, then continue | [#116](https://github.com/ProAgentStore/platform/issues/116) ✅ |
 
 ## Verdict
 
-**Can this class of agent be built by config alone today? Mostly — the spine, not yet the whole
-sweep.** `geocode → single Places searchNearby → map(reshape/derive) → filter(no-website) →
-dedupe_upsert → leads`, with a per-record audit trail, is **100% declarative and proven
-end-to-end**. What still needs code (until #113–#116 land): the **grid** multi-cell sweep
-(#113+#114), **reachability** to catch dead sites (#115+#114), **geo fields** from Google
-address components (#116), and the **socials/email** enrich (#115+#99). None require touching
-the running agent — they are runner/step-library primitives.
+**Can this class of agent be built by config alone today? Yes — the WHOLE sweep, not just the
+spine.** `geocode → fan_out(grid) → http_request(searchNearby per cell, forEach) → flatten →
+map(reshape/derive + geo from typed addressComponents) → enrich(http_reachable) →
+filter(no-website OR unreachable) → dedupe_upsert → leads`, with a per-record audit trail, is
+**100% declarative and proven end-to-end**. The four expressibility gaps the capstone first hit
+(#113–#116) are closed as **general runner / step-library primitives** — `flatten`, dotted-item
+access, `enrich`-merge, and type-predicate `getPath` — each useful well beyond this pipeline. The
+same `enrich` primitive also makes the #99 socials/email stage expressible. None of it touched
+the running lead-finder agent; the lead-finder class is now buildable with zero bespoke code.

workers/api/src/lib/pipelines/lead-finder.json

@@ -3,7 +3,7 @@
   "params": {
     "city": { "type": "string", "description": "City to sweep, e.g. \"Sydney, NSW\" — geocoded to the search centre." },
     "type": { "type": "string", "description": "Google Places includedType, e.g. \"cafe\" / \"restaurant\"." },
-    "radius": { "type": "number", "description": "searchNearby radius in metres around the centre (e.g. 900)." }
+    "radius": { "type": "number", "description": "searchNearby radius in metres around each grid cell (e.g. 900)." }
   },
   "steps": [
     {
@@ -13,9 +13,20 @@
         "address": { "$param": "city" }
       }
     },
+    {
+      "tool": "fan_out",
+      "bind": "grid",
+      "inputs": {
+        "mode": "grid",
+        "center": { "lat": { "$ref": "geo.lat" }, "lng": { "$ref": "geo.lng" } },
+        "extentKm": 2,
+        "stepKm": 1
+      }
+    },
     {
       "tool": "http_request",
-      "bind": "places",
+      "bind": "pages",
+      "forEach": { "$ref": "grid.cells" },
       "inputs": {
         "method": "POST",
         "url": "https://places.googleapis.com/v1/places:searchNearby",
@@ -29,35 +40,56 @@
           "maxResultCount": 20,
           "locationRestriction": {
             "circle": {
-              "center": { "latitude": { "$ref": "geo.lat" }, "longitude": { "$ref": "geo.lng" } },
+              "center": { "latitude": { "$param": "item.lat" }, "longitude": { "$param": "item.lng" } },
               "radius": { "$param": "radius" }
             }
           }
         },
         "responseMap": "places[].{place_id:id,name:displayName.text,address:formattedAddress,phone:nationalPhoneNumber,websiteUri:websiteUri,lat:location.latitude,lng:location.longitude,maps_url:googleMapsUri,addressComponents:addressComponents}"
       }
     },
+    {
+      "tool": "flatten",
+      "bind": "flat",
+      "inputs": {
+        "items": { "$ref": "pages" },
+        "path": "data"
+      }
+    },
     {
       "tool": "map",
       "bind": "shaped",
       "inputs": {
-        "items": { "$ref": "places.data" },
-        "derive": { "category": { "$param": "type" }, "status": "new", "website_status": "none" },
+        "items": { "$ref": "flat.items" },
+        "derive": { "category": { "$param": "type" }, "status": "new" },
         "extract": {
-          "country": "addressComponents.country",
-          "state": "addressComponents.state",
-          "city": "addressComponents.locality",
-          "suburb": "addressComponents.suburb"
+          "country": "addressComponents[types~=country].longText",
+          "state": "addressComponents[types~=administrative_area_level_1].longText",
+          "city": "addressComponents[types~=locality].longText",
+          "suburb": "addressComponents[types~=sublocality].longText"
         }
       }
     },
+    {
+      "tool": "enrich",
+      "bind": "enriched",
+      "inputs": {
+        "items": { "$ref": "shaped.items" },
+        "tool": "http_reachable",
+        "input": { "url": { "$item": "websiteUri" } },
+        "as": "reachable",
+        "concurrency": 4
+      }
+    },
     {
       "tool": "filter",
       "bind": "leads",
       "inputs": {
-        "items": { "$ref": "shaped.items" },
+        "items": { "$ref": "enriched.items" },
+        "any": true,
         "where": [
-          { "field": "websiteUri", "op": "missing" }
+          { "field": "websiteUri", "op": "missing" },
+          { "field": "reachable.ok", "op": "eq", "value": false }
         ]
       }
     },