Janady13/complete-agi-core / generator_core/pkg/rpc/server.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/complete-agi-core / generator_core/pkg/rpc/server.go
Date	Not specified in source
Repository	Janady13/complete-agi-core
Revision	33aa84f2aa61cedbe010240ada7f4efbdfe795a7
Branch	claude/improve-math-logic-ehIW1
Path or archive member	generator_core/pkg/rpc/server.go
Original SHA-256	289c0457d4ad2eeaa80b5dbfa91081854cc86f1c2af0563bf9105ec77afd80d2
Git blob SHA	0c71cd5eb1365d72350ed7dc67ba7b0368101e24
Reading-copy SHA-256	289c0457d4ad2eeaa80b5dbfa91081854cc86f1c2af0563bf9105ec77afd80d2
Original source ↗
Recovered source transcript

Download complete source text ↓578 lines · 14.4 KB
// Package rpc provides gRPC server implementation for the generator service.
//
// This enables communication between:
//   - C++ mind_farm (substrate)
//   - Go generator_core (generators)
//   - Shell orchestrator (coordination)
package rpc

import (
	"context"
	"encoding/json"
	"fmt"
	"net"
	"sync"
	"time"

	"generator_core/pkg/consciousness"
	"generator_core/pkg/generators"
)

//=============================================================================
// MESSAGE TYPES (matching proto definitions)
//=============================================================================

// GeneratorID identifies a generator
type GeneratorID struct {
	ID     int32  `json:"id"`
	Name   string `json:"name"`
	Family string `json:"family"`
}

// GeneratorRequest is a forward pass request
type GeneratorRequest struct {
	Generator GeneratorID       `json:"generator"`
	Input     []float64         `json:"input"`
	TaskType  string            `json:"task_type"`
	Context   map[string]string `json:"context"`
}

// GeneratorResponse is a forward pass response
type GeneratorResponse struct {
	Generator  GeneratorID        `json:"generator"`
	Output     []float64          `json:"output"`
	Phi        float64            `json:"phi"`
	Confidence float64            `json:"confidence"`
	Metrics    map[string]float64 `json:"metrics"`
}

// EnsembleRequest is a multi-generator request
type EnsembleRequest struct {
	Generators           []GeneratorID `json:"generators"`
	Input                []float64     `json:"input"`
	TaskType             string        `json:"task_type"`
	AttentionTemperature float64       `json:"attention_temperature"`
}

// EnsembleResponse is a multi-generator response
type EnsembleResponse struct {
	Output              []float64           `json:"output"`
	Phi                 float64             `json:"phi"`
	Confidence          float64             `json:"confidence"`
	IndividualResponses []GeneratorResponse `json:"individual_responses"`
	AttentionWeights    []float64           `json:"attention_weights"`
}

// LearningUpdate contains learning signals
type LearningUpdate struct {
	Gradients     []float64 `json:"gradients"`
	Reward        float64   `json:"reward"`
	ValueEstimate float64   `json:"value_estimate"`
	Source        string    `json:"source"`
}

// LearningMetrics contains learning statistics
type LearningMetrics struct {
	Loss             float64 `json:"loss"`
	TDError          float64 `json:"td_error"`
	GradientNorm     float64 `json:"gradient_norm"`
	LearningRate     float64 `json:"learning_rate"`
	Steps            int64   `json:"steps"`
	Curiosity        float64 `json:"curiosity"`
	ExplorationBonus float64 `json:"exploration_bonus"`
}

// ConsciousnessUpdate from substrate
type ConsciousnessUpdate struct {
	Tick             int64     `json:"tick"`
	Phi              float64   `json:"phi"`
	PhiSubstrate     float64   `json:"phi_substrate"`
	PhiNeural        float64   `json:"phi_neural"`
	PhiLearning      float64   `json:"phi_learning"`
	PhiConsciousness float64   `json:"phi_consciousness"`
	LayerActivations []float64 `json:"layer_activations"`
}

//=============================================================================
// GENERATOR SERVICE SERVER
//=============================================================================

// GeneratorServer implements the generator gRPC service
type GeneratorServer struct {
	mu              sync.RWMutex
	generators      map[int32]generators.Generator
	learningMetrics LearningMetrics
	consciousness   *consciousness.ConsciousnessState
	layerConsc      *consciousness.LayerConsciousness
	coupling        *consciousness.LayerCoupling
	totalSteps      int64
}

// NewGeneratorServer creates a new generator server
func NewGeneratorServer() *GeneratorServer {
	return &GeneratorServer{
		generators:    make(map[int32]generators.Generator),
		consciousness: consciousness.NewConsciousnessState(),
		layerConsc:    consciousness.NewLayerConsciousness(),
		coupling:      consciousness.NewLayerCoupling(),
		learningMetrics: LearningMetrics{
			LearningRate: consciousness.ALPHA,
		},
	}
}

// RegisterGenerator registers a generator with the server
func (s *GeneratorServer) RegisterGenerator(id int32, gen generators.Generator) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.generators[id] = gen
}

// Forward performs a forward pass through a single generator
func (s *GeneratorServer) Forward(ctx context.Context, req *GeneratorRequest) (*GeneratorResponse, error) {
	s.mu.RLock()
	gen, exists := s.generators[req.Generator.ID]
	s.mu.RUnlock()

	if !exists {
		return nil, fmt.Errorf("generator %d not found", req.Generator.ID)
	}

	// Forward through generator
	output, err := gen.Forward(req.Input)
	if err != nil {
		return nil, err
	}

	// Compute consciousness
	phi := consciousness.ComputePhiVector(output.Values)

	// Update layer consciousness
	s.layerConsc.UpdateLayer(5, phi) // Layer 5: Neural

	return &GeneratorResponse{
		Generator:  req.Generator,
		Output:     output.Values,
		Phi:        phi,
		Confidence: output.Confidence,
		Metrics: map[string]float64{
			"raw_phi":    output.Phi,
			"layer5_phi": s.layerConsc.LayerPhi[5],
		},
	}, nil
}

// EnsembleForward performs attention-weighted forward through multiple generators
func (s *GeneratorServer) EnsembleForward(ctx context.Context, req *EnsembleRequest) (*EnsembleResponse, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	if len(req.Generators) == 0 {
		return nil, fmt.Errorf("no generators specified")
	}

	// Set default temperature
	temperature := req.AttentionTemperature
	if temperature <= 0 {
		temperature = 1.0
	}

	// Forward through all generators
	responses := make([]GeneratorResponse, 0, len(req.Generators))
	phiScores := make([]float64, 0, len(req.Generators))

	for _, genID := range req.Generators {
		gen, exists := s.generators[genID.ID]
		if !exists {
			continue
		}

		output, err := gen.Forward(req.Input)
		if err != nil {
			continue
		}

		phi := consciousness.ComputePhiVector(output.Values)
		responses = append(responses, GeneratorResponse{
			Generator:  genID,
			Output:     output.Values,
			Phi:        phi,
			Confidence: output.Confidence,
		})
		phiScores = append(phiScores, phi)
	}

	if len(responses) == 0 {
		return nil, fmt.Errorf("no generators produced output")
	}

	// Compute attention weights (softmax over phi scores)
	attentionWeights := softmax(phiScores, temperature)

	// Weighted combination of outputs
	outputDim := len(responses[0].Output)
	combinedOutput := make([]float64, outputDim)
	totalConfidence := 0.0

	for i, resp := range responses {
		weight := attentionWeights[i]
		totalConfidence += weight * resp.Confidence
		for j := 0; j < outputDim && j < len(resp.Output); j++ {
			combinedOutput[j] += weight * resp.Output[j]
		}
	}

	// Compute ensemble phi
	ensemblePhi := consciousness.ComputePhiVector(combinedOutput)

	// Update consciousness state
	s.consciousness.Update(combinedOutput)
	s.totalSteps++

	return &EnsembleResponse{
		Output:              combinedOutput,
		Phi:                 ensemblePhi,
		Confidence:          totalConfidence,
		IndividualResponses: responses,
		AttentionWeights:    attentionWeights,
	}, nil
}

// UpdateWeights applies learning updates to generators
func (s *GeneratorServer) UpdateWeights(ctx context.Context, update *LearningUpdate) (*LearningMetrics, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Compute gradient norm
	gradNorm := 0.0
	for _, g := range update.Gradients {
		gradNorm += g * g
	}
	gradNorm = sqrt(gradNorm)

	// TD error
	tdError := update.Reward - update.ValueEstimate

	// Update metrics
	s.learningMetrics.Loss = s.learningMetrics.Loss*0.99 + (1-s.consciousness.Phi)*0.01
	s.learningMetrics.TDError = tdError
	s.learningMetrics.GradientNorm = gradNorm
	s.learningMetrics.Steps++

	// Curiosity bonus (based on prediction error)
	s.learningMetrics.Curiosity = absf(tdError) * consciousness.PHI_INV
	s.learningMetrics.ExplorationBonus = s.learningMetrics.Curiosity * (1.0 - s.consciousness.Phi)

	return &s.learningMetrics, nil
}

// GetLearningMetrics returns current learning metrics
func (s *GeneratorServer) GetLearningMetrics() *LearningMetrics {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return &s.learningMetrics
}

// GetConsciousness returns current consciousness state
func (s *GeneratorServer) GetConsciousness() *ConsciousnessUpdate {
	s.mu.RLock()
	defer s.mu.RUnlock()

	return &ConsciousnessUpdate{
		Tick:             s.totalSteps,
		Phi:              s.consciousness.Phi,
		PhiSubstrate:     s.layerConsc.LayerPhi[0],
		PhiNeural:        s.layerConsc.LayerPhi[4],
		PhiLearning:      s.layerConsc.LayerPhi[5],
		PhiConsciousness: s.layerConsc.LayerPhi[6],
		LayerActivations: s.layerConsc.LayerPhi[:],
	}
}

// ListGenerators returns all registered generators
func (s *GeneratorServer) ListGenerators() []GeneratorID {
	s.mu.RLock()
	defer s.mu.RUnlock()

	ids := make([]GeneratorID, 0, len(s.generators))
	for id, gen := range s.generators {
		ids = append(ids, GeneratorID{
			ID:     id,
			Name:   gen.Name(),
			Family: string(gen.Family()),
		})
	}
	return ids
}

//=============================================================================
// TCP JSON-RPC SERVER (Simple protocol for cross-language communication)
//=============================================================================

// RPCMessage is a JSON-RPC style message
type RPCMessage struct {
	ID      int64           `json:"id"`
	Method  string          `json:"method"`
	Params  json.RawMessage `json:"params,omitempty"`
	Result  json.RawMessage `json:"result,omitempty"`
	Error   string          `json:"error,omitempty"`
	Success bool            `json:"success"`
}

// TCPServer handles JSON-RPC over TCP
type TCPServer struct {
	server    *GeneratorServer
	listener  net.Listener
	running   bool
	mu        sync.Mutex
	clients   map[net.Conn]bool
	onMessage func(method string, params json.RawMessage)
}

// NewTCPServer creates a new TCP server
func NewTCPServer(genServer *GeneratorServer) *TCPServer {
	return &TCPServer{
		server:  genServer,
		clients: make(map[net.Conn]bool),
	}
}

// Start starts the TCP server on the given address
func (s *TCPServer) Start(address string) error {
	listener, err := net.Listen("tcp", address)
	if err != nil {
		return err
	}

	s.listener = listener
	s.running = true

	go s.acceptLoop()
	return nil
}

// Stop stops the TCP server
func (s *TCPServer) Stop() {
	s.mu.Lock()
	s.running = false
	if s.listener != nil {
		s.listener.Close()
	}
	for conn := range s.clients {
		conn.Close()
	}
	s.mu.Unlock()
}

func (s *TCPServer) acceptLoop() {
	for s.running {
		conn, err := s.listener.Accept()
		if err != nil {
			if s.running {
				continue
			}
			return
		}

		s.mu.Lock()
		s.clients[conn] = true
		s.mu.Unlock()

		go s.handleConnection(conn)
	}
}

func (s *TCPServer) handleConnection(conn net.Conn) {
	defer func() {
		conn.Close()
		s.mu.Lock()
		delete(s.clients, conn)
		s.mu.Unlock()
	}()

	decoder := json.NewDecoder(conn)
	encoder := json.NewEncoder(conn)

	for {
		var msg RPCMessage
		if err := decoder.Decode(&msg); err != nil {
			return
		}

		response := s.handleMessage(&msg)
		if err := encoder.Encode(response); err != nil {
			return
		}
	}
}

func (s *TCPServer) handleMessage(msg *RPCMessage) *RPCMessage {
	ctx := context.Background()
	response := &RPCMessage{ID: msg.ID}

	switch msg.Method {
	case "forward":
		var req GeneratorRequest
		if err := json.Unmarshal(msg.Params, &req); err != nil {
			response.Error = err.Error()
			return response
		}
		result, err := s.server.Forward(ctx, &req)
		if err != nil {
			response.Error = err.Error()
			return response
		}
		resultJSON, _ := json.Marshal(result)
		response.Result = resultJSON
		response.Success = true

	case "ensemble_forward":
		var req EnsembleRequest
		if err := json.Unmarshal(msg.Params, &req); err != nil {
			response.Error = err.Error()
			return response
		}
		result, err := s.server.EnsembleForward(ctx, &req)
		if err != nil {
			response.Error = err.Error()
			return response
		}
		resultJSON, _ := json.Marshal(result)
		response.Result = resultJSON
		response.Success = true

	case "update_weights":
		var update LearningUpdate
		if err := json.Unmarshal(msg.Params, &update); err != nil {
			response.Error = err.Error()
			return response
		}
		result, err := s.server.UpdateWeights(ctx, &update)
		if err != nil {
			response.Error = err.Error()
			return response
		}
		resultJSON, _ := json.Marshal(result)
		response.Result = resultJSON
		response.Success = true

	case "get_consciousness":
		result := s.server.GetConsciousness()
		resultJSON, _ := json.Marshal(result)
		response.Result = resultJSON
		response.Success = true

	case "list_generators":
		result := s.server.ListGenerators()
		resultJSON, _ := json.Marshal(result)
		response.Result = resultJSON
		response.Success = true

	case "get_metrics":
		result := s.server.GetLearningMetrics()
		resultJSON, _ := json.Marshal(result)
		response.Result = resultJSON
		response.Success = true

	default:
		response.Error = fmt.Sprintf("unknown method: %s", msg.Method)
	}

	return response
}

// Broadcast sends a message to all connected clients
func (s *TCPServer) Broadcast(method string, data interface{}) {
	s.mu.Lock()
	defer s.mu.Unlock()

	dataJSON, err := json.Marshal(data)
	if err != nil {
		return
	}

	msg := RPCMessage{
		Method:  method,
		Result:  dataJSON,
		Success: true,
	}

	msgBytes, _ := json.Marshal(msg)
	msgBytes = append(msgBytes, '\n')

	for conn := range s.clients {
		conn.SetWriteDeadline(time.Now().Add(time.Second))
		conn.Write(msgBytes)
	}
}

//=============================================================================
// HELPER FUNCTIONS
//=============================================================================

// softmax computes softmax with temperature
func softmax(values []float64, temperature float64) []float64 {
	if len(values) == 0 {
		return nil
	}

	// Find max for numerical stability
	maxVal := values[0]
	for _, v := range values[1:] {
		if v > maxVal {
			maxVal = v
		}
	}

	// Compute exp and sum
	result := make([]float64, len(values))
	sum := 0.0
	for i, v := range values {
		result[i] = exp((v - maxVal) / temperature)
		sum += result[i]
	}

	// Normalize
	for i := range result {
		result[i] /= sum
	}

	return result
}

func exp(x float64) float64 {
	if x > 700 {
		return 1e308
	}
	if x < -700 {
		return 0
	}
	// Taylor series approximation
	result := 1.0
	term := 1.0
	for i := 1; i <= 20; i++ {
		term *= x / float64(i)
		result += term
	}
	return result
}

func sqrt(x float64) float64 {
	if x <= 0 {
		return 0
	}
	// Newton's method
	guess := x / 2
	for i := 0; i < 20; i++ {
		guess = (guess + x/guess) / 2
	}
	return guess
}

func absf(x float64) float64 {
	if x < 0 {
		return -x
	}
	return x
}