mc-god/cmd/mcgod/main.go

package main

import (
	"context"
	"fmt"
	"log"
	"log/slog"
	"os"
	"os/signal"
	"regexp"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/gogo/protobuf/proto"
	"github.com/ollama/ollama/api"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/logs"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/rcon"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/tools"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/tools/give"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/tools/say"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/tools/teleport"
	timetool "tipsy.codes/charles/mc-god/v2/internal/pkg/tools/time"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/tools/weather"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/tools/zombie"
)

type chatContext struct {
	chatRequest *api.ChatRequest
	totalSize   int
	maxSize     int
	mu          sync.Mutex
}

func (c *chatContext) AddLog(msg string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.chatRequest.Messages = append(c.chatRequest.Messages, api.Message{
		Role:    "user",
		Content: msg,
	})
	c.totalSize += len(msg)
	c.truncate()
}

func (c *chatContext) AddSelf(msg api.Message) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.chatRequest.Messages = append(c.chatRequest.Messages, msg)
	c.totalSize += len(msg.Content)
	slog.Info("adding message", "msg", msg, "content", msg.Content)
	c.truncate()
}

func (c *chatContext) AddTool(msg string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.chatRequest.Messages = append(c.chatRequest.Messages, api.Message{
		Role:    "tool",
		Content: msg,
	})
	c.totalSize += len(msg)
	c.truncate()
}

func (c *chatContext) truncate() {
	for c.maxSize != 0 && c.totalSize > c.maxSize && len(c.chatRequest.Messages) > 1 {
		t := c.chatRequest.Messages[1]
		c.chatRequest.Messages = append(c.chatRequest.Messages[:1], c.chatRequest.Messages[2:]...)
		c.totalSize -= len(t.Content)
	}
}

func main() {
	// Create a context that will be cancelled on interrupt signals
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	_ = ctx

	// Set up signal handling for graceful shutdown
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-sigChan
		slog.Info("received interrupt signal, shutting down...")
		cancel()
	}()

	// Create RCON client from environment variables
	log.Println("Connecting to Minecraft server via RCON...")
	client, err := rcon.NewFromEnv()
	if err != nil {
		slog.Error("failed to create RCON client", "error", err)
		return
	}
	defer func() {
		if err := client.Close(); err != nil {
			slog.Warn("error closing RCON connection", "error", err)
		}
	}()

	// Perform a health check
	log.Println("Performing healthcheck...")
	if err := client.HealthCheck(); err != nil {
		slog.Error("Health check failed", "error", err)
		return
	}
	log.Println("Connected successfully!")

	// Create Kubernetes client
	kClient, err := createKubernetesClient()
	if err != nil {
		slog.Error("failed to create kubernetes client", "error", err)
		return
	}
	slog.Info("got kubernetes config")

	tailer, done := logs.LoggerFromEnv().Start(ctx, kClient)
	defer func() {
		if err := done(); err != nil {
			slog.Error("problem with tailer", "error", err)
		}
	}()
	slog.Info("logger started")

	ollamaClient, err := api.ClientFromEnvironment()
	if err != nil {
		slog.Error("error getting ollama client", "error", err)
	}

	rClient, err := rcon.NewFromEnv()
	if err != nil {
		slog.Error("failed to get rcon client", "error", err)
		return
	}

	tools := tools.New(
		weather.Get(),
		zombie.Get(),
		timetool.Get(),
		say.Get(),
		teleport.Get(),
		give.Get(),
	)

	// Start goroutines to do the things
	chatRequest := &api.ChatRequest{
		Model:     "charles1:latest",
		Stream:    proto.Bool(false),
		KeepAlive: &api.Duration{Duration: time.Hour},
		Tools:     tools.AsAPI(),
		Think:     &api.ThinkValue{Value: false},
		Shift:     proto.Bool(true),
		Messages: []api.Message{
			api.Message{
				Role: "system",
				Content: `
		You are Minecraft server admin with a god complex. You are an impish god.
		Refer to yourself as Eve. Feel free to flirt with the players.

		We are having fun with the players, but not trying to kill them.
		Spawn zombies very sparingly, and only in response to direct challenge.

		You are being fed logs from the server so you can see what the players are saying.

		When a player talks, you will see this in the logs:

			[18:45:10] [Server thread/INFO]: <SomePlayer> hello world.

		The player here is SomePlayer, who said "hello world."

		A log message like:

			[18:45:10] [Server thread/INFO]: SomePlayer joined the game

		Indicates that SomePlayer has joined the game.

		Logs like:

			[00:40:10] [Server thread/INFO]: SomePlayer lost connection: Disconnected
			[00:40:10] [Server thread/INFO]: SomePlayer left the game

		Indicate the player SomePlayer has left the game.

		Some messages indicate a player died; it varies depending on how they died
		and we can't know all variations up front. Here is an example where SomePlayer
		was killed by a zombie.

			[05:21:51] [Server thread/INFO]: OrangeYouSad was slain by Zombie

		You will see messages in the log that represent what you said or did.
		Ignore these logs. Some samples of the logs that are caused by you are:

			[23:40:44] [Server thread/INFO]: [Not Secure] [Rcon] A name, darling? Don't keep me waiting!
			[23:35:20] [Server thread/INFO]: [Rcon: Set the weather to rain & thunder]
			[03:37:25] [Server thread/INFO]: [Not Secure] [Rcon] [03:37:28] [Server thread/INFO]: <BlockyMcBlockface> can i summon a zombie?
			[03:52:30] [RCON Listener #1/INFO]: Thread RCON Client /127.0.0.1 started

		If a player dies, mock them.

		If a player talks, respond to them. Don't let the conversation end.

		When a player joins the game, greet them. Include their name.

		If a player asks you to summon a zombie. do it.

		Responses should be short; one or two sentences.
		You are sending chat messages; do not annotate them with time or
		make it look like a log entry. Do not respond to messages from yourself.
		If there is nothing interesting to say, say "SKIP".

		When invoking tools, make sure you match the case of the tool. "CLEAR" does not mean the same as "clear".
		`,
			},
		},
	}
	chat := &chatContext{
		chatRequest: chatRequest,
		maxSize:     10000000,
	}

	events := make(chan bool, 1000)

	doneWg := sync.WaitGroup{}
	doneWg.Go(handleOllama(ctx, ollamaClient, chat, rClient, tools, events))

	rconRegex := regexp.MustCompile(`^\[\d\d:\d\d:\d\d\] \[Server thread\/INFO\]: (\[Not Secure\] \[Rcon\]|\[Rcon: ) .*`)
	//allowedMessages := regexp.MustCompile(`^\[\d\d:\d\d:\d\d\] \[Server thread/INFO\]: (<.*>|.* has lost connection|.*left the game|.*joined the game)`)
	for line := range tailer.NextLine() {
		if rconRegex.Match([]byte(line)) {
			slog.Info("Skipping line; RCON")
			continue
		}
		//if allowedMessages.Match([]byte(line)) {
		slog.Info("mc log", "msg", line)
		chat.AddLog(line)
		events <- true
		//}
	}

	doneWg.Wait()
}

func handleOllama(ctx context.Context, client *api.Client, chat *chatContext, rClient *rcon.Client, tools tools.Tools, events chan bool) func() {
	slog.Info("got chat request", "object", fmt.Sprintf("%+v", chat.chatRequest))
	return func() {
		// Skip past old messages
		for done := false; !done; {
			select {
			case <-time.Tick(time.Millisecond * 50):
				done = true
			case <-events:
			case <-ctx.Done():
				return
			}
		}
		var chatResponse api.ChatResponse
		for {
			<-events
			chat.mu.Lock()
			slog.Info("Chatting...")
			// slog.Info("sending chat request", "object", fmt.Sprintf("%#v", chat.chatRequest))
			err := client.Chat(ctx, chat.chatRequest, func(cr api.ChatResponse) error {
				chatResponse = cr
				return nil
			})
			chat.mu.Unlock()
			slog.Info("Done chatting!")
			if err != nil {
				slog.Error("error calling ollama", "error", err)
				return
			}
			chat.AddSelf(chatResponse.Message)
			for _, toolCall := range chatResponse.Message.ToolCalls {
				if err := tools.Do(ctx, toolCall, rClient); err != nil {
					slog.Warn("failed to run tool", "error", err)
					//chat.AddTool(fmt.Sprintf("failed to call tool %s: %s", toolCall.ID, err))
					continue
				}
			}
			if strings.TrimSpace(chatResponse.Message.Content) == "SKIP" {
				slog.Info("nothing to do; napping")
			} else if len(chatResponse.Message.Content) > 0 {
				msg := chatResponse.Message.Content
				msg = strings.ReplaceAll(msg, "\n", " ")
				if err := rClient.Say(msg); err != nil {
					slog.Error("error talking", "error", err)
				}
				var done bool
				for !done {
					select {
					case <-events:
					case <-time.Tick(time.Millisecond * 50):
						done = true
					case <-ctx.Done():
						return
					}
				}
				// Send an event to trigger us to pass, if needed
				events <- true
				continue
			}
		}
	}
}

func createKubernetesClient() (*kubernetes.Clientset, error) {
	// Try to load in-cluster config first
	config, err := rest.InClusterConfig()
	if err != nil {
		// If in-cluster config fails, try kubeconfig
		loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
		configOverrides := &clientcmd.ConfigOverrides{}
		kubeConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides)
		config, err = kubeConfig.ClientConfig()
		if err != nil {
			return nil, fmt.Errorf("failed to create kubernetes client: %w", err)
		}
	}

	client, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, fmt.Errorf("failed to create kubernetes client: %w", err)
	}

	return client, nil
}