fix: fetch open runs by key and run_id

app/crm/outreach/db/accessors/enrollment.py

@@ -5,7 +5,7 @@
 """
 
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Sequence, Tuple
 
 import asyncpg
 
@@ -252,9 +252,14 @@ async def get_run(
 
 
 async def open_runs_for_customer(
-    merchant_id: str, customer_id: str
+    merchant_id: str,
+    customer_id: str,
+    keys: Sequence[Tuple[str, str]] = (),
+    run_ids: Sequence[str] = (),
 ) -> List[EnrollmentRun]:
-    query, values = open_runs_for_customer_query(merchant_id, customer_id)
+    query, values = open_runs_for_customer_query(
+        merchant_id, customer_id, keys, run_ids
+    )
     async with crm_connection() as conn:
         rows = await conn.fetch(query, *values)
     return [decode_run(row) for row in rows]

app/crm/outreach/db/queries/enrollment.py

@@ -5,7 +5,7 @@
 
 import json
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Sequence, Tuple
 
 from app.crm.outreach.db.queries.step import flush_arm
 from app.crm.outreach.db.queries.tables import (
@@ -354,19 +354,65 @@ def record_run_error_query(
 
 
 def open_runs_for_customer_query(
-    merchant_id: str, customer_id: str
+    merchant_id: str,
+    customer_id: str,
+    keys: Sequence[Tuple[str, str]] = (),
+    run_ids: Sequence[str] = (),
 ) -> Tuple[str, List[Any]]:
     """The consumer's per-run read (rollout phase 13): this customer's
     open runs across every plan, so goals and listening are judged
     against each run's PINNED version. The customer index carries it; a
-    customer with nothing open costs one empty indexed read."""
+    customer with nothing open costs one empty indexed read.
+
+    ``keys`` adds the open runs of a keyed plan by (workflow_id,
+    enrollment_key) — the letter's value for the key the plan declared.
+    A merchant's own customer id can arrive with a second phone, which
+    resolves to a second customer; by customer alone that letter would
+    never reach the run its key names (Flipkart, 13 Oct 2026). The open
+    partial unique carries this half.
+
+    ``run_ids`` adds the runs the letter names by id — our own call
+    report's enrollment_id, the run whose send a reply answers — exact
+    whichever phone was dialled. A primary-key read.
+
+    A run found more than one way is read once; every half is still only
+    a CANDIDATE — each run's goal key and square match decide whether the
+    letter is about it. A letter that names neither (most) reads by
+    customer alone — the release statement, unchanged."""
+    if not keys and not run_ids:
+        query = f"""
+            SELECT {_RUN_COLUMNS}
+            FROM {ENROLLMENT_TABLE}
+            WHERE merchant_id = $1 AND customer_id = $2 AND status <> 'exited'
+            ORDER BY entered_at, id
+        """
+        return query, [merchant_id, customer_id]
     query = f"""
         SELECT {_RUN_COLUMNS}
         FROM {ENROLLMENT_TABLE}
-        WHERE merchant_id = $1 AND customer_id = $2 AND status <> 'exited'
+        WHERE id IN (
+            SELECT id FROM {ENROLLMENT_TABLE}
+            WHERE merchant_id = $1 AND customer_id = $2 AND status <> 'exited'
+            UNION
+            SELECT e.id
+            FROM {ENROLLMENT_TABLE} e
+            JOIN unnest($3::uuid[], $4::text[]) AS k(workflow_id, enrollment_key)
+              ON e.workflow_id = k.workflow_id
+             AND e.enrollment_key = k.enrollment_key
+            WHERE e.merchant_id = $1 AND e.status <> 'exited'
+            UNION
+            SELECT id FROM {ENROLLMENT_TABLE}
+            WHERE merchant_id = $1 AND id = ANY($5::uuid[]) AND status <> 'exited'
+        )
         ORDER BY entered_at, id
     """
-    return query, [merchant_id, customer_id]
+    return query, [
+        merchant_id,
+        customer_id,
+        [workflow_id for workflow_id, _ in keys],
+        [key for _, key in keys],
+        list(run_ids),
+    ]
 
 
 def resume_run_by_id_query(

app/crm/outreach/entry.py

@@ -23,7 +23,7 @@
 source-event check and the open-run unique — not by that rollback.
 """
 
-from typing import Any, Dict, Optional, Sequence, Tuple
+from typing import Any, Dict, List, Optional, Sequence, Tuple
 
 from app.core.config.dynamic import CRM_CONTEXT_VALUE_MAX_CHARS
 from app.core.logger import logger
@@ -130,15 +130,24 @@ async def consume_attributed_event(
     can never see."""
     if customer_id is None:
         return  # not about a person: nothing to admit, end or wake
-    open_runs = await enrollment_accessor.open_runs_for_customer(
-        event.merchant_id, customer_id
-    )
-    goal_patch = _goal_patch(event) if open_runs else None
+    plans = [
+        (flow, WorkflowDefinition.model_validate(flow.definition))
+        for flow in await workflow_accessor.live_workflows(event.merchant_id)
+    ]
     # Whose send this letter answers, resolved ONCE for the letter rather
     # than once per run: a reply is addressed, not matched (see
     # reply_attribution). None for anything that answers nothing of ours,
-    # and then every square below behaves as it always did.
-    addressed = await addressed_run(event) if open_runs else None
+    # and then every square below behaves as it always did. Before the
+    # read, so the run she answered is read even when her number resolved
+    # to another customer.
+    addressed = await addressed_run(event)
+    open_runs = await enrollment_accessor.open_runs_for_customer(
+        event.merchant_id,
+        customer_id,
+        keys=_run_keys(plans, event),
+        run_ids=_named_runs(event, addressed),
+    )
+    goal_patch = _goal_patch(event) if open_runs else None
     if addressed is not None and addressed.run_id not in {str(r.id) for r in open_runs}:
         # She answered a run that has since ended (its goal fired, it timed
         # out, she was ejected). Narrowing to a run nobody is holding would
@@ -167,9 +176,7 @@ async def consume_attributed_event(
             continue  # exited: there is nothing left to wake
         await _wake_on_reply(run, definition, event, variables, addressed)
 
-    flows = await workflow_accessor.live_workflows(event.merchant_id)
-    for flow in flows:
-        definition = WorkflowDefinition.model_validate(flow.definition)
+    for flow, definition in plans:
         for door in definition.entries:
             if door.topic == event.topic and _where_matches(door, event):
                 await _try_enrol(
@@ -186,6 +193,47 @@ async def consume_attributed_event(
                 break  # topics are unique across a plan's doors
 
 
+def _run_keys(
+    plans: Sequence[Tuple[Workflow, WorkflowDefinition]], event: RawEvent
+) -> List[Tuple[str, str]]:
+    """PURE: (workflow_id, key) for every live keyed plan this letter
+    carries a key for — the runs it names whichever customer it resolved
+    to. A merchant's own customer id arriving with a second phone resolves
+    to a second customer; by customer alone, her goal and her progress
+    would never reach the run her key opened (Flipkart, 13 Oct 2026)."""
+    keys: List[Tuple[str, str]] = []
+    for flow, definition in plans:
+        for field in {door.key for door in definition.entries if door.key}:
+            value = field_value(
+                event.payload,
+                canonical_path(field),
+                derive_for(event.source, event.topic),
+            )
+            if value not in (None, ""):
+                keys.append((str(flow.id), str(value)))
+    return keys
+
+
+def _named_runs(event: RawEvent, addressed: Optional[Addressed]) -> List[str]:
+    """PURE: the runs this letter names by id — the run our own call report
+    says placed the call (enrollment_id, stamped on the lead by the call
+    square), and the run whose send a reply answers (reply_attribution).
+    Exact whichever phone was dialled or customer it resolved to: a run
+    whose number changed would otherwise never hear its own call's outcome
+    or the answer to its own message.
+
+    Only our own reports are trusted to name a run: an enrollment_id in a
+    merchant's payload is its data, not our pointer. Both ids come from uuid
+    columns; a lead no run placed carries none, and an empty id names
+    nothing."""
+    named: List[str] = []
+    if event.source in CALL_REPORT_SOURCES:
+        named.append(str(event.payload.get("enrollment_id") or ""))
+    if addressed is not None:
+        named.append(addressed.run_id)
+    return [run_id for run_id in named if run_id]
+
+
 async def _end_on_goal(
     run: EnrollmentRun,
     definition: WorkflowDefinition,
@@ -613,10 +661,10 @@ async def _repeat_door(
     never find a v3 run. That version's door for this topic; if it has
     none (the door is newer than the run), its first door — the patch's
     guard makes a wrong square a no-op, never a wrong write. The run was
-    read at the top of this pass; when it is not among her open runs (a
-    keyed run that resolved to another customer, or a sibling tick opened
-    it after the read) the latest door stands in — exactly the
-    pre-pinning behaviour."""
+    read at the top of this pass — by her customer and by its key, so a
+    keyed run on another customer is among them; when it is not (a
+    sibling tick opened it after the read) the latest door stands in —
+    exactly the pre-pinning behaviour."""
     for run in open_runs:
         if (
             str(run.workflow_id) == str(flow.id)

app/crm/outreach/reply_attribution.py

@@ -23,9 +23,11 @@
 What it narrows, and what it cannot: a customer with three open orders taps
 CONFIRM on one of them, and the other two runs must not hear it — her single
 tap once resolved all three and cancelled a confirmed order (10 Sep 2026,
-docs/crm/reply-run-matching.md). Attribution only ever NARROWS, so a letter
-it cannot address behaves exactly as it did before this existed. That is why
-no published plan has to change and no open run has to be re-pinned.
+docs/crm/reply-run-matching.md). Attribution narrows her other runs out, and
+it also FINDS the run it names: entry.py reads that run by id beside her own,
+so a run whose number resolved to another customer still hears its answer. A
+letter it cannot address behaves exactly as it did before this existed. That
+is why no published plan has to change and no open run has to be re-pinned.
 
 A source opts in by declaring a ``replied_to`` field — the id of the message
 this one answers, in whatever the provider calls it (Meta's context.id). A

docs/crm/reply-run-matching.md

@@ -62,8 +62,11 @@ own sends, and there it needs no declaration at all.
 
 ## What stays exactly as it was
 
-Attribution only ever NARROWS, so anything it cannot address behaves as it always did. That is
-why **no published plan has to change** and no run has to be re-pinned:
+Attribution narrows her other runs out, and it also **finds** the run it names: the consumer
+reads that run by id beside her own open runs, so a run whose number later resolved to a
+different customer still hears the answer to its own message. Anything it cannot address
+behaves as it always did. That is why **no published plan has to change** and no run has to be
+re-pinned:
 
 - **She typed instead of tapping.** No thread on the letter, nothing to look up, every
   listening square hears it — as before.

tests/crm/test_reply_attribution.py

@@ -135,7 +135,9 @@ def __init__(self, flow: Workflow, runs: List[EnrollmentRun]) -> None:
     async def live_workflows(self, merchant_id: str) -> List[Workflow]:
         return []
 
-    async def open_runs_for_customer(self, merchant_id, customer_id):
+    async def open_runs_for_customer(
+        self, merchant_id, customer_id, keys=(), run_ids=()
+    ):
         return list(self.runs)
 
     async def get_definition(self, merchant_id, workflow_id, version):
@@ -434,3 +436,35 @@ def test_the_word_it_filters_on_is_the_word_the_send_queues_with() -> None:
 
     assert attribution._WORKFLOW in SOURCE_KINDS
     assert f'source_kind="{attribution._WORKFLOW}"' in inspect.getsource(send_node)
+
+
+def test_a_reply_finds_its_run_when_her_number_resolved_to_another_customer(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The run's number changed to one that resolves to another customer:
+    by customer alone the read finds nothing. The reply names the run that
+    sent the message, so the read takes that run by id and it hears its
+    answer."""
+    flow = _flow()
+    run = _run(flow)
+
+    class _Elsewhere(_Spine):
+        async def open_runs_for_customer(
+            self, merchant_id, customer_id, keys=(), run_ids=()
+        ):
+            return [r for r in self.runs if str(r.id) in run_ids]
+
+    spine = _Elsewhere(flow, [run])
+    _install(
+        monkeypatch,
+        spine,
+        SendBehind(
+            source_kind="workflow",
+            source_id=str(run.id),
+            dedupe_key=f"{run.id}:confirm",
+        ),
+    )
+
+    _consume(_tap())
+
+    assert {r for r, _ in spine.resumes} == {str(run.id)}

tests/crm/test_worker_runtime.py

@@ -280,7 +280,7 @@ async def live_workflows(merchant_id: str) -> List[Workflow]:
         return [flow] if merchant_id == flow.merchant_id else []
 
     async def open_runs_for_customer(
-        merchant_id: str, customer_id: str
+        merchant_id: str, customer_id: str, keys: Any = (), run_ids: Any = ()
     ) -> List[EnrollmentRun]:
         return [run] if merchant_id == flow.merchant_id else []