fix(engine): bound the flowline layer at both ends

.github/workflows/checks.yml

@@ -42,6 +42,9 @@ jobs:
       - name: Trace direction
         run: npm run check-trace-direction
 
+      - name: Flowline scope
+        run: npm run check-flowline-scope
+
       - name: Cache key
         run: npm run check-cache-key
 

package.json

@@ -13,6 +13,7 @@
     "check-cache-key": "tsx scripts/check-cache-key.mts",
     "check-step-labels": "tsx scripts/check-step-labels.mts",
     "check-trace-direction": "tsx scripts/check-trace-direction.mts",
+    "check-flowline-scope": "tsx scripts/check-flowline-scope.mts",
     "check-substance-labels": "tsx scripts/check-substance-labels.mts",
     "check-query-joins": "tsx scripts/check-query-joins.mts",
     "query-matrix": "tsx scripts/query-matrix.mts",

scripts/check-flowline-scope.mts

@@ -0,0 +1,94 @@
+// Self-check: the supporting flowline layer must be bounded at BOTH ends.
+//
+//   npx tsx scripts/check-flowline-scope.mts
+//
+// buildFusedFlowlineQuery traces a transitive closure out of the anchor cells.
+// Left one-sided it runs to the end of the network, so an anchor near a
+// drainage divide drags in the neighbouring basin: "facilities upstream from
+// PFOS samples in York County, ME" drew 122 segments of the Merrimack and 21
+// of the Winnipesaukee, in a basin no York sample drains from. Nothing looked
+// broken -- the map just had extra rivers on it, and they were real rivers.
+//
+// The fix intersects that closure with the set of flowlines that actually
+// reach the resolved targets. This asserts the intersection is present in
+// every shape that draws the layer, bounded and unbounded, so a future edit to
+// either branch cannot drop it silently.
+//
+// No network: the pipeline steps only build query strings.
+import assert from 'node:assert/strict';
+import { planPipeline, type PipelineContext } from '../src/engine/planner';
+import type { AnalysisQuestion, EntityType, SpatialRelationship } from '../src/types/query';
+
+const ENTITY_TYPES: EntityType[] = [
+  'samples',
+  'facilities',
+  'waterBodies',
+  'wells',
+  'aquifers',
+  'streams',
+];
+const HYDROLOGY: SpatialRelationship['type'][] = ['downstream', 'upstream'];
+// Unbounded, and one bounded value to cover the aggregate branch.
+const DISTANCES: (number | undefined)[] = [undefined, 25];
+
+const context = (): PipelineContext => ({
+  question: {} as AnalysisQuestion,
+  targetIris: ['https://example.org/t1', 'https://example.org/t2'],
+  anchorIris: ['https://example.org/a1', 'https://example.org/a2'],
+  results: {},
+});
+
+let checks = 0;
+let drawn = 0;
+
+for (const blockA of ENTITY_TYPES) {
+  for (const blockC of ENTITY_TYPES) {
+    for (const rel of HYDROLOGY) {
+      for (const maxDistanceKm of DISTANCES) {
+        const question: AnalysisQuestion = {
+          blockA: { type: blockA },
+          relationship: { type: rel, maxDistanceKm },
+          blockC: { type: blockC },
+        };
+        const step = planPipeline(question).find((s) => s.type === 'GET_FLOWLINE_GEOMETRIES');
+        // Skipped when a side is streams: those flowlines are the answer set.
+        if (!step) {
+          assert.ok(
+            blockA === 'streams' || blockC === 'streams',
+            `${blockA} ${rel} ${blockC}: no flowline step, but neither side is streams`,
+          );
+          continue;
+        }
+        drawn++;
+        const sparql = step.buildQuery(context());
+        const where = `${blockA} ${rel} ${blockC} ${maxDistanceKm ?? 'unbounded'}`;
+
+        // The target IRIs have to reach the query, or the closure is one-sided.
+        assert.ok(
+          sparql.includes('https://example.org/t1'),
+          `${where}: flowline query does not bind the resolved targets`,
+        );
+        // Both ends present: a closure out of the anchor cells, and a closure
+        // into the targets, joined on ?flowline.
+        assert.ok(
+          sparql.includes('?_flTarget spatial:connectedTo ?s2celltarget'),
+          `${where}: missing the target reach clause`,
+        );
+        assert.match(
+          sparql,
+          /\?flowline hyf:downstreamFlowPathTC \?_flTarget|\?_flTarget hyf:downstreamFlowPathTC \?flowline/,
+          `${where}: target reach clause does not close on ?flowline`,
+        );
+        // The reach clause must be a bare TC. `TC?` was measured: it recovers
+        // 0 flowlines and costs 8s on the York question.
+        assert.ok(
+          !sparql.includes('downstreamFlowPathTC? ?_flTarget'),
+          `${where}: reflexive target reach, measured to add nothing`,
+        );
+        checks += 4;
+      }
+    }
+  }
+}
+
+console.log(`flowline scope: ${drawn} shapes draw the layer, ${checks} checks passed`);

src/engine/planner.ts

@@ -328,8 +328,12 @@ function buildFusedSteps(question: AnalysisQuestion): PipelineStep[] {
       buildQuery: (ctx, scope) =>
         buildFusedFlowlineQuery({
           anchor: anchorBlock,
+          target: targetBlock,
           direction: traceDirection,
           anchorIris: scope?.anchorIris ?? ctx.anchorIris,
+          // Never sliced: the target set bounds the trace, so a partial list
+          // would silently shrink the drawn network rather than split it.
+          targetIris: ctx.targetIris,
           maxDistanceKm: relationship.maxDistanceKm,
         }),
       initialScopes: (ctx) => iriScopes(ctx.anchorIris, 'anchor'),

src/engine/templates/fusedQueries.ts

@@ -581,11 +581,63 @@ export function buildFusedWellQuery(opts: FusedWellSideOpts): string {
 
 export interface FusedFlowlineOpts {
   anchor: EntityBlock;
+  target: EntityBlock;
   direction: 'downstream' | 'upstream';
   anchorIris: string[];
+  targetIris: string[];
   maxDistanceKm?: number;
 }
 
+// Restricts the traced closure to flowlines that actually connect the anchors
+// to the targets. Without it the trace is one-sided: it runs from the anchor
+// cells to the end of the network, so an anchor sitting near a drainage divide
+// drags in the whole of the neighbouring basin. On "facilities upstream from
+// PFOS samples in York County, ME" that drew 122 segments of the Merrimack and
+// 21 of the Winnipesaukee, ~130km away in a basin no York sample drains from.
+//
+// Written as an intersection of two DISTINCT closures rather than the direct
+// `?flowline hyf:downstreamFlowPathTC? ?_flEnd` membership test: that form
+// leaves ?flowline unbound on the left and QLever times out joining on it
+// (31s, "Join on ?flowline"). The intersection runs in the same 7s the
+// one-sided query took.
+//
+// The closure is strict on both ends, so the segment a target sits on is not
+// drawn. Making the target side reflexive (`TC?`) was measured and recovers
+// nothing (0 extra flowlines) while costing 8s, because that segment already
+// arrives via another target upstream of it.
+function targetReachClause(opts: FusedFlowlineOpts): string {
+  // ponytail: target IRIs inlined whole. Only anchors are sliced today
+  // (iriScopes/divideIriList in the planner); add target slicing if a question
+  // ever pushes this past the gateway's body limit.
+  const targetValues = `VALUES ${entityIriVar(opts.target, 'C')} { ${opts.targetIris
+    .map(wrapUri)
+    .join(' ')} }`;
+  // Filters stripped: the pinned IRIs are already the filtered answer set, so
+  // every filter clause here is a join that can only re-confirm what the VALUES
+  // list states. Passing the bare block rather than opts.target keeps this to
+  // the cell-membership triple. Measured on the York question: re-deriving the
+  // substance filter cost 1s of the 8 and changed nothing.
+  const targetBind = bindEntityInCell({ type: opts.target.type }, '?s2celltarget', 'C', false);
+  const reach =
+    opts.direction === 'downstream'
+      ? `?flowline hyf:downstreamFlowPathTC ?_flTarget .`
+      : `?_flTarget hyf:downstreamFlowPathTC ?flowline .`;
+
+  return `{
+        SELECT DISTINCT ?flowline WHERE {
+          {
+            SELECT DISTINCT ?s2celltarget WHERE {
+              ${targetValues}
+              ?s2celltarget rdf:type kwg-ont:S2Cell_Level13 .
+              ${targetBind}
+            }
+          }
+          ?_flTarget spatial:connectedTo ?s2celltarget .
+          ${reach}
+        }
+      }`;
+}
+
 // Returns flowline geometries traced from the anchor entities the pipeline
 // already resolved in FIND_ANCHOR_IRIS.
 //
@@ -624,13 +676,20 @@ export function buildFusedFlowlineQuery(opts: FusedFlowlineOpts): string {
             spatial:connectedTo ?s2cellus .
         ?flowline hyf:downstreamFlowPathTC ?downstream_flowline .`;
 
+  const reachesTarget = targetReachClause(opts);
+
   // Unbounded: the flowline set is the plain transitive closure.
   if (!opts.maxDistanceKm) {
     return `
     ${PREFIXES}
     SELECT DISTINCT ?flowline ?flowlineWKT ?fl_type ?streamName WHERE {
-      ${seedCells}
-      ${flowlinePattern}
+      {
+        SELECT DISTINCT ?flowline WHERE {
+          ${seedCells}
+          ${flowlinePattern}
+        }
+      }
+      ${reachesTarget}
       ?flowline geo:hasGeometry/geo:asWKT ?flowlineWKT ;
                 nhdplusv2:hasFTYPE ?fl_type .
       OPTIONAL { ?flowline rdfs:label ?streamName }
@@ -682,6 +741,7 @@ export function buildFusedFlowlineQuery(opts: FusedFlowlineOpts): string {
           BIND(xsd:float(?_plen) + xsd:float(?_extra) AS ?_ptotal)
         } GROUP BY ?flowline
       }
+      ${reachesTarget}
       ?flowline geo:hasGeometry/geo:asWKT ?flowlineWKT ;
                 nhdplusv2:hasFTYPE ?fl_type .
       OPTIONAL { ?flowline rdfs:label ?streamName }