diff --git a/workers/api/src/lib/connectors/http.test.ts b/workers/api/src/lib/connectors/http.test.ts
index c37a4d33..338e389a 100644
--- a/workers/api/src/lib/connectors/http.test.ts
+++ b/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");
diff --git a/workers/api/src/lib/connectors/http.ts b/workers/api/src/lib/connectors/http.ts
index 12ece960..6b50f138 100644
--- a/workers/api/src/lib/connectors/http.ts
+++ b/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") {
diff --git a/workers/api/src/lib/pipeline.test.ts b/workers/api/src/lib/pipeline.test.ts
index 1e8a1200..314d8ee1 100644
--- a/workers/api/src/lib/pipeline.test.ts
+++ b/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", () => {
diff --git a/workers/api/src/lib/pipeline.ts b/workers/api/src/lib/pipeline.ts
index c68317bc..7ca61e74 100644
--- a/workers/api/src/lib/pipeline.ts
+++ b/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);
diff --git a/workers/api/src/lib/pipelines/README.md b/workers/api/src/lib/pipelines/README.md
index bac75bba..dccb86f8 100644
--- a/workers/api/src/lib/pipelines/README.md
+++ b/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.
diff --git a/workers/api/src/lib/pipelines/lead-finder.json b/workers/api/src/lib/pipelines/lead-finder.json
index 7562ff6a..602144c4 100644
--- a/workers/api/src/lib/pipelines/lead-finder.json
+++ b/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,7 +40,7 @@
           "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" }
             }
           }
@@ -37,27 +48,48 @@
         "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 }
         ]
       }
     },
diff --git a/workers/api/src/lib/pipelines/lead-finder.test.ts b/workers/api/src/lib/pipelines/lead-finder.test.ts
index e77e9746..435e722e 100644
--- a/workers/api/src/lib/pipelines/lead-finder.test.ts
+++ b/workers/api/src/lib/pipelines/lead-finder.test.ts
@@ -1,66 +1,143 @@
 // CAPSTONE PROOF (epic #94): the lead-finder as a PURE DECLARATIVE PIPELINE — the JSON in
 // lead-finder.json is data, not code — driven end-to-end through the REAL runner
-// (executePipelineStep) and the REAL step handlers (map/filter from steps.ts). Only the two
-// true I/O boundaries are mocked: the outbound HTTP (geocode + Places searchNearby +
-// reachability) and the collection sink (dedupe_upsert's Durable Object). Everything between
-// them — input resolution ($ref/$param), step threading (bind), the map reshape, the
-// no-website filter, and the per-record audit trail — is exercised for real.
+// (executePipelineStep) and the REAL step handlers (fan_out/flatten/map/enrich/filter from
+// steps.ts). Only the two true I/O boundaries are mocked: the outbound HTTP (geocode +
+// per-cell Places searchNearby + reachability) and the collection sink (dedupe_upsert's
+// Durable Object). Everything between them — input resolution ($ref/$param, incl. the #114
+// dotted-item + #116 type-predicate + #113 flatten + #115 enrich-merge primitives), step
+// threading (bind/forEach), the grid fan-out, the map reshape, the no-website-OR-unreachable
+// filter, and the per-record audit trail — is exercised for real.
 //
-// It also PROVES, with assertions, the four expressibility gaps this pipeline hit (documented
-// in README.md with follow-up issue numbers), so the green test hides nothing.
+// This is the FULL SWEEP (not the earlier "spine only"): the four gaps the #94 capstone first
+// hit (#113–#116) are now CLOSED, so the tests that used to assert-the-limitation now assert-
+// it-works. The green test hides nothing.
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import leadFinder from "./lead-finder.json" with { type: "json" };
 
-// The REAL pure step handlers — we route map/filter to these so the transform logic is
+// The REAL pure step handlers — we route the transform steps to these so the logic is
 // genuinely tested, not faked.
 import { STEP_TOOLS } from "../steps.js";
 import { getPath } from "../connectors/http.js";
 
-// Mock the tool-registry boundary the runner dispatches through. getRegistryTool must know
-// every tool the JSON names (so validatePipeline passes); runRegistryTool routes pure tools
-// to their real handlers and stubs the I/O tools with realistic responses.
-const KNOWN = new Set(["geocode", "http_request", "map", "filter", "dedupe_upsert", "http_reachable"]);
+// Mock the tool-registry boundary the runner (and the enrich step) dispatch through.
+// getRegistryTool must know every tool the JSON names (so validatePipeline passes);
+// runRegistryTool routes pure tools to their real handlers and stubs the I/O tools.
+const KNOWN = new Set(["geocode", "fan_out", "http_request", "flatten", "map", "enrich", "filter", "dedupe_upsert", "http_reachable"]);
 
 // Captured sink writes so we can assert what landed in the `leads` collection.
 let upserted: Array<Record<string, unknown>> = [];
 
 const realHandler = (name: string) => STEP_TOOLS.find((t) => t.name === name)!.handler;
 
-const runRegistryTool = vi.fn(async (name: string, _ctx: unknown, input: Record<string, unknown>) => {
-	// ── real pure transforms ──────────────────────────────────────────────
-	if (name === "map" || name === "filter") {
-		const r = await realHandler(name)({} as never, input);
+// A realistic Google Places searchNearby response — three cafes near Newtown, WITH real
+// addressComponents arrays (typed-component shape). Their location matches specific grid
+// cells so the per-cell mock returns them cell-by-cell (and the overlap proves dedupe).
+//   • Corner Espresso — NO websiteUri            → qualifies (no website)
+//   • Old Roasters    — websiteUri present, DEAD  → qualifies (unreachable)
+//   • Bean Machine    — websiteUri present, LIVE  → NOT a lead
+function addr(locality: string) {
+	return [
+		{ longText: locality, types: ["locality", "political"] },
+		{ longText: "Marrickville", types: ["sublocality", "political"] },
+		{ longText: "New South Wales", types: ["administrative_area_level_1", "political"] },
+		{ longText: "Australia", types: ["country", "political"] },
+	];
+}
+
+const BUSINESSES: Record<string, Array<Record<string, unknown>>> = {
+	// keyed by a stable cell token (rounded lat,lng) → the raw Places `places` for that cell.
+	cellA: [
+		{
+			id: "ChIJ_no_site",
+			displayName: { text: "Corner Espresso" },
+			formattedAddress: "1 King St, Newtown NSW 2042",
+			nationalPhoneNumber: "0298001111",
+			// NO websiteUri
+			location: { latitude: -33.8951, longitude: 151.179 },
+			googleMapsUri: "https://maps.google.com/?cid=1",
+			addressComponents: addr("Newtown"),
+		},
+	],
+	cellB: [
+		{
+			id: "ChIJ_dead_site",
+			displayName: { text: "Old Roasters" },
+			formattedAddress: "2 Enmore Rd, Newtown NSW 2042",
+			nationalPhoneNumber: "0298002222",
+			websiteUri: "https://oldroasters.example", // present but DEAD
+			location: { latitude: -33.8972, longitude: 151.1766 },
+			googleMapsUri: "https://maps.google.com/?cid=2",
+			addressComponents: addr("Newtown"),
+		},
+		{
+			id: "ChIJ_live_site",
+			displayName: { text: "Bean Machine" },
+			formattedAddress: "3 Australia St, Newtown NSW 2042",
+			nationalPhoneNumber: "0298003333",
+			websiteUri: "https://beanmachine.example", // LIVE
+			location: { latitude: -33.8938, longitude: 151.1802 },
+			googleMapsUri: "https://maps.google.com/?cid=3",
+			addressComponents: addr("Newtown"),
+		},
+	],
+	// cellDup re-returns the no-website business (grid cells overlap) → dedupe must collapse it.
+	cellDup: [
+		{
+			id: "ChIJ_no_site",
+			displayName: { text: "Corner Espresso" },
+			formattedAddress: "1 King St, Newtown NSW 2042",
+			nationalPhoneNumber: "0298001111",
+			location: { latitude: -33.8951, longitude: 151.179 },
+			googleMapsUri: "https://maps.google.com/?cid=1",
+			addressComponents: addr("Newtown"),
+		},
+	],
+};
+
+// Which cell (by grid index) returns which businesses. Everything else returns an empty page.
+// The grid in the JSON is extentKm 2 / stepKm 1 → 25 cells; we seed three of them.
+let cellCounter = 0;
+function placesForCell(_body: unknown): Array<Record<string, unknown>> {
+	// The runner calls http_request once per cell in grid order; seed the first three cells.
+	const idx = cellCounter++;
+	if (idx === 0) return BUSINESSES.cellA;
+	if (idx === 1) return BUSINESSES.cellB;
+	if (idx === 2) return BUSINESSES.cellDup;
+	return [];
+}
+
+const runRegistryTool = vi.fn(async (name: string, ctx: unknown, input: Record<string, unknown>) => {
+	// ── real pure / step transforms ────────────────────────────────────────
+	if (name === "map" || name === "filter" || name === "flatten" || name === "fan_out" || name === "enrich") {
+		const r = await realHandler(name)(ctx as never, input);
 		return { name, content: r.content, success: r.success };
 	}
 	// ── mocked I/O boundary ───────────────────────────────────────────────
 	if (name === "geocode") {
-		// city → centre. A realistic Google-geocode-shaped result.
 		return { name, content: JSON.stringify({ lat: -33.8915, lng: 151.1795, country: "Australia", state: "New South Wales", locality: "Newtown", formatted: "Newtown NSW 2042, Australia" }), success: true };
 	}
 	if (name === "http_request") {
-		// Google Places searchNearby, AFTER the pipeline's responseMap projection
-		// ("places[].{place_id:id,name:displayName.text,...,addressComponents:addressComponents}").
-		// Three businesses: one with NO website, one with a DEAD site, one LIVE.
-		const projected = PLACES_RAW.places.map((p) => ({
+		// Per-cell Places searchNearby, AFTER the pipeline's responseMap projection. Each cell
+		// returns its slice of businesses (as {status,data:[…]} — the http_request envelope).
+		const projected = placesForCell(input.body).map((p) => ({
 			place_id: p.id,
-			name: p.displayName.text,
+			name: (p.displayName as { text: string }).text,
 			address: p.formattedAddress,
 			phone: p.nationalPhoneNumber ?? null,
 			websiteUri: p.websiteUri ?? null,
-			lat: p.location.latitude,
-			lng: p.location.longitude,
+			lat: (p.location as { latitude: number }).latitude,
+			lng: (p.location as { longitude: number }).longitude,
 			maps_url: p.googleMapsUri,
-			addressComponents: p.addressComponents, // raw Google array (see gap #4)
+			addressComponents: p.addressComponents, // raw Google typed-component array (#116)
 		}));
 		return { name, content: JSON.stringify({ status: 200, data: projected }), success: true };
 	}
 	if (name === "http_reachable") {
 		const url = String(input.url ?? "");
-		const alive = url === "https://beanmachine.example"; // the live one; dead site is unreachable
+		const alive = url === "https://beanmachine.example"; // only the live one is up
 		return { name, content: JSON.stringify({ ok: alive, code: alive ? 200 : null }), success: true };
 	}
 	if (name === "dedupe_upsert") {
-		// Sink: dedupe by place_id + record what landed. Simulates the DO collection write.
 		const items = (Array.isArray(input.items) ? input.items : []) as Array<Record<string, unknown>>;
 		const seen = new Set(upserted.map((r) => r[String(input.key)]));
 		let inserted = 0;
@@ -81,58 +158,10 @@ vi.mock("../tool-registry.js", () => ({
 	runRegistryTool: (...args: unknown[]) => (runRegistryTool as unknown as (...a: unknown[]) => unknown)(...args),
 }));
 
-// Import AFTER the mock so pipeline.ts binds the mocked runRegistryTool.
-import { attachAudit, auditStepEntry, executePipelineStep, stepBind, validatePipeline, type PipelineDef, type StepResult, type AuditEntry } from "../pipeline.js";
+// Import AFTER the mock so pipeline.ts + steps.ts bind the mocked runRegistryTool.
+import { attachAudit, auditStepEntry, executePipelineStep, stepBind, validatePipeline, resolveInputValue, type PipelineDef, type StepResult, type AuditEntry } from "../pipeline.js";
 import type { Env } from "../../types.js";
 
-// A realistic Google Places searchNearby response (three cafes near Newtown).
-const PLACES_RAW = {
-	places: [
-		{
-			id: "ChIJ_no_site",
-			displayName: { text: "Corner Espresso" },
-			formattedAddress: "1 King St, Newtown NSW 2042",
-			nationalPhoneNumber: "0298001111",
-			// NO websiteUri at all → qualifies (no website)
-			location: { latitude: -33.8951, longitude: 151.179 },
-			googleMapsUri: "https://maps.google.com/?cid=1",
-			addressComponents: [
-				{ longText: "Newtown", types: ["locality"] },
-				{ longText: "New South Wales", types: ["administrative_area_level_1"] },
-				{ longText: "Australia", types: ["country"] },
-			],
-		},
-		{
-			id: "ChIJ_dead_site",
-			displayName: { text: "Old Roasters" },
-			formattedAddress: "2 Enmore Rd, Newtown NSW 2042",
-			nationalPhoneNumber: "0298002222",
-			websiteUri: "https://oldroasters.example", // present but DEAD → qualifies (unreachable)
-			location: { latitude: -33.8972, longitude: 151.1766 },
-			googleMapsUri: "https://maps.google.com/?cid=2",
-			addressComponents: [
-				{ longText: "Newtown", types: ["locality"] },
-				{ longText: "New South Wales", types: ["administrative_area_level_1"] },
-				{ longText: "Australia", types: ["country"] },
-			],
-		},
-		{
-			id: "ChIJ_live_site",
-			displayName: { text: "Bean Machine" },
-			formattedAddress: "3 Australia St, Newtown NSW 2042",
-			nationalPhoneNumber: "0298003333",
-			websiteUri: "https://beanmachine.example", // LIVE → NOT a lead
-			location: { latitude: -33.8938, longitude: 151.1802 },
-			googleMapsUri: "https://maps.google.com/?cid=3",
-			addressComponents: [
-				{ longText: "Newtown", types: ["locality"] },
-				{ longText: "New South Wales", types: ["administrative_area_level_1"] },
-				{ longText: "Australia", types: ["country"] },
-			],
-		},
-	],
-};
-
 const env = {} as Env;
 const ctx = { env, userId: "u1", instanceId: "i1" };
 const params = { city: "Newtown, NSW", type: "cafe", radius: 900 };
@@ -140,6 +169,7 @@ const params = { city: "Newtown, NSW", type: "cafe", radius: 900 };
 beforeEach(() => {
 	runRegistryTool.mockClear();
 	upserted = [];
+	cellCounter = 0;
 });
 
 // Drive the full JSON pipeline through the real runner, step by step, exactly as the durable
@@ -158,7 +188,7 @@ async function drivePipeline(def: PipelineDef) {
 	return { outputs, trail, results };
 }
 
-describe("lead-finder declarative pipeline (capstone #94)", () => {
+describe("lead-finder declarative pipeline (capstone #94 — FULL SWEEP)", () => {
 	it("the JSON validates against the real runner contract (validatePipeline → null)", () => {
 		expect(validatePipeline(leadFinder)).toBeNull();
 	});
@@ -170,37 +200,69 @@ describe("lead-finder declarative pipeline (capstone #94)", () => {
 		expect(def.sink?.keyField).toBe("place_id");
 	});
 
-	it("composes end-to-end: source → map → filter → dedupe → sink yields the RIGHT leads", async () => {
+	it("uses the full-sweep step chain (geocode→fan_out→http_request/forEach→flatten→map→enrich→filter→dedupe)", () => {
+		const def = leadFinder as unknown as PipelineDef;
+		expect(def.steps.map((s) => s.tool)).toEqual([
+			"geocode", "fan_out", "http_request", "flatten", "map", "enrich", "filter", "dedupe_upsert",
+		]);
+		// the Places request runs once PER grid cell (forEach over grid.cells).
+		const places = def.steps.find((s) => s.tool === "http_request")!;
+		expect(places.forEach).toEqual({ $ref: "grid.cells" });
+		// per-cell body plugs the cell's lat/lng via the #114 dotted-item convention.
+		const center = (places.inputs as any).body.locationRestriction.circle.center;
+		expect(center.latitude).toEqual({ $param: "item.lat" });
+		expect(center.longitude).toEqual({ $param: "item.lng" });
+	});
+
+	it("composes the FULL sweep end-to-end: grid → flatten → geo map → reachability enrich → filter → dedupe → sink", async () => {
 		const def = leadFinder as unknown as PipelineDef;
 		const { outputs } = await drivePipeline(def);
 
-		// source: Places returned all three businesses (flat, via responseMap). http_request's
-		// output is the {status,data} envelope; the JSON chains the next step off `places.data`.
-		expect(((outputs.places as { data: unknown[] }).data).length).toBe(3);
+		// fan_out produced the grid; http_request ran once per cell → an ARRAY of per-cell
+		// envelopes; flatten (path:"data") collapsed them to a flat list of per-PLACE records.
+		const grid = outputs.grid as { cells: unknown[] };
+		expect(grid.cells.length).toBe(25); // extent 2 / step 1 → (2*2+1)^2
+		expect(Array.isArray(outputs.pages)).toBe(true);
+		const flat = (outputs.flat as { items: unknown[] }).items;
+		// 3 distinct businesses across cells + 1 duplicate (Corner Espresso re-seen) = 4 records.
+		expect(flat.length).toBe(4);
 
-		// filter (keep no-website): the composable spine drops the two businesses that HAVE a
-		// websiteUri, keeping only the true no-website lead. filter's output is the {items,count}
-		// envelope; the JSON chains the sink off `leads.items`.
-		const leads = (outputs.leads as { items: Array<Record<string, unknown>> }).items;
-		expect(leads.map((l) => l.place_id)).toEqual(["ChIJ_no_site"]);
+		// map: geo fields populated from Google's typed addressComponents (the #116 fix).
+		const shaped = (outputs.shaped as { items: Array<Record<string, unknown>> }).items;
+		const corner = shaped.find((r) => r.place_id === "ChIJ_no_site")!;
+		expect(corner.city).toBe("Newtown");
+		expect(corner.suburb).toBe("Marrickville");
+		expect(corner.state).toBe("New South Wales");
+		expect(corner.country).toBe("Australia");
+		expect(corner.category).toBe("cafe");
+		expect(corner.status).toBe("new");
 
-		// sink: exactly that lead was upserted into `leads`.
-		expect(upserted.map((r) => r.place_id)).toEqual(["ChIJ_no_site"]);
+		// enrich: reachability probed per record + MERGED back under `reachable` (the #115 fix).
+		const enriched = (outputs.enriched as { items: Array<Record<string, unknown>> }).items;
+		const live = enriched.find((r) => r.place_id === "ChIJ_live_site")!;
+		const dead = enriched.find((r) => r.place_id === "ChIJ_dead_site")!;
+		expect(live.reachable).toMatchObject({ ok: true, code: 200 });
+		expect(dead.reachable).toMatchObject({ ok: false, code: null });
 
-		// map carried the derived + passthrough fields onto the record.
-		const lead = leads[0];
-		expect(lead.name).toBe("Corner Espresso");
-		expect(lead.category).toBe("cafe");
-		expect(lead.status).toBe("new");
-		expect(lead.website_status).toBe("none");
-		expect(lead.maps_url).toBe("https://maps.google.com/?cid=1");
+		// filter (no-website OR unreachable): BOTH the no-website AND the dead-site survive;
+		// the live-site is excluded. This is the full epic filter, not just the "no-website" half.
+		// The duplicate Corner Espresso also survives the filter (it's still no-website) — dedupe
+		// happens at the SINK, not in the filter — so filter keeps 3 (2 unique + 1 dup).
+		const leads = (outputs.leads as { items: Array<Record<string, unknown>> }).items;
+		const keptIds = leads.map((l) => l.place_id).sort();
+		expect(keptIds).toEqual(["ChIJ_dead_site", "ChIJ_no_site", "ChIJ_no_site"]);
+		expect(keptIds).not.toContain("ChIJ_live_site");
+
+		// sink: dedupe_upsert by place_id collapsed the duplicate Corner Espresso → each lead once.
+		expect(upserted.map((r) => r.place_id).sort()).toEqual(["ChIJ_dead_site", "ChIJ_no_site"]);
 	});
 
 	it("dedupe by place_id: re-running the pipeline does not double-insert", async () => {
 		const def = leadFinder as unknown as PipelineDef;
 		await drivePipeline(def);
-		await drivePipeline(def); // second sweep, same place_ids
-		expect(upserted.map((r) => r.place_id)).toEqual(["ChIJ_no_site"]); // still one, not two
+		cellCounter = 0; // second sweep re-seeds the same cells
+		await drivePipeline(def);
+		expect(upserted.map((r) => r.place_id).sort()).toEqual(["ChIJ_dead_site", "ChIJ_no_site"]);
 	});
 
 	it("attaches a per-record audit trail (attachAudit) with a step-by-step decision log", async () => {
@@ -208,12 +270,11 @@ describe("lead-finder declarative pipeline (capstone #94)", () => {
 		const { outputs, trail } = await drivePipeline(def);
 		const leads = (outputs.leads as { items: unknown[] }).items;
 		const records = attachAudit(leads, trail, def.sink!.collection);
-		expect(records).toHaveLength(1);
+		expect(records.length).toBe(3); // the two unique leads + the pre-dedupe duplicate
 		const audit = records[0].audit as AuditEntry[];
 		// trail = one entry per pipeline step + a final sink line.
 		expect(audit.length).toBe(def.steps.length + 1);
 		expect(audit[audit.length - 1]).toMatchObject({ step: "sink", detail: 'upserted into "leads"' });
-		// each entry has the {step, detail, at} shape the /data tab renders.
 		for (const e of audit) {
 			expect(typeof e.step).toBe("string");
 			expect(typeof e.detail).toBe("string");
@@ -221,46 +282,59 @@ describe("lead-finder declarative pipeline (capstone #94)", () => {
 		}
 	});
 
-	// ── HONEST GAP PROOFS — the four places this class of agent does NOT yet fully compose.
-	//    Each is documented in README.md and filed as a #94 child issue.
+	// ── CLOSED-GAP PROOFS — the four places this class of agent could NOT compose before,
+	//    now proven to WORK. Each was a #94 child issue (#113–#116); the assertions FLIP from
+	//    asserting-the-limitation to asserting-it-works.
 
-	it("GAP #1 (grid fan-out flatten, issue #113): forEach http_request → array-of-arrays; map can't flatten it", async () => {
-		// If the source were a per-cell grid sweep (forEach over grid.cells), each cell yields
-		// an ARRAY of places, so the bound output is [[place…],[place…]]. map() treats each inner
-		// ARRAY as a non-record → {}, producing one row per CELL, not per place. No flatten step
-		// / $ref grammar exists to collapse it. (This is why the shipped JSON uses a single
-		// searchNearby at the centre, not the recipe's grid.)
-		const aoa = [[{ place_id: "a" }], [{ place_id: "b" }, { place_id: "c" }]];
-		const r = JSON.parse((await realHandler("map")({} as never, { items: aoa })).content);
-		expect(r.count).toBe(2); // 2 cells, NOT 3 places
+	it("GAP #1 CLOSED (grid fan-out flatten, #113): array-of-arrays collapses to per-place records", async () => {
+		// A forEach http_request binds an ARRAY of per-cell envelopes; flatten(path:"data")
+		// lifts each cell's places and concatenates → one flat record per place, not per cell.
+		const perCell = [
+			{ status: 200, data: [{ place_id: "a" }] },
+			{ status: 200, data: [{ place_id: "b" }, { place_id: "c" }] },
+		];
+		const r = JSON.parse((await realHandler("flatten")({} as never, { items: perCell, path: "data" })).content);
+		expect(r.count).toBe(3); // 3 PLACES (was: 2 cells)
+		expect(r.items.map((x: { place_id: string }) => x.place_id)).toEqual(["a", "b", "c"]);
 	});
 
-	it("GAP #2 (forEach item has no dotted access, issue #114): can't read item.lat / item.websiteUri in a body", async () => {
-		const { resolveInputValue } = await import("../pipeline.js");
-		const scope = { outputs: {}, params: {}, item: { lat: -33.9, websiteUri: "x" } };
-		expect(resolveInputValue({ $param: "item" }, scope)).toEqual({ lat: -33.9, websiteUri: "x" }); // whole item OK
-		expect(resolveInputValue({ $param: "item.lat" }, scope)).toBeUndefined(); // dotted access NOT OK
+	it("GAP #2 CLOSED (forEach dotted item, #114): item.lat / item.websiteUri resolve in a body", () => {
+		const scope = { outputs: {}, params: {}, item: { lat: -33.9, websiteUri: "https://x" } };
+		expect(resolveInputValue({ $param: "item" }, scope)).toEqual({ lat: -33.9, websiteUri: "https://x" });
+		expect(resolveInputValue({ $param: "item.lat" }, scope)).toBe(-33.9); // dotted access WORKS
+		expect(resolveInputValue({ $param: "item.websiteUri" }, scope)).toBe("https://x");
 	});
 
-	it("GAP #3 (enrich-merge, issue #115): http_reachable forEach is a PARALLEL array, never merged back onto places", () => {
-		// Even setting aside gap #2, http_reachable over `shaped` yields [{ok,code},…] as its own
-		// bound output. No zip/join step exists, and map/filter each read a single `items` array,
-		// so the `reachable` flag can't be correlated back to its place. → the "unreachable" half
-		// of "keep no-website OR unreachable" cannot be expressed; only "no-website" composes.
-		const reach = [{ ok: true }, { ok: false }];
-		const places = [{ place_id: "a" }, { place_id: "b" }];
-		expect(reach.length).toBe(places.length); // parallel, but nothing joins them by index
+	it("GAP #3 CLOSED (enrich-merge, #115): http_reachable merges back onto each place by identity", async () => {
+		// enrich runs http_reachable per item and writes {ok,code} under `reachable` on a COPY —
+		// so filter can test reachable.ok. No index-fragile zip; the flag rides ITS own place.
+		const places = [
+			{ place_id: "a", websiteUri: "https://beanmachine.example" }, // live in the mock
+			{ place_id: "b", websiteUri: "https://oldroasters.example" }, // dead in the mock
+		];
+		const enriched = JSON.parse(
+			(await realHandler("enrich")(ctx as never, { items: places, tool: "http_reachable", input: { url: { $item: "websiteUri" } }, as: "reachable" })).content,
+		);
+		expect(enriched.items.find((x: any) => x.place_id === "a").reachable).toMatchObject({ ok: true });
+		expect(enriched.items.find((x: any) => x.place_id === "b").reachable).toMatchObject({ ok: false });
 	});
 
-	it("GAP #4 (addressComponents extraction, issue #116): map can't type-select Google's addressComponents array", async () => {
-		// The JSON's map.extract {city:"addressComponents.locality",…} uses getPath, which has no
-		// "find the element whose types[] includes 'locality'" grammar — so against REAL Google
-		// data those geo fields resolve to null. Proven directly on the projected record.
-		const ac = PLACES_RAW.places[0].addressComponents;
-		expect(getPath(ac, "locality")).toBeUndefined(); // type-predicate lookup unsupported
+	it("GAP #4 CLOSED (addressComponents type-select, #116): map populates geo from Google's typed array", async () => {
+		const ac = BUSINESSES.cellA[0].addressComponents;
+		// type-predicate getPath selects the element whose types[] contains the token.
+		expect(getPath(ac, "[types~=locality].longText")).toBeUndefined(); // needs a named array field
 		const shaped = JSON.parse(
-			(await realHandler("map")({} as never, { items: [{ addressComponents: ac }], extract: { city: "addressComponents.locality" } })).content,
+			(await realHandler("map")({} as never, {
+				items: [{ addressComponents: ac }],
+				extract: {
+					city: "addressComponents[types~=locality].longText",
+					state: "addressComponents[types~=administrative_area_level_1].longText",
+					country: "addressComponents[types~=country].longText",
+				},
+			})).content,
 		);
-		expect(shaped.items[0].city).toBeNull(); // geo field does NOT populate from real Google shape
+		expect(shaped.items[0].city).toBe("Newtown");
+		expect(shaped.items[0].state).toBe("New South Wales");
+		expect(shaped.items[0].country).toBe("Australia");
 	});
 });
diff --git a/workers/api/src/lib/steps.test.ts b/workers/api/src/lib/steps.test.ts
index a7643c46..60669044 100644
--- a/workers/api/src/lib/steps.test.ts
+++ b/workers/api/src/lib/steps.test.ts
@@ -7,8 +7,10 @@ import type { ConnectorClient } from "./connectors/client.js";
 // runRegistryTool for the #97 runner, and via POST …/tools/:name with no bespoke route).
 const mapT = getRegistryTool("map")!;
 const filterT = getRegistryTool("filter")!;
+const flattenT = getRegistryTool("flatten")!;
 const dedupeT = getRegistryTool("dedupe_upsert")!;
 const fanOutT = getRegistryTool("fan_out")!;
+const enrichT = getRegistryTool("enrich")!;
 const reachableT = getRegistryTool("http_reachable")!;
 const geocodeT = getRegistryTool("geocode")!;
 const extractT = getRegistryTool("extract_contacts")!;
@@ -24,7 +26,7 @@ afterEach(() => vi.restoreAllMocks());
 // ── registration ──────────────────────────────────────────────────────────────
 describe("step library — registration", () => {
 	it("registers all steps as standard-tier, non-connector tools", () => {
-		for (const t of [mapT, filterT, dedupeT, fanOutT, reachableT, geocodeT, extractT]) {
+		for (const t of [mapT, filterT, flattenT, dedupeT, fanOutT, enrichT, reachableT, geocodeT, extractT]) {
 			expect(t).toBeDefined();
 			expect(t.tier).toBe("standard");
 			expect(t.connector).toBeUndefined();
@@ -113,6 +115,39 @@ describe("filter", () => {
 	});
 });
 
+// ── 2b. flatten (issue #113) ───────────────────────────────────────────────────
+describe("flatten", () => {
+	it("collapses an array-of-arrays (grid fan-out result) into one flat record list", async () => {
+		// The exact grid shape: one page of places per cell → [[…],[…]].
+		const aoa = [[{ place_id: "a" }], [{ place_id: "b" }, { place_id: "c" }]];
+		const r = await flattenT.handler(baseCtx, { items: aoa });
+		const parsed = parse(r.content);
+		expect(parsed.count).toBe(3); // 3 places, NOT 2 cells (the GAP #1 fix)
+		expect(parsed.items.map((x: any) => x.place_id)).toEqual(["a", "b", "c"]);
+	});
+
+	it("default depth 1 leaves deeper nesting intact; depth:2 flattens further", async () => {
+		const nested = [[[1], [2]], [[3]]];
+		expect(parse((await flattenT.handler(baseCtx, { items: nested })).content).items).toEqual([[1], [2], [3]]);
+		expect(parse((await flattenT.handler(baseCtx, { items: nested, depth: 2 })).content).items).toEqual([1, 2, 3]);
+	});
+
+	it("depth:0 is a no-op copy; a flat array passes through unchanged", async () => {
+		expect(parse((await flattenT.handler(baseCtx, { items: [[1], [2]], depth: 0 })).content).items).toEqual([[1], [2]]);
+		expect(parse((await flattenT.handler(baseCtx, { items: [1, 2, 3] })).content).items).toEqual([1, 2, 3]);
+	});
+
+	it("path lifts a sub-array from each envelope before concatenating (per-cell http_request results)", async () => {
+		// The real grid shape: a forEach binds one http_request {status,data:[…]} per cell.
+		const perCell = [
+			{ status: 200, data: [{ place_id: "a" }] },
+			{ status: 200, data: [{ place_id: "b" }, { place_id: "c" }] },
+		];
+		const r = await flattenT.handler(baseCtx, { items: perCell, path: "data" });
+		expect(parse(r.content).items.map((x: any) => x.place_id)).toEqual(["a", "b", "c"]);
+	});
+});
+
 // ── 3. dedupe_upsert ─────────────────────────────────────────────────────────────
 // Mock the instance DO: GET records?where= returns a seen map keyed by place_id; POST
 // inserts; PUT updates. Proves insert-vs-update routing + that we never double-insert a
@@ -248,6 +283,70 @@ describe("fan_out — pages (cursor drive + concurrency cap)", () => {
 	});
 });
 
+// ── 4b. enrich (issue #115) ─────────────────────────────────────────────────────
+describe("enrich", () => {
+	it("runs a tool per item ($item template) and merges each result under `as` (via a pure tool)", async () => {
+		// enrich with the pure `map` tool: derive a constant per item, written under `tag`.
+		const r = await enrichT.handler(baseCtx, {
+			items: [{ id: 1 }, { id: 2 }],
+			tool: "map",
+			input: { items: [{}], derive: { flag: "x" } },
+			as: "tag",
+		});
+		const parsed = parse(r.content);
+		expect(parsed.count).toBe(2);
+		// original fields preserved on a COPY; the tool result lands under `as`.
+		expect(parsed.items[0].id).toBe(1);
+		expect(parsed.items[0].tag).toMatchObject({ count: 1 });
+	});
+
+	it("the reachability enrich-merge (the lead-finder 'is this site up' half)", async () => {
+		// enrich with http_reachable; {$item:"websiteUri"} feeds each item's url. Mock the wire:
+		// one live site, one dead host.
+		vi.spyOn(globalThis, "fetch").mockImplementation(async (u: any) => {
+			if (String(u).includes("live")) return new Response("hi", { status: 200 });
+			throw new Error("ECONNREFUSED");
+		});
+		const r = await enrichT.handler(baseCtx, {
+			items: [
+				{ place_id: "a", websiteUri: "https://live.test" },
+				{ place_id: "b", websiteUri: "https://dead.test" },
+			],
+			tool: "http_reachable",
+			input: { url: { $item: "websiteUri" }, timeoutMs: 500 },
+			as: "reachable",
+			concurrency: 2,
+		});
+		const items = parse(r.content).items;
+		// each place now carries {ok,code} under `reachable` — correlated to ITS url.
+		expect(items.find((x: any) => x.place_id === "a").reachable).toMatchObject({ ok: true, code: 200 });
+		expect(items.find((x: any) => x.place_id === "b").reachable).toMatchObject({ ok: false, code: null });
+	});
+
+	it("filter can then test the merged field → 'keep no-website OR unreachable' fully composes", async () => {
+		// The enrich output above, run through filter with the OR predicate.
+		const enriched = [
+			{ place_id: "a", websiteUri: undefined, reachable: undefined }, // no website
+			{ place_id: "b", websiteUri: "https://x", reachable: { ok: false } }, // dead
+			{ place_id: "c", websiteUri: "https://y", reachable: { ok: true } }, // healthy → drop
+		];
+		const r = await filterT.handler(baseCtx, {
+			items: enriched,
+			any: true,
+			where: [
+				{ field: "websiteUri", op: "missing" },
+				{ field: "reachable.ok", op: "eq", value: false },
+			],
+		});
+		expect(parse(r.content).items.map((x: any) => x.place_id)).toEqual(["a", "b"]);
+	});
+
+	it("fails without tool/as", async () => {
+		expect((await enrichT.handler(baseCtx, { items: [{ a: 1 }], as: "x" } as any)).success).toBe(false);
+		expect((await enrichT.handler(baseCtx, { items: [{ a: 1 }], tool: "map" } as any)).success).toBe(false);
+	});
+});
+
 // ── 5. http_reachable ────────────────────────────────────────────────────────────
 describe("http_reachable", () => {
 	it("classifies a live server (200) as ok with its code", async () => {
diff --git a/workers/api/src/lib/steps.ts b/workers/api/src/lib/steps.ts
index b23f82ff..2053466f 100644
--- a/workers/api/src/lib/steps.ts
+++ b/workers/api/src/lib/steps.ts
@@ -49,6 +49,34 @@ function fail(content: string): RegistryToolResult {
 	return { content, success: false };
 }
 
+/** Concatenate an array-of-arrays into a flat array, `depth` levels deep (default 1). Used by
+ *  the `flatten` step (#113) to collapse a grid fan-out's per-cell page arrays into one list. */
+function flattenDeep(items: unknown[], depth: number): unknown[] {
+	if (depth <= 0) return items.slice();
+	const out: unknown[] = [];
+	for (const el of items) {
+		if (Array.isArray(el)) out.push(...flattenDeep(el, depth - 1));
+		else out.push(el);
+	}
+	return out;
+}
+
+/**
+ * Resolve an `enrich` input template against one item (#115). A `{$item:"path"}` node reads a
+ * dotted field off the item (the documented per-item ergonomics — the sibling of the pipeline
+ * runner's `{$param:"item.<path>"}` from #114); everything else passes through, recursing into
+ * objects/arrays so nested `$item` references work.
+ */
+function resolveItemTemplate(tpl: unknown, item: unknown): unknown {
+	if (tpl === null || typeof tpl !== "object") return tpl;
+	if (Array.isArray(tpl)) return tpl.map((v) => resolveItemTemplate(v, item));
+	const obj = tpl as Record<string, unknown>;
+	if (typeof obj.$item === "string") return getPath(item, obj.$item);
+	const out: Record<string, unknown> = {};
+	for (const [k, v] of Object.entries(obj)) out[k] = resolveItemTemplate(v, item);
+	return out;
+}
+
 // ── 1. map ───────────────────────────────────────────────────────────────────
 // Pure reshape of each item: rename fields, derive constants/copies, and extract nested
 // values by the same dotted / `arr[]` grammar as the #95 responseMap (reused via getPath).
@@ -295,6 +323,34 @@ export const STEP_TOOLS: ToolDef[] = [
 		},
 	},
 
+	// 2b ─ flatten (issue #113)
+	{
+		name: "flatten",
+		tier: "standard",
+		scope: "read",
+		description:
+			"Concatenate an array-of-arrays into a flat array (pure, no I/O). `depth` (default 1) is how many nesting levels to collapse. Optional `path`: when a `forEach` binds an array of RESULT ENVELOPES (e.g. one http_request `{status,data:[…]}` per grid cell), `path` pulls that sub-array (\"data\") from each element FIRST, so the grid fan-out (`fan_out` grid → `forEach` http_request → flatten) collapses to one flat list of records for `map`/`filter`/`dedupe_upsert`. Returns the flattened array.",
+		jsonSchema: {
+			type: "object",
+			properties: {
+				items: { type: "array", description: "The array to flatten (an array-of-arrays, or an array of result envelopes when `path` is set)." },
+				depth: { type: "number", description: "Levels to flatten (default 1)." },
+				path: { type: "string", description: 'Optional dotted path pulled from each element before concatenating, e.g. "data".' },
+			},
+			required: [],
+		},
+		handler: async (_ctx, input) => {
+			const raw = Array.isArray(input.items) ? input.items : asArray(input.items);
+			// With `path`, lift the sub-array out of each envelope element first (array of
+			// {…,data:[…]} → array-of-arrays); then the same depth flatten collapses it.
+			const path = typeof input.path === "string" && input.path ? input.path : "";
+			const items = path ? raw.map((el) => getPath(el, path)) : raw;
+			const depth = input.depth === undefined ? 1 : Math.max(0, Math.floor(Number(input.depth) || 0));
+			const out = flattenDeep(items, depth);
+			return ok(JSON.stringify({ items: out, count: out.length }, null, 2));
+		},
+	},
+
 	// 3 ─ dedupe_upsert
 	{
 		name: "dedupe_upsert",
@@ -417,6 +473,47 @@ export const STEP_TOOLS: ToolDef[] = [
 		},
 	},
 
+	// 4b ─ enrich (issue #115)
+	{
+		name: "enrich",
+		tier: "standard",
+		scope: "read",
+		description:
+			"Call a tool once PER record and merge each result back onto the record — the general 'enrich a list with a per-item connector call' primitive. `tool` is dispatched with `input`, a template resolved against each item: a `{\"$item\":\"path\"}` node reads that dotted field off the item (e.g. `{url:{\"$item\":\"websiteUri\"}}`). The tool's parsed result is written under key `as` on a COPY of the item. Concurrency-capped (default 4, max 20). Powers 'keep no-website OR unreachable' (enrich with http_reachable) and #99 socials/email (enrich with web_search → extract_contacts). Returns {items,count}.",
+		jsonSchema: {
+			type: "object",
+			properties: {
+				items: { type: "array", description: "Records to enrich (a single object is treated as a one-element array)." },
+				tool: { type: "string", description: "Registry tool to run once per item, e.g. \"http_reachable\"." },
+				input: { type: "object", description: 'Input template for `tool`; `{"$item":"path"}` reads a dotted field off the item.' },
+				as: { type: "string", description: "Key on each item to write the tool's result under, e.g. \"reachable\"." },
+				concurrency: { type: "number", description: "Max concurrent per-item calls (default 4, max 20)." },
+			},
+			required: ["tool", "as"],
+		},
+		handler: async (ctx, input) => {
+			const items = asArray(input.items).filter(isRecord) as Record<string, unknown>[];
+			const tool = String(input.tool || "");
+			const as = String(input.as || "");
+			if (!tool || !as) return fail("enrich needs `tool` and `as`.");
+			const template = isRecord(input.input) ? input.input : {};
+			const concurrency = Number(input.concurrency) || 4;
+			const { runRegistryTool } = await import("./tool-registry.js");
+			const out = await mapWithConcurrency(items, concurrency, async (item) => {
+				const perItemInput = resolveItemTemplate(template, item) as Record<string, unknown>;
+				const r = await runRegistryTool(tool, ctx, perItemInput);
+				let result: unknown;
+				try {
+					result = JSON.parse(r.content);
+				} catch {
+					result = r.content;
+				}
+				return { ...item, [as]: result };
+			});
+			return ok(JSON.stringify({ items: out, count: out.length }, null, 2));
+		},
+	},
+
 	// 5 ─ http_reachable
 	{
 		name: "http_reachable",