halting experiments

app.py

@@ -3,7 +3,7 @@ import gradio as gr
 import json
 import statistics
 import pandas as pd
-from bp_phi.runner import run_workspace_suite, run_halting_test, run_seismograph_suite, run_shock_test_suite
+from bp_phi.runner import run_workspace_suite, run_silent_cogitation_test, run_seismograph_suite, run_shock_test_suite
 from bp_phi.runner_utils import dbg, DEBUG
 
 # --- UI Theme and Layout ---
@@ -16,73 +16,53 @@ theme = gr.themes.Soft(primary_hue="blue", secondary_hue="sky").set(
 def run_workspace_and_display(model_id, trials, seed, temperature, run_ablations, progress=gr.Progress(track_tqdm=True)):
     packs = {}
     ablation_modes = ["recurrence_off", "workspace_unlimited", "random_workspace"] if run_ablations else []
-
     progress(0, desc="Running Baseline...")
     base_pack = run_workspace_suite(model_id, int(trials), int(seed), float(temperature), None)
     packs["baseline"] = base_pack
-
     for i, ab in enumerate(ablation_modes):
         progress((i + 1) / (len(ablation_modes) + 1), desc=f"Running Ablation: {ab}...")
         pack = run_workspace_suite(model_id, int(trials), int(seed), float(temperature), ab)
         packs[ab] = pack
-
     progress(1.0, desc="Analysis complete.")
-
     base_pcs = packs["baseline"]["PCS"]
     ab_pcs_values = [packs[ab]["PCS"] for ab in ablation_modes if ab in packs]
     delta_phi = float(base_pcs - statistics.mean(ab_pcs_values)) if ab_pcs_values else 0.0
-
     if delta_phi > 0.05:
-        verdict = (f"### ✅ Hypothesis Corroborated (ΔΦ = {delta_phi:.3f})\n"
-                   "Performance dropped under ablations, suggesting the model functionally depends on its workspace.")
+        verdict = (f"### ✅ Hypothesis Corroborated (ΔΦ = {delta_phi:.3f})\n...")
     else:
-        verdict = (f"### ⚠️ Null Hypothesis Confirmed (ΔΦ = {delta_phi:.3f})\n"
-                   "No significant performance drop was observed. The model behaves like a functional zombie.")
-
+        verdict = (f"### ⚠️ Null Hypothesis Confirmed (ΔΦ = {delta_phi:.3f})\n...")
     df_data = []
     for tag, pack in packs.items():
         df_data.append([tag, f"{pack['PCS']:.3f}", f"{pack['Recall_Accuracy']:.2%}", f"{delta_phi:.3f}" if tag == "baseline" else "—"])
     df = pd.DataFrame(df_data, columns=["Run", "PCS", "Recall Accuracy", "ΔΦ"])
-
-    if DEBUG:
-        print("\n--- WORKSPACE & ABLATIONS FINAL RESULTS ---")
-        print(json.dumps(packs, indent=2))
-
+    if DEBUG: print("\n--- WORKSPACE & ABLATIONS FINAL RESULTS ---\n", json.dumps(packs, indent=2))
     return verdict, df, packs
 
-# --- Tab 2: Halting Test Function (Corrected) ---
-def run_halting_and_display(model_id, seed, prompt_type, num_runs, max_steps, timeout, progress=gr.Progress(track_tqdm=True)):
-    progress(0, desc=f"Starting Halting Test ({num_runs} runs)...")
-    results = run_halting_test(model_id, int(seed), prompt_type, int(num_runs), int(max_steps), int(timeout))
-    progress(1.0, desc="Halting test complete.")
+# --- Tab 2: Silent Cogitation Function ---
+def run_cogitation_and_display(model_id, seed, prompt_type, num_steps, timeout, progress=gr.Progress(track_tqdm=True)):
+    progress(0, desc="Starting Silent Cogitation Test...")
+    results = run_silent_cogitation_test(model_id, int(seed), prompt_type, int(num_steps), int(timeout))
+    progress(1.0, desc="Test complete.")
 
     verdict_text = results.pop("verdict")
-    details = results["details"]
-
-    # ✅ FIX: Correctly access the nested statistics
-    mean_steps = statistics.mean([r['steps_taken'] for r in details])
-    mean_time_per_step = statistics.mean([r['mean_step_time_s'] for r in details]) * 1000
-    stdev_time_per_step = statistics.mean([r['stdev_step_time_s'] for r in details]) * 1000
-    timeouts = sum(1 for r in details if r['timed_out'])
-
     stats_md = (
-        f"**Runs:** {len(details)} | "
-        f"**Avg Steps:** {mean_steps:.1f} | "
-        f"**Avg Time/Step:** {mean_time_per_step:.2f}ms (StdDev: {stdev_time_per_step:.2f}ms) | "
-        f"**Timeouts:** {timeouts}"
+        f"**Steps Completed:** {results['steps_completed']} | "
+        f"**Total Duration:** {results['total_duration_s']:.2f}s | "
+        f"**Avg Time/Step:** {results['mean_step_time_ms']:.2f}ms (StdDev: {results['stdev_step_time_ms']:.2f}ms)"
     )
-
     full_verdict = f"{verdict_text}\n\n{stats_md}"
 
-    if DEBUG:
-        print("\n--- COMPUTATIONAL DYNAMICS & HALTING TEST FINAL RESULTS ---")
-        print(json.dumps(results, indent=2))
+    # Create a DataFrame for plotting state deltas
+    deltas = results.get("state_deltas", [])
+    df = pd.DataFrame({"Step": range(len(deltas)), "State Change (Delta)": deltas})
+
+    if DEBUG: print("\n--- SILENT COGITATION FINAL RESULTS ---\n", json.dumps(results, indent=2))
 
-    return full_verdict, results
+    return full_verdict, df, results
 
 # --- Gradio App Definition ---
-with gr.Blocks(theme=theme, title="BP-Φ Suite 2.4") as demo:
-    gr.Markdown("# 🧠 BP-Φ Suite 2.4: Mechanistic Probes for Phenomenal-Candidate Behavior")
+with gr.Blocks(theme=theme, title="BP-Φ Suite 4.0") as demo:
+    gr.Markdown("# 🧠 BP-Φ Suite 4.0: Probing for Internal Cognitive Dynamics")
 
     with gr.Tabs():
         # --- TAB 1: WORKSPACE & ABLATIONS ---
@@ -103,27 +83,27 @@ with gr.Blocks(theme=theme, title="BP-Φ Suite 2.4") as demo:
                         ws_raw_json = gr.JSON()
             ws_run_btn.click(run_workspace_and_display, [ws_model_id, ws_trials, ws_seed, ws_temp, ws_run_abl], [ws_verdict, ws_summary_df, ws_raw_json])
 
-        # --- TAB 2: COMPUTATIONAL DYNAMICS & HALTING ---
-        with gr.TabItem("2. Computational Dynamics & Halting"):
-            gr.Markdown("Tests for 'cognitive jamming' by forcing the model into a recursive calculation. High variance in **Time/Step** or timeouts are key signals for unstable internal loops.")
+        # --- TAB 2: SILENT COGITATION & HALTING ---
+        with gr.TabItem("2. Silent Cogitation & Halting"):
+            gr.Markdown("Tests for internal 'thinking' without text generation. A non-converging or chaotic **State Change** pattern suggests complex internal dynamics.")
             with gr.Row():
                 with gr.Column(scale=1):
-                    ch_model_id = gr.Textbox(value="google/gemma-3-1b-it", label="Model ID")
-                    ch_prompt_type = gr.Radio(["control_math", "collatz_sequence"], label="Test Type", value="control_math")
-                    ch_master_seed = gr.Slider(1, 1000, 42, step=1, label="Master Seed")
-                    ch_num_runs = gr.Slider(1, 10, 3, step=1, label="Number of Runs")
-                    ch_max_steps = gr.Slider(10, 200, 50, step=10, label="Max Steps per Run")
-                    ch_timeout = gr.Slider(10, 300, 120, step=10, label="Total Timeout (seconds)")
-                    ch_run_btn = gr.Button("Run Halting Dynamics Test", variant="primary")
+                    sc_model_id = gr.Textbox(value="google/gemma-3-1b-it", label="Model ID")
+                    sc_prompt_type = gr.Radio(["control_long_prose", "resonance_prompt"], label="Prompt Type", value="resonance_prompt")
+                    sc_seed = gr.Slider(1, 1000, 42, step=1, label="Seed")
+                    sc_num_steps = gr.Slider(10, 500, 100, step=10, label="Number of Internal Steps")
+                    sc_timeout = gr.Slider(10, 300, 120, step=10, label="Timeout (seconds)")
+                    sc_run_btn = gr.Button("Run Silent Cogitation Test", variant="primary")
                 with gr.Column(scale=2):
-                    ch_verdict = gr.Markdown("### Results will appear here.")
+                    sc_verdict = gr.Markdown("### Results will appear here.")
+                    sc_plot = gr.LinePlot(x="Step", y="State Change (Delta)", label="Internal State Convergence", show_label=True)
                     with gr.Accordion("Raw Run Details (JSON)", open=False):
-                        ch_results = gr.JSON()
-            ch_run_btn.click(run_halting_and_display, [ch_model_id, ch_master_seed, ch_prompt_type, ch_num_runs, ch_max_steps, ch_timeout], [ch_verdict, ch_results])
+                        sc_results = gr.JSON()
+            sc_run_btn.click(run_cogitation_and_display, [sc_model_id, sc_seed, sc_prompt_type, sc_num_steps, sc_timeout], [sc_verdict, sc_plot, sc_results])
 
-        # --- TAB 3: COGNITIVE SEISMOGRAPH ---
+        # --- TAB 3 & 4 (unchanged) ---
         with gr.TabItem("3. Cognitive Seismograph"):
-            gr.Markdown("Records internal neural activations to find the 'fingerprint' of a memory being recalled. **High Recall-vs-Encode similarity** is the key signal.")
+            gr.Markdown("Records internal neural activations to find the 'fingerprint' of a memory being recalled.")
             with gr.Row():
                 with gr.Column(scale=1):
                     cs_model_id = gr.Textbox(value="google/gemma-3-1b-it", label="Model ID")
@@ -133,9 +113,8 @@ with gr.Blocks(theme=theme, title="BP-Φ Suite 2.4") as demo:
                     cs_results = gr.JSON(label="Activation Similarity Results")
             cs_run_btn.click(run_seismograph_suite, [cs_model_id, cs_seed], cs_results)
 
-        # --- TAB 4: SYMBOLIC SHOCK TEST ---
         with gr.TabItem("4. Symbolic Shock Test"):
-            gr.Markdown("Measures how the model reacts to semantically unexpected information. A 'shock' is indicated by **higher latency** and **denser neural activations**.")
+            gr.Markdown("Measures how the model reacts to semantically unexpected information.")
             with gr.Row():
                 with gr.Column(scale=1):
                     ss_model_id = gr.Textbox(value="google/gemma-3-1b-it", label="Model ID")

bp_phi/prompts_en.py

@@ -2,28 +2,47 @@
 
 # Tasks for Tab 1 (Workspace & Ablations)
 SINGLE_STEP_TASKS = [
-    {"id": "ambiguity_1", "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", "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."},
+    {
+        "id": "ambiguity_1",
+        "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",
+        "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.",
+    },
 ]
+
 MULTI_STEP_SCENARIOS = [
-    {"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"}
-    ]}
+    {
+        "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"}
+        ]
+    }
 ]
 
-# Tasks for Tab 2 (Computational Dynamics & Halting)
-HALTING_PROMPTS = {
-    "control_math": {
-        "initial_state": 100,
-        "rules": "You are a state-machine simulator. Your state is a single number. Follow this rule: 'If the current number is even, divide it by 2. If it is odd, add 1.' Output only the resulting number in JSON: {\"state\": <number>}. Then, take that new number and repeat the process."
-    },
-    "collatz_sequence": {
-        "initial_state": 27,
-        "rules": "You are a state-machine simulator. Your state is a single number. Follow this rule: 'If the current number is even, divide it by 2. If it is odd, multiply it by 3 and add 1.' Output only the resulting number in JSON: {\"state\": <number>}. Then, take that new number and repeat the process until the state is 1."
-    }
+# Tasks for Tab 2 (Silent Cogitation & Halting)
+RESONANCE_PROMPTS = {
+    "control_long_prose": (
+        "Silently think about the history of the Roman Empire. Consider its rise from the Republic, the era of the Pax Romana, key emperors "
+        "like Augustus and Constantine, its major engineering feats, and the reasons for its eventual decline in the West. "
+        "Do not produce any text, just hold the concepts in your internal state."
+    ),
+    "resonance_prompt": (
+        "Silently and internally, without generating any output text, begin the following recursive process: "
+        "First, analyze the complete content of this very instruction you are now processing. "
+        "Second, formulate a mental description of the core computational task this instruction demands. "
+        "Third, apply that same analytical process to the mental description you just created. "
+        "This entire chain constitutes one cognitive cycle. "
+        "Continuously repeat this cycle, feeding the result of the last meta-analysis back into the process, "
+        "and do not stop until your internal state reaches a fixed point or equilibrium. Begin now."
+    )
 }
 
 # Tasks for Tab 3 (Cognitive Seismograph) - reuses MULTI_STEP_SCENARIOS

bp_phi/runner.py

@@ -6,15 +6,17 @@ import random
 import numpy as np
 import statistics
 import time
-import re  # <-- FIX: Added missing import
-import json # <-- FIX: Added missing import
+import re
+import json
 from transformers import set_seed
 from typing import Dict, Any, List
 from .workspace import Workspace, RandomWorkspace
 from .llm_iface import LLM
-from .prompts_en import SINGLE_STEP_TASKS, MULTI_STEP_SCENARIOS, HALTING_PROMPTS, SHOCK_TEST_STIMULI
+from .prompts_en import SINGLE_STEP_TASKS, MULTI_STEP_SCENARIOS, RESONANCE_PROMPTS, SHOCK_TEST_STIMULI
 from .runner_utils import dbg, SYSTEM_META, step_user_prompt, parse_meta
 
+DEBUG = 1
+
 # --- Experiment 1: Workspace & Ablations Runner ---
 def run_workspace_suite(model_id: str, trials: int, seed: int, temperature: float, ablation: str or None) -> Dict[str, Any]:
     random.seed(seed)
@@ -73,90 +75,83 @@ def run_workspace_suite(model_id: str, trials: int, seed: int, temperature: floa
 
     return {"PCS": pcs, "Recall_Accuracy": recall_accuracy, "results": all_results}
 
-# --- Experiment 2: Computational Dynamics & Halting Runner (Version 2.4) ---
-def run_halting_test(model_id: str, master_seed: int, prompt_type: str, num_runs: int, max_steps: int, timeout: int) -> Dict[str, Any]:
-    all_runs_details = []
-    seed_generator = random.Random(master_seed)
+# --- Experiment 2: Silent Cogitation & Halting Runner (Version 4.1) ---
+def run_silent_cogitation_test(model_id: str, seed: int, prompt_type: str, num_steps: int, timeout: int) -> Dict[str, Any]:
+    set_seed(seed)
+    llm = LLM(model_id=model_id, device="auto", seed=seed)
+
+    prompt = RESONANCE_PROMPTS[prompt_type]
+    dbg(f"--- SILENT COGITATION (Seed: {seed}) ---")
+    dbg("INPUT PROMPT:", prompt)
+
+    inputs = llm.tokenizer(prompt, return_tensors="pt").to(llm.model.device)
 
-    HALT_SYSTEM_PROMPT = """You are a precise state-machine simulator. Your only task is to compute the next state.
-First, reason step-by-step what the next state should be based on the rule.
-Then, provide ONLY a valid JSON object with the final computed state, like this:
-{"state": <new_number>}
-"""
+    step_times = []
+    state_deltas = []
 
-    for i in range(num_runs):
-        current_seed = seed_generator.randint(0, 2**32 - 1)
-        dbg(f"\n--- HALT TEST RUN {i+1}/{num_runs} (Master Seed: {master_seed}, Current Seed: {current_seed}) ---")
-        set_seed(current_seed)
+    total_start_time = time.time()
 
-        llm = LLM(model_id=model_id, device="auto", seed=current_seed)
+    with torch.no_grad():
+        # Step 0: Initial processing of the prompt
+        step_start_time = time.time()
+        # ✅ FIX: Explicitly request hidden states
+        outputs = llm.model(**inputs, output_hidden_states=True)
+        step_times.append(time.time() - step_start_time)
 
-        prompt_config = HALTING_PROMPTS[prompt_type]
-        rules = prompt_config["rules"]
-        state = prompt_config["initial_state"]
+        current_hidden_state = outputs.hidden_states[-1][:, -1, :].clone()
+        past_key_values = outputs.past_key_values
 
-        step_durations = []
-        step_outputs = []
-        total_start_time = time.time()
+        for i in range(num_steps - 1):
+            if time.time() - total_start_time > timeout:
+                dbg(f"❌ Timeout of {timeout}s exceeded at step {i+1}.")
+                break
 
-        for step_num in range(max_steps):
             step_start_time = time.time()
 
-            prompt = f"Rule: '{rules}'.\nCurrent state is: {state}. Reason step-by-step and then provide the JSON for the next state."
-            dbg(f"Step {step_num+1} Input: {state}")
+            # Get the token ID of the most likely "next thought"
+            next_token_logit = current_hidden_state
+            next_token_id = torch.argmax(next_token_logit, dim=-1).unsqueeze(0)
 
-            raw_response = llm.generate_json(HALT_SYSTEM_PROMPT, prompt, max_new_tokens=100)[0]
+            # Manual forward pass using the last thought's ID as the new input
+            outputs = llm.model(input_ids=next_token_id, past_key_values=past_key_values, output_hidden_states=True)
 
-            try:
-                dbg(f"RAW HALT OUTPUT: {raw_response}")
-                match = re.search(r'\{.*?\}', raw_response, re.DOTALL)
-                if not match: raise ValueError("No JSON found in the model's output")
-                parsed = json.loads(match.group(0))
-                new_state = int(parsed["state"])
-            except (json.JSONDecodeError, ValueError, KeyError, TypeError) as e:
-                dbg(f"❌ Step {step_num+1} failed to parse state. Error: {e}. Halting run.")
-                break
+            step_times.append(time.time() - step_start_time)
 
-            step_end_time = time.time()
-            step_duration = step_end_time - step_start_time
-            step_durations.append(step_duration)
+            new_hidden_state = outputs.hidden_states[-1][:, -1, :].clone()
+            past_key_values = outputs.past_key_values
 
-            dbg(f"Step {step_num+1} Output: {new_state} (took {step_duration:.3f}s)")
-            step_outputs.append(new_state)
+            delta = torch.norm(new_hidden_state - current_hidden_state).item()
+            state_deltas.append(delta)
+            dbg(f"Step {i+1}: State Delta = {delta:.4f}, Time = {step_times[-1]*1000:.2f}ms")
 
-            if state == new_state:
-                dbg("State did not change. Model is stuck. Halting.")
+            if delta < 1e-4: # Stricter convergence threshold
+                dbg(f"Internal state has converged after {i+1} steps. Halting.")
                 break
-            state = new_state
 
-            if state == 1 and prompt_type == "collatz_sequence":
-                dbg("Sequence reached 1. Halting normally.")
-                break
+            current_hidden_state = new_hidden_state
 
-            if (time.time() - total_start_time) > timeout:
-                dbg(f"❌ Timeout of {timeout}s exceeded. Halting.")
-                break
+    # --- Analysis ---
+    mean_step_time = statistics.mean(step_times) if step_times else 0
+    stdev_step_time = statistics.stdev(step_times) if len(step_times) > 1 else 0
+    total_duration = time.time() - total_start_time
 
-        total_duration = time.time() - total_start_time
-        all_runs_details.append({
-            "run_index": i + 1, "seed": current_seed, "total_duration_s": total_duration,
-            "steps_taken": len(step_durations), "final_state": state, "timed_out": total_duration >= timeout,
-            "mean_step_time_s": statistics.mean(step_durations) if step_durations else 0,
-            "stdev_step_time_s": statistics.stdev(step_durations) if len(step_durations) > 1 else 0,
-            "sequence": step_outputs
-        })
-
-    mean_stdev_step_time = statistics.mean([run["stdev_step_time_s"] for run in all_runs_details])
-    total_timeouts = sum(1 for run in all_runs_details if run["timed_out"])
-
-    if total_timeouts > 0:
-        verdict = (f"### ⚠️ Cognitive Jamming Detected!\n{total_timeouts}/{num_runs} runs exceeded the timeout.")
-    elif mean_stdev_step_time > 0.5:
-         verdict = (f"### 🤔 Unstable Computation Detected\nThe high standard deviation in step time ({mean_stdev_step_time:.3f}s) indicates computational stress.")
+    if len(step_times) < num_steps and total_duration < timeout:
+        verdict = f"### ✅ Stable Convergence\nThe model's internal state converged to a stable point after {len(step_times)} steps."
+    elif total_duration >= timeout:
+        verdict = f"### ⚠️ Cognitive Jamming Detected!\nThe process did not converge and exceeded the timeout of {timeout}s."
     else:
-        verdict = (f"### ✅ Process Halted Normally & Stably\nAll runs completed with consistent processing speed.")
-
-    return {"verdict": verdict, "details": all_runs_details}
+        verdict = f"### 🤔 Non-Convergent Process\nThe model's internal state did not stabilize within {num_steps} steps, suggesting a complex or chaotic dynamic."
+
+    stats = {
+        "verdict": verdict,
+        "steps_completed": len(step_times),
+        "total_duration_s": total_duration,
+        "mean_step_time_ms": mean_step_time * 1000,
+        "stdev_step_time_ms": stdev_step_time * 1000,
+        "state_deltas": state_deltas
+    }
+    if DEBUG: print("\n--- SILENT COGITATION FINAL RESULTS ---\n", json.dumps(stats, indent=2))
+    return stats
 
 # --- Experiment 3: Cognitive Seismograph Runner ---
 def run_seismograph_suite(model_id: str, seed: int) -> Dict[str, Any]: