^i1dZddlZddlZddlmZddlmZmZddlmZddl m Z dZ eejdd Z d ZGd d Zy) u MemGPTEscalation — Opus escalation handler. When MVFL correction loop exhausts 3 attempts, this handler builds full failure context and dispatches to Opus for resolution. N) dataclass)datetimetimezone)Path)Optionalzclaude-opus-4-6EVENTS_LOG_PATHz5/mnt/e/genesis-system/data/observability/events.jsonlzYou are the Genesis Maximum-Effort Resolver. Previous agents have failed {attempt_count} times. Full failure history: {failure_history} Resolve this definitively. Provide a complete, correct output.c ZeZdZdZd dZ d dededeedefdZd ede fd Z d eddfd Z y)MemGPTEscalationu Opus escalation handler for the MVFL correction pipeline. When the CorrectionLoop exhausts MAX_CORRECTION_ATTEMPTS, this class takes over: it builds a rich context prompt from all failure history and dispatches the task to claude-opus-4-6 for a definitive resolution. The dispatch function is injected for full testability — no real API calls are made in unit tests. Nc||_y)z Args: dispatch_fn: async callable (model, system_prompt, task_payload) -> dict If None, returns a stub escalated response (used in tests). N) _dispatch)self dispatch_fns 4/mnt/e/genesis-system/core/mvfl/memgpt_escalation.py__init__zMemGPTEscalation.__init__#s % task_payload failed_outputfailure_historyreturnc K|jdd}||n|g}t|}|j|}tj ||}|j d|t |tjtjjd |jr!|jt ||d{} | S|dt dt d |d S7#t$r} |d d t| t d cYd} ~ Sd} ~ wwxYww)uQ Escalate to Opus with full failure context. Builds a system prompt that includes all failed attempt details, then dispatches to ESCALATION_MODEL. Logs an "mvfl_escalation" event to the observability log regardless of outcome. Args: task_payload: Original task that failed (must contain 'task_id'). failed_output: The last failed output from the CorrectionLoop. failure_history: List of all failed attempt dicts. If not provided, [failed_output] is used as a single-item history. Returns: Standard output dict from Opus dispatch, OR: - {"status": "escalated", ...} — when no dispatch_fn configured - {"status": "error", "error": "MVFL_ESCALATION_FAILED", ...} — on exception task_idunknownN) attempt_countrmvfl_escalation)eventrmodelr timestamp escalatedzEscalation to z, requested (no dispatch function configured))rstatusroutputrerrorMVFL_ESCALATION_FAILED)rrr! error_detailr)getlen_build_history_stringESCALATION_SYSTEM_PROMPTformat _log_eventESCALATION_MODELrnowrutc isoformatr Exceptionstr) r rrrrhistoryr history_str system_promptresultes rescalatezMemGPTEscalation.escalate+s 0""9i8%4%@/}oG 009 177''8 &%*!hll3==?    ~~#~~.> |\\  ')-()9(:;<<&3   ] "!1 #A)   sTBD&CCCDCDC D&D:D;DDDr0c"g}t|dD]n\}}|jd|dt|tr0|j D]\}}|jd|d|[|jd|pdj |S)a Format failure history list into a numbered, human-readable string. Each attempt is shown as a numbered block with all key-value pairs (or a raw string representation for non-dict entries). Args: history: List of attempt dicts or raw values. Returns: Multi-line string suitable for inclusion in the Opus system prompt. z --- Attempt z ---z z:  ) enumerateappend isinstancedictitemsjoin)r r0partsiattemptkvs rr&z&MemGPTEscalation._build_history_stringss#GQ/ -JAw LL<s$/ 0'4(#MMO0DAqLL2aS1#/0 r'^,  -yyrrc tjjddttd5}|j t j |dzdddy#1swYyxYw#t$rYywxYw)u Append a JSON event line to the observability events log. Never raises — telemetry must not break the pipeline. T)parentsexist_okar8N)rparentmkdiropenwritejsondumpsr.)r rfs rr)zMemGPTEscalation._log_eventsl    " " ( ( ( Eos+ 2q 5)D01 2 2 2   s.1A0(A$A0$A-)A0-A00 A<;A<)N) __name__ __module__ __qualname____doc__rr<rlistr5r/r&r)rrr r sl %+/ FFF"$ F  FP T c .   rr )rRrLos dataclassesrrrpathlibrtypingrr*getenvrr'r rTrrrZsT  !'$yryy!24klmE}}r