
    +j#                         d Z ddlZddlZddlZddlZddlmZmZmZ ddl	m
Z
mZmZmZmZ  ej        e          Z G d d          Z G d d	          ZdS )
u  
LightClaw — 可靠发送器 (ReliableEmitter)
Mirrors: src/socket/reliable-emitter.ts

设计哲学（对齐 TS 版）：
    - emit_fire_and_forget(): 同步发送，不等 ACK，不重试。用于控制帧。
    - emit_with_ack(): 可靠发送 + 自动重试。返回 asyncio.Future<bool>。
    - 所有 ACK 通过 callback 派发，不阻塞任何协程。
    - pause/resume 控制断线期间的重试调度。
    - 每条消息注入 idempotencyKey，服务端/前端据此去重。
    N)CallableDictOptional   )EMIT_ACK_TIMEOUTEMIT_MAX_RETRIESEMIT_PENDING_MAXEMIT_RETRY_BASE_DELAYEMIT_RETRY_MAX_DELAYc                   P    e Zd ZdZdZdedee         dededee	gdf         f
d	Z
dS )
_PendingMsgzAOne outbound message waiting for server ACK (reliable path only).)emit_idmsg_ideventdataretry_count
created_atretry_timerresolver   r   r   r   r   Nc                     || _         || _        || _        || _        d| _        t          j                    | _        d | _        || _	        d S )Nr   )
r   r   r   r   r   time	monotonicr   r   r   )selfr   r   r   r   r   s         K/home/agentuser/.hermes/plugins/lightclawbot/src/socket/reliable_emitter.py__init__z_PendingMsg.__init__&   sJ     
	.**:>    )__name__
__module____qualname____doc__	__slots__strr   dictr   boolr    r   r   r   r      st        KKI
  	
  4&$,'     r   r   c                   ,   e Zd ZdZ	 ddeeegdf         deeeeeee	         gdf         gdf         defdZ
ded	ed
dfdZ	 dded	edee         d
efdZddZddZddZd
efdZded
dfdZded
dfdZded
dfdZded
dfdZded
efdZddZdS )ReliableEmitteru,  
    业务可靠性层 — 在 NativeSocketMixin 的 callback 之上加重试逻辑。
    Mirrors: ReliableEmitter in reliable-emitter.ts

    使用方式::

        emitter = ReliableEmitter(
            ws_emit=adapter._ws_emit,
            ws_emit_with_timeout=adapter._ws_emit_with_timeout,
            prefix="[lightclaw]",
        )
        # fire-and-forget (控制帧)
        emitter.emit_fire_and_forget("message:private", data)

        # 可靠发送 (关键帧)
        ok = await emitter.emit_with_ack("message:private", data, msg_id)
    [lightclaw]ws_emitNws_emit_with_timeoutprefixc                 Z    || _         || _        || _        i | _        d| _        d| _        d S )NFr   )_ws_emit_ws_emit_with_timeout_prefix_pending_paused_counter)r   r)   r*   r+   s       r   r   zReliableEmitter.__init__K   s3      %9"02r   r   r   returnc                 j    |                                  }i |d|i}|                     ||           dS )u   Send one frame — no ACK, no retry, no blocking.

        Injects ``idempotencyKey`` for client-side dedup/ordering, then
        calls the raw socket emit.  Replaces the old
        ``asyncio.create_task(emit_nowait(...))`` pattern.
        idempotencyKeyN)_gen_idr-   )r   r   r   r   enricheds        r   emit_fire_and_forgetz$ReliableEmitter.emit_fire_and_forget]   s@     ,,..6d6,g66eX&&&&&r   r   c                   	K   |                                  }i |d|i}|                                  t          j                    }|                                	dt
          ddf	fd}t          |||||          }|| j        |<   t          	                    d| j
        |||           | j        s|                     |           	 d{V S )u   Reliable send — ACK + auto-retry.  Returns True on ACK, False on give-up.

        Mirrors TS ``emitWithAck``.  Internally bridges the callback-based
        socket layer to an ``asyncio.Future`` so callers can ``await``.
        r5   okr3   Nc                 ^                                     s                    |            d S d S N)done
set_result)r:   futures    r   _resolvez/ReliableEmitter.emit_with_ack.<locals>._resolve|   s7    ;;== &!!"%%%%%& &r   z-%s emit_with_ack: emitId=%s msgId=%s event=%s)r6   _evict_if_neededasyncioget_running_loopcreate_futurer$   r   r0   loggerinfor/   r1   _do_emit)
r   r   r   r   r   r7   loopr@   entryr?   s
            @r   emit_with_ackzReliableEmitter.emit_with_ackh   s      ,,..6d6,g66 	'))'+'9'9';';	& 	&$ 	& 	& 	& 	& 	& 	& GVUHhGG!&g;L'65	
 	
 	

 | 	!MM%   ||||||r   c                    | j         rdS d| _         | j                                        D ])}|j         |j                                         d|_        *t
                              d| j        t          | j                             dS )z7Call on disconnect: clear all retry timers, keep queue.NTz%s Paused (%d pending))	r1   r0   valuesr   cancelrE   rF   r/   lenr   rI   s     r   pausezReliableEmitter.pause   s    < 	F]))++ 	) 	)E ,!((***$(!,dlC<N<NOOOOOr   c                     | j         sdS d| _         t                              d| j        t	          | j                             | j                                        D ]}|                     |           dS )z0Call on reconnect: re-emit all pending messages.NFz-%s Resumed, re-emitting %d pending message(s))r1   rE   rF   r/   rN   r0   rL   rG   rO   s     r   resumezReliableEmitter.resume   s    | 	F;L#dm,,	
 	
 	
 ]))++ 	! 	!EMM%    	! 	!r   c                 4   t          | j                                                  D ]7}|j        |j                                         |                    d           8| j                                         t                              d| j	                   dS )z6Call on shutdown: cancel all timers, fail all futures.NFz%s Destroyed)
listr0   rL   r   rM   r   clearrE   rF   r/   rO   s     r   destroyzReliableEmitter.destroy   s    $-..0011 	! 	!E ,!((***MM%    NDL11111r   c                     | xj         dz  c_         dt          t          j                    dz            dd| j         dS )N   _re_  016d_012d)r2   intr   )r   s    r   r6   zReliableEmitter._gen_id   sC    Ic$)++,--IIIT]IIIIr   rI   c                      t          t          dz            }dt          t                   ddf fd}                     j        j        ||           dS )zSend one frame via the socket timeout+callback path.

        Mirrors TS ``doEmit``: calls ``socket.timeout(ms).emit(event, data, cb)``.
        The callback resolves the pending entry on ACK or schedules retry on timeout.
        rZ   errr3   Nc                     j         j        vrd S |                     j                    d S t                              dj        j         j        | j                                                  d S )Nz2%s ACK error: emitId=%s msgId=%s err=%s retries=%d)	r   r0   _confirmrE   warningr/   r   r   _schedule_retry)r`   rI   r   s    r   _on_ack_or_timeoutz4ReliableEmitter._do_emit.<locals>._on_ack_or_timeout   s    }DM11{em,,,,, HL%-s%  
 $$U+++++r   )r^   r   r   	Exceptionr.   r   r   )r   rI   
timeout_msre   s   ``  r   rG   zReliableEmitter._do_emit   s     )D011
	,HY$7 	,D 	, 	, 	, 	, 	, 	, 	,  	""KZ1C	
 	
 	
 	
 	
r   r   c                     | j                             |d          }|dS |j         |j                                         d|_        |                    d           dS )u3   ACK received — remove from pending, resolve True.NT)r0   popr   rM   r   )r   r   rI   s      r   rb   zReliableEmitter._confirm   sa    !!'400=F($$&&& $Edr   c           	      `   | j         rdS |j        t          k    r| j                            |j        d           t                              d| j        |j        |j        |j	        t          j                    |j        z
             |                    d           dS |xj        dz  c_        |                     |j                  }t                              d| j        |j        ||j        |j	                   t!          j                    }|                    || j        |          |_        dS )z8Exponential backoff retry. Mirrors TS ``scheduleRetry``.Nz=%s Gave up after %d retries: emitId=%s msgId=%s elapsed=%.1fsFrX   z)%s Retry #%d in %.1fs: emitId=%s msgId=%s)r1   r   r   r0   ri   r   rE   errorr/   r   r   r   r   r   _retry_delayrF   rB   rC   
call_later_on_retry_timerr   )r   rI   delayrH   s       r   rd   zReliableEmitter._schedule_retry   s   < 	F 000MemT222LLOe/  5#33  
 MM%   FQ!!%"3447L%+UEM5<	
 	
 	

 ')) OOE43GOOr   c                 p    d|_         |j        | j        vrdS | j        rdS |                     |           dS )z8Timer callback: re-emit if still pending and not paused.N)r   r   r0   r1   rG   rO   s     r   rn   zReliableEmitter._on_retry_timer   sE     =--F< 	Fer   r   c                 x    t           d|dz
  z  z  }t          j                    }t          ||z   t                    S )Nr   rX   )r
   randomminr   )r   r   basejitters       r   rl   zReliableEmitter._retry_delay   s5    $kAo(>?4&="6777r   c                    t          | j                  t          k    rt          t	          | j                  d           }|d S | j                            |          }|j        |j                                         t          	                    d| j
        |j        |j                   |                    d           t          | j                  t          k    d S d S )Nz%%s Evicted oldest: emitId=%s msgId=%sF)rN   r0   r	   nextiterri   r   rM   rE   rc   r/   r   r   r   )r   	oldest_idrI   s      r   rA   z ReliableEmitter._evict_if_needed   s    $-  $444T$-00$77I M%%i00E ,!((***NN7emU\   MM%    $-  $4444444r   )r(   r<   )r3   N)r   r   r   r    r   r"   r#   r^   r   rf   r   r8   r$   rJ   rP   rR   rV   r6   r   rG   rb   rd   rn   floatrl   rA   r%   r   r   r'   r'   8   s$        0 $ 3+t+, '$Xx	':&;T&ABCTI
    $	'# 	'T 	'd 	' 	' 	' 	' !%	# ## # 	#
 
# # # #J	P 	P 	P 	P
! 
! 
! 
!2 2 2 2J J J J J
k 
d 
 
 
 
8     P[ PT P P P P2[ T    8 8 8 8 8 8
! ! ! ! ! !r   r'   )r    rB   loggingrr   r   typingr   r   r   configr   r   r	   r
   r   	getLoggerr   rE   r   r'   r%   r   r   <module>r      s  
 
     + + + + + + + + + +              
	8	$	$       4T! T! T! T! T! T! T! T! T! T!r   