99normal callback attempt may try the alternate dialect. This prevents an AT
1010query from following a framed query (or vice versa) on one stream while keeping
1111the established one-``set>server``-per-attempt contract intact.
12+ A short heartbeat learned during FC1 permits a bounded FC2 identity upgrade on
13+ that now-known framed socket; it does not permit claiming the short identity.
1214"""
1315
1416from __future__ import annotations
2527from .identity_probe import (
2628 PROBE_AT_DTUPN ,
2729 PROBE_FRAMED_FC1 ,
30+ PROBE_FRAMED_FC2 ,
2831 silent_probe_kind_for_protocol ,
2932)
3033from .silent_session_probe import SessionObservation
@@ -281,6 +284,48 @@ async def async_negotiate(
281284 session_protocol = protocol ,
282285 identity_probe_kind = probe_kind ,
283286 )
287+ post = next (
288+ (observation for observation in channel .snapshot_session_observations ()
289+ if observation .session_id == selected .session_id ),
290+ selected ,
291+ )
292+ if (
293+ not identity_source_is_strong (post .identity_source )
294+ and unknown_wire
295+ and protocol == "eybond_framed"
296+ and post .identity_source == "framed_heartbeat"
297+ and validated_collector_pn (post .collector_pn )
298+ and post .state not in _TERMINAL_STATES
299+ and not post .state .startswith ("closed" )
300+ ):
301+ # FC1 can reveal only a short PN on an initially silent socket.
302+ # This proves the framed dialect, NOT ownership. Upgrade that
303+ # exact stream with FC2, rather than discarding the observation
304+ # and trying AT on the next connection. Never mix dialects or
305+ # extend the caller's deadline; a failed upgrade stays weak.
306+ unknown_wire = False
307+ if not pn_is_same_identity (durable_pn , post .collector_pn ):
308+ return ExactSessionIdentityResult (
309+ NEGOTIATION_FOREIGN_IDENTITY ,
310+ session_id = selected .session_id ,
311+ collector_pn = post .collector_pn ,
312+ session_protocol = protocol ,
313+ identity_source = post .identity_source ,
314+ )
315+ remaining = deadline - loop .time ()
316+ if remaining > 0 :
317+ probe_kind = PROBE_FRAMED_FC2
318+ try :
319+ identified_pn = await asyncio .wait_for (
320+ channel .async_identify_exact_session (
321+ selected .session_id ,
322+ session_protocol = protocol ,
323+ identity_probe_kind = probe_kind ,
324+ ),
325+ timeout = remaining ,
326+ )
327+ except asyncio .TimeoutError :
328+ identified_pn = ""
284329 except asyncio .CancelledError as original_cancel :
285330 if unknown_wire :
286331 try :
@@ -292,7 +337,17 @@ async def async_negotiate(
292337 pass
293338 raise original_cancel
294339
295- strong_pn = validated_collector_pn (identified_pn )
340+ post = next (
341+ (observation for observation in channel .snapshot_session_observations ()
342+ if observation .session_id == selected .session_id ),
343+ selected ,
344+ )
345+ strong_pn = (
346+ validated_collector_pn (post .collector_pn )
347+ if identity_source_is_strong (post .identity_source )
348+ and pn_is_same_identity (identified_pn , post .collector_pn )
349+ else ""
350+ )
296351 if strong_pn :
297352 if not pn_is_same_identity (durable_pn , strong_pn ):
298353 return ExactSessionIdentityResult (
@@ -302,14 +357,6 @@ async def async_negotiate(
302357 session_protocol = protocol ,
303358 probe_kind = probe_kind ,
304359 )
305- post = next (
306- (
307- observation
308- for observation in channel .snapshot_session_observations ()
309- if observation .session_id == selected .session_id
310- ),
311- selected ,
312- )
313360 self ._reset_unknown_candidate ()
314361 return ExactSessionIdentityResult (
315362 NEGOTIATION_IDENTIFIED ,
@@ -337,7 +384,7 @@ async def async_negotiate(
337384 NEGOTIATION_PROBE_FAILED ,
338385 session_id = selected .session_id ,
339386 session_protocol = protocol ,
340- identity_source = selected .identity_source ,
387+ identity_source = post .identity_source ,
341388 probe_kind = probe_kind ,
342389 )
343390
0 commit comments