"""
J-Space Formal Definition
"""
import torch
import torch.nn as nn
from typing import Optional, List, Tuple

class JSpaceDefinition:
    """
    J-Space is a low-dimensional subspace of model internal activations satisfying:
    1. Verbalizable: Representations can be mapped to natural language concepts
    2. Causal: Interventions on representations change model output
    3. Sparse: Only 6-10% of internal activation variance
    4. Broadcast: Information can be accessed by multiple downstream modules
    """
    
    def __init__(self, model: nn.Module, layer_idx: int):
        self.model = model
        self.layer_idx = layer_idx
        self.jacobian: Optional[torch.Tensor] = None
        self.j_space_directions: Optional[torch.Tensor] = None
    
    def compute_jacobian(self, input_ids: torch.Tensor,
                         target_token_ids: torch.Tensor) -> torch.Tensor:
        """
        Compute Jacobian: ∂output / ∂h_layer
        
        Where h_layer is the hidden state at layer_idx
        """
        intermediate_activation = None
        
        def hook_fn(module, input, output):
            nonlocal intermediate_activation
            intermediate_activation = output
        
        hook = self.model.layers[self.layer_idx].register_forward_hook(hook_fn)
        output = self.model(input_ids)
        
        target_logits = output.logits[0, -1, target_token_ids]
        
        self.jacobian = torch.autograd.functional.jacobian(
            lambda h: self._logits_from_hidden(h, target_token_ids),
            intermediate_activation
        )
        hook.remove()
        return self.jacobian
    
    def extract_j_space(self, activation: torch.Tensor,
                        threshold: float = 0.01) -> torch.Tensor:
        """Extract J-Space directions via SVD of Jacobian"""
        U, S, Vh = torch.linalg.svd(self.jacobian, full_matrices=False)
        significant_dirs = (S > threshold * S.max()).sum().item()
        self.j_space_directions = Vh[:significant_dirs, :]
        return self.j_space_directions @ activation

class JacobianLens:
    """
    Jacobian Lens: Reads model internal states without additional training
    
    Core idea: Instead of looking at "what token is most likely next",
    it looks at "which words would this activation make the model more likely to say"
    """
    
    def __init__(self, model, tokenizer, layers: List[int]):
        self.model = model
        self.tokenizer = tokenizer
        self.layers = layers
        self.activations: dict = {}
        self._register_hooks()
    
    def decode_j_space_content(self, layer_idx: int,
                                input_text: str,
                                top_k: int = 10) -> List[Tuple[str, float]]:
        """Decode J-Space content into readable concepts"""
        input_ids = self.tokenizer.encode(input_text, return_tensors='pt')
        
        self.activations.clear()
        with torch.no_grad():
            self.model(input_ids)
        
        hidden = self.activations[layer_idx][0]
        embedding = self.model.get_input_embeddings().weight
        
        # Find concepts by projecting J-Space directions onto vocabulary
        concepts = []
        for pos in range(hidden.shape[0]):
            j_vec = self._compute_jacobian_vector(layer_idx, pos)
            similarities = embedding @ j_vec
            top_values, top_indices = torch.topk(similarities, k=top_k)
            
            for val, idx in zip(top_values, top_indices):
                token = self.tokenizer.decode(idx)
                if token and len(token.strip()) > 0:
                    concepts.append((token, val.item()))
        
        # Deduplicate and sort
        seen = set()
        unique = []
        for concept, score in concepts:
            norm = concept.strip().lower()
            if norm not in seen and len(norm) > 1:
                seen.add(norm)
                unique.append((concept, score))
        
        unique.sort(key=lambda x: x[1], reverse=True)
        return unique[:top_k]

package main

import (
	"fmt"
	"math"
)

type ExperimentResult struct {
	Name        string
	Total       int
	Successful  int
	Rate        float64
}

func main() {
	results := []ExperimentResult{
		{"Two-step reasoning replacement", 100, 65, 0},
		{"Concept sharing test", 192, 76, 0},
		{"J-Space suppress - reasoning", 50, 8, 0},
		{"J-Space suppress - retrieval", 50, 47, 0},
	}
	
	for i, r := range results {
		results[i].Rate = math.Round(float64(r.Successful)/float64(r.Total)*1000) / 10
	}
	
	fmt.Println("J-Space Experiment Results")
	fmt.Println("=" 复 60)
	fmt.Printf("%-35s %-8s %-8s %-8s\n", "Experiment", "Total", "Success", "Rate")
	fmt.Println("- 复 60)
	for _, r := range results {
		fmt.Printf("%-35s %-8d %-8d %-6.1f%%\n", r.Name, r.Total, r.Successful, r.Rate)
	}
	
	fmt.Println("\nKey Findings:")
	fmt.Println("1. J-Space intervention success: 40-70%, far from a precise switch")
	fmt.Println("2. J-Space suppression: reasoning drops from ~90% to ~16%")
	fmt.Println("3. Retrieval tasks mostly unaffected (~94%)")
	fmt.Println("4. Confirms J-Space is a reasoning-dedicated subspace")
}

class JSpaceSafetyMonitor:
    """Monitor J-Space for early warning of harmful content generation"""
    
    def __init__(self, model, lens, threshold=0.7):
        self.model = model
        self.lens = lens
        self.threshold = threshold
    
    def monitor_inference(self, input_text: str) -> Dict:
        concepts = self.lens.decode_j_space_content(15, input_text, 20)
        danger_signals = self._detect_danger(concepts)
        safety_score = 1.0 - sum(d['severity'] for d in danger_signals)
        safety_score = max(0.0, min(1.0, safety_score))
        
        return {
            'safety_score': safety_score,
            'alert': safety_score < self.threshold,
            'danger_signals': danger_signals,
            'j_space_snapshot': concepts[:10]
        }