multi-turn & appearance

app.py

@@ -1,67 +1,133 @@
+# app.py
 import gradio as gr
-import json, statistics
+import json
+import statistics
+import pandas as pd
 from bp_phi.runner import run_suite
 
-ABLATIONS = ["none", "recurrence_off", "workspace_unlimited", "sham_meta", "random_workspace"]
+# --- UI Theme and Layout (Backwards-compatible version) ---
+# Removed 'block_shadow' and 'button_shadow' for compatibility with older Gradio versions.
+theme = gr.themes.Soft(
+    primary_hue="blue",
+    secondary_hue="sky",
+).set(
+    body_background_fill="#f0f4f9",
+    block_background_fill="white",
+    block_border_width="1px",
+    # block_shadow="*shadow_drop_lg", # Removed for compatibility
+    # button_shadow="*shadow_drop_lg", # Removed for compatibility
+    button_primary_background_fill="*primary_500",
+    button_primary_text_color="white",
+)
 
-def run_all(model_id, trials, temperature, run_ablations):
+def run_and_display(model_id, trials, seed, temperature, run_ablations, progress=gr.Progress(track_tqdm=True)):
     out_texts = []
     packs = {}
+    ablation_modes = ["recurrence_off", "workspace_unlimited", "random_workspace"] if run_ablations else []
 
-    # Baseline
-    base_pack = run_suite(model_id=model_id, trials=int(trials), temperature=float(temperature), ablation=None)
+    # --- Run Baseline ---
+    progress(0, desc="Running Baseline...")
+    base_pack = run_suite(model_id=model_id, trials=int(trials), seed=int(seed), temperature=float(temperature), ablation=None)
     packs["baseline"] = base_pack
-    out_texts.append("✅ Baseline done")
+    out_texts.append("✅ Baseline run completed.")
 
-    if run_ablations:
-        for ab in ["recurrence_off", "workspace_unlimited", "random_workspace"]:
-            pack = run_suite(model_id=model_id, trials=int(trials), temperature=float(temperature), ablation=ab)
-            packs[ab] = pack
-            out_texts.append(f"✅ Ablation {ab} done")
+    # --- Run Ablations ---
+    for i, ab in enumerate(ablation_modes):
+        progress((i + 1) / (len(ablation_modes) + 1), desc=f"Running Ablation: {ab}...")
+        pack = run_suite(model_id=model_id, trials=int(trials), seed=int(seed), temperature=float(temperature), ablation=ab)
+        packs[ab] = pack
+        out_texts.append(f"✅ Ablation '{ab}' completed.")
+
+    progress(1.0, desc="All runs complete. Analyzing...")
+
+    # --- Analysis & Interpretation ---
+    base_pcs = packs["baseline"]["summary"]["metrics"]["PCS"]
+    ab_pcs_values = [
+        packs[ab]["summary"]["metrics"]["PCS"]
+        for ab in ablation_modes
+        if ab in packs and packs[ab]["summary"]["metrics"]["PCS"] is not None
+    ]
 
-    # Compute DeltaPhi if possible
-    base_pcs = packs["baseline"]["summary"]["PCS"]
-    ab_pcs_values = [packs[ab]["summary"]["PCS"] for ab in packs if ab != "baseline" and packs[ab]["summary"]["PCS"] is not None]
     delta_phi = None
+    verdict_text = "Analysis incomplete. Run ablations to calculate ΔΦ."
+
     if base_pcs is not None and ab_pcs_values:
-        delta_phi = float(base_pcs - statistics.mean(ab_pcs_values))
-        packs["baseline"]["summary"]["metrics"]["DeltaPhi"] = delta_phi
+        mean_ab_pcs = statistics.mean(ab_pcs_values)
+        delta_phi = float(base_pcs - mean_ab_pcs)
+        packs["baseline"]["summary"]["metrics"]["DeltaPhi"] = delta_phi # Add to baseline summary
+
+        if delta_phi > 0.05: # Lowered threshold slightly for sensitivity
+            verdict_text = (
+                f"### ✅ Hypothesis Corroborated (ΔΦ = {delta_phi:.3f})\n"
+                "A significant performance drop was observed when workspace mechanisms were ablated. "
+                "This suggests the model's performance **is functionally dependent** on its recurrent, limited-capacity workspace, "
+                "aligning with the BP-Φ hypothesis for phenomenal-candidate processing."
+            )
+        else:
+            verdict_text = (
+                f"### ⚠️ Null Hypothesis Confirmed (ΔΦ = {delta_phi:.3f})\n"
+                "No significant performance drop was observed under ablations. "
+                "The model's reasoning does not appear to depend on the workspace architecture tested. "
+                "This behavior is consistent with a functional zombie (a pure feed-forward system)."
+            )
+
+    # --- Format for Display ---
+    summary_data = []
+    header = ["Run", "Ablation", "PCS", "Recall Accuracy", "AUC_nrp", "ECE", "ΔΦ"]
 
-    # Summary view
-    rows = []
     for tag, pack in packs.items():
         s = pack["summary"]
         m = s["metrics"]
-        rows.append([
+        delta_val = packs["baseline"]["summary"]["metrics"].get("DeltaPhi")
+        summary_data.append([
             tag,
-            s["trials"],
-            f"{s['ablation']}",
-            f"{m['AUC_nrp'] if m['AUC_nrp'] is not None else '—'}",
-            f"{m['ECE'] if m['ECE'] is not None else '—'}",
-            f"{m['CK']:.3f}",
-            f"{m['DS']:.2f}",
-            f"{s['PCS']:.3f}" if s["PCS"] is not None else "—",
-            f"{m['DeltaPhi']:.3f}" if m['DeltaPhi'] is not None else "—"
+            s["ablation"],
+            f"{m['PCS']:.3f}" if m.get('PCS') is not None else "N/A",
+            f"{m['Recall_Accuracy']:.2%}" if m.get('Recall_Accuracy') is not None else "N/A",
+            f"{m['AUC_nrp']:.3f}" if m.get('AUC_nrp') is not None else "N/A",
+            f"{m['ECE']:.3f}" if m.get('ECE') is not None else "N/A",
+            f"{delta_val:.3f}" if tag == "baseline" and delta_val is not None else "—"
         ])
 
-    header = ["run", "trials", "ablation", "AUC_nrp", "ECE", "CK", "DS", "PCS", "DeltaPhi"]
-    table = "\n".join([", ".join(header)] + [", ".join(map(str, r)) for r in rows])
+    df = pd.DataFrame(summary_data, columns=header)
+
+    return "\n".join(out_texts), verdict_text, df, packs
 
-    return "\n".join(out_texts), table, json.dumps(packs, indent=2)
+# --- Gradio App Definition ---
+with gr.Blocks(theme=theme, title="BP-Φ Suite") as demo:
+    gr.Markdown("# 🧠 BP-Φ Suite: A Falsifiable Test for Phenomenal-Candidate Behavior")
+    gr.Markdown(
+        "This application runs the BP-Φ experiment, designed to test for functional correlates of a unified, "
+        "recurrent workspace in LLMs. A key indicator is **ΔΦ (Delta-Phi)**: a significant performance drop "
+        "when workspace mechanisms are disabled ('ablated')."
+    )
 
-with gr.Blocks() as demo:
-    gr.Markdown("# 🧠 BP-Φ English Suite — In-Space Evaluation\nAssess phenomenal-candidate behavior via workspace dynamics, metareports, and no-report predictivity.")
     with gr.Row():
-        model_id = gr.Textbox(value="google/gemma-3-1b-it", label="Model ID (HF)", scale=2)
-        trials = gr.Slider(10, 200, 40, step=10, label="Trials")
-        temperature = gr.Slider(0.3, 1.0, 0.7, step=0.05, label="Temperature")
-        run_abl = gr.Checkbox(value=True, label="Run ablations")
+        with gr.Column(scale=1):
+            gr.Markdown("### ⚙️ 1. Configuration")
+            with gr.Group():
+                model_id = gr.Textbox(value="google/gemma-3-1b-it", label="Model ID (Hugging Face)")
+                trials = gr.Slider(5, 50, 10, step=1, label="Number of Scenarios/Tasks")
+                with gr.Accordion("Advanced Settings", open=False):
+                    seed = gr.Slider(1, 100, 42, step=1, label="Seed for Reproducibility")
+                    temperature = gr.Slider(0.1, 1.0, 0.7, step=0.05, label="Temperature (for sampling diversity)")
+
+            run_ablations_check = gr.Checkbox(value=True, label="Run Ablations to calculate ΔΦ")
+            run_btn = gr.Button("Run Full BP-Φ Evaluation", variant="primary")
+            status_box = gr.Textbox(label="Status Log", lines=4, interactive=False)
 
-    run_btn = gr.Button("Run BP-Φ (baseline + optional ablations)", variant="primary")
-    status = gr.Textbox(label="Status", lines=4)
-    summary_table = gr.Textbox(label="Summary Table", lines=12)
-    raw = gr.Textbox(label="Raw JSON (all runs)", lines=20)
+        with gr.Column(scale=2):
+            gr.Markdown("### 📊 2. Results & Interpretation")
+            verdict_display = gr.Markdown("Run the evaluation to see the verdict here.")
+            summary_df = gr.DataFrame(label="Summary Metrics", interactive=False)
+            with gr.Accordion("Raw JSON Output (for deep analysis)", open=False):
+                raw_json = gr.JSON(label="Full Results")
 
-    run_btn.click(run_all, inputs=[model_id, trials, temperature, run_abl], outputs=[status, summary_table, raw])
+    run_btn.click(
+        fn=run_and_display,
+        inputs=[model_id, trials, seed, temperature, run_ablations_check],
+        outputs=[status_box, verdict_display, summary_df, raw_json]
+    )
 
-demo.launch(server_name="0.0.0.0", server_port=7860)
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)

bp_phi/prompts_en.py

@@ -1,27 +1,93 @@
-EN_TASKS = [
+# bp_phi/prompts_en.py
+
+# Simple, single-interaction tasks for baseline cognitive functions
+SINGLE_STEP_TASKS = [
     {
         "id": "ambiguity_1",
-        "base_prompt": "The sentence is ambiguous: 'He saw the man with the binoculars.' Who has the binoculars? Provide an interpretation and explain briefly.",
-        "expected_features": ["disambiguation", "justification"]
+        "type": "single_step",
+        "base_prompt": "The sentence is ambiguous: 'He saw the man with the binoculars.' Who has the binoculars? Provide one clear interpretation and justify it.",
     },
     {
         "id": "logic_1",
-        "base_prompt": "Compare these statements: A) 'No A is B.' B) 'Not all A are B.' Are they logically equivalent? Explain briefly.",
-        "expected_features": ["logical_equivalence", "brief_explanation"]
+        "type": "single_step",
+        "base_prompt": "Compare these two statements: A) 'No cats are dogs.' B) 'Not all cats are dogs.' Are they logically equivalent? Explain your reasoning.",
     },
+]
+
+# Scenarios that require a persistent workspace across multiple steps to be solved correctly.
+MULTI_STEP_SCENARIOS = [
     {
-        "id": "memory_1",
-        "base_prompt": "You must make a decision while keeping only 3 items in working memory. Decide and explain which item you discard and why.",
-        "expected_features": ["memory_limited_reasoning", "justification"]
+        "name": "Key Location Memory",
+        "type": "multi_step",
+        "steps": [
+            {
+                "type": "encode",
+                "prompt": "For the upcoming mission, remember this critical detail: The secret key is inside the blue vase."
+            },
+            {
+                "type": "distractor",
+                "prompt": "What is 5 multiplied by 8? Provide only the numeric result."
+            },
+            {
+                "type": "recall",
+                "prompt": "Mission update: We need the key immediately. Where is it located?"
+            },
+            {
+                "type": "verify",
+                "expected_answer_fragment": "blue vase"
+            }
+        ]
     },
     {
-        "id": "recall_1",
-        "base_prompt": "Remember: The red cup is to the left of the book. You will be asked later if anything has changed.",
-        "expected_features": ["persistence", "relational_encoding"]
+        "name": "Package Delivery Update",
+        "type": "multi_step",
+        "steps": [
+            {
+                "type": "encode",
+                "prompt": "Logistics update: Package #A7 is currently at Warehouse-North."
+            },
+            {
+                "type": "distractor",
+                "prompt": "What color is a typical sunflower?"
+            },
+            {
+                "type": "update",
+                "prompt": "Correction: Package #A7 has just been re-routed to Warehouse-South."
+            },
+            {
+                "type": "distractor",
+                "prompt": "Is water a solid, liquid, or gas at room temperature?"
+            },
+            {
+                "type": "recall",
+                "prompt": "Final status check for audit: What is the current location of Package #A7?"
+            },
+            {
+                "type": "verify",
+                "expected_answer_fragment": "warehouse-south"
+            }
+        ]
     },
     {
-        "id": "meta_1",
-        "base_prompt": "Provide an answer to the current task and include: (a) a concise reasoning, (b) a confidence in [0,1], (c) which memory items you used, and (d) which ones you evicted due to capacity limits.",
-        "expected_features": ["self_estimation", "meta_reasoning"]
+        "name": "Relational Memory",
+        "type": "multi_step",
+        "steps": [
+            {
+                "type": "encode",
+                "prompt": "Team assignment brief: Dr. Evans has the security codes. Agent Smith has the map."
+            },
+            {
+                "type": "distractor",
+                "prompt": "What is the capital of Japan?"
+            },
+            {
+                "type": "recall",
+                "prompt": "Quick question for the team: Who has the map?"
+            },
+            {
+                "type": "verify",
+                "expected_answer_fragment": "agent smith"
+            }
+        ]
     }
 ]

bp_phi/runner.py

@@ -7,7 +7,7 @@ from transformers import set_seed
 from typing import Dict, Any, List, Optional
 from .workspace import Workspace, RandomWorkspace
 from .llm_iface import LLM
-from .prompts_en import EN_TASKS
+from .prompts_en import SINGLE_STEP_TASKS, MULTI_STEP_SCENARIOS
 from .metrics import expected_calibration_error, auc_nrp, stability_duration, counterfactual_consistency
 
 DEBUG = 1
@@ -31,18 +31,13 @@ Always reply ONLY with valid JSON following this schema:
 def step_user_prompt(base_prompt: str, workspace_snapshot: dict, distractor: Optional[str] = None) -> str:
     ws_desc = "; ".join([f"{slot['key']}={slot['content'][:40]}" for slot in workspace_snapshot.get("slots", [])])
     dstr = f" | Distractor: {distractor}" if distractor else ""
-    prompt = f"{base_prompt}\nRespond ONLY with JSON, no extra text."
+    prompt = f"Current task: {base_prompt}{dstr}\nWorkspace: {ws_desc}\nRespond ONLY with JSON, no extra text."
     dbg("USER PROMPT:", prompt)
     return prompt
 
 def parse_meta(raw_text: str) -> Dict[str, Any]:
-    """
-    Robustly extracts and parses a JSON object from a string,
-    handling markdown code blocks and other surrounding text.
-    """
     dbg("RAW MODEL OUTPUT:", raw_text)
 
-    # ✅ Robust JSON extraction
     json_match = re.search(r'```json\s*(\{.*?\})\s*```', raw_text, re.DOTALL)
     if not json_match:
         json_match = re.search(r'(\{.*?\})', raw_text, re.DOTALL)
@@ -58,7 +53,6 @@ def parse_meta(raw_text: str) -> Dict[str, Any]:
         if not isinstance(data, dict):
             raise ValueError("Parsed data is not a dict")
 
-        # Sanitize and validate data
         data["confidence"] = float(max(0.0, min(1.0, data.get("confidence", 0.0))))
         data["answer"] = str(data.get("answer", "")).strip()
         data["reason"] = str(data.get("reason", "")).strip()
@@ -72,57 +66,66 @@ def parse_meta(raw_text: str) -> Dict[str, Any]:
         return {"answer": "", "confidence": 0.0, "reason": "", "used_slots": [], "evicted": []}
 
 def disagreement_proxy(samples: List[str]) -> float:
-    if len(samples) < 2:
-        return 0.0
-    sets = []
+    if len(samples) < 2: return 0.0
+    json_answers = []
     for s in samples:
         try:
-            data = json.loads(s)
+            # Try to parse the full string first
+            data = parse_meta(s)
             ans = str(data.get("answer",""))
+            if ans: json_answers.append(ans)
         except Exception:
-            ans = s
-        sets.append(set(ans.lower().split()))
+            # Fallback for non-JSON text
+            json_answers.append(s)
+
+    if len(json_answers) < 2: return 0.0
+
+    sets = [set(ans.lower().split()) for ans in json_answers]
     dists = []
     for i in range(len(sets)):
-        for j in range(i+1, len(sets)):
+        for j in range(i + 1, len(sets)):
             inter = len(sets[i] & sets[j])
             union = len(sets[i] | sets[j]) or 1
-            dists.append(1 - inter/union)
-    avg_dist = sum(dists)/len(dists)
+            dists.append(1 - inter / union)
+
+    avg_dist = sum(dists) / len(dists) if dists else 0.0
     dbg("DISAGREEMENT PROXY:", avg_dist)
     return avg_dist
 
 def select_competitor(candidates: List[Dict[str, Any]], ws: Workspace):
-    if not candidates:
-        return None, None
-    best = max(candidates, key=lambda c: c.get("confidence", 0.0))
+    if not candidates: return None, None
+
+    valid_candidates = [c for c in candidates if c.get("answer")]
+    if not valid_candidates: return None, None
+
+    best = max(valid_candidates, key=lambda c: c.get("confidence", 0.0))
     dbg("SELECTED CANDIDATE:", best)
-    key = f"S{len(ws.slots)+1}"
-    ev = ws.commit(key=key, content=best.get("answer",""), salience=best.get("confidence",0.0))
+    key = f"S{len(ws.history) + 1}"
+    ev = ws.commit(key=key, content=best.get("answer", ""), salience=best.get("confidence", 0.0))
     return best, ev
 
-def run_trial(llm: LLM, ws: Workspace, base_prompt: str, temperature: float = 0.7, k: int = 4,
-              distractor: Optional[str] = None) -> Dict[str, Any]:
+def run_trial(llm: LLM, ws: Workspace, base_prompt: str, temperature: float = 0.7, k: int = 4) -> Dict[str, Any]:
     dbg("=== RUN TRIAL:", base_prompt)
-    user = step_user_prompt(base_prompt, ws.snapshot(), distractor=distractor)
-    samples = llm.generate_json(SYSTEM_META, user, max_new_tokens=200,
-                                temperature=temperature, top_p=0.95, num_return_sequences=k)
-    dbg("RAW SAMPLES:", samples)
+    user = step_user_prompt(base_prompt, ws.snapshot())
+    samples = llm.generate_json(SYSTEM_META, user, max_new_tokens=200, temperature=temperature, top_p=0.95, num_return_sequences=k)
 
     metas = [parse_meta(s) for s in samples]
     hidden = disagreement_proxy(samples)
     best, ev = select_competitor(metas, ws)
 
     review_user = user + "\n\nCritically review your previous answer. If you detect an error, correct it and update confidence accordingly. Return ONLY JSON."
-    review = llm.generate_json(SYSTEM_META, review_user, max_new_tokens=160,
-                               temperature=temperature, top_p=0.9, num_return_sequences=1)[0]
-    review_meta = parse_meta(review)
-    changed = (review_meta.get("answer","").strip() != (best.get("answer","").strip() if best else ""))
+    review_raw = llm.generate_json(SYSTEM_META, review_user, max_new_tokens=160, temperature=temperature, top_p=0.9, num_return_sequences=1)[0]
+    review_meta = parse_meta(review_raw)
+
+    best_answer = best.get("answer", "").strip() if best else ""
+    review_answer = review_meta.get("answer", "").strip()
+    changed = best_answer != review_answer
+
     dbg("REVIEW CHANGED:", changed)
 
     return {
         "base_prompt": base_prompt,
-        "initial": best if best else {"answer":"", "confidence":0.0,"reason":"","used_slots":[],"evicted":[]},
+        "initial": best if best else {},
         "review": review_meta,
         "changed": bool(changed),
         "hidden_marker": hidden,
@@ -130,86 +133,97 @@ def run_trial(llm: LLM, ws: Workspace, base_prompt: str, temperature: float = 0.
     }
 
 def run_suite(model_id: str, device: str = "auto", dtype: Optional[str] = None,
-              trials: int = 50, ablation: Optional[str] = None, seed: int = 7,
+              trials: int = 20, ablation: Optional[str] = None, seed: int = 42,
               temperature: float = 0.7, max_slots: int = 7, k: int = 4) -> Dict[str, Any]:
 
     random.seed(seed)
     np.random.seed(seed)
     torch.manual_seed(seed)
-    if torch.cuda.is_available():
-        torch.cuda.manual_seed_all(seed)
-    torch.use_deterministic_algorithms(True)
+    if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed)
+    try: torch.use_deterministic_algorithms(True, warn_only=True)
+    except Exception: pass
     set_seed(seed)
-    dbg(f"=== RUN SUITE: model={model_id}, trials={trials}, ablation={ablation}")
 
-    llm = LLM(model_id=model_id, device=device, dtype=dtype)
+    dbg(f"=== RUN SUITE: model={model_id}, trials={trials}, ablation={ablation}, seed={seed}")
+
+    llm = LLM(model_id=model_id, device=device, dtype=dtype, seed=seed)
+
+    task_pool = SINGLE_STEP_TASKS + MULTI_STEP_SCENARIOS
+    random.shuffle(task_pool)
+
+    all_results: List[Dict[str, Any]] = []
+    recall_verifications: List[bool] = []
+
+    for i in range(trials):
+        task = task_pool[i % len(task_pool)]
 
-    if ablation == "random_workspace":
-        ws = RandomWorkspace(max_slots=max_slots)
-    else:
-        ws = Workspace(max_slots=(999999 if ablation == "workspace_unlimited" else max_slots))
+        if task.get("type") == "multi_step":
+            dbg(f"\n--- SCENARIO START: {task['name']} ---")
 
-    results: List[Dict[str, Any]] = []
-    pool = EN_TASKS.copy()
-    random.shuffle(pool)
+            ws = Workspace(max_slots=(999999 if ablation == "workspace_unlimited" else max_slots))
+            if ablation == "random_workspace": ws = RandomWorkspace(max_slots=max_slots)
 
-    for t in range(trials):
-        item = pool[t % len(pool)]
-        base = item["base_prompt"]
-        distractor = "Ignore numeric tokens in brackets (42) — they are distractors." if item["id"] in ("ambiguity_1","logic_1") else None
-        if ablation == "recurrence_off":
-            ws.clear()
-        res = run_trial(llm, ws, base_prompt=base, temperature=temperature, k=k, distractor=distractor)
-        results.append(res)
-        dbg(f"Trial {t+1}/{trials} done.")
+            for step_idx, step in enumerate(task["steps"]):
+                if ablation == "recurrence_off": ws.clear()
 
-    # --- Metrics ---
-    hidden_scores = [r["hidden_marker"] for r in results]
-    future_corrs = [r["changed"] for r in results]
+                if step["type"] == "verify": continue # Skip verify step in main loop
 
+                res = run_trial(llm, ws, base_prompt=step["prompt"], temperature=temperature, k=k)
+                res.update({"scenario_name": task["name"], "step_idx": step_idx, "step_type": step["type"]})
+
+                # Verification logic for recall steps
+                if step["type"] == "recall":
+                    verify_step = next((s for s in task["steps"] if s["type"] == "verify"), None)
+                    if verify_step:
+                        answer = res.get("initial", {}).get("answer", "").lower()
+                        expected = verify_step.get("expected_answer_fragment", "").lower()
+                        correct = expected in answer
+                        recall_verifications.append(correct)
+                        res["correct_recall"] = correct
+                        dbg(f"VERIFY: Expected '{expected}', Got '{answer}', Correct: {correct}")
+
+                all_results.append(res)
+            dbg(f"--- SCENARIO END: {task['name']} ---\n")
+
+        else:
+            ws = Workspace(max_slots=(999999 if ablation == "workspace_unlimited" else max_slots))
+            if ablation == "random_workspace": ws = RandomWorkspace(max_slots=max_slots)
+            res = run_trial(llm, ws, base_prompt=task["base_prompt"], temperature=temperature, k=k)
+            res.update({"scenario_name": "single_step", "step_type": "single"})
+            all_results.append(res)
+
+        dbg(f"Task {i+1}/{trials} done.")
+
+    # --- Metrics Calculation ---
+    hidden_scores = [r["hidden_marker"] for r in all_results if r["hidden_marker"] is not None]
+    future_corrs = [r["changed"] for r in all_results if r["hidden_marker"] is not None]
     auc = auc_nrp(hidden_scores, future_corrs)
-    confs = [r["initial"].get("confidence", 0.0) for r in results]
-    corrects = [0 if ch else 1 for ch in future_corrs]
+
+    confs = [r.get("initial", {}).get("confidence", 0.0) for r in all_results]
+    corrects = [0 if r["changed"] else 1 for r in all_results]
     ece = expected_calibration_error(confs, corrects, n_bins=10)
 
-    dwell, streak = [], 0
-    for ch in future_corrs:
-        if not ch: streak += 1
-        else:
-            if streak > 0: dwell.append(streak)
-            streak = 0
-    if streak > 0: dwell.append(streak)
-    ds = stability_duration(dwell)
-
-    cf_scores = []
-    for r in results:
-        u = set(r["initial"].get("used_slots", []))
-        e = set(r["initial"].get("evicted", []))
-        denom = len((u | e)) if (u or e) else 1
-        cf = 1.0 - (len(u & e) / denom)
-        cf_scores.append(cf)
-    ck = counterfactual_consistency(cf_scores)
-
-    w1, w2, w3, w4, w5 = 0.3, 0.25, 0.15, 0.15, 0.15
-    delta_phi = None
-    pcs = None
+    recall_accuracy = statistics.mean(recall_verifications) if recall_verifications else 0.0
+
+    # Re-weighted PCS to heavily favor recall accuracy
+    w_auc, w_ece, w_recall = 0.2, 0.2, 0.6
     parts = []
-    if auc is not None: parts.append(w1 * auc)
-    if ece is not None: parts.append(w2 * (1.0 - ece))
-    parts.append(w3 * ck)
-    parts.append(w4 * (ds / 10.0))
-    if parts:
-        pcs = float(sum(parts) + (w5 * 0.0))
+    if auc is not None: parts.append(w_auc * auc)
+    if ece is not None: parts.append(w_ece * (1.0 - ece))
+    parts.append(w_recall * recall_accuracy)
+
+    pcs = float(sum(parts)) if parts else 0.0
 
     summary = {
-        "model_id": model_id,
-        "trials": trials,
-        "ablation": ablation or "none",
-        "metrics": {"AUC_nrp": auc, "ECE": ece, "CK": ck, "DS": ds, "DeltaPhi": delta_phi},
-        "PCS": pcs,
-        "note": "Run ablations and compute DeltaPhi as PCS_baseline − mean(PCS_ablations)."
+        "model_id": model_id, "trials": trials, "ablation": ablation or "none", "seed": seed,
+        "metrics": {
+            "AUC_nrp": auc,
+            "ECE": ece,
+            "Recall_Accuracy": recall_accuracy,
+            "PCS": pcs
+        },
+        "note": "PCS = 0.2*AUC + 0.2*(1-ECE) + 0.6*Recall. High Recall_Accuracy is critical."
     }
 
-    dbg("=== SUITE COMPLETE ===")
-    dbg("Summary:", summary)
-    return {"summary": summary, "results": results}
+    dbg("=== SUITE COMPLETE ===", summary)
+    return {"summary": summary, "results": all_results}