
    ߩ^j1                         d Z ddlZddlZddlZddlmZmZ  ej        e          Z	dZ
dZ G d d          Z G d d	          Z G d
 d          ZdS )u!  Per-session turn lease — serializes the [load history → run → flush] region.

Why this exists (#64934): the gateway's busy guards are keyed by ROUTING KEY
(``_active_sessions`` in the adapter, ``_running_agents`` in the runner), but
the durable transcript is owned by SESSION_ID — and ``switch_session()`` makes
the key→id mapping many-to-one (``/resume`` of a named session from a second
chat/topic, CLI-continuity rebinding, async-delegation completion pinning,
Telegram topic-binding tip-walks). Two routing keys mapped to one session_id
run concurrent turns on two different agent objects, so no per-key guard ever
sees the collision. The two turns then interleave their flushes on one
transcript: rows persist in completion order instead of arrival order, the
identity-marker dedup over shared history dicts can swallow a row outright,
and the second turn runs on a history base that never saw the first turn's
exchange — leaving a permanent ``user;user`` alternation wedge that
``repair_message_sequence`` re-repairs on every request forever.

The lease closes that route by serializing per RESOLVED session_id: it is
acquired after session resolution is final (post ``switch_session``/tip-walk),
immediately before the transcript load, and released in the dispatch layer's
``finally`` on every exit path. Same-key messages never reach the acquisition
point while a turn runs (both routing-key guards hold them), so the lock is
uncontended everywhere except the alias-key route — where the second turn now
waits for the first turn's flush and logs one WARNING naming the session and
both routing keys (pairing with the cross-agent tripwire in
``agent/agent_runtime_helpers.note_turn_start``).

Safety properties:

- **Generation-scoped, identity-checked release.** A token records its owner
  (routing key, run generation) and release only frees the lease when that
  exact token is the current holder — a stale unwind can never release a
  newer turn's lease (the #28686 ownership lesson applied). Release is
  idempotent.
- **Fail-open on timeout.** A stuck holder degrades to today's unserialized
  behavior with a loud ERROR after the configured wait — never a wedged
  session. A degraded token holds nothing and releases nothing.
- **Bounded registry.** The per-session lease map is size-capped; eviction
  only ever removes idle (unheld, uncontended) entries, never a live lease.

Known limits (deliberate, flagged on #64934):

- A CLI process sharing the session via CLI-continuity is outside any
  in-process lock — that pair needs a DB-level lease (separate design).
- Mid-turn compression rotation leaves a small alias window: the tip-walk can
  resolve a fresh child id while the parent-holding turn is still in flight.
  The mid-turn binding-sync sites are the right place to alias the lease in a
  follow-up.
    N)DictOptionali   g      @c                   B    e Zd ZdZdZ	 ddededededd	f
d
ZdefdZ	d	S )TurnLeaseTokena
  Handle returned by :meth:`SessionTurnLeaseRegistry.acquire`.

    ``degraded`` means the acquire timed out and the turn is proceeding
    UNSERIALIZED (fail-open); such a token holds nothing and its release is a
    no-op. ``released`` makes release idempotent.
    
session_id	owner_key
generationdegradedreleasedFr   r	   r
   r   returnNc                 L    || _         || _        || _        || _        d| _        d S )NFr   )selfr   r	   r
   r   s        4/home/ice/.hermes/hermes-agent/gateway/turn_lease.py__init__zTurnLeaseToken.__init__O   s+     %"$     c                 X    d| j         d| j        d| j         d| j         d| j         dS )NzTurnLeaseToken(session_id=z, owner_key=z, generation=z, degraded=z, released=)r   r   s    r   __repr__zTurnLeaseToken.__repr__\   sg    C C CC C8<C CC C26-C C C	
r   )F)
__name__
__module____qualname____doc__	__slots__strintboolr   r    r   r   r   r   E   s          RI    	
  
   
# 
 
 
 
 
 
r   r   c                   6    e Zd ZdZddZedefd            ZdS )_SessionLease)lockholderacquired_at	last_usedr   Nc                     t          j                    | _        d | _        d| _        t          j                    | _        d S )N        )asyncioLockr"   r#   r$   timer%   r   s    r   r   z_SessionLease.__init__g   s/    LNN	04r   c                 H    | j         du o| j                                         S )z?True when this lease can be evicted: nobody holds or awaits it.N)r#   r"   lockedr   s    r   idlez_SessionLease.idlem   s'     {d"=49+;+;+=+='==r   r   N)r   r   r   r   r   propertyr   r-   r   r   r   r!   r!   d   sT        >I% % % % >d > > > X> > >r   r!   c                       e Zd ZdZefdeddfdZdefdZdede	fdZ
dd	Zdd
dedededee         dee         f
dZdee         dedefdZdee         defdZdS )SessionTurnLeaseRegistryu  Asyncio lease per resolved session_id serializing transcript turns.

    Process-local and single-event-loop by design — the same visibility scope
    as the routing-key guards it extends. All methods must be called from the
    gateway's event loop.
    max_entriesr   Nc                 X    i | _         t          dt          |                    | _        d S )N   )_leasesmaxr   _max_entries)r   r2   s     r   r   z!SessionTurnLeaseRegistry.__init__{   s)    133{#3#344r   c                 *    t          | j                  S N)lenr5   r   s    r   __len__z SessionTurnLeaseRegistry.__len__   s    4<   r   r   c                     | j                             |          }|,|                                  t                      }|| j         |<   t	          j                    |_        |S r9   )r5   get_evict_idler!   r*   r%   )r   r   leases      r   _get_or_createz'SessionTurnLeaseRegistry._get_or_create   sU      ,,=!OOE',DL$)++r   c                     t           j                   j        z
  dz   }|dk    rdS t          d  j                                        D              fd          }|d|         D ]} j                            |d           dS )u   Drop oldest idle entries so a new lease fits under the cap.

        Never evicts a held or contended lease — correctness beats the cap.
        r4   r   Nc              3   .   K   | ]\  }}|j         |V  d S r9   )r-   ).0sidr?   s      r   	<genexpr>z7SessionTurnLeaseRegistry._evict_idle.<locals>.<genexpr>   s-      FFZS%5:FSFFFFFFr   c                 (    j         |          j        S r9   )r5   r%   )rD   r   s    r   <lambda>z6SessionTurnLeaseRegistry._evict_idle.<locals>.<lambda>   s    DL-7 r   )key)r:   r5   r7   sorteditemspop)r   overflowidle_idsrD   s   `   r   r>   z$SessionTurnLeaseRegistry._evict_idle   s    
 t|$$t'881<q==FFF4<#5#5#7#7FFF7777
 
 
 IXI& 	( 	(CLS$''''	( 	(r   timeoutr	   r
   rO   c          
      
  K   |sdS |r|dk    rt          |          nt          }t          ||t          |                    }|                     |          }|j                                        r[|j        }t          	                    d||||r|j
        nd|r|j        nd|j        rt          j                    |j        z
  nd           	 t          j        |j                                        |           d{V  nW# t          j        $ rE |j        }t                              d|||||r|j
        nd|r|j        nd           d|_        |cY S w xY w||_        t          j                    |_        |j        |_        |S )	u   Acquire the turn lease for ``session_id``, waiting if held.

        Returns a :class:`TurnLeaseToken` — degraded when the wait timed out
        (fail-open: caller proceeds unserialized). Returns ``None`` for a
        falsy ``session_id``.
        Nr   u   turn lease contention on session %s: routing key %s (gen %s) waiting behind in-flight turn held by routing key %s (gen %s, held %.0fs) — two routing keys are mapped to one session_id (#64934); serializing this turn behind the previous turn's flush?g      rN   u   turn lease wait timed out after %.0fs on session %s (waiter: routing key %s gen %s; holder: routing key %s gen %s) — failing open: this turn runs UNSERIALIZED against the stuck holder rather than wedging the session; transcript writes may interleaveT)floatDEFAULT_LEASE_WAITr   r   r@   r"   r,   r#   loggerwarningr	   r
   r$   r*   r(   wait_foracquireTimeoutErrorerrorr   r%   )	r   r   r	   r
   rO   waittokenr?   r#   s	            r   rW   z SessionTurnLeaseRegistry.acquire   s       	4!(PWq[[uW~~~>Pz9c*ooFF##J//: 	\FNN
 $*3  %+4!!383DN	e///$  	"5:#5#5#7#7FFFFFFFFFFF# 	 	 	\FLL(
 $*3  %+4!!   "ENLLL!	$  IKK+s   3D AEEr[   new_session_idc           
         ||j         s|j        s|r||j        k    rdS | j                            |j                  }|	|j        |urdS | j                            |          }|V||urR|j        sK|j        }t                              d|j        ||j	        |j
        |r|j	        nd|r|j
        nd|           dS || j        |<   t          j                    |_        ||_        dS )u  Alias a HELD lease onto ``new_session_id`` after mid-turn rotation.

        Compression can rotate the durable session_id while a turn is in
        flight (session-hygiene pre-compression, in-agent compression). The
        turn's flush then targets the NEW id — so the serialization boundary
        must follow it, or an alias routing key resolving the new id (e.g. a
        topic tip-walk landing on the fresh child) could start a concurrent
        turn the lease never sees. This closes the rotation-alias window
        flagged on #64934.

        Mechanism: the SAME ``_SessionLease`` object is registered under the
        new id (the old mapping stays until it goes idle and is evicted), so
        acquirers on either id serialize against one lock — no lock state is
        moved, no asyncio internals are touched. Only the current holder can
        rebind (identity-checked like release), and the token follows to the
        new id so release frees the shared object.

        Edge: if the new id already has a live lease of its own (another
        turn is running on the target session), the two serialization
        domains cannot be merged mid-wait — log loudly and keep the token on
        the old id. Fail-open, never deadlock: a holder cannot wait mid-turn.
        NFu  turn lease rebind blocked: session %s rotated to %s mid-turn (holder: routing key %s gen %s) but the target session's lease is already live (holder: routing key %s gen %s) — keeping the lease on the old id; transcript writes on %s may interleave (#64934 rotation-alias edge)rQ   T)r   r   r   r5   r=   r#   r-   rT   rU   r	   r
   r*   r%   )r   r[   r\   r?   existingr#   s         r   rebindzSessionTurnLeaseRegistry.rebind   s   0 M~ ~  "  !1115  !122=EL555<##N33HE$9$9(-$9_FNN>
   $*3  %+4!!   5',^$)++)tr   c                    ||j         s|j        rdS d|_        | j                            |j                  }|dS |j        |ur.t                              d|j        |j        |j	                   dS d|_        d|_
        t          j                    |_        |j                                        r|j                                         dS )ul  Release ``token``'s lease. Idempotent; ownership-checked.

        Returns True only when this exact token was the current holder and
        the lock was freed. A degraded token, a re-release, or a stale token
        whose slot has since been granted to a newer turn are all safe
        no-ops — a stale unwind can never release a newer turn's lease.
        NFTzYturn lease release skipped on session %s: token (key %s gen %s) is not the current holderr'   )r   r   r5   r=   r   r#   rT   debugr	   r
   r$   r*   r%   r"   r,   release)r   r[   r?   s      r   rb   z SessionTurnLeaseRegistry.release  s     =EN=en=5  !122=5<u$$LL4     5)++: 	!J   tr   r.   )r   r   r   r   DEFAULT_MAX_LEASESr   r   r;   r   r!   r@   r>   r   rR   r   rW   r   r_   rb   r   r   r   r1   r1   s   sN         += 5 5C 5 5 5 5 5! ! ! ! !     ( ( ( (* $(; ; ;; 	;
 ; %; 
.	!; ; ; ;z9H^4 9c 9d 9 9 9 9vXn5 $      r   r1   )r   r(   loggingr*   typingr   r   	getLoggerr   rT   rc   rS   r   r!   r1   r   r   r   <module>rg      s   / /b    ! ! ! ! ! ! ! !		8	$	$  
  
 
 
 
 
 
 
 
>> > > > > > > >{ { { { { { { { { {r   