!iAdZddlmZddlZddlZddlmZmZddlm Z ddl m Z eGddZ e d Z d ZGd d Zy) u core/evolution/shadow_arena.py Story 8.04: ShadowArena — Containerized Test Sandbox Tests proposed architectural changes against historical failed sagas in isolation. Fetches historical failed sagas from Postgres, injects their inputs into Redis (Shadow Mode), re-runs them under the proposed code branch, evaluates axiom compliance, and decides whether the proposal is ready for a PR. Usage:: arena = ShadowArena(pg_connection=pg, redis_client=redis, axiomatic_tests=axtest) result = arena.evaluate_proposal( proposal_branch="core.evolution.candidate_v2", test_saga_ids=["saga-001", "saga-002"], ) if result.ready_for_pr: print("Proposal cleared for PR.") ) annotationsN) dataclassfield)Path)AnycbeZdZUdZded<eeZded<eeZ ded<d Z d ed <y ) ArenaResulta$Result of a ShadowArena evaluation run. Attributes: pass_rate: Fraction of sagas that passed under the proposal (new_success / total_sagas_tested). axiom_violations: List of axiom violation IDs found in the proposed branch. improved_metrics: Dict with at minimum ``old_success_rate`` and ``new_success_rate`` keys, plus deltas. ready_for_pr: True iff ``pass_rate >= 0.8`` AND ``axiom_violations == []``. float pass_rate)default_factory list[str]axiom_violationsdict[str, Any]improved_metricsFbool ready_for_prN) __name__ __module__ __qualname____doc____annotations__rlistrdictrr4/mnt/e/genesis-system/core/evolution/shadow_arena.pyr r #s9 "'"=i=',T'BnBL$rr z@/mnt/e/genesis-system/data/observability/shadow_arena_runs.jsonlzSHADOW:ceZdZdZ d ddZ ddZddZddZ ddZddZ dd Z e dd Z e dd Z dd Zy) ShadowArenaa[Containerised test sandbox for proposed Genesis architectural changes. All external dependencies (Postgres, Redis, AxiomaticTests) are dependency-injected so the class is fully testable without real services. Args: pg_connection: A Postgres connection object exposing ``.cursor()``. If ``None``, saga fetching will return empty results. redis_client: A Redis client exposing ``.set(key, value)`` and ``.get(key)``. If ``None``, shadow injection is skipped. axiomatic_tests: An ``AxiomaticTests`` instance (from story 8.02). If ``None``, axiom checks are skipped (no violations added). log_path: Where to append JSONL arena run records. Defaults to ``data/observability/shadow_arena_runs.jsonl``. Ncj||_||_||_|rt||_yt|_y)N)_pg_redis_axiomatic_testsr_DEFAULT_LOG_PATH _log_path)self pg_connection redis_clientaxiomatic_testslog_paths r__init__zShadowArena.__init__Ss.!" /+3h9Jrc|j|}|j||j||\}}|j|}|j ||}t |}t d|jD} |dkDr| |z nd} | dk\xrt |dk(} t| ||| } |j|||| | S)aEvaluate a proposed branch against historical failed sagas. Steps ----- 1. Fetch saga records from Postgres by ID. 2. Inject saga inputs into Redis under SHADOW prefix (Shadow Mode). 3. Re-run sagas using the proposed branch code, recording pass/fail per saga. 4. Run axiomatic checks on the proposed branch. 5. Compute improvement metrics (old vs new success rate). 6. Determine ``ready_for_pr``. 7. Persist full run record to ``shadow_arena_runs.jsonl``. Args: proposal_branch: Dotted Python module path of the proposed branch (e.g. ``"core.evolution.candidate_v2"``). test_saga_ids: IDs of historical failed sagas to replay. Returns: ArenaResult with pass_rate, axiom_violations, improved_metrics, and ready_for_pr. c3&K|] }|sd ywNr.0oks r z0ShadowArena.evaluate_proposal..s<2Q<rg?)r rrr) _fetch_sagas_inject_to_shadow_run_in_shadow _check_axioms_compute_metricslensumvaluesr _write_log) r%proposal_branch test_saga_idssagas old_results new_resultsrrtotalpassedr rresults revaluate_proposalzShadowArena.evaluate_proposalcs6!!-0 u%$(#6#6#N [ --o> 00kJK <!3!3!5<<&+aiFUNS  C'FC0@,AQ,F --%   vF rc`|jr|s|Dcgc]}|idd c}Sg} |jj}djdgt|z}|j d|dt ||j D]8}|\}}} |j|t|tr|nit| d: |D chc]} | d } } |D]}|| vs|j|idd|Scc}w#t$r|Dcgc]}|idd ncc}w}}Y^wxYwcc} w)aFetch saga records from Postgres by ID. Returns a list of saga dicts. Each dict must contain at minimum: ``{"saga_id": str, "inputs": dict, "success": bool}``. If no Postgres connection is available, returns synthetic records (all marked as failed) so the arena can still operate. Args: saga_ids: The list of saga IDs to fetch. Returns: List of saga record dicts. F)saga_idinputssuccess,z%sz=SELECT saga_id, inputs, success FROM sagas WHERE saga_id IN ()rH) r cursorjoinr:executetuplefetchallappend isinstancerr Exception) r%saga_idssidr@rM placeholdersrowrHrIrJs found_idss rr5zShadowArena._fetch_sagassfxxx$ 2%@  ') XX__&F88TFS]$:;L NNOP\~]^_h ( +.( &(264(@fb#G} ,11aQy\1 1 OC)# uMN O C( $ 2%@E 2s) C>B"D D+ D( DD('D(c|jr|sy|D]C}t|d}tj|d}|jj ||Ey)aJWrite saga inputs to Redis under the SHADOW prefix. Shadow Mode blocks external calls by routing all saga input reads through prefixed Redis keys rather than live sources. Key format: ``SHADOW:`` Args: sagas: The list of saga dicts previously fetched from Postgres. NrHrI)r! SHADOW_PREFIXjsondumpsset)r%r@sagakeyvalues rr6zShadowArena._inject_to_shadowsV{{%  (D"ODO#45CJJtH~.E KKOOC ' (rcJi}i}|j|}|D]r}|d}t|jdd||<|Ft|dd}t |r( ||jdi} t| ||<hd||<nd||<t||fS#t $rd||<YwxYw)uSimulate re-running sagas under the proposed branch. For each saga: - ``old_result`` is the recorded historical outcome (``saga["success"]``). - ``new_result`` simulates what the proposed branch would produce. The simulation strategy uses Python's import system to load the proposed module (if importable); if import fails the saga is counted as failing. For each saga the proposed module is asked for a ``run_saga(saga_inputs)`` callable. If that doesn't exist, we fall back to treating the saga as passing (the branch may not directly handle every saga type — that is fine). Args: proposal_branch: Dotted module path of the proposed branch. sagas: List of saga dicts. Returns: Tuple of (old_results, new_results) where each is a dict mapping saga_id → bool (True = passed / success). rHrJFNrun_sagarIT)_try_import_modulergetgetattrcallablerT) r%r>r@rArBproposed_moduler`rVrun_fnoutcomes rr7zShadowArena._run_in_shadows4(* ') 11/B )Dy/C#DHHY$>?K * *dCF#1"((B)?"@+/= C( (,K$$) C % )(K''%1+0 C(1s&BB"!B"c|jgS|j|}|jj|i}|jDcgc]}|jc}Scc}w)uRun AxiomaticTests against the proposed branch source code. Reads the source file corresponding to ``proposal_branch`` (dotted module path → file path), then invokes ``AxiomaticTests.run_all(code_content=..., state_content={})``. Args: proposal_branch: Dotted module path (e.g. ``"core.evolution.candidate"``). Returns: List of violated axiom ID strings. Empty list if no violations. ) code_content state_content)r"_read_module_sourcerun_all violationsaxiom_id)r%r>rm axiom_resultvs rr8zShadowArena._check_axioms sf  (I//@ ,,44%5 %1$;$;.W>R2q>r3c3&K|] }|sd ywr-rr/s rr2z/ShadowArena._compute_metrics..Xr}r3)r:r;r<round)r%rArBrCold_passnew_passold_ratenew_rates rr9zShadowArena._compute_metrics8s(K  A:$'$'"#"#  >;#5#5#7>>>;#5#5#7>>e#e#!&h 2 %h 28h.2&&   rclddl} |j|S#tttt f$rYywxYw)zAttempt to import a dotted module path. Returns the module object on success, or ``None`` on failure. Args: dotted_path: e.g. ``"core.evolution.candidate_v2"`` rN) importlib import_module ImportErrorModuleNotFoundErrorAttributeErrorrT) dotted_pathrs rrezShadowArena._try_import_modulejs9  **;7 70.)L  s 33ctd}|jdd}||dz ||z dz g}|D]'}|js |jdcSy#t$rY6wxYw) aTResolve a dotted module path to its source file and read it. Converts dots to slashes and tries both ``.py`` and ``/__init__.py`` variants under ``/mnt/e/genesis-system``. Returns an empty string if the file cannot be located. Args: dotted_path: e.g. ``"core.evolution.candidate_v2"`` z/mnt/e/genesis-system./z.pyz __init__.pyutf-8encoding)rreplaceexists read_textOSError)rbase rel_parts candidates candidates rrozShadowArena._read_module_sourceys+,''S1 i[$ $ 9 } , $ I!$...@@ sA A'&A'c tj||t||j|j|j|j d} |j jjdd|j jdd5}|jtj|dzdddy#1swYyxYw#t$rYywxYw) a5Append a full arena run record to the JSONL log file. Args: proposal_branch: Dotted module path of the proposal. test_saga_ids: Requested saga IDs. sagas: Fetched saga records. result: The ArenaResult produced by this run. ) timestampr>r? sagas_fetchedr rrrT)parentsexist_okarr N)timer:r rrrr$parentmkdiropenwriter]r^r)r%r>r?r@rErecordfhs rr=zShadowArena._write_logs .* Z)) & 7 7 & 7 7"//    NN ! ! ' 't ' D$$S7$; 4rF+d23 4 4 4   s1AC(C<CC CC CC)NNNN)r)zstr | Path | NonereturnNone)r>strr?r rr )rUr rlist[dict[str, Any]])r@rrr)r>rr@rrz'tuple[dict[str, bool], dict[str, bool]])r>rrr )rAdict[str, bool]rBrrr)rr)rrrr) r>rr?r r@rrEr rr)rrrrr*rFr5r6r7r8r9 staticmethodreror=rrrrrBs$&* K $ K  K ::!:  :@2h(&4(4($4( 1 4(l=0, $, %,   , d  2!$     rr)r __future__rr]r dataclassesrrpathlibrtypingrr r#r\rrrrrsX*# (  .[\ ppr