fix(db): prevent nested-loop observation scoring

hindsight-api-slim/hindsight_api/engine/db/ops_postgresql.py

@@ -936,6 +936,14 @@ async def expand_observations(
         # grow hubs far past that, re-measure before assuming this is still the
         # right shape.
         #
+        # Even with that traceable column, bank-wide unit_id statistics can still
+        # collapse the connected_sources estimate to one row (#4163). An inner
+        # join may then become a nested loop over every candidate/source pair.
+        # Keep scored materialized and use a FULL JOIN: PostgreSQL cannot execute
+        # an equality FULL JOIN as a nested loop, so it must hash or merge even
+        # when the estimate is wrong. Unmatched rows form a single NULL-id group
+        # which the following inner join to candidates naturally discards.
+        #
         # Entity/source traversal and semantic/causal expansion run as ONE query
         # (#3857): the observation entity arm is fused into the semantic/causal CTE
         # query behind an 'entity' source discriminator, like the non-observation
@@ -987,11 +995,11 @@ async def expand_observations(
                   AND mu.source_memory_ids && ca.source_ids
                   {window.clause("mu")}
             ),
-            scored AS (
+            scored AS MATERIALIZED (
                 SELECT c.id, COUNT(DISTINCT cs.source_id)::float AS score
                 FROM candidates c
                 CROSS JOIN LATERAL unnest(c.source_memory_ids) AS s(source_id)
-                JOIN connected_sources cs ON cs.source_id = s.source_id
+                FULL OUTER JOIN connected_sources cs ON cs.source_id = s.source_id
                 GROUP BY c.id
             ),
             observation_entity_expanded AS (

hindsight-api-slim/tests/test_observation_expansion_single_fetch.py

@@ -418,6 +418,29 @@ async def test_postgresql_fused_query_is_one_statement():
     assert sql.count("$1::uuid[]") >= 4, "the seed bind reaches every arm"
 
 
+@pytest.mark.asyncio
+async def test_postgresql_scoring_join_excludes_nested_loop_plan():
+    """The scored join stays hash/merge-capable when source estimates collapse (#4163)."""
+    conn = AsyncMock()
+    conn.fetch.return_value = []
+
+    await PostgreSQLOps().expand_observations(
+        conn,
+        "memory_units",
+        "unit_entities",
+        "memory_links",
+        [uuid.uuid4()],
+        100,
+        200,
+        UpdatedWindow(after=None, before=None, first_param_index=3),
+    )
+
+    normalized_sql = " ".join(conn.fetch.await_args.args[0].split())
+    assert "scored AS MATERIALIZED" in normalized_sql
+    assert "FULL OUTER JOIN connected_sources cs ON cs.source_id = s.source_id" in normalized_sql
+    assert "GROUP BY c.id" in normalized_sql
+
+
 @pytest.mark.asyncio
 async def test_oracle_fused_query_is_one_statement():
     """Structural (no Oracle runtime): Oracle emits ONE statement, three arms.