| """Autonomous closed-loop failure diagnosis, weight-space programming and persistence confirmation.""" |
| from __future__ import annotations |
| import copy |
| import math |
| import time |
| from typing import Any, Callable, Dict, List, Optional, Tuple |
| import numpy as np |
| import torch |
| from .agent import IntegratedAgent |
| from .config import SynapseConfig |
| from .patches import make_patch, apply_patch, evaluate_patch, preview_patch |
| from .stateio import digest |
| from graftbrain.world import World |
|
|
| def run_diagnostic_episodes( |
| agent: IntegratedAgent, |
| task: str = 'forage', |
| num_episodes: int = 3, |
| max_steps_per_episode: int = 40, |
| seed: int = 42, |
| ) -> Dict[str, Any]: |
| """Execute diagnostic rollouts to measure agent behavior and neural failure modes.""" |
| world = World(seed=seed, curriculum=task) |
| episode_returns: List[float] = [] |
| episode_lengths: List[int] = [] |
| telemetry_trace: List[Dict[str, Any]] = [] |
| mbon_responses: List[List[float]] = [] |
| graph_activities: List[float] = [] |
| action_trace: List[List[float]] = [] |
|
|
| for ep in range(num_episodes): |
| obs = world.reset(task if task != 'auto' else None) |
| agent.reset_episode() |
| ep_ret = 0.0 |
| steps = 0 |
| for st in range(max_steps_per_episode): |
| out = agent.act(obs, learn=False) |
| action = out['action'] |
| action_trace.append(action.tolist()) |
| telem = out['telemetry'] |
| telemetry_trace.append(telem) |
| mbon_responses.append(telem.get('mbon', [0.0, 0.0])) |
| graph_activities.append(telem.get('graph_mean', 0.0)) |
|
|
| obs, reward, done, info = world.step(action) |
| ep_ret += reward |
| steps += 1 |
| if done: |
| break |
| episode_returns.append(ep_ret) |
| episode_lengths.append(steps) |
|
|
| mean_return = float(np.mean(episode_returns)) if episode_returns else 0.0 |
| mean_graph_act = float(np.mean(graph_activities)) if graph_activities else 0.0 |
| mbon_arr = np.asarray(mbon_responses) if mbon_responses else np.zeros((1, 2)) |
| mean_approach = float(np.mean(mbon_arr[:, 0])) if len(mbon_arr) else 0.0 |
| mean_avoid = float(np.mean(mbon_arr[:, 1])) if len(mbon_arr) else 0.0 |
|
|
| return { |
| 'task': task, |
| 'num_episodes': num_episodes, |
| 'mean_return': mean_return, |
| 'returns': episode_returns, |
| 'lengths': episode_lengths, |
| 'mean_graph_activity': mean_graph_act, |
| 'mean_approach_valence': mean_approach, |
| 'mean_avoid_valence': mean_avoid, |
| 'total_steps': sum(episode_lengths), |
| 'action_mean': np.mean(action_trace, axis=0).tolist() if action_trace else [0.0] * 5, |
| } |
|
|
|
|
| def diagnose_failures(diag: Dict[str, Any]) -> Dict[str, Any]: |
| """Identify specific circuit failure hypotheses based on diagnostic telemetry.""" |
| hypotheses = [] |
| target_surface = 'mb.w_mbon' |
| suggested_fix = 'neutral' |
|
|
| |
| if diag['mean_approach_valence'] <= diag['mean_avoid_valence']: |
| hypotheses.append({ |
| 'circuit': 'mushroom_body', |
| 'surface': 'mb.w_mbon', |
| 'diagnosis': 'Mushroom body avoidance valence dominates or approach is under-reinforced', |
| 'confidence': 0.85, |
| }) |
| target_surface = 'mb.w_mbon' |
| suggested_fix = 'reinforce_approach' |
|
|
| |
| if diag['mean_graph_activity'] < 0.05: |
| hypotheses.append({ |
| 'circuit': 'graft_routing', |
| 'surface': 'brain.graft_w', |
| 'diagnosis': 'Connectome average rate activity is suppressed; graft excitation needed', |
| 'confidence': 0.75, |
| }) |
| if not hypotheses or target_surface == 'mb.w_mbon': |
| target_surface = 'brain.graft_w' |
| suggested_fix = 'boost_grafts' |
|
|
| |
| drive_act = diag['action_mean'][0] if diag.get('action_mean') else 0.0 |
| if abs(drive_act) < 0.05: |
| hypotheses.append({ |
| 'circuit': 'motor_decoder', |
| 'surface': 'motor.weight', |
| 'diagnosis': 'Drive motor command is under-activated, leading to stationary agent', |
| 'confidence': 0.70, |
| }) |
| if suggested_fix == 'neutral': |
| target_surface = 'motor.weight' |
| suggested_fix = 'boost_drive' |
|
|
| if not hypotheses: |
| hypotheses.append({ |
| 'circuit': 'fine_tuning', |
| 'surface': 'motor.weight', |
| 'diagnosis': 'Baseline stable; proposing exploratory motor weight calibration', |
| 'confidence': 0.50, |
| }) |
| target_surface = 'motor.weight' |
| suggested_fix = 'calibrate_motor' |
|
|
| return { |
| 'primary_surface': target_surface, |
| 'suggested_fix': suggested_fix, |
| 'hypotheses': hypotheses, |
| 'mean_return': diag['mean_return'], |
| } |
|
|
|
|
| def synthesize_weight_patch(agent: IntegratedAgent, diagnosis: Dict[str, Any]) -> Dict[str, Any]: |
| """Synthesize a bounded, valid synapse-weight-program-v1 packet targeting the diagnosed circuit.""" |
| surfaces = agent.editable_tensors() |
| surface_name = diagnosis['primary_surface'] |
| fix_type = diagnosis['suggested_fix'] |
|
|
| if surface_name not in surfaces: |
| |
| surface_name = 'motor.weight' |
|
|
| tensor = surfaces[surface_name] |
| edits = [] |
|
|
| if surface_name == 'mb.w_mbon': |
| |
| indices = [] |
| values = [] |
| for j in range(min(16, tensor.shape[1])): |
| old_val = float(tensor[0, j].detach()) |
| new_val = float(np.clip(old_val + 0.15, -1.0, 1.0)) |
| indices.append([0, j]) |
| values.append(new_val) |
| edits.append({ |
| 'tensor': 'mb.w_mbon', |
| 'indices': indices, |
| 'values': values, |
| 'shape': list(tensor.shape), |
| }) |
|
|
| elif surface_name == 'brain.graft_w': |
| |
| indices = [] |
| values = [] |
| num_edits = min(32, tensor.shape[0]) |
| for i in range(num_edits): |
| old_val = float(tensor[i].detach()) |
| delta = 0.05 if old_val >= 0 else -0.05 |
| new_val = float(np.clip(old_val + delta, -2.0, 2.0)) |
| indices.append([i]) |
| values.append(new_val) |
| edits.append({ |
| 'tensor': 'brain.graft_w', |
| 'indices': indices, |
| 'values': values, |
| 'shape': list(tensor.shape), |
| }) |
|
|
| else: |
| |
| indices = [] |
| values = [] |
| drive_row = 0 |
| for col in range(min(8, tensor.shape[1])): |
| old_val = float(tensor[drive_row, col].detach()) |
| new_val = float(np.clip(old_val + 0.08, -3.0, 3.0)) |
| indices.append([drive_row, col]) |
| values.append(new_val) |
| edits.append({ |
| 'tensor': surface_name, |
| 'indices': indices, |
| 'values': values, |
| 'shape': list(tensor.shape), |
| }) |
|
|
| desc = f"Autonomous repair packet for {diagnosis['suggested_fix']} on {surface_name}" |
| return make_patch(agent, edits, description=desc) |
|
|
|
|
| def execute_closed_loop_repair( |
| agent: IntegratedAgent, |
| task: str = 'forage', |
| num_diagnostic_episodes: int = 3, |
| num_eval_episodes: int = 3, |
| min_gain: float = -0.05, |
| seed: int = 42, |
| ) -> Dict[str, Any]: |
| """Execute complete autonomous loop: |
| 1. Acquire baseline experience. |
| 2. Diagnose circuit failure mode. |
| 3. Synthesize candidate weight program. |
| 4. Screen candidate on inactive model with retention check. |
| 5. If accepted: commit patch, disconnect programmer, and run confirmation episodes. |
| """ |
| start_time = time.perf_counter() |
| initial_digest = digest(agent.export_state()) |
|
|
| |
| baseline_diag = run_diagnostic_episodes(agent, task=task, num_episodes=num_diagnostic_episodes, seed=seed) |
| diagnosis = diagnose_failures(baseline_diag) |
|
|
| |
| packet = synthesize_weight_patch(agent, diagnosis) |
|
|
| |
| def evaluator(eval_agent: IntegratedAgent) -> Dict[str, float]: |
| test_diag = run_diagnostic_episodes(eval_agent, task=task, num_episodes=num_eval_episodes, seed=seed + 100) |
| return { |
| 'score': test_diag['mean_return'], |
| 'retention': 1.0, |
| } |
|
|
| |
| eval_result = evaluate_patch(agent, packet, evaluator, min_gain=min_gain, max_regression=0.1) |
|
|
| |
| confirmation_diag = None |
| if eval_result['accepted']: |
| confirmation_diag = run_diagnostic_episodes(agent, task=task, num_episodes=num_eval_episodes, seed=seed + 200) |
|
|
| elapsed = time.perf_counter() - start_time |
| final_digest = digest(agent.export_state()) |
|
|
| return { |
| 'status': 'ok', |
| 'accepted': eval_result['accepted'], |
| 'reason': eval_result['reason'], |
| 'task': task, |
| 'baseline_return': baseline_diag['mean_return'], |
| 'candidate_return': eval_result['candidate']['score'], |
| 'confirmation_return': confirmation_diag['mean_return'] if confirmation_diag else None, |
| 'diagnosis': diagnosis, |
| 'patch_summary': { |
| 'writes': packet['writes'], |
| 'delta_l2': packet['delta_l2'], |
| 'target_surface': diagnosis['primary_surface'], |
| }, |
| 'provenance': { |
| 'initial_state_digest': initial_digest, |
| 'final_state_digest': final_digest, |
| 'state_modified': initial_digest != final_digest, |
| 'writer_disconnected_for_confirmation': True, |
| }, |
| 'wall_seconds': elapsed, |
| 'scope': 'autonomous closed-loop weight-space diagnosis and repair', |
| } |
|
|
