xj.dZddlmZddlZddlZddlZddlZddlZddlm Z ddl m Z ddl m Z  ddlZn #e$rdZYnwxYwdZdd hZejd Zd&d Zd'd(dZd)dZe dZddd*dZddd+dZd,dZd-d Zddd.d!Zd/d#Zd0d%ZdS)1aCrash-safe WebUI turn journal helpers. The journal is deliberately tiny: one JSONL file per session, append-only events, and read helpers that tolerate malformed lines. Recovery and repair can then reason about submitted turns without depending on in-memory stream state. ) annotationsN)contextmanager)Path)Iterable _turn_journal completed interruptedz^[A-Za-z0-9_.-]+$returnrc,ddlm}t|S)Nr SESSION_DIR) api.modelsr rr s 3/Users/bertmahoney/hermes-webui/api/turn_journal.py_default_session_dirrs"&&&&&&    session_idstr session_dir Path | Nonec t|pd}|r"d|vsd|vst|st d|t |n t }|tz |dz S)N/\zinvalid session_idz.jsonl)rstrip_SESSION_ID_RE fullmatch ValueErrorrrTURN_JOURNAL_DIR_NAME)rrsidroots r _journal_pathr!"s jB   % % ' 'C /#** >3K3KC3P3P -... + 74   =Q=S=SD ' 'S... 88rctjdtjdtjjddS)Nz%Y%m%dT%H%M%SZ- )timestrftimegmtimeuuiduuid4hexrr _make_turn_idr,*s>m,dkmm<< V Vtz||?OPSQSPS?T V VVrc#^KtdVdStj|tj dVtj|tjdS#tj|tjwxYw)aSerialize multi-process journal writes when advisory locks exist. ``O_APPEND`` keeps normal same-process appends simple, but a long JSONL event can exceed POSIX's small atomic-write boundary. On Unix, take an advisory lock around the single event write+fsync so two WebUI worker processes cannot interleave large submitted-message payloads into corrupted JSONL. Platforms without ``fcntl`` keep the previous best-effort append behavior. N)_fcntlflockfilenoLOCK_EXLOCK_UN)file_objs r_journal_file_lockr4.s~  L""FN3338  X__&&77777 X__&&7777s A993B,reventdictct|tstdt|dpd}|st dt|}|ddt||d<|dt|d tj t|| }|j d d tj|d ddz}tj|tjtjztjzd}tj|dd5}t+|5|||tj|dddn #1swxYwYdddn #1swxYwY tj|j tj} tj| tj| n#tj| wxYwn#t8$rYnwxYw|S)zAppend one turn journal event and fsync it before returning. The returned event is the exact payload written, with default ``version``, ``session_id``, ``turn_id``, and ``created_at`` fields filled in. zevent must be a dictr6rzevent is requiredversionrturn_id created_atr5T)parentsexist_okF),:) ensure_ascii separators iautf-8encodingN) isinstancer7 TypeErrorrgetrr setdefaultr,r%r!parentmkdirjsondumpsosopenO_CREATO_APPENDO_WRONLYfdopenr4writeflushfsyncr0 O_DIRECTORYcloseOSError) rr6r event_namepayloadpathlinefdfhdir_fds rappend_turn_journal_eventrcBs eT " "0.///UYYw''-2..4466J .,---5kkG y!$$$ OOGL y-//222 |TY[[111  = = =DKdT222 :gEj I I ID PD rzBK/"+=u E EB 2sW - - -"  # # " " HHTNNN HHJJJ HRYY[[ ! ! ! " " " " " " " " " " " " " " """"""""""""""""  bn55  HV    HV    BHV           NsaHAG0$ H0G4 4H7G4 8HH H $I;8I! I;!I77I;; JJcPt||}g}g} |d}n##t$rt |ggdcYSwxYwt |dD]\}}|s tj|}n-#tj $r| ||dYYwxYwt|tr| || ||dt |||dS)zDRead a session journal, returning valid events plus malformed lines.r5rErF)revents malformedr:)start)r_raw) r! read_text splitlinesFileNotFoundErrorr enumeraterrNloadsJSONDecodeErrorappendrHr7) rrr^rerflinesline_norhr6s rread_turn_journalrrlsd  = = =DFIN00;;== NNN!*oo"MMMMMN!%q111 < < yy{{   JsOOEE#      gc:: ; ; ; H  eT " " < MM%   gc:: ; ; ; ;j//V) T TTs#(AA A B$$'C CreIterable[dict]"tuple[dict[str, dict], list[dict]]c2i}i}|D]}t|tst|dpd}|sQt |r)||g|||}|Jt|dpdt|dpdkr|||<d| D}||fS)aReturn the latest event per ``turn_id`` and any terminal-collision entries. The first element is the latest event per turn_id (same overwrite-by-timestamp behaviour as before). The second element is a list of collision records, one per turn_id that had more than one terminal event. Each collision record contains ``turn_id`` and the ``events`` list (in ascending created_at order). A collision means the same logical turn recorded both ``completed`` and ``interrupted`` terminal events -- the derived state still picks the latest by timestamp, but callers can now detect and audit the double-terminal situation explicitly rather than having it silently collapse. r;rNr<rcdg|]-\}}t|dk|t|dd.S)r:cJt|dpdS)Nr<r)floatrJ)es rz7derive_turn_journal_states...s eAEE,DWDWD\[\>]>]r)key)r;re)lensorted).0tidevtss r z.derive_turn_journal_states..sL C t99q==6$4]4]#^#^#^__ ==r) rHr7rrJris_terminal_turn_eventrKrorxitems)restatesterminal_eventsr6r;previous collisionss rderive_turn_journal_statesrs2!F-/O $ $%&&  eii **0b117799   !% ( ( B  & &w 3 3 : :5 A A A::g&&  uUYY|%<%<%ABBeHLLYeLfLfLkjkFlFlll#F7O(..00J : r stream_id str | NonecTt|pd}|sdSd}|D]{}t|tst|dpd|krAt|dpd}|r|}||S)Nrrr;)rrrHr7rJ)rerstreamlatestr6r;s r_latest_turn_id_for_streamrs b ! ! ' ' ) )F tF%&&   uyy%%+ , , 6 6 eii **0b117799  F Mrc t|}t||d<|ds=t||}t |dpg|}|r||d<t |||S)zDAppend a lifecycle event for the turn associated with ``stream_id``.rr;r5re)r7rrJrrrrc)rrr6rr]journalr;s r$append_turn_journal_event_for_streamrs5kkGy>>GK ;;y ! !)#JKHHH,W[[-B-B-Hb)TT  )!(GI  $Zk R R RRr list[str]ct|tz }|sgStd|dDS)Nc3LK|]}||jV dSN)is_filestem)r~r^s r z0iter_turn_journal_session_ids..s1VVt||~~V$)VVVVVVrz*.jsonl)rrexistsr}glob)r journal_dirs riter_turn_journal_session_idsrsY{##&;;K      VV (8(8(C(CVVV V VVrboolc\t|pidpdtvS)Nr6r)rrJ_TERMINAL_EVENTS)r6s rrrs-    ))/R 0 04D DDr)r rr)rrrrr r)r r)rrr6r7rrr r7)rrrrr r7)rersr rt)rersrrr r) rrrrr6r7rrr r7)rrr r)r6r7r r) __doc__ __future__rrNrPrer%r( contextlibrpathlibrtypingrfcntlr. ImportErrorrrcompilerrr!r,r4rcrrrrrrrr+rrrs  #""""" %%%%%% FFF(/011 99999WWWW888. $ ''''''TFJUUUUUU0$$$$L    * $ SSSSSS$WWWWEEEEEEs 5??