From 6a2b4dfa46313b93aa8a7bb1502d9a6c063370b9 Mon Sep 17 00:00:00 2001 From: toki Date: Sun, 3 May 2026 13:09:52 +0900 Subject: [PATCH] =?UTF-8?q?=EC=A0=95=EB=A6=AC:=20edge=20=EC=95=B1=EC=9D=98?= =?UTF-8?q?=20main.go=EC=97=90=EC=84=9C=20=EB=B6=88=ED=95=84=EC=9A=94?= =?UTF-8?q?=ED=95=9C=20=EC=BD=94=EB=93=9C=20=EC=A0=95=EB=A6=AC=20=EB=B0=8F?= =?UTF-8?q?=20=EC=BD=98=EC=86=94=20=EB=AA=A8=EB=93=88=20=EB=B6=84=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/edge/cmd/edge/console.go | 169 ++++++++++++++++++++++++++++++++++ apps/edge/cmd/edge/main.go | 160 -------------------------------- 2 files changed, 169 insertions(+), 160 deletions(-) create mode 100644 apps/edge/cmd/edge/console.go diff --git a/apps/edge/cmd/edge/console.go b/apps/edge/cmd/edge/console.go new file mode 100644 index 0000000..c6cfee6 --- /dev/null +++ b/apps/edge/cmd/edge/console.go @@ -0,0 +1,169 @@ +package main + +import ( + "bufio" + "context" + "fmt" + "io" + "strings" + "time" + + "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" + + edgenode "iop/apps/edge/internal/node" + "iop/apps/edge/internal/transport" + "iop/packages/config" + "iop/packages/observability" + iop "iop/proto/gen/iop" +) + +func runConsole(ctx context.Context, cfg *config.EdgeConfig, in io.Reader, out io.Writer) error { + logger, err := observability.NewLogger(cfg.Logging.Level, cfg.Logging.Pretty) + if err != nil { + return err + } + defer func() { _ = logger.Sync() }() + + registry := edgenode.NewRegistry() + nodeStore, err := edgenode.LoadFromConfig(cfg.Nodes) + if err != nil { + return fmt.Errorf("edge: seed node store: %w", err) + } + + server, err := transport.NewServer(cfg.Server.Listen, registry, nodeStore, logger) + if err != nil { + return err + } + events := make(chan *iop.RunEvent, 256) + server.SetRunEventHandler(func(event *iop.RunEvent) { + select { + case events <- event: + default: + logger.Warn("run event dropped", zap.String("run_id", event.GetRunId())) + } + }) + + if err := server.Start(ctx); err != nil { + return err + } + defer func() { _ = server.Stop() }() + + go func() { + if err := observability.ServeMetrics(cfg.Metrics.Port); err != nil { + logger.Warn("metrics server exited", zap.Error(err)) + } + }() + + fmt.Fprintf(out, "IOP Edge console listening on %s\n", cfg.Server.Listen) + fmt.Fprintf(out, "Console target adapter=%s model=%s\n", cfg.Console.Adapter, cfg.Console.Model) + fmt.Fprintln(out, "Start node.sh on another host, then type a message here.") + fmt.Fprintln(out, "Commands: /nodes, /exit") + + scanner := bufio.NewScanner(in) + for { + fmt.Fprint(out, "edge> ") + if !scanner.Scan() { + fmt.Fprintln(out) + return scanner.Err() + } + + message := strings.TrimSpace(scanner.Text()) + switch strings.ToLower(message) { + case "": + continue + case "/exit", "/quit", "exit", "quit": + fmt.Fprintln(out, "bye") + return nil + case "/nodes": + printNodes(out, registry) + default: + if err := sendConsoleRun(ctx, registry, events, out, cfg.Console.Adapter, cfg.Console.Model, message); err != nil { + fmt.Fprintf(out, "error: %v\n", err) + } + } + } +} + +func printNodes(out io.Writer, registry *edgenode.Registry) { + nodes := registry.All() + if len(nodes) == 0 { + fmt.Fprintln(out, "no nodes connected") + return + } + for _, node := range nodes { + fmt.Fprintf(out, "- %s (%s)\n", node.NodeID, node.Alias) + } +} + +func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events <-chan *iop.RunEvent, out io.Writer, adapter, model, message string) error { + entry, err := registry.Pick() + if err != nil { + return err + } + + input, err := structpb.NewStruct(map[string]any{"prompt": message}) + if err != nil { + return err + } + + runID := fmt.Sprintf("manual-%d", time.Now().UnixNano()) + req := &iop.RunRequest{ + RunId: runID, + Adapter: adapter, + Model: model, + Input: input, + TimeoutSec: 30, + Metadata: map[string]string{ + "source": "edge-console", + }, + } + + nodeAlias := entry.Alias + fmt.Fprintf(out, "[edge] sent run_id=%s node=%s adapter=%s model=%s\n", runID, nodeAlias, adapter, model) + if err := entry.Client.Send(req); err != nil { + return err + } + + timer := time.NewTimer(35 * time.Second) + defer timer.Stop() + var response strings.Builder + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return fmt.Errorf("timed out waiting for node response") + case event := <-events: + if event.GetRunId() != runID { + continue + } + switch event.GetType() { + case "start": + fmt.Fprintf(out, "[node-%s-event] start run_id=%s\n", nodeAlias, runID) + case "delta": + response.WriteString(event.GetDelta()) + case "complete": + fmt.Fprintf(out, "[node-%s-event] complete run_id=%s detail=%q\n", nodeAlias, runID, event.GetMessage()) + printNodeMessage(out, nodeAlias, response.String()) + return nil + case "error": + fmt.Fprintf(out, "[node-%s-event] error run_id=%s detail=%q\n", nodeAlias, runID, event.GetError()) + printNodeMessage(out, nodeAlias, response.String()) + return fmt.Errorf("node reported error") + default: + fmt.Fprintf(out, "[node-%s-event] %s run_id=%s detail=%q\n", nodeAlias, event.GetType(), runID, event.GetMessage()) + } + } + } +} + +func printNodeMessage(out io.Writer, alias, message string) { + message = strings.TrimSpace(message) + if message == "" { + fmt.Fprintf(out, "[node-%s-message] \n", alias) + return + } + fmt.Fprintf(out, "[node-%s-message] %s\n", alias, message) +} diff --git a/apps/edge/cmd/edge/main.go b/apps/edge/cmd/edge/main.go index 29b1f40..0f0b7b6 100644 --- a/apps/edge/cmd/edge/main.go +++ b/apps/edge/cmd/edge/main.go @@ -1,25 +1,15 @@ package main import ( - "bufio" "context" "fmt" - "io" "os" - "strings" - "time" "github.com/spf13/cobra" "go.uber.org/fx" - "go.uber.org/zap" - "google.golang.org/protobuf/types/known/structpb" "iop/apps/edge/internal/bootstrap" - edgenode "iop/apps/edge/internal/node" - "iop/apps/edge/internal/transport" "iop/packages/config" - "iop/packages/observability" - iop "iop/proto/gen/iop" ) var cfgFile string @@ -69,153 +59,3 @@ func consoleCmd() *cobra.Command { }, } } - -func runConsole(ctx context.Context, cfg *config.EdgeConfig, in io.Reader, out io.Writer) error { - logger, err := observability.NewLogger(cfg.Logging.Level, cfg.Logging.Pretty) - if err != nil { - return err - } - defer func() { _ = logger.Sync() }() - - registry := edgenode.NewRegistry() - nodeStore, err := edgenode.LoadFromConfig(cfg.Nodes) - if err != nil { - return fmt.Errorf("edge: seed node store: %w", err) - } - - server, err := transport.NewServer(cfg.Server.Listen, registry, nodeStore, logger) - if err != nil { - return err - } - events := make(chan *iop.RunEvent, 256) - server.SetRunEventHandler(func(event *iop.RunEvent) { - select { - case events <- event: - default: - logger.Warn("run event dropped", zap.String("run_id", event.GetRunId())) - } - }) - - if err := server.Start(ctx); err != nil { - return err - } - defer func() { _ = server.Stop() }() - - go func() { - if err := observability.ServeMetrics(cfg.Metrics.Port); err != nil { - logger.Warn("metrics server exited", zap.Error(err)) - } - }() - - fmt.Fprintf(out, "IOP Edge console listening on %s\n", cfg.Server.Listen) - fmt.Fprintf(out, "Console target adapter=%s model=%s\n", cfg.Console.Adapter, cfg.Console.Model) - fmt.Fprintln(out, "Start node.sh on another host, then type a message here.") - fmt.Fprintln(out, "Commands: /nodes, /exit") - - scanner := bufio.NewScanner(in) - for { - fmt.Fprint(out, "edge> ") - if !scanner.Scan() { - fmt.Fprintln(out) - return scanner.Err() - } - - message := strings.TrimSpace(scanner.Text()) - switch strings.ToLower(message) { - case "": - continue - case "/exit", "/quit", "exit", "quit": - fmt.Fprintln(out, "bye") - return nil - case "/nodes": - printNodes(out, registry) - default: - if err := sendConsoleRun(ctx, registry, events, out, cfg.Console.Adapter, cfg.Console.Model, message); err != nil { - fmt.Fprintf(out, "error: %v\n", err) - } - } - } -} - -func printNodes(out io.Writer, registry *edgenode.Registry) { - nodes := registry.All() - if len(nodes) == 0 { - fmt.Fprintln(out, "no nodes connected") - return - } - for _, node := range nodes { - fmt.Fprintf(out, "- %s (%s)\n", node.NodeID, node.Alias) - } -} - -func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events <-chan *iop.RunEvent, out io.Writer, adapter, model, message string) error { - entry, err := registry.Pick() - if err != nil { - return err - } - - input, err := structpb.NewStruct(map[string]any{"prompt": message}) - if err != nil { - return err - } - - runID := fmt.Sprintf("manual-%d", time.Now().UnixNano()) - req := &iop.RunRequest{ - RunId: runID, - Adapter: adapter, - Model: model, - Input: input, - TimeoutSec: 30, - Metadata: map[string]string{ - "source": "edge-console", - }, - } - - nodeAlias := entry.Alias - fmt.Fprintf(out, "[edge] sent run_id=%s node=%s adapter=%s model=%s\n", runID, nodeAlias, adapter, model) - if err := entry.Client.Send(req); err != nil { - return err - } - - timer := time.NewTimer(35 * time.Second) - defer timer.Stop() - var response strings.Builder - - for { - select { - case <-ctx.Done(): - return ctx.Err() - case <-timer.C: - return fmt.Errorf("timed out waiting for node response") - case event := <-events: - if event.GetRunId() != runID { - continue - } - switch event.GetType() { - case "start": - fmt.Fprintf(out, "[node-%s-event] start run_id=%s\n", nodeAlias, runID) - case "delta": - response.WriteString(event.GetDelta()) - case "complete": - fmt.Fprintf(out, "[node-%s-event] complete run_id=%s detail=%q\n", nodeAlias, runID, event.GetMessage()) - printNodeMessage(out, nodeAlias, response.String()) - return nil - case "error": - fmt.Fprintf(out, "[node-%s-event] error run_id=%s detail=%q\n", nodeAlias, runID, event.GetError()) - printNodeMessage(out, nodeAlias, response.String()) - return fmt.Errorf("node reported error") - default: - fmt.Fprintf(out, "[node-%s-event] %s run_id=%s detail=%q\n", nodeAlias, event.GetType(), runID, event.GetMessage()) - } - } - } -} - -func printNodeMessage(out io.Writer, alias, message string) { - message = strings.TrimSpace(message) - if message == "" { - fmt.Fprintf(out, "[node-%s-message] \n", alias) - return - } - fmt.Fprintf(out, "[node-%s-message] %s\n", alias, message) -}