# simulate.py (SimulationLayer) V13.3.3+ Attainment Snapshot from __future__ import annotations from typing import Any, Dict import copy import traceback from .terminals import DSLTerminal class SimulationLayer: def simulate(self, match_result: Any, decomposition: Any, mined_knowledge: Dict[str, Any] = None) -> Dict[str, Any]: terminal = DSLTerminal() if not mined_knowledge: mined_knowledge = {} query_text = mined_knowledge.get("QUERY_TEXT") or getattr(decomposition, "text_part", "") or "" ops_list = mined_knowledge.get("OPS") or [ {"op": "EXTRACT_CLAIMS", "args": {}}, {"op": "REGISTER_DEFS", "args": {"source": "mined"}}, {"op": "LEXICAL_MATCH", "args": {}}, {"op": "VERIFY_WITH_RULESET", "args": {}}, {"op": "VERIFY_NECESSITY", "args": {}}, {"op": "SOLVE_CHOICE", "args": {}}, ] # ✅ Fix: Extract these BEFORE using them in session_vars mined_rules = mined_knowledge.get("MINED_RULES", []) mined_definitions = mined_knowledge.get("DEFINITIONS", {}) session_vars = { "choices": getattr(decomposition, "slots", {}).get("choices", {}), "frame_props": getattr(decomposition, "slots", {}).get("frame_props", []), "refute_mode": getattr(decomposition, "slots", {}).get("refute_mode", False), "need_exactness": getattr(decomposition, "slots", {}).get("need_exactness", False), "intent_candidates": getattr(decomposition, "slots", {}).get("intent_candidates", []), "entity": getattr(decomposition, "slots", {}).get("entity", ""), "mined_rules": mined_rules, "mined_definitions": mined_definitions, "query_text": query_text, "claims": [], "verified_map": [], "rule_hits": [], "refutation_hits": [], "evidence_count": 0, } steps_log = [] verified_results = [] evidence_count = 0 # ★追加：Reasoning Attainment 用コンテナ attainment = { "total_ops": [op.get("op", "UNKNOWN") for op in ops_list], "completed": [], "failed_at": None, "error": None, "traceback": None, "snapshots": {}, # op名 -> スナップショット } print(f"⚙️ [SIM] OPS List: {[op.get('op') for op in ops_list]}") def snapshot_state(op_name: str, res: Dict[str, Any] | None): """ UIに出したい「直前までの状態」を最小限で保存 """ attainment["snapshots"][op_name] = { "op": op_name, "verdict": (res or {}).get("verdict"), "result": (res or {}).get("result"), "frame_props": list(session_vars.get("frame_props", [])), "choices": list((session_vars.get("choices") or {}).keys()), "registered_defs_count": len(session_vars.get("mined_definitions") or {}), "claims_count": len(session_vars.get("claims") or []), "rule_hits_count": len(session_vars.get("rule_hits") or []), "refutation_hits_count": len(session_vars.get("refutation_hits") or []), "evidence_count": evidence_count, } final_verdict = "insufficient_evidence" for op_data in ops_list: op_name = op_data.get("op", "UNKNOWN") print(f"⚙️ [SIM] Op: {op_name}") try: # Context Injection if op_name.upper() == "SOLVE_CHOICE": op_data.setdefault("args", {}) op_data["args"]["choices"] = session_vars["choices"] op_data["args"]["question"] = query_text session_vars["evidence_count"] = evidence_count res = terminal.execute_op(op_data, session_vars) if res is None: res = {"verdict": "rejected", "error": "None result"} # Capture final verdict from significant ops if op_name.upper() in ["SOLVE_CHOICE", "GENERATE_EXPLANATION", "APPLY_SCHEMA"]: final_verdict = res.get("verdict", "insufficient_evidence") # --- 通常パイプライン --- if "claims" in res: session_vars["claims"] = res["claims"] if "verified_map" in res: session_vars["verified_map"] = res["verified_map"] if "rule_hits" in res: session_vars["rule_hits"].extend(res["rule_hits"]) if "refutation_hits" in res: session_vars["refutation_hits"].extend(res["refutation_hits"]) # ログ verdict = res.get("verdict") result_txt = res.get("result", "") if verdict == "accepted": steps_log.append(f"✅ PASS: {op_name} -> {result_txt}") print(f" → Verdict: {verdict}") print(f" → Result: {result_txt}") verified_results.append(result_txt) if res.get("verified"): added = int(res.get("count", 1)) evidence_count += added steps_log.append(f" ⭐ Logic Evidence +{added} (Total: {evidence_count})") print(f" → Evidence added: +{added} (Total: {evidence_count})") elif verdict == "insufficient_evidence": reason = res.get('reason', result_txt) steps_log.append(f"⚠️ WAIT: {op_name} -> {reason}") print(f" → Verdict: {verdict}") print(f" → Reason: {reason}") else: reason = res.get('error') or res.get('reason') or "Unknown error" steps_log.append(f"❌ FAIL: {op_name} ({reason})") print(f" → Verdict: {verdict} (FAIL)") print(f" → Reason: {reason}") # ★追加：完了として記録 & スナップショット attainment["completed"].append(op_name) snapshot_state(op_name, res) except Exception as e: # ★追加：どこで落ちたか確定 attainment["failed_at"] = op_name attainment["error"] = str(e) attainment["traceback"] = traceback.format_exc(limit=2) print(f"🔥 [SIM] ERROR at {op_name}: {e}") print(traceback.format_exc()) # “直前まで” の状態を保存（resが無いのでNone） snapshot_state(op_name, None) # ログ上の表示（あなたの "logic error" の根拠にする） steps_log.append(f"❌ FAIL: {op_name} (logic error: {e})") break content = { "definition": f"生産型推論シミュレーション完了 (Evidence: {evidence_count}).", "logic_check": steps_log, "conclusion": "\n".join(verified_results) if verified_results else "", "summary": f"AXIS-V13.3.x Verified {evidence_count} logic units.", # ★追加：UI表示用 "reasoning_attainment": attainment, # ✅ Fix: Structured Metadata for Controller "meta": { "proofs_count": len(session_vars.get("rule_hits", [])), "refutations_count": len(session_vars.get("refutation_hits", [])), "evidence_count": evidence_count, "final_verdict": final_verdict } } return content