diff --git a/README.md b/README.md index 5a4f457..ed1a47c 100644 --- a/README.md +++ b/README.md @@ -219,12 +219,21 @@ NomadCode ## Roadmap +Edge의 외부 입력 표면은 두 방식을 병행 지원한다. + +- **OpenAI-compatible HTTP API**: 외부 모델 클라이언트, Cline류 도구, 단순 chat completion/inference 호환을 위한 입력이다. 이 경계에서는 호환성을 위해 `model`을 받되, Edge 내부에서는 `adapter + target`으로 변환한다. +- **A2A JSON-RPC HTTP API**: NomadCode Core나 외부 agent가 작업을 위임하고 `Task` 상태, artifact, cancel/polling을 공유해야 할 때 사용하는 agent delegation 입력이다. 이 경계도 Edge 내부에서는 `adapter + target`으로 변환한다. + +두 입력 방식은 Edge의 inbound/input surface로 함께 관리하고, 내부 실행은 기존 `RunRequest`/`RunEvent` 흐름으로 수렴시킨다. IOP native protocol은 이 둘을 대체하는 외부 호환 API가 아니라, Control Plane/Portal/운영 CLI가 Edge/Node 운영 제어와 lifecycle event를 다루기 위한 wire protocol로 유지한다. + ### Phase 1. Edge-Node execution skeleton - Edge-Node 소켓 연결 - Node adapter execution - CLI adapter 검증 - `adapter + target` 개념 정리 +- OpenAI-compatible serving 표면은 외부 표준 호환 경로로 유지하고, 내부에서는 `model`을 `adapter + target`으로 변환하는 경계를 검증 +- A2A input surface는 기존 OpenAI-compatible input과 같은 Edge input 관리 계층에 추가하고, 1차로 `message/send`, `tasks/get`, `tasks/cancel`을 지원 ### Phase 2. Runtime and Automation unification @@ -232,6 +241,9 @@ NomadCode - CLI Agent 실행 안정화 - Runtime / Automation 실행 흐름 공통화 - Node local execution history와 Edge event aggregation 경계 정리 +- IOP native execution protocol을 기존 protobuf `RunRequest`, `RunEvent`, `NodeCommandRequest`, `EdgeNodeEvent` 계열 계약 위에서 정식화 +- native protocol에는 node 선택, `adapter + target`, logical session, background run, cancel/terminate-session, capabilities/status/session/transport command, node lifecycle event처럼 OpenAI-compatible API로 표현하기 어려운 기능을 둠 +- Edge input surface는 OpenAI-compatible 입력을 모델/채팅 호환 경로로, A2A 입력을 agent 작업 위임과 상태 공유 경로로 분리해 유지 ### Phase 3. Control Plane @@ -241,6 +253,9 @@ NomadCode - Edge 설정 변경 - Edge 명령 전달 - 이벤트 수신 +- Portal-Control Plane, Control Plane-Edge 통신은 IOP Wire Protocol(protobuf-socket)을 기준으로 확장하고, 브라우저 직접 TCP가 어려운 경우 server-side bridge를 둠 +- OpenAI-compatible HTTP API는 외부 모델 클라이언트 호환 표면으로 유지하되, 운영/제어/세션 기능의 기본 프로토콜로 확장하지 않음 +- A2A HTTP API는 Core/외부 agent가 Edge 실행 그룹에 작업을 위임하는 표준 입력으로 유지하되, Control Plane의 Edge 운영 제어 프로토콜로 확장하지 않음 ### Phase 4. Multi-Edge operations @@ -248,6 +263,8 @@ NomadCode - Edge별 runtime/automation 상태 표시 - Edge 단위 실행 이력과 상태 집계 - Edge 단위 작업 실행과 제어 +- multi-edge 운영 명령과 이벤트 집계는 IOP native protocol을 기준으로 설계하고, OpenAI-compatible 표면은 특정 Edge/adapter로 라우팅되는 inference 호환 경로로 제한 +- A2A 표면은 특정 Edge/adapter로 위임되는 agent task 경로로 제한하고, multi-edge scheduling과 fleet-wide state ownership은 Control Plane/native protocol에서 결정 ### Phase 5. Policy, History, Audit @@ -260,6 +277,8 @@ NomadCode - 내부 통신은 TCP/protobuf 기반 소켓 흐름을 우선한다. - gRPC 도입, WebSocket 기본 transport 전환, actor/FSM/plugin framework 도입은 현재 단계의 기본 방향이 아니다. - OpenAI API 호환 계층은 외부 호환을 위한 표면이며, 내부 실행 모델 전체를 대표하지 않는다. +- A2A API 계층은 agent 간 작업 위임과 상태 공유를 위한 표면이며, 단순 모델 호출 호환은 OpenAI-compatible API를 사용한다. +- IOP native protocol은 OpenAI-compatible API나 A2A API를 대체하는 것이 아니라, Edge/Node 운영 제어와 CLI/session/command/event 같은 IOP 고유 기능을 제공하는 병행 표면이다. - 앱별 README에는 현재 수동 테스트나 구현 세부가 더 많이 남아 있을 수 있다. 루트 README는 전체 방향과 경계를 설명하는 문서로 유지한다. ## Non-Goals for Current Stage diff --git a/agent-ops/rules/project/rules.md b/agent-ops/rules/project/rules.md index 529fcc2..1c93e3c 100644 --- a/agent-ops/rules/project/rules.md +++ b/agent-ops/rules/project/rules.md @@ -36,7 +36,7 @@ - 메트릭/헬스: Prometheus HTTP handler - 저장소: `modernc.org/sqlite` - 메시지 계약: `google.golang.org/protobuf`, `proto/iop/*.proto` -- 내부 소켓: `git.toki-labs.com/toki/common-proto-socket/go` +- 내부 소켓: `git.toki-labs.com/toki/proto-socket/go` ## 프로젝트 특화 컨벤션 diff --git a/apps/edge/README.md b/apps/edge/README.md index d94631f..86a18d9 100644 --- a/apps/edge/README.md +++ b/apps/edge/README.md @@ -99,6 +99,24 @@ edge> /terminate-session terminated session default node=test-node ``` +## Edge CLI + +운영 호스트에서 `iop-edge` 바이너리는 다음 명령을 제공한다. + +```bash +iop-edge serve --config /etc/iop/edge.yaml +iop-edge console --config /etc/iop/edge.yaml +iop-edge version +iop-edge config print --config /etc/iop/edge.yaml +iop-edge config check --config /etc/iop/edge.yaml + +# host setup (systemd unit, user/group, data dir, config 템플릿 준비) +sudo iop-edge setup --dry-run --binary /usr/local/bin/iop-edge +sudo iop-edge setup --enable --start --binary /usr/local/bin/iop-edge +``` + +`setup`의 `--config` 기본값은 `/etc/iop/edge.yaml`이다. root persistent `--config`의 dev 기본값(`configs/edge.yaml`)은 `serve`, `console`, `config print/check` 같은 dev 실행 경로에만 적용된다. 기존 설정 파일은 그대로 두며 `--overwrite-config`를 지정해야 덮어쓴다. + ## Console 명령 | 명령 | 설명 | @@ -137,10 +155,16 @@ edge-node transport 연결은 **node id당 1개** TCP 연결만 유지한다. `bin/node.sh`의 edge 대기 시간은 `IOP_NODE_WAIT_TIMEOUT`, 확인 간격은 `IOP_NODE_WAIT_INTERVAL`, 접속 주소는 `IOP_EDGE_ADDR` 환경 변수로 임시 override할 수 있다. -## OpenAI-Compatible Serving +## Edge Input Surfaces + +Edge 외부 입력은 OpenAI-compatible HTTP API와 A2A JSON-RPC HTTP API 두 방식으로 정리한다. 둘 다 Edge에서 받은 외부 요청을 내부 `adapter + target` 실행으로 변환하고, 실제 실행은 `apps/edge/internal/service`와 edge-node transport를 통해 처리한다. + +### OpenAI-Compatible Serving `configs/edge.yaml`의 `openai` 섹션을 켜면 edge가 별도 HTTP listener를 열고 외부 agent가 OpenAI API 형태로 접속할 수 있다. 1차 구현은 Ollama를 주 테스트 경로로 삼는다. +이 표면은 외부 모델 클라이언트 호환을 위한 표준 경로다. Edge/Node 운영 제어, CLI logical session, background run, cancel/terminate-session, capabilities/status/session/transport command, node lifecycle event 같은 IOP 고유 기능은 OpenAI-compatible 요청에 억지로 싣지 않고 IOP native protocol(protobuf-socket) 계열에서 다룬다. + ```yaml openai: enabled: true @@ -172,3 +196,36 @@ curl -s http://127.0.0.1:8080/v1/chat/completions \ ``` vLLM은 같은 `adapter + target` 경계를 쓰도록 설정과 `/v1/models` capability 조회 skeleton만 먼저 열어두었다. 실제 completion 실행은 다음 단계에서 붙인다. + +### A2A Agent Input + +A2A JSON-RPC HTTP API는 NomadCode Core나 외부 agent가 Edge 실행 그룹에 작업을 위임하고 `Task` 상태, artifact, cancel/polling을 공유해야 할 때 사용하는 입력 표면이다. OpenAI-compatible API가 단순 모델/chat completion 호환에 맞는 경로라면, A2A는 agent-to-agent 작업 위임과 상태 공유에 맞는 경로다. + +1차 지원 범위는 `message/send`, `tasks/get`, `tasks/cancel`이다. `message/stream`, push notification, `tasks/list`는 후속 단계에서 다룬다. 구현은 OpenAI-compatible serving과 같은 Edge input 관리 계층에서 lifecycle/config를 묶고, 내부 실행은 동일하게 `adapter + target`으로 변환한다. + +#### 설정 예시 + +```yaml +a2a: + enabled: true + listen: "0.0.0.0:8081" + path: "/a2a" + node: "node0" # 빈 값이면 단일 연결 node 자동 선택 + adapter: "cli" + target: "claude" + session_id: "a2a" + timeout_sec: 120 + bearer_token: "" # 빈 값이면 auth 비활성화 +``` + +#### 지원 메서드 + +| 메서드 | 설명 | +|--------|------| +| `message/send` | 실행 요청. `configuration.blocking=true`(기본)이면 완료까지 대기 후 최종 Task 반환. `false`이면 working Task를 즉시 반환하고 background 수집. | +| `tasks/get` | task ID로 현재 Task 상태/artifact/history 조회. | +| `tasks/cancel` | 실행 중인 Task에 cancel 요청. `CANCEL_ACTION_CANCEL_RUN`을 node에 전송. | + +#### Agent card + +`GET /.well-known/agent.json`으로 최소 agent metadata를 제공한다. diff --git a/apps/edge/cmd/edge/main.go b/apps/edge/cmd/edge/main.go index 49ec3e3..abe3235 100644 --- a/apps/edge/cmd/edge/main.go +++ b/apps/edge/cmd/edge/main.go @@ -7,9 +7,12 @@ import ( "github.com/spf13/cobra" "go.uber.org/fx" + "gopkg.in/yaml.v3" "iop/apps/edge/internal/bootstrap" "iop/packages/config" + "iop/packages/hostsetup" + "iop/packages/version" ) var cfgFile string @@ -27,7 +30,7 @@ func rootCmd() *cobra.Command { Short: "IOP Edge — execution group controller", } root.PersistentFlags().StringVarP(&cfgFile, "config", "c", "configs/edge.yaml", "config file path") - root.AddCommand(serveCmd(), consoleCmd()) + root.AddCommand(serveCmd(), consoleCmd(), versionCmd(), configCmd(), setupCmd()) return root } @@ -59,3 +62,80 @@ func consoleCmd() *cobra.Command { }, } } + +func versionCmd() *cobra.Command { + return &cobra.Command{ + Use: "version", + Short: "Print edge binary version", + RunE: func(cmd *cobra.Command, _ []string) error { + fmt.Fprintln(cmd.OutOrStdout(), version.Version) + return nil + }, + } +} + +func configCmd() *cobra.Command { + c := &cobra.Command{ + Use: "config", + Short: "Inspect edge config", + } + c.AddCommand(configPrintCmd(), configCheckCmd()) + return c +} + +func configPrintCmd() *cobra.Command { + return &cobra.Command{ + Use: "print", + Short: "Print effective edge config as YAML", + RunE: func(cmd *cobra.Command, _ []string) error { + cfg, err := config.LoadEdge(cfgFile) + if err != nil { + return fmt.Errorf("load config: %w", err) + } + enc := yaml.NewEncoder(cmd.OutOrStdout()) + defer enc.Close() + return enc.Encode(cfg) + }, + } +} + +func configCheckCmd() *cobra.Command { + return &cobra.Command{ + Use: "check", + Short: "Validate edge config can be loaded", + RunE: func(cmd *cobra.Command, _ []string) error { + if _, err := config.LoadEdge(cfgFile); err != nil { + return fmt.Errorf("load config: %w", err) + } + fmt.Fprintf(cmd.OutOrStdout(), "OK %s\n", cfgFile) + return nil + }, + } +} + +func setupCmd() *cobra.Command { + var opts hostsetup.SetupOptions + c := &cobra.Command{ + Use: "setup", + Short: "Prepare host environment and systemd unit for iop-edge", + RunE: func(cmd *cobra.Command, _ []string) error { + if cmd.Flags().Changed("config") { + opts.ConfigPath = cfgFile + } else { + opts.ConfigPath = "/etc/iop/edge.yaml" + } + return hostsetup.Run(cmd.Context(), hostsetup.EdgeSpec(), opts, cmd.OutOrStdout()) + }, + } + c.Flags().StringVar(&opts.BinaryPath, "binary", "", "path to iop-edge binary (defaults to current executable)") + c.Flags().StringVar(&opts.DataDir, "data-dir", "", "data directory (defaults to spec)") + c.Flags().StringVar(&opts.UnitPath, "unit", "", "systemd unit path (defaults to spec)") + c.Flags().StringVar(&opts.User, "user", "", "service user") + c.Flags().StringVar(&opts.Group, "group", "", "service group") + c.Flags().BoolVar(&opts.Enable, "enable", false, "enable the systemd unit") + c.Flags().BoolVar(&opts.Start, "start", false, "start the systemd unit") + c.Flags().BoolVar(&opts.Restart, "restart", false, "restart the systemd unit") + c.Flags().BoolVar(&opts.DryRun, "dry-run", false, "print plan and previews without making changes") + c.Flags().BoolVar(&opts.OverwriteConfig, "overwrite-config", false, "overwrite an existing config file") + return c +} diff --git a/apps/edge/cmd/edge/main_test.go b/apps/edge/cmd/edge/main_test.go new file mode 100644 index 0000000..9592827 --- /dev/null +++ b/apps/edge/cmd/edge/main_test.go @@ -0,0 +1,120 @@ +package main + +import ( + "bytes" + "path/filepath" + "strings" + "testing" + + "iop/packages/version" +) + +func TestRootCmdIncludesOperationalCommands(t *testing.T) { + root := rootCmd() + want := map[string]bool{ + "serve": false, + "console": false, + "version": false, + "config": false, + "setup": false, + } + for _, c := range root.Commands() { + if _, ok := want[c.Name()]; ok { + want[c.Name()] = true + } + } + for name, ok := range want { + if !ok { + t.Errorf("root command missing %q", name) + } + } + + cfg, _, err := root.Find([]string{"config"}) + if err != nil { + t.Fatalf("find config: %v", err) + } + subs := map[string]bool{"print": false, "check": false} + for _, c := range cfg.Commands() { + if _, ok := subs[c.Name()]; ok { + subs[c.Name()] = true + } + } + for name, ok := range subs { + if !ok { + t.Errorf("config subcommand missing %q", name) + } + } +} + +func TestVersionCmdPrintsVersion(t *testing.T) { + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + root.SetArgs([]string{"version"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v", err) + } + got := strings.TrimSpace(out.String()) + if got != version.Version { + t.Fatalf("version output = %q, want %q", got, version.Version) + } +} + +func TestConfigCheckCmdLoadsEdgeConfig(t *testing.T) { + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + root.SetArgs([]string{"config", "check", "--config", "../../../../configs/edge.yaml"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v\n%s", err, out.String()) + } + if !strings.Contains(out.String(), "OK") { + t.Fatalf("expected OK in output, got %q", out.String()) + } +} + +func TestSetupDryRunUsesEdgeDefaults(t *testing.T) { + cfgFile = "configs/edge.yaml" // reset to root default + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + root.SetArgs([]string{"setup", "--dry-run", "--binary", "/usr/local/bin/iop-edge"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v\n%s", err, out.String()) + } + s := out.String() + for _, want := range []string{ + "/etc/iop/edge.yaml", + "/var/lib/iop/edge", + "/etc/systemd/system/iop-edge.service", + "/usr/local/bin/iop-edge serve --config /etc/iop/edge.yaml", + "--- unit preview ---", + "--- config preview ---", + } { + if !strings.Contains(s, want) { + t.Errorf("dry-run output missing %q\n---\n%s", want, s) + } + } + if strings.Contains(s, "configs/edge.yaml") { + t.Errorf("setup default leaked root dev config path: %s", s) + } +} + +func TestSetupDryRunAcceptsExplicitConfig(t *testing.T) { + cfgFile = "configs/edge.yaml" // reset + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + custom := filepath.Join(t.TempDir(), "edge.yaml") + root.SetArgs([]string{"--config", custom, "setup", "--dry-run", "--binary", "/usr/local/bin/iop-edge"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v\n%s", err, out.String()) + } + if !strings.Contains(out.String(), custom) { + t.Errorf("explicit config not used; output:\n%s", out.String()) + } +} diff --git a/apps/edge/internal/bootstrap/runtime.go b/apps/edge/internal/bootstrap/runtime.go index 392cf4c..93b5a4f 100644 --- a/apps/edge/internal/bootstrap/runtime.go +++ b/apps/edge/internal/bootstrap/runtime.go @@ -8,8 +8,8 @@ import ( "go.uber.org/zap" edgeevents "iop/apps/edge/internal/events" + edgeinput "iop/apps/edge/internal/input" edgenode "iop/apps/edge/internal/node" - edgeopenai "iop/apps/edge/internal/openai" edgeservice "iop/apps/edge/internal/service" "iop/apps/edge/internal/transport" "iop/packages/config" @@ -28,7 +28,7 @@ type Runtime struct { EventBus *edgeevents.Bus Service *edgeservice.Service Server *transport.Server - OpenAI *edgeopenai.Server + Input *edgeinput.Manager } func NewRuntime(cfg *config.EdgeConfig) (*Runtime, error) { @@ -45,7 +45,7 @@ func NewRuntime(cfg *config.EdgeConfig) (*Runtime, error) { bus := edgeevents.NewBus() svc := edgeservice.New(registry, bus) - openaiServer := edgeopenai.NewServer(cfg.OpenAI, svc, logger.Named("openai")) + inputManager := edgeinput.NewManager(*cfg, svc, logger.Named("input")) server, err := transport.NewServer(cfg.Server.Listen, registry, nodeStore, logger) if err != nil { @@ -60,7 +60,7 @@ func NewRuntime(cfg *config.EdgeConfig) (*Runtime, error) { EventBus: bus, Service: svc, Server: server, - OpenAI: openaiServer, + Input: inputManager, }, nil } @@ -74,7 +74,7 @@ func (r *Runtime) Start(ctx context.Context) error { if err := r.Server.Start(ctx); err != nil { return err } - if err := r.OpenAI.Start(ctx); err != nil { + if err := r.Input.Start(ctx); err != nil { _ = r.Server.Stop() return err } @@ -89,11 +89,11 @@ func (r *Runtime) Start(ctx context.Context) error { func (r *Runtime) Stop() error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - openAIErr := r.OpenAI.Stop(ctx) + inputErr := r.Input.Stop(ctx) err := r.Server.Stop() _ = r.Logger.Sync() if err == nil { - err = openAIErr + err = inputErr } return err } diff --git a/apps/edge/internal/bootstrap/runtime_test.go b/apps/edge/internal/bootstrap/runtime_test.go index 6aae19e..85cff3d 100644 --- a/apps/edge/internal/bootstrap/runtime_test.go +++ b/apps/edge/internal/bootstrap/runtime_test.go @@ -56,6 +56,23 @@ func TestRuntimeWiresTransportHandlers(t *testing.T) { } } +func TestNewRuntimeWiresInputManager(t *testing.T) { + cfg := newTestConfig() + rt, err := NewRuntime(cfg) + if err != nil { + t.Fatalf("NewRuntime: %v", err) + } + if rt.Input == nil { + t.Fatal("Input manager is nil") + } + if rt.Input.OpenAI == nil { + t.Fatal("Input.OpenAI is nil") + } + if rt.Input.A2A == nil { + t.Fatal("Input.A2A is nil") + } +} + func TestNewRuntimeRejectsDuplicateToken(t *testing.T) { cfg := &config.EdgeConfig{ Server: config.EdgeServerConf{Listen: "127.0.0.1:0"}, diff --git a/apps/edge/internal/input/a2a/server.go b/apps/edge/internal/input/a2a/server.go new file mode 100644 index 0000000..a3b7156 --- /dev/null +++ b/apps/edge/internal/input/a2a/server.go @@ -0,0 +1,303 @@ +package a2a + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net" + "net/http" + "strings" + "time" + + "go.uber.org/zap" + + edgeservice "iop/apps/edge/internal/service" + "iop/packages/config" +) + +type runService interface { + SubmitRun(context.Context, edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) + CancelRun(context.Context, edgeservice.CancelRunRequest) (edgeservice.CommandResult, error) +} + +type Server struct { + cfg config.EdgeA2AConf + svc runService + tasks *TaskStore + logger *zap.Logger + server *http.Server +} + +func NewServer(cfg config.EdgeA2AConf, svc runService, logger *zap.Logger) *Server { + if logger == nil { + logger = zap.NewNop() + } + return &Server{cfg: cfg, svc: svc, tasks: newTaskStore(), logger: logger} +} + +func (s *Server) Enabled() bool { + return s != nil && s.cfg.Enabled +} + +func (s *Server) Start(ctx context.Context) error { + if !s.Enabled() { + return nil + } + if s.cfg.Listen == "" { + s.cfg.Listen = "0.0.0.0:8081" + } + path := s.cfg.Path + if path == "" { + path = "/a2a" + } + + mux := http.NewServeMux() + mux.HandleFunc("/.well-known/agent.json", s.handleAgentCard) + mux.HandleFunc(path, s.withAuth(s.handleRPC)) + + s.server = &http.Server{ + Addr: s.cfg.Listen, + Handler: mux, + ReadHeaderTimeout: 5 * time.Second, + } + ln, err := net.Listen("tcp", s.cfg.Listen) + if err != nil { + return fmt.Errorf("a2a server listen %s: %w", s.cfg.Listen, err) + } + go func() { + <-ctx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = s.server.Shutdown(shutdownCtx) + }() + go func() { + s.logger.Info("a2a server listening", zap.String("addr", s.cfg.Listen), zap.String("path", path)) + if err := s.server.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + s.logger.Warn("a2a server exited", zap.Error(err)) + } + }() + return nil +} + +func (s *Server) Stop(ctx context.Context) error { + if s == nil || s.server == nil { + return nil + } + return s.server.Shutdown(ctx) +} + +// RPCHandlerForTest returns the auth-wrapped JSON-RPC handler for unit testing +// without starting a real listener. +func (s *Server) RPCHandlerForTest() http.HandlerFunc { + return s.withAuth(s.handleRPC) +} + +func (s *Server) withAuth(next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if s.cfg.BearerToken != "" { + auth := r.Header.Get("Authorization") + expected := "Bearer " + s.cfg.BearerToken + if auth != expected { + writeRPCError(w, nil, http.StatusUnauthorized, ErrCodeInvalidRequest, "unauthorized") + return + } + } + next(w, r) + } +} + +func (s *Server) handleAgentCard(w http.ResponseWriter, _ *http.Request) { + card := map[string]any{ + "name": "IOP Edge A2A Agent", + "description": "IOP Edge A2A input surface", + "url": fmt.Sprintf("http://%s%s", s.cfg.Listen, s.cfg.Path), + "version": "1.0.0", + "capabilities": map[string]any{ + "streaming": false, + }, + } + writeJSON(w, http.StatusOK, card) +} + +func (s *Server) handleRPC(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeRPCError(w, nil, http.StatusMethodNotAllowed, ErrCodeInvalidRequest, "method not allowed") + return + } + defer r.Body.Close() + + var req JSONRPCRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeRPCError(w, nil, http.StatusOK, ErrCodeParseError, "parse error") + return + } + if req.JSONRPC != "2.0" { + writeRPCError(w, req.ID, http.StatusOK, ErrCodeInvalidRequest, "jsonrpc must be \"2.0\"") + return + } + + switch req.Method { + case "message/send": + s.handleMessageSend(w, r, req) + case "tasks/get": + s.handleTasksGet(w, req) + case "tasks/cancel": + s.handleTasksCancel(w, r, req) + default: + writeRPCError(w, req.ID, http.StatusOK, ErrCodeMethodNotFound, "method not found: "+req.Method) + } +} + +func (s *Server) handleMessageSend(w http.ResponseWriter, r *http.Request, rpcReq JSONRPCRequest) { + var params MessageSendParams + if err := json.Unmarshal(rpcReq.Params, ¶ms); err != nil { + writeRPCError(w, rpcReq.ID, http.StatusOK, ErrCodeInvalidParams, "invalid params") + return + } + + prompt := extractPrompt(params.Message) + if strings.TrimSpace(prompt) == "" { + writeRPCError(w, rpcReq.ID, http.StatusOK, ErrCodeInvalidParams, "message parts are empty") + return + } + + // blocking defaults to true when configuration is absent or Blocking field is omitted. + blocking := params.Configuration == nil || params.Configuration.Blocking == nil || *params.Configuration.Blocking + + handle, err := s.svc.SubmitRun(r.Context(), edgeservice.SubmitRunRequest{ + NodeRef: s.cfg.NodeRef, + Adapter: s.resolveAdapter(), + Target: s.resolveTarget(), + SessionID: s.resolveSessionID(), + Prompt: prompt, + TimeoutSec: s.resolveTimeoutSec(), + Metadata: map[string]string{ + "source": "a2a", + "blocking": fmt.Sprintf("%t", blocking), + }, + }) + if err != nil { + writeRPCError(w, rpcReq.ID, http.StatusOK, ErrCodeInvalidRequest, err.Error()) + return + } + + taskID := s.tasks.create(handle.RunID, params.Message) + + var snapshot *Task + if blocking { + snapshot = s.tasks.collectBlocking(taskID, handle) + } else { + s.tasks.collectBackground(taskID, handle) + // Get a snapshot after background drain is started; background goroutine + // owns the internal pointer from this point, so we never share it. + snapshot, _ = s.tasks.get(taskID) + } + + writeRPCResult(w, rpcReq.ID, snapshot) +} + +func (s *Server) handleTasksGet(w http.ResponseWriter, rpcReq JSONRPCRequest) { + var params TaskQueryParams + if err := json.Unmarshal(rpcReq.Params, ¶ms); err != nil { + writeRPCError(w, rpcReq.ID, http.StatusOK, ErrCodeInvalidParams, "invalid params") + return + } + task, ok := s.tasks.get(params.ID) + if !ok { + writeRPCError(w, rpcReq.ID, http.StatusOK, ErrCodeTaskNotFound, "task not found: "+params.ID) + return + } + writeRPCResult(w, rpcReq.ID, task) +} + +func (s *Server) handleTasksCancel(w http.ResponseWriter, r *http.Request, rpcReq JSONRPCRequest) { + var params TaskIDParams + if err := json.Unmarshal(rpcReq.Params, ¶ms); err != nil { + writeRPCError(w, rpcReq.ID, http.StatusOK, ErrCodeInvalidParams, "invalid params") + return + } + task, ok := s.tasks.get(params.ID) + if !ok { + writeRPCError(w, rpcReq.ID, http.StatusOK, ErrCodeTaskNotFound, "task not found: "+params.ID) + return + } + if task.Status.State != StateWorking { + writeRPCError(w, rpcReq.ID, http.StatusOK, ErrCodeNotCancelable, "task is not in working state") + return + } + _, err := s.svc.CancelRun(r.Context(), edgeservice.CancelRunRequest{ + NodeRef: s.cfg.NodeRef, + RunID: params.ID, + Adapter: s.resolveAdapter(), + Target: s.resolveTarget(), + SessionID: s.resolveSessionID(), + }) + if err != nil { + writeRPCError(w, rpcReq.ID, http.StatusOK, ErrCodeInvalidRequest, err.Error()) + return + } + writeRPCResult(w, rpcReq.ID, map[string]string{"id": params.ID}) +} + +func (s *Server) resolveAdapter() string { + if s.cfg.Adapter != "" { + return s.cfg.Adapter + } + return "cli" +} + +func (s *Server) resolveTarget() string { + return s.cfg.Target +} + +func (s *Server) resolveSessionID() string { + if s.cfg.SessionID != "" { + return s.cfg.SessionID + } + return edgeservice.DefaultSessionID +} + +func (s *Server) resolveTimeoutSec() int { + if s.cfg.TimeoutSec > 0 { + return s.cfg.TimeoutSec + } + return edgeservice.DefaultTimeoutSec +} + +func extractPrompt(msg Message) string { + var b strings.Builder + for _, p := range msg.Parts { + if p.Type == "text" && p.Text != "" { + if b.Len() > 0 { + b.WriteString("\n") + } + b.WriteString(p.Text) + } + } + return b.String() +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +func writeRPCResult(w http.ResponseWriter, id any, result any) { + writeJSON(w, http.StatusOK, JSONRPCResponse{ + JSONRPC: "2.0", + Result: result, + ID: id, + }) +} + +func writeRPCError(w http.ResponseWriter, id any, httpStatus, code int, message string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(httpStatus) + _ = json.NewEncoder(w).Encode(JSONRPCResponse{ + JSONRPC: "2.0", + Error: &JSONRPCError{Code: code, Message: message}, + ID: id, + }) +} diff --git a/apps/edge/internal/input/a2a/server_test.go b/apps/edge/internal/input/a2a/server_test.go new file mode 100644 index 0000000..910d8cb --- /dev/null +++ b/apps/edge/internal/input/a2a/server_test.go @@ -0,0 +1,376 @@ +package a2a_test + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "go.uber.org/zap" + + "iop/apps/edge/internal/input/a2a" + edgeservice "iop/apps/edge/internal/service" + "iop/packages/config" + iop "iop/proto/gen/iop" +) + +// fakeService is a minimal runService implementation for testing. +type fakeService struct { + submitFn func(context.Context, edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) + cancelFn func(context.Context, edgeservice.CancelRunRequest) (edgeservice.CommandResult, error) +} + +func (f *fakeService) SubmitRun(ctx context.Context, req edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) { + if f.submitFn != nil { + return f.submitFn(ctx, req) + } + return nil, nil +} + +func (f *fakeService) CancelRun(ctx context.Context, req edgeservice.CancelRunRequest) (edgeservice.CommandResult, error) { + if f.cancelFn != nil { + return f.cancelFn(ctx, req) + } + return edgeservice.CommandResult{}, nil +} + +// newTestServer builds a disabled-by-default a2a.Server for handler-level testing. +// We call the handlers directly via httptest rather than starting a real listener. +func newTestServer(cfg config.EdgeA2AConf, svc runServiceIface) *a2a.Server { + return a2a.NewServer(cfg, svc, zap.NewNop()) +} + +type runServiceIface interface { + SubmitRun(context.Context, edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) + CancelRun(context.Context, edgeservice.CancelRunRequest) (edgeservice.CommandResult, error) +} + +func rpcPost(t *testing.T, handler http.Handler, body any) *httptest.ResponseRecorder { + t.Helper() + b, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal body: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/a2a", bytes.NewReader(b)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + return w +} + +func decodeRPCResponse(t *testing.T, w *httptest.ResponseRecorder) a2a.JSONRPCResponse { + t.Helper() + var resp a2a.JSONRPCResponse + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decode response: %v", err) + } + return resp +} + +func TestHandleMessageSendDispatchesRun(t *testing.T) { + dispatched := false + svc := &fakeService{ + submitFn: func(_ context.Context, req edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) { + dispatched = true + if req.Metadata["source"] != "a2a" { + t.Errorf("expected source=a2a, got %q", req.Metadata["source"]) + } + return feedHandle("run-1", &iop.RunEvent{RunId: "run-1", Type: "complete"}), nil + }, + } + + cfg := config.EdgeA2AConf{Enabled: true, Listen: "127.0.0.1:0", Path: "/a2a", TimeoutSec: 5} + srv := a2a.NewServer(cfg, svc, zap.NewNop()) + + mux := http.NewServeMux() + mux.HandleFunc("/a2a", srv.RPCHandlerForTest()) + + sendBody := a2a.JSONRPCRequest{ + JSONRPC: "2.0", + Method: "message/send", + Params: mustMarshal(t, a2a.MessageSendParams{ + Message: a2a.Message{ + Role: "user", + Parts: []a2a.Part{{Type: "text", Text: "hi"}}, + }, + Configuration: &a2a.SendConfig{Blocking: boolPtr(true)}, + }), + ID: 1, + } + w := rpcPost(t, mux, sendBody) + if w.Code != http.StatusOK { + t.Fatalf("message/send status: %d body: %s", w.Code, w.Body.String()) + } + resp := decodeRPCResponse(t, w) + if resp.Error != nil { + t.Fatalf("message/send error: %+v", resp.Error) + } + if !dispatched { + t.Error("SubmitRun was not called") + } +} + +func TestHandleGetTaskReturnsStoredTask(t *testing.T) { + runID := "run-get-1" + svc := &fakeService{ + submitFn: func(_ context.Context, _ edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) { + return feedHandle(runID, + &iop.RunEvent{RunId: runID, Type: "delta", Delta: "result text"}, + &iop.RunEvent{RunId: runID, Type: "complete"}, + ), nil + }, + } + cfg := config.EdgeA2AConf{Enabled: false, Path: "/a2a", TimeoutSec: 5} + srv := a2a.NewServer(cfg, svc, zap.NewNop()) + + // Inject a task by calling message/send handler directly via httptest. + sendBody := a2a.JSONRPCRequest{ + JSONRPC: "2.0", + Method: "message/send", + Params: mustMarshal(t, a2a.MessageSendParams{ + Message: a2a.Message{ + Role: "user", + Parts: []a2a.Part{{Type: "text", Text: "hello"}}, + }, + Configuration: &a2a.SendConfig{Blocking: boolPtr(true)}, + }), + ID: 1, + } + + mux := http.NewServeMux() + mux.HandleFunc("/a2a", srv.RPCHandlerForTest()) + w := rpcPost(t, mux, sendBody) + if w.Code != http.StatusOK { + t.Fatalf("message/send status: %d body: %s", w.Code, w.Body.String()) + } + sendResp := decodeRPCResponse(t, w) + if sendResp.Error != nil { + t.Fatalf("message/send error: %+v", sendResp.Error) + } + + // Extract task ID from result. + taskMap, ok := sendResp.Result.(map[string]any) + if !ok { + t.Fatalf("expected map result, got %T", sendResp.Result) + } + taskID, _ := taskMap["id"].(string) + if taskID == "" { + t.Fatal("task id is empty") + } + + // Now tasks/get. + getBody := a2a.JSONRPCRequest{ + JSONRPC: "2.0", + Method: "tasks/get", + Params: mustMarshal(t, a2a.TaskQueryParams{ID: taskID}), + ID: 2, + } + w2 := rpcPost(t, mux, getBody) + if w2.Code != http.StatusOK { + t.Fatalf("tasks/get status: %d", w2.Code) + } + getResp := decodeRPCResponse(t, w2) + if getResp.Error != nil { + t.Fatalf("tasks/get error: %+v", getResp.Error) + } + taskMap2, _ := getResp.Result.(map[string]any) + status, _ := taskMap2["status"].(map[string]any) + state, _ := status["state"].(string) + if state != "completed" { + t.Errorf("expected completed, got %q", state) + } +} + +func TestHandleCancelTask(t *testing.T) { + runID := "run-cancel-1" + cancelCalled := false + svc := &fakeService{ + submitFn: func(_ context.Context, _ edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) { + // Return a handle that stays working (never completes) for non-blocking. + events := make(chan *iop.RunEvent, 1) + nodeEvents := make(chan *iop.EdgeNodeEvent, 1) + return &edgeservice.RunHandle{ + RunDispatch: edgeservice.RunDispatch{RunID: runID, TimeoutSec: 60}, + RunStream: edgeservice.RunStream{Events: events, NodeEvents: nodeEvents}, + }, nil + }, + cancelFn: func(_ context.Context, req edgeservice.CancelRunRequest) (edgeservice.CommandResult, error) { + cancelCalled = true + if req.RunID != runID { + t.Errorf("expected RunID=%q, got %q", runID, req.RunID) + } + return edgeservice.CommandResult{}, nil + }, + } + cfg := config.EdgeA2AConf{Enabled: false, Path: "/a2a", TimeoutSec: 60} + srv := a2a.NewServer(cfg, svc, zap.NewNop()) + + mux := http.NewServeMux() + mux.HandleFunc("/a2a", srv.RPCHandlerForTest()) + + // Submit non-blocking so task stays in working state. + sendBody := a2a.JSONRPCRequest{ + JSONRPC: "2.0", + Method: "message/send", + Params: mustMarshal(t, a2a.MessageSendParams{ + Message: a2a.Message{ + Role: "user", + Parts: []a2a.Part{{Type: "text", Text: "work"}}, + }, + Configuration: &a2a.SendConfig{Blocking: boolPtr(false)}, + }), + ID: 1, + } + w := rpcPost(t, mux, sendBody) + if w.Code != http.StatusOK { + t.Fatalf("message/send: %d %s", w.Code, w.Body.String()) + } + sendResp := decodeRPCResponse(t, w) + if sendResp.Error != nil { + t.Fatalf("message/send error: %+v", sendResp.Error) + } + taskMap, _ := sendResp.Result.(map[string]any) + taskID, _ := taskMap["id"].(string) + + // Cancel it. + cancelBody := a2a.JSONRPCRequest{ + JSONRPC: "2.0", + Method: "tasks/cancel", + Params: mustMarshal(t, a2a.TaskIDParams{ID: taskID}), + ID: 2, + } + w2 := rpcPost(t, mux, cancelBody) + if w2.Code != http.StatusOK { + t.Fatalf("tasks/cancel: %d %s", w2.Code, w2.Body.String()) + } + cancelResp := decodeRPCResponse(t, w2) + if cancelResp.Error != nil { + t.Fatalf("tasks/cancel error: %+v", cancelResp.Error) + } + if !cancelCalled { + t.Error("CancelRun was not called") + } +} + +func TestRejectsBadAuth(t *testing.T) { + svc := &fakeService{} + cfg := config.EdgeA2AConf{Enabled: false, Path: "/a2a", BearerToken: "secret"} + srv := a2a.NewServer(cfg, svc, zap.NewNop()) + + mux := http.NewServeMux() + mux.HandleFunc("/a2a", srv.RPCHandlerForTest()) + + body := a2a.JSONRPCRequest{JSONRPC: "2.0", Method: "message/send", ID: 1} + b, _ := json.Marshal(body) + req := httptest.NewRequest(http.MethodPost, "/a2a", bytes.NewReader(b)) + req.Header.Set("Authorization", "Bearer wrong") + w := httptest.NewRecorder() + mux.ServeHTTP(w, req) + if w.Code != http.StatusUnauthorized { + t.Errorf("expected 401, got %d", w.Code) + } +} + +func TestRejectsUnknownMethod(t *testing.T) { + svc := &fakeService{} + cfg := config.EdgeA2AConf{Enabled: false, Path: "/a2a"} + srv := a2a.NewServer(cfg, svc, zap.NewNop()) + + mux := http.NewServeMux() + mux.HandleFunc("/a2a", srv.RPCHandlerForTest()) + + body := a2a.JSONRPCRequest{JSONRPC: "2.0", Method: "unknown/method", ID: 1} + w := rpcPost(t, mux, body) + resp := decodeRPCResponse(t, w) + if resp.Error == nil { + t.Fatal("expected error for unknown method") + } + if resp.Error.Code != a2a.ErrCodeMethodNotFound { + t.Errorf("expected code %d, got %d", a2a.ErrCodeMethodNotFound, resp.Error.Code) + } +} + +// TestBlockingDefault verifies the four cases for the blocking field: +// - no configuration → blocking true +// - configuration: {} (empty object, Blocking nil) → blocking true +// - configuration: {"blocking": true} → blocking true +// - configuration: {"blocking": false} → blocking false +func TestBlockingDefault(t *testing.T) { + cases := []struct { + name string + cfg *a2a.SendConfig + wantBlock bool + }{ + {"no configuration", nil, true}, + {"empty configuration", &a2a.SendConfig{}, true}, + {"explicit true", &a2a.SendConfig{Blocking: boolPtr(true)}, true}, + {"explicit false", &a2a.SendConfig{Blocking: boolPtr(false)}, false}, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + events := make(chan *iop.RunEvent, 2) + nodeEvents := make(chan *iop.EdgeNodeEvent, 1) + var dispatched bool + svc := &fakeService{ + submitFn: func(_ context.Context, req edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) { + dispatched = true + got := req.Metadata["blocking"] + want := "true" + if !tc.wantBlock { + want = "false" + } + if got != want { + t.Errorf("blocking metadata: got %q want %q", got, want) + } + if tc.wantBlock { + events <- &iop.RunEvent{RunId: "r", Type: "complete"} + } + return &edgeservice.RunHandle{ + RunDispatch: edgeservice.RunDispatch{RunID: "r", TimeoutSec: 5}, + RunStream: edgeservice.RunStream{Events: events, NodeEvents: nodeEvents}, + }, nil + }, + } + cfg := config.EdgeA2AConf{Enabled: false, Path: "/a2a", TimeoutSec: 5} + srv := a2a.NewServer(cfg, svc, zap.NewNop()) + mux := http.NewServeMux() + mux.HandleFunc("/a2a", srv.RPCHandlerForTest()) + + body := a2a.JSONRPCRequest{ + JSONRPC: "2.0", + Method: "message/send", + Params: mustMarshal(t, a2a.MessageSendParams{ + Message: a2a.Message{Role: "user", Parts: []a2a.Part{{Type: "text", Text: "hi"}}}, + Configuration: tc.cfg, + }), + ID: 1, + } + w := rpcPost(t, mux, body) + if w.Code != http.StatusOK { + t.Fatalf("status: %d body: %s", w.Code, w.Body.String()) + } + resp := decodeRPCResponse(t, w) + if resp.Error != nil { + t.Fatalf("rpc error: %+v", resp.Error) + } + if !dispatched { + t.Error("SubmitRun was not called") + } + }) + } +} + +func mustMarshal(t *testing.T, v any) json.RawMessage { + t.Helper() + b, err := json.Marshal(v) + if err != nil { + t.Fatalf("marshal: %v", err) + } + return b +} + +func boolPtr(b bool) *bool { return &b } diff --git a/apps/edge/internal/input/a2a/task_store.go b/apps/edge/internal/input/a2a/task_store.go new file mode 100644 index 0000000..59c1e8b --- /dev/null +++ b/apps/edge/internal/input/a2a/task_store.go @@ -0,0 +1,163 @@ +package a2a + +import ( + "strings" + "sync" + "time" + + edgeservice "iop/apps/edge/internal/service" + iop "iop/proto/gen/iop" +) + +// TaskStore is an in-memory registry of A2A tasks indexed by task (run) ID. +type TaskStore struct { + mu sync.RWMutex + tasks map[string]*Task +} + +func newTaskStore() *TaskStore { + return &TaskStore{tasks: make(map[string]*Task)} +} + +func (s *TaskStore) get(id string) (*Task, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + t, ok := s.tasks[id] + if !ok { + return nil, false + } + return snapshotTask(t), true +} + +// snapshotTask returns a deep copy of t so callers cannot observe future mutations. +// Must be called with s.mu held for reading (or from a context where t cannot be +// concurrently modified, e.g. already holding the write lock). +func snapshotTask(t *Task) *Task { + cp := Task{ + ID: t.ID, + Status: TaskStatus{State: t.Status.State}, + } + if t.Status.Message != nil { + msgCopy := copyMessage(*t.Status.Message) + cp.Status.Message = &msgCopy + } + if len(t.Artifacts) > 0 { + cp.Artifacts = make([]Artifact, len(t.Artifacts)) + for i, a := range t.Artifacts { + cp.Artifacts[i] = Artifact{ + Name: a.Name, + Parts: append([]Part(nil), a.Parts...), + } + } + } + if len(t.History) > 0 { + cp.History = make([]Message, len(t.History)) + for i, m := range t.History { + cp.History[i] = copyMessage(m) + } + } + return &cp +} + +func copyMessage(m Message) Message { + return Message{ + Role: m.Role, + Parts: append([]Part(nil), m.Parts...), + } +} + +// create stores a new working Task and returns the task ID (not the internal pointer). +func (s *TaskStore) create(id string, userMsg Message) string { + t := &Task{ + ID: id, + Status: TaskStatus{State: StateWorking}, + History: []Message{userMsg}, + } + s.mu.Lock() + s.tasks[id] = t + s.mu.Unlock() + return id +} + +// collectBackground drains a RunHandle in a background goroutine and updates the stored Task. +func (s *TaskStore) collectBackground(taskID string, handle *edgeservice.RunHandle) { + go func() { + defer handle.Close() + s.drain(taskID, handle) + }() +} + +// collectBlocking drains a RunHandle synchronously and returns the final Task snapshot. +func (s *TaskStore) collectBlocking(taskID string, handle *edgeservice.RunHandle) *Task { + defer handle.Close() + return s.drain(taskID, handle) +} + +func (s *TaskStore) drain(taskID string, handle *edgeservice.RunHandle) *Task { + var text strings.Builder + timeout := time.NewTimer(handle.WaitTimeout()) + defer timeout.Stop() + + for { + select { + case nodeEvent := <-handle.NodeEvents: + if edgeservice.IsNodeDisconnected(nodeEvent) { + return s.setFailed(taskID, "node disconnected") + } + case event := <-handle.Events: + if event == nil { + continue + } + switch event.GetType() { + case "delta": + text.WriteString(event.GetDelta()) + case "complete": + artifact := Artifact{Parts: []Part{{Type: "text", Text: text.String()}}} + s.mu.Lock() + if t, ok := s.tasks[taskID]; ok { + t.Artifacts = []Artifact{artifact} + t.Status = TaskStatus{State: StateCompleted} + } + s.mu.Unlock() + t, _ := s.get(taskID) + return t + case "error": + msg := errorMsg(event) + return s.setFailed(taskID, msg) + case "cancelled": + s.mu.Lock() + if t, ok := s.tasks[taskID]; ok { + t.Status = TaskStatus{State: StateCanceled} + } + s.mu.Unlock() + t, _ := s.get(taskID) + return t + } + case <-timeout.C: + return s.setFailed(taskID, "run timed out") + } + } +} + +func (s *TaskStore) setFailed(taskID, msg string) *Task { + s.mu.Lock() + if t, ok := s.tasks[taskID]; ok { + t.Status = TaskStatus{State: StateFailed, Message: &Message{ + Role: "agent", + Parts: []Part{{Type: "text", Text: msg}}, + }} + } + s.mu.Unlock() + t, _ := s.get(taskID) + return t +} + +func errorMsg(event *iop.RunEvent) string { + if m := event.GetError(); m != "" { + return m + } + if m := event.GetMessage(); m != "" { + return m + } + return "run failed" +} diff --git a/apps/edge/internal/input/a2a/task_store_test.go b/apps/edge/internal/input/a2a/task_store_test.go new file mode 100644 index 0000000..187c38d --- /dev/null +++ b/apps/edge/internal/input/a2a/task_store_test.go @@ -0,0 +1,187 @@ +package a2a_test + +import ( + "context" + "net/http" + "testing" + + "go.uber.org/zap" + + "iop/apps/edge/internal/input/a2a" + edgeservice "iop/apps/edge/internal/service" + "iop/packages/config" + iop "iop/proto/gen/iop" +) + +// feedHandle builds a RunHandle pre-loaded with the given events. +func feedHandle(runID string, events ...*iop.RunEvent) *edgeservice.RunHandle { + ch := make(chan *iop.RunEvent, len(events)) + for _, e := range events { + ch <- e + } + nodeEvents := make(chan *iop.EdgeNodeEvent, 1) + return &edgeservice.RunHandle{ + RunDispatch: edgeservice.RunDispatch{RunID: runID, TimeoutSec: 5}, + RunStream: edgeservice.RunStream{Events: ch, NodeEvents: nodeEvents}, + } +} + +func checkBlockingResult(t *testing.T, handle *edgeservice.RunHandle, check func(map[string]any)) { + t.Helper() + svc := &fakeService{ + submitFn: func(_ context.Context, _ edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) { + return handle, nil + }, + } + cfg := config.EdgeA2AConf{Enabled: false, Path: "/a2a", TimeoutSec: 5} + srv := a2a.NewServer(cfg, svc, zap.NewNop()) + mux := http.NewServeMux() + mux.HandleFunc("/a2a", srv.RPCHandlerForTest()) + + body := a2a.JSONRPCRequest{ + JSONRPC: "2.0", + Method: "message/send", + Params: mustMarshal(t, a2a.MessageSendParams{ + Message: a2a.Message{ + Role: "user", + Parts: []a2a.Part{{Type: "text", Text: "ping"}}, + }, + Configuration: &a2a.SendConfig{Blocking: boolPtr(true)}, + }), + ID: 1, + } + w := rpcPost(t, mux, body) + if w.Code != http.StatusOK { + t.Fatalf("status: %d body: %s", w.Code, w.Body.String()) + } + resp := decodeRPCResponse(t, w) + if resp.Error != nil { + t.Fatalf("rpc error: %+v", resp.Error) + } + task, ok := resp.Result.(map[string]any) + if !ok { + t.Fatalf("expected map result, got %T", resp.Result) + } + check(task) +} + +func TestTaskStoreMapsRunEventsToCompletedTask(t *testing.T) { + runID := "run-ts-1" + checkBlockingResult(t, + feedHandle(runID, + &iop.RunEvent{RunId: runID, Type: "delta", Delta: "hello "}, + &iop.RunEvent{RunId: runID, Type: "delta", Delta: "world"}, + &iop.RunEvent{RunId: runID, Type: "complete"}, + ), + func(task map[string]any) { + status := task["status"].(map[string]any) + if status["state"] != "completed" { + t.Errorf("expected completed, got %q", status["state"]) + } + artifacts, ok := task["artifacts"].([]any) + if !ok || len(artifacts) == 0 { + t.Fatal("expected non-empty artifacts") + } + parts := artifacts[0].(map[string]any)["parts"].([]any) + text := parts[0].(map[string]any)["text"].(string) + if text != "hello world" { + t.Errorf("expected %q, got %q", "hello world", text) + } + }, + ) +} + +func TestTaskStoreMapsErrorToFailedTask(t *testing.T) { + runID := "run-ts-2" + checkBlockingResult(t, + feedHandle(runID, + &iop.RunEvent{RunId: runID, Type: "error", Error: "something broke"}, + ), + func(task map[string]any) { + status := task["status"].(map[string]any) + if status["state"] != "failed" { + t.Errorf("expected failed, got %q", status["state"]) + } + }, + ) +} + +func TestMessageSendBlockingReturnsCompletedTask(t *testing.T) { + runID := "run-ts-3" + checkBlockingResult(t, + feedHandle(runID, + &iop.RunEvent{RunId: runID, Type: "delta", Delta: "done"}, + &iop.RunEvent{RunId: runID, Type: "complete"}, + ), + func(task map[string]any) { + status := task["status"].(map[string]any) + if status["state"] != "completed" { + t.Errorf("expected completed, got %q", status["state"]) + } + }, + ) +} + +func TestCancelTaskSendsCancelRun(t *testing.T) { + runID := "run-ts-cancel" + cancelCalled := false + + events := make(chan *iop.RunEvent, 1) + nodeEvents := make(chan *iop.EdgeNodeEvent, 1) + handle := &edgeservice.RunHandle{ + RunDispatch: edgeservice.RunDispatch{RunID: runID, TimeoutSec: 60}, + RunStream: edgeservice.RunStream{Events: events, NodeEvents: nodeEvents}, + } + + svc := &fakeService{ + submitFn: func(_ context.Context, _ edgeservice.SubmitRunRequest) (*edgeservice.RunHandle, error) { + return handle, nil + }, + cancelFn: func(_ context.Context, req edgeservice.CancelRunRequest) (edgeservice.CommandResult, error) { + cancelCalled = true + if req.RunID != runID { + t.Errorf("expected RunID=%q, got %q", runID, req.RunID) + } + return edgeservice.CommandResult{}, nil + }, + } + + cfg := config.EdgeA2AConf{Enabled: false, Path: "/a2a", TimeoutSec: 60} + srv := a2a.NewServer(cfg, svc, zap.NewNop()) + mux := http.NewServeMux() + mux.HandleFunc("/a2a", srv.RPCHandlerForTest()) + + sendBody := a2a.JSONRPCRequest{ + JSONRPC: "2.0", + Method: "message/send", + Params: mustMarshal(t, a2a.MessageSendParams{ + Message: a2a.Message{ + Role: "user", + Parts: []a2a.Part{{Type: "text", Text: "work"}}, + }, + Configuration: &a2a.SendConfig{Blocking: boolPtr(false)}, + }), + ID: 1, + } + w := rpcPost(t, mux, sendBody) + sendResp := decodeRPCResponse(t, w) + if sendResp.Error != nil { + t.Fatalf("message/send: %+v", sendResp.Error) + } + taskID := sendResp.Result.(map[string]any)["id"].(string) + + cancelBody := a2a.JSONRPCRequest{ + JSONRPC: "2.0", + Method: "tasks/cancel", + Params: mustMarshal(t, a2a.TaskIDParams{ID: taskID}), + ID: 2, + } + w2 := rpcPost(t, mux, cancelBody) + cancelResp := decodeRPCResponse(t, w2) + if cancelResp.Error != nil { + t.Fatalf("tasks/cancel: %+v", cancelResp.Error) + } + if !cancelCalled { + t.Error("CancelRun was not called") + } +} diff --git a/apps/edge/internal/input/a2a/types.go b/apps/edge/internal/input/a2a/types.go new file mode 100644 index 0000000..0b5df2a --- /dev/null +++ b/apps/edge/internal/input/a2a/types.go @@ -0,0 +1,92 @@ +package a2a + +import "encoding/json" + +// JSON-RPC 2.0 envelope types. + +type JSONRPCRequest struct { + JSONRPC string `json:"jsonrpc"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` + ID any `json:"id"` +} + +type JSONRPCResponse struct { + JSONRPC string `json:"jsonrpc"` + Result any `json:"result,omitempty"` + Error *JSONRPCError `json:"error,omitempty"` + ID any `json:"id"` +} + +type JSONRPCError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// A2A method parameter types. + +type MessageSendParams struct { + Message Message `json:"message"` + Configuration *SendConfig `json:"configuration,omitempty"` +} + +type SendConfig struct { + // Blocking uses *bool so that omitted and explicit false can be distinguished. + // nil means "use default" (true); explicit false means non-blocking. + Blocking *bool `json:"blocking,omitempty"` +} + +type TaskQueryParams struct { + ID string `json:"id"` +} + +type TaskIDParams struct { + ID string `json:"id"` +} + +// A2A core types. + +type Message struct { + Role string `json:"role"` + Parts []Part `json:"parts"` +} + +type Part struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` +} + +type Task struct { + ID string `json:"id"` + Status TaskStatus `json:"status"` + Artifacts []Artifact `json:"artifacts,omitempty"` + History []Message `json:"history,omitempty"` +} + +type TaskStatus struct { + State string `json:"state"` + Message *Message `json:"message,omitempty"` +} + +type Artifact struct { + Name string `json:"name,omitempty"` + Parts []Part `json:"parts"` +} + +// JSON-RPC error codes. +const ( + ErrCodeParseError = -32700 + ErrCodeInvalidRequest = -32600 + ErrCodeMethodNotFound = -32601 + ErrCodeInvalidParams = -32602 + ErrCodeTaskNotFound = -32001 + ErrCodeNotCancelable = -32002 +) + +// A2A task states. +const ( + StateWorking = "working" + StateCompleted = "completed" + StateFailed = "failed" + StateCanceled = "canceled" +) diff --git a/apps/edge/internal/input/manager.go b/apps/edge/internal/input/manager.go new file mode 100644 index 0000000..9bad880 --- /dev/null +++ b/apps/edge/internal/input/manager.go @@ -0,0 +1,48 @@ +package input + +import ( + "context" + "time" + + "go.uber.org/zap" + + edgea2a "iop/apps/edge/internal/input/a2a" + edgeopenai "iop/apps/edge/internal/openai" + edgeservice "iop/apps/edge/internal/service" + "iop/packages/config" +) + +// Manager owns the lifecycle of all Edge inbound input servers (OpenAI-compatible and A2A). +type Manager struct { + OpenAI *edgeopenai.Server + A2A *edgea2a.Server +} + +// NewManager creates a Manager wiring both input servers. +func NewManager(cfg config.EdgeConfig, svc *edgeservice.Service, logger *zap.Logger) *Manager { + openaiServer := edgeopenai.NewServer(cfg.OpenAI, svc, logger.Named("openai")) + a2aServer := edgea2a.NewServer(cfg.A2A, svc, logger.Named("a2a")) + return &Manager{OpenAI: openaiServer, A2A: a2aServer} +} + +func (m *Manager) Start(ctx context.Context) error { + if err := m.OpenAI.Start(ctx); err != nil { + return err + } + if err := m.A2A.Start(ctx); err != nil { + stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = m.OpenAI.Stop(stopCtx) + return err + } + return nil +} + +func (m *Manager) Stop(ctx context.Context) error { + a2aErr := m.A2A.Stop(ctx) + openaiErr := m.OpenAI.Stop(ctx) + if a2aErr != nil { + return a2aErr + } + return openaiErr +} diff --git a/apps/edge/internal/input/manager_test.go b/apps/edge/internal/input/manager_test.go new file mode 100644 index 0000000..b55b1ae --- /dev/null +++ b/apps/edge/internal/input/manager_test.go @@ -0,0 +1,50 @@ +package input_test + +import ( + "context" + "testing" + + edgeinput "iop/apps/edge/internal/input" + edgeservice "iop/apps/edge/internal/service" + "iop/packages/config" + + "go.uber.org/zap" +) + +func newTestService() *edgeservice.Service { + return edgeservice.New(nil, nil) +} + +func TestManagerOwnsOpenAIAndA2AInputs(t *testing.T) { + cfg := config.EdgeConfig{ + OpenAI: config.EdgeOpenAIConf{Enabled: false}, + A2A: config.EdgeA2AConf{Enabled: false}, + } + mgr := edgeinput.NewManager(cfg, newTestService(), zap.NewNop()) + if mgr == nil { + t.Fatal("NewManager returned nil") + } + if mgr.OpenAI == nil { + t.Fatal("Manager.OpenAI is nil") + } + if mgr.A2A == nil { + t.Fatal("Manager.A2A is nil") + } +} + +func TestManagerStartStopDisabled(t *testing.T) { + cfg := config.EdgeConfig{ + OpenAI: config.EdgeOpenAIConf{Enabled: false}, + A2A: config.EdgeA2AConf{Enabled: false}, + } + mgr := edgeinput.NewManager(cfg, newTestService(), zap.NewNop()) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + if err := mgr.Start(ctx); err != nil { + t.Fatalf("Start disabled: %v", err) + } + if err := mgr.Stop(ctx); err != nil { + t.Fatalf("Stop disabled: %v", err) + } +} diff --git a/apps/edge/internal/node/registry.go b/apps/edge/internal/node/registry.go index abe6be3..6cb57ec 100644 --- a/apps/edge/internal/node/registry.go +++ b/apps/edge/internal/node/registry.go @@ -7,7 +7,7 @@ import ( "strings" "sync" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" ) // NodeEntry represents one connected node. diff --git a/apps/edge/internal/node/registry_test.go b/apps/edge/internal/node/registry_test.go index 0b5ff2c..9c1b7d0 100644 --- a/apps/edge/internal/node/registry_test.go +++ b/apps/edge/internal/node/registry_test.go @@ -3,7 +3,7 @@ package node_test import ( "testing" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" edgenode "iop/apps/edge/internal/node" ) diff --git a/apps/edge/internal/service/service.go b/apps/edge/internal/service/service.go index 48812ba..0d41d6c 100644 --- a/apps/edge/internal/service/service.go +++ b/apps/edge/internal/service/service.go @@ -5,7 +5,7 @@ import ( "fmt" "time" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "google.golang.org/protobuf/types/known/structpb" edgeevents "iop/apps/edge/internal/events" @@ -189,6 +189,40 @@ func (s *Service) SubmitRun(_ context.Context, req SubmitRunRequest) (*RunHandle }, nil } +type CancelRunRequest struct { + NodeRef string + RunID string + Adapter string + Target string + SessionID string +} + +func BuildCancelRunRequest(req CancelRunRequest) *iop.CancelRequest { + return &iop.CancelRequest{ + RunId: req.RunID, + Adapter: req.Adapter, + Target: req.Target, + SessionId: NormalizeSessionID(req.SessionID), + Action: iop.CancelAction_CANCEL_ACTION_CANCEL_RUN, + } +} + +func (s *Service) CancelRun(_ context.Context, req CancelRunRequest) (CommandResult, error) { + entry, err := s.ResolveNode(req.NodeRef) + if err != nil { + return CommandResult{}, err + } + cancelReq := BuildCancelRunRequest(req) + if err := entry.Client.Send(cancelReq); err != nil { + return CommandResult{}, err + } + return CommandResult{ + NodeID: entry.NodeID, + NodeLabel: nodeLabel(entry), + SessionID: cancelReq.GetSessionId(), + }, nil +} + type TerminateSessionRequest struct { NodeRef string Adapter string diff --git a/apps/edge/internal/service/service_test.go b/apps/edge/internal/service/service_test.go index 3fb7104..84a3ec9 100644 --- a/apps/edge/internal/service/service_test.go +++ b/apps/edge/internal/service/service_test.go @@ -203,6 +203,55 @@ func TestSubmitRunReturnsDispatchMetadata(t *testing.T) { } } +func TestBuildCancelRunRequest(t *testing.T) { + cases := []struct { + name string + req edgeservice.CancelRunRequest + wantSessionID string + }{ + { + name: "explicit session", + req: edgeservice.CancelRunRequest{ + RunID: "run-123", + Adapter: "cli", + Target: "claude", + SessionID: "session-a", + }, + wantSessionID: "session-a", + }, + { + name: "empty session normalizes to default", + req: edgeservice.CancelRunRequest{ + RunID: "run-456", + Adapter: "ollama", + Target: "llama3", + SessionID: "", + }, + wantSessionID: edgeservice.DefaultSessionID, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + proto := edgeservice.BuildCancelRunRequest(tc.req) + if proto.GetRunId() != tc.req.RunID { + t.Errorf("RunId: got %q want %q", proto.GetRunId(), tc.req.RunID) + } + if proto.GetAdapter() != tc.req.Adapter { + t.Errorf("Adapter: got %q want %q", proto.GetAdapter(), tc.req.Adapter) + } + if proto.GetTarget() != tc.req.Target { + t.Errorf("Target: got %q want %q", proto.GetTarget(), tc.req.Target) + } + if proto.GetSessionId() != tc.wantSessionID { + t.Errorf("SessionId: got %q want %q", proto.GetSessionId(), tc.wantSessionID) + } + if proto.GetAction() != iop.CancelAction_CANCEL_ACTION_CANCEL_RUN { + t.Errorf("Action: got %v want CANCEL_ACTION_CANCEL_RUN", proto.GetAction()) + } + }) + } +} + func TestResolveNode_AllowsSingleNodeFallback(t *testing.T) { reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "node-1"}) diff --git a/apps/edge/internal/transport/integration_test.go b/apps/edge/internal/transport/integration_test.go index 7a98c7d..9aa79ee 100644 --- a/apps/edge/internal/transport/integration_test.go +++ b/apps/edge/internal/transport/integration_test.go @@ -10,7 +10,7 @@ import ( "go.uber.org/zap" "google.golang.org/protobuf/proto" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" edgenode "iop/apps/edge/internal/node" "iop/apps/edge/internal/transport" diff --git a/apps/edge/internal/transport/server.go b/apps/edge/internal/transport/server.go index 0ad9df5..0e7f527 100644 --- a/apps/edge/internal/transport/server.go +++ b/apps/edge/internal/transport/server.go @@ -7,7 +7,7 @@ import ( "sync" "sync/atomic" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "go.uber.org/zap" "google.golang.org/protobuf/proto" diff --git a/apps/edge/internal/transport/server_test.go b/apps/edge/internal/transport/server_test.go index b225ace..3af43c6 100644 --- a/apps/edge/internal/transport/server_test.go +++ b/apps/edge/internal/transport/server_test.go @@ -3,7 +3,7 @@ package transport import ( "testing" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "google.golang.org/protobuf/proto" edgenode "iop/apps/edge/internal/node" diff --git a/apps/node/README.md b/apps/node/README.md index ab32590..da3767d 100644 --- a/apps/node/README.md +++ b/apps/node/README.md @@ -30,6 +30,9 @@ go build -o bin/node ./apps/node/cmd/node # 버전 확인 ./bin/node version +# 설정 파일 로드 검증 +./bin/node config check --config configs/node.yaml + # 설정 확인 ./bin/node config print --config configs/node.yaml @@ -37,6 +40,41 @@ go build -o bin/node ./apps/node/cmd/node ./bin/node serve --config configs/node.yaml ``` +## 호스트 환경 준비 (setup) + +운영 설치 공식 경로는 `setup` 명령으로 일원화한다. + +```bash +sudo iop-node setup --enable --start +``` + +배포 전에 결과를 검토하려면 `--dry-run`을 쓴다. + +```bash +sudo iop-node setup --dry-run --binary /usr/local/bin/iop-node +``` + +`setup`은 다음을 담당한다. + +- 실행 user/group(`iop`) 준비 +- `/etc/iop`와 `/var/lib/iop/node` 디렉터리 준비 +- 설정 파일이 없을 때만 기본 템플릿 생성 (`--overwrite-config`로 강제 갱신) +- systemd unit 생성 또는 갱신 (`/etc/systemd/system/iop-node.service`) +- `systemctl daemon-reload` +- 옵션에 따른 `--enable`, `--start`, `--restart` + +`setup`의 `--config` 기본값은 `/etc/iop/node.yaml`이다. 개발용 명령(`serve`, `config print/check`)의 root persistent `--config` 기본값(`configs/node.yaml`)과는 다르다. + +현재 구현된 node CLI 표면은 다음과 같다. + +```text +iop-node serve +iop-node setup +iop-node config print +iop-node config check +iop-node version +``` + 원격 edge에 붙는 수동 테스트는 `configs/node.yaml`의 `transport.edge_addr`와 `transport.token`을 먼저 맞춘 뒤 repo root에서 실행한다. ```bash diff --git a/apps/node/cmd/node/main.go b/apps/node/cmd/node/main.go index fa99a8d..f2eb2e0 100644 --- a/apps/node/cmd/node/main.go +++ b/apps/node/cmd/node/main.go @@ -10,6 +10,7 @@ import ( "iop/apps/node/internal/bootstrap" "iop/packages/config" + "iop/packages/hostsetup" "iop/packages/version" ) @@ -29,7 +30,7 @@ func rootCmd() *cobra.Command { } root.PersistentFlags().StringVarP(&cfgFile, "config", "c", "configs/node.yaml", "config file path") - root.AddCommand(serveCmd(), versionCmd(), configCmd()) + root.AddCommand(serveCmd(), versionCmd(), configCmd(), setupCmd()) return root } @@ -54,8 +55,8 @@ func versionCmd() *cobra.Command { return &cobra.Command{ Use: "version", Short: "Print IOP node version", - Run: func(_ *cobra.Command, _ []string) { - fmt.Println(version.Version) + Run: func(cmd *cobra.Command, _ []string) { + fmt.Fprintln(cmd.OutOrStdout(), version.Version) }, } } @@ -65,21 +66,72 @@ func configCmd() *cobra.Command { Use: "config", Short: "Config management commands", } + cfgGroup.AddCommand(configPrintCmd(), configCheckCmd()) + return cfgGroup +} - printCmd := &cobra.Command{ +func configPrintCmd() *cobra.Command { + return &cobra.Command{ Use: "print", Short: "Print the resolved configuration", - RunE: func(_ *cobra.Command, _ []string) error { + RunE: func(cmd *cobra.Command, _ []string) error { cfg, err := config.Load(cfgFile) if err != nil { return fmt.Errorf("load config: %w", err) } - enc := yaml.NewEncoder(os.Stdout) + enc := yaml.NewEncoder(cmd.OutOrStdout()) enc.SetIndent(2) return enc.Encode(cfg) }, } - - cfgGroup.AddCommand(printCmd) - return cfgGroup +} + +func configCheckCmd() *cobra.Command { + return &cobra.Command{ + Use: "check", + Short: "Validate node config can be loaded", + RunE: func(cmd *cobra.Command, _ []string) error { + if _, err := config.Load(cfgFile); err != nil { + return fmt.Errorf("load config: %w", err) + } + fmt.Fprintf(cmd.OutOrStdout(), "OK %s\n", cfgFile) + return nil + }, + } +} + +func setupCmd() *cobra.Command { + var opts hostsetup.SetupOptions + c := &cobra.Command{ + Use: "setup", + Short: "Prepare host environment and systemd unit for iop-node", + RunE: func(cmd *cobra.Command, _ []string) error { + if cmd.Flags().Changed("config") { + opts.ConfigPath = cfgFile + } else { + opts.ConfigPath = "/etc/iop/node.yaml" + } + return hostsetup.Run(cmd.Context(), hostsetup.NodeSpec(), opts, cmd.OutOrStdout()) + }, + } + // setup --help shows /etc/iop/node.yaml as the --config default, not the dev default. + c.SetHelpFunc(func(cmd *cobra.Command, args []string) { + if f := cmd.Root().PersistentFlags().Lookup("config"); f != nil { + prev := f.DefValue + f.DefValue = "/etc/iop/node.yaml" + defer func() { f.DefValue = prev }() + } + cmd.Root().HelpFunc()(cmd, args) + }) + c.Flags().StringVar(&opts.BinaryPath, "binary", "", "path to iop-node binary (defaults to current executable)") + c.Flags().StringVar(&opts.DataDir, "data-dir", "", "data directory (defaults to spec)") + c.Flags().StringVar(&opts.UnitPath, "unit", "", "systemd unit path (defaults to spec)") + c.Flags().StringVar(&opts.User, "user", "", "service user") + c.Flags().StringVar(&opts.Group, "group", "", "service group") + c.Flags().BoolVar(&opts.Enable, "enable", false, "enable the systemd unit") + c.Flags().BoolVar(&opts.Start, "start", false, "start the systemd unit") + c.Flags().BoolVar(&opts.Restart, "restart", false, "restart the systemd unit") + c.Flags().BoolVar(&opts.DryRun, "dry-run", false, "print plan and previews without making changes") + c.Flags().BoolVar(&opts.OverwriteConfig, "overwrite-config", false, "overwrite an existing config file") + return c } diff --git a/apps/node/cmd/node/main_test.go b/apps/node/cmd/node/main_test.go new file mode 100644 index 0000000..181c5d4 --- /dev/null +++ b/apps/node/cmd/node/main_test.go @@ -0,0 +1,166 @@ +package main + +import ( + "bytes" + "path/filepath" + "strings" + "testing" + + "iop/packages/version" +) + +func TestRootCmdIncludesOperationalCommands(t *testing.T) { + root := rootCmd() + want := map[string]bool{ + "serve": false, + "version": false, + "config": false, + "setup": false, + } + for _, c := range root.Commands() { + if _, ok := want[c.Name()]; ok { + want[c.Name()] = true + } + } + for name, ok := range want { + if !ok { + t.Errorf("root command missing %q", name) + } + } + + cfg, _, err := root.Find([]string{"config"}) + if err != nil { + t.Fatalf("find config: %v", err) + } + subs := map[string]bool{"print": false, "check": false} + for _, c := range cfg.Commands() { + if _, ok := subs[c.Name()]; ok { + subs[c.Name()] = true + } + } + for name, ok := range subs { + if !ok { + t.Errorf("config subcommand missing %q", name) + } + } +} + +func TestVersionCmdPrintsVersion(t *testing.T) { + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + root.SetArgs([]string{"version"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v", err) + } + got := strings.TrimSpace(out.String()) + if got != version.Version { + t.Fatalf("version output = %q, want %q", got, version.Version) + } +} + +func TestConfigCheckCmdLoadsNodeConfig(t *testing.T) { + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + root.SetArgs([]string{"config", "check", "--config", "../../../../configs/node.yaml"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v\n%s", err, out.String()) + } + if !strings.Contains(out.String(), "OK") { + t.Fatalf("expected OK in output, got %q", out.String()) + } +} + +func TestConfigPrintCmdPrintsResolvedConfig(t *testing.T) { + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + root.SetArgs([]string{"config", "print", "--config", "../../../../configs/node.yaml"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v\n%s", err, out.String()) + } + if out.Len() == 0 { + t.Fatal("expected non-empty YAML output") + } +} + +func TestSetupDryRunUsesNodeDefaults(t *testing.T) { + cfgFile = "configs/node.yaml" // reset to root default + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + root.SetArgs([]string{"setup", "--dry-run", "--binary", "/usr/local/bin/iop-node"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v\n%s", err, out.String()) + } + s := out.String() + for _, want := range []string{ + "/etc/iop/node.yaml", + "/var/lib/iop/node", + "/etc/systemd/system/iop-node.service", + "/usr/local/bin/iop-node serve --config /etc/iop/node.yaml", + "--- unit preview ---", + "--- config preview ---", + } { + if !strings.Contains(s, want) { + t.Errorf("dry-run output missing %q\n---\n%s", want, s) + } + } + if strings.Contains(s, "configs/node.yaml") { + t.Errorf("setup default leaked root dev config path: %s", s) + } +} + +func TestSetupDryRunAcceptsExplicitConfig(t *testing.T) { + cfgFile = "configs/node.yaml" // reset + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + custom := filepath.Join(t.TempDir(), "node.yaml") + root.SetArgs([]string{"--config", custom, "setup", "--dry-run", "--binary", "/usr/local/bin/iop-node"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v\n%s", err, out.String()) + } + if !strings.Contains(out.String(), custom) { + t.Errorf("explicit config not used; output:\n%s", out.String()) + } +} + +func TestSetupDryRunConfigSubcmdPosition(t *testing.T) { + cfgFile = "configs/node.yaml" // reset + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + custom := filepath.Join(t.TempDir(), "node.yaml") + root.SetArgs([]string{"setup", "--config", custom, "--dry-run", "--binary", "/usr/local/bin/iop-node"}) + if err := root.Execute(); err != nil { + t.Fatalf("execute: %v\n%s", err, out.String()) + } + if !strings.Contains(out.String(), custom) { + t.Errorf("subcommand-position --config not used; output:\n%s", out.String()) + } +} + +func TestSetupHelpShowsNodeDefault(t *testing.T) { + cfgFile = "configs/node.yaml" // reset + root := rootCmd() + var out bytes.Buffer + root.SetOut(&out) + root.SetErr(&out) + root.SetArgs([]string{"setup", "--help"}) + _ = root.Execute() + s := out.String() + if strings.Contains(s, `"configs/node.yaml"`) { + t.Errorf("setup --help must not show dev default configs/node.yaml; output:\n%s", s) + } + if !strings.Contains(s, "/etc/iop/node.yaml") { + t.Errorf("setup --help must show /etc/iop/node.yaml; output:\n%s", s) + } +} diff --git a/apps/node/internal/bootstrap/module_test.go b/apps/node/internal/bootstrap/module_test.go index 4b22255..8821795 100644 --- a/apps/node/internal/bootstrap/module_test.go +++ b/apps/node/internal/bootstrap/module_test.go @@ -9,7 +9,7 @@ import ( "testing" "time" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "go.uber.org/fx" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/structpb" diff --git a/apps/node/internal/transport/client.go b/apps/node/internal/transport/client.go index fdc65e2..0852fb3 100644 --- a/apps/node/internal/transport/client.go +++ b/apps/node/internal/transport/client.go @@ -7,7 +7,7 @@ import ( "strconv" "time" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "go.uber.org/zap" iop "iop/proto/gen/iop" diff --git a/apps/node/internal/transport/heartbeat_test.go b/apps/node/internal/transport/heartbeat_test.go index 6e0808e..91418a3 100644 --- a/apps/node/internal/transport/heartbeat_test.go +++ b/apps/node/internal/transport/heartbeat_test.go @@ -8,7 +8,7 @@ import ( "testing" "time" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "google.golang.org/protobuf/proto" iop "iop/proto/gen/iop" diff --git a/apps/node/internal/transport/integration_test.go b/apps/node/internal/transport/integration_test.go index 3eb8de9..3465ac1 100644 --- a/apps/node/internal/transport/integration_test.go +++ b/apps/node/internal/transport/integration_test.go @@ -10,7 +10,7 @@ import ( "go.uber.org/zap" "google.golang.org/protobuf/proto" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "iop/apps/node/internal/transport" eventpkg "iop/packages/events" diff --git a/apps/node/internal/transport/parser.go b/apps/node/internal/transport/parser.go index 13992d1..3dc5944 100644 --- a/apps/node/internal/transport/parser.go +++ b/apps/node/internal/transport/parser.go @@ -1,7 +1,7 @@ package transport import ( - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "google.golang.org/protobuf/proto" iop "iop/proto/gen/iop" diff --git a/apps/node/internal/transport/parser_test.go b/apps/node/internal/transport/parser_test.go index 7bd8a50..54aa329 100644 --- a/apps/node/internal/transport/parser_test.go +++ b/apps/node/internal/transport/parser_test.go @@ -3,7 +3,7 @@ package transport import ( "testing" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "google.golang.org/protobuf/proto" iop "iop/proto/gen/iop" diff --git a/apps/node/internal/transport/session.go b/apps/node/internal/transport/session.go index 84e59b8..ba1482b 100644 --- a/apps/node/internal/transport/session.go +++ b/apps/node/internal/transport/session.go @@ -4,7 +4,7 @@ import ( "context" "sync" - toki "git.toki-labs.com/toki/common-proto-socket/go" + toki "git.toki-labs.com/toki/proto-socket/go" "go.uber.org/zap" "google.golang.org/protobuf/proto" diff --git a/configs/edge.yaml b/configs/edge.yaml index 3e92fe9..52b5533 100644 --- a/configs/edge.yaml +++ b/configs/edge.yaml @@ -15,6 +15,16 @@ logging: metrics: port: 9092 +a2a: + enabled: false + listen: "0.0.0.0:8081" + path: "/a2a" + node: "" + adapter: "cli" + session_id: "a2a" + timeout_sec: 120 + bearer_token: "" + openai: enabled: true listen: "0.0.0.0:8080" diff --git a/docs/architecture.md b/docs/architecture.md index 7f2fbc5..1347bc3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -62,6 +62,17 @@ ops console과 HTTP/API를 구분한다: ops console은 edge-local diagnostic su 외부 OpenAI-compatible API에서는 호환성을 위해 `model` 필드가 남을 수 있다. 그러나 내부 실행 계약에서는 `target`을 우선한다. +## 외부 API 표면 + +Edge의 외부 입력 표면은 두 방식을 병행한다. + +- **OpenAI-compatible HTTP API**: 외부 모델 클라이언트, Cline류 도구, 단순 chat completion/inference 호환을 위한 입력이다. 이 경계에서는 `model`을 받을 수 있지만 Edge 내부에서는 `adapter + target`으로 변환한다. +- **A2A JSON-RPC HTTP API**: NomadCode Core나 외부 agent가 작업을 위임하고 `Task` 상태, artifact, cancel/polling을 공유해야 할 때 사용하는 agent delegation 입력이다. 이 경계도 Edge 내부에서는 `adapter + target`으로 변환한다. + +두 입력 방식은 Edge inbound/input surface로 함께 관리하고, 내부 실행은 기존 `RunRequest`/`RunEvent` 흐름으로 수렴시킨다. + +IOP native protocol은 Control Plane, Portal, 운영 CLI, 고급 자동화 클라이언트를 위한 IOP 고유 wire protocol이다. 기존 protobuf-socket 흐름을 기준으로 `RunRequest`, `RunEvent`, `NodeCommandRequest`, `EdgeNodeEvent` 계열 계약을 확장한다. OpenAI-compatible API와 A2A API는 외부 호환 입력으로 유지하지만, node 선택, logical session, background run, terminate-session, capabilities/status/session/transport command, node lifecycle event 같은 운영 기능을 억지로 `model`이나 vendor extension에 싣지 않는다. 이런 기능은 IOP native protocol에서 명시적인 필드와 이벤트로 다룬다. + ## apps/node 내부 구조 ```text @@ -101,6 +112,8 @@ type Adapter interface { - Node는 Edge에 연결한다. - Control Plane은 향후 Edge와 소켓 기반 연결을 맺는다. - Control Plane이 Node에 직접 연결하거나 매 요청마다 Node를 직접 스케줄링하는 구조는 목표가 아니다. +- Portal-Control Plane, Control Plane-Edge, Edge-Node의 장기 통신 기준은 IOP Wire Protocol(protobuf-socket)이다. 브라우저가 직접 TCP를 사용할 수 없는 구간은 Web Portal이 직접 wire client가 되기보다 Control Plane의 server-side bridge를 통해 연결한다. +- OpenAI-compatible HTTP API와 A2A JSON-RPC HTTP API는 Edge 외부 입력 표면으로 유지하며, Control Plane 운영 제어 프로토콜의 기본값으로 삼지 않는다. ## 배포 철학 @@ -108,7 +121,7 @@ Control Plane과 Web은 중앙 운영면으로 묶어 compose 기반 서비스 Edge와 Node는 Docker 이미지로 만들지 않고 호스트 단일 바이너리로 배포한다. Jenkins는 Edge/Node 바이너리만 빌드하고, 각 호스트는 배포된 바이너리의 `setup` 명령을 통해 systemd 실행 환경을 준비한다. -운영 CLI는 분기된 설치 방식을 만들지 않는다. 초기 공식 경로는 `iop-edge setup`과 `iop-node setup` 하나로 고정하고, 검토나 CI 확인은 별도 `render` 명령이 아니라 `--dry-run` 옵션으로 흡수한다. +운영 CLI는 분기된 설치 방식을 만들지 않는다. 공식 경로는 `iop-edge setup`과 `iop-node setup` 하나로 고정하고, 검토나 CI 확인은 별도 `render` 명령이 아니라 `--dry-run` 옵션으로 흡수한다. ```text build: @@ -118,11 +131,11 @@ control host: docker compose -> postgres, redis, control-plane, web edge host: - iop-edge setup -> systemd unit + config/data directories + iop-edge setup -> systemd unit + config/data directories [구현 완료] systemd -> iop-edge serve --config /etc/iop/edge.yaml node host: - iop-node setup -> systemd unit + config/data directories + iop-node setup -> systemd unit + config/data directories [구현 완료] systemd -> iop-node serve --config /etc/iop/node.yaml ``` diff --git a/docs/deploy-dev.md b/docs/deploy-dev.md index f89467c..ddc2207 100644 --- a/docs/deploy-dev.md +++ b/docs/deploy-dev.md @@ -130,29 +130,34 @@ make test-e2e 이 명령은 보조 확인이며, 필드 배포 완료 기준은 실제 edge/node 바이너리 실행 흐름 확인이다. -## 다음 결정 지점 +## Edge setup CLI -### Edge/Node setup CLI - -다음 구현 단계에서는 Edge/Node 바이너리에 `setup` 명령을 추가한다. 공식 운영 경로는 하나로 고정한다. +Edge 바이너리는 호스트 환경 준비를 `setup` 명령으로 일원화한다. 공식 운영 경로는 하나다. ```bash -sudo iop-edge setup --config /etc/iop/edge.yaml --enable --start -sudo iop-node setup --config /etc/iop/node.yaml --enable --start +sudo iop-edge setup --enable --start +``` + +배포 전에 결과를 검토하려면 `--dry-run`을 쓴다. + +```bash +sudo iop-edge setup --dry-run --binary /usr/local/bin/iop-edge ``` `setup`은 다음을 담당한다. -- 실행 user/group 준비 -- `/etc/iop`와 `/var/lib/iop/{edge,node}` 디렉터리 준비 -- 설정 파일이 없을 때만 기본 템플릿 생성 -- systemd unit 생성 또는 갱신 +- 실행 user/group(`iop`) 준비 +- `/etc/iop`와 `/var/lib/iop/edge` 디렉터리 준비 +- 설정 파일이 없을 때만 기본 템플릿 생성 (`--overwrite-config`로 강제 갱신) +- systemd unit 생성 또는 갱신 (`/etc/systemd/system/iop-edge.service`) - `systemctl daemon-reload` -- 옵션에 따른 `enable`, `start`, `restart` +- 옵션에 따른 `--enable`, `--start`, `--restart` + +`setup`의 `--config` 기본값은 `/etc/iop/edge.yaml`이다. dev 단계 명령(`serve`, `console`, `config print/check`)의 root persistent `--config` 기본값(`configs/edge.yaml`)과는 다르다. 별도 `render`, `service install`, `service status` 명령은 초기 범위에 넣지 않는다. 검토와 CI 확인은 `setup --dry-run`으로 흡수하고, 상태/로그/재시작은 `systemctl`과 `journalctl`을 기준 운영 도구로 둔다. -초기 CLI 표면은 다음 정도로 제한한다. +현재 구현된 edge CLI 표면은 다음과 같다. ```text iop-edge serve @@ -161,7 +166,40 @@ iop-edge setup iop-edge config print iop-edge config check iop-edge version +``` +`setup`은 여러 번 실행해도 같은 결과로 수렴한다. 기존 설정 파일은 기본적으로 덮어쓰지 않고, 강제 갱신이 필요하면 `--overwrite-config`를 지정한다. + +## Node setup CLI + +Node 바이너리는 호스트 환경 준비를 `setup` 명령으로 일원화한다. 공식 운영 경로는 하나다. + +```bash +sudo iop-node setup --enable --start +``` + +배포 전에 결과를 검토하려면 `--dry-run`을 쓴다. + +```bash +sudo iop-node setup --dry-run --binary /usr/local/bin/iop-node +``` + +`setup`은 다음을 담당한다. + +- 실행 user/group(`iop`) 준비 +- `/etc/iop`와 `/var/lib/iop/node` 디렉터리 준비 +- 설정 파일이 없을 때만 기본 템플릿 생성 (`--overwrite-config`로 강제 갱신) +- systemd unit 생성 또는 갱신 (`/etc/systemd/system/iop-node.service`) +- `systemctl daemon-reload` +- 옵션에 따른 `--enable`, `--start`, `--restart` + +`setup`의 `--config` 기본값은 `/etc/iop/node.yaml`이다. dev 단계 명령(`serve`, `config print/check`)의 root persistent `--config` 기본값(`configs/node.yaml`)과는 다르다. + +별도 `render`, `service install`, `service status` 명령은 초기 범위에 넣지 않는다. 검토와 CI 확인은 `setup --dry-run`으로 흡수하고, 상태/로그/재시작은 `systemctl`과 `journalctl`을 기준 운영 도구로 둔다. + +현재 구현된 node CLI 표면은 다음과 같다. + +```text iop-node serve iop-node setup iop-node config print @@ -169,4 +207,4 @@ iop-node config check iop-node version ``` -구현 시 `setup`은 여러 번 실행해도 같은 결과로 수렴해야 한다. 기존 설정 파일은 기본적으로 덮어쓰지 않고, 강제 갱신이 필요하면 명시 옵션을 별도로 둔다. +`setup`은 여러 번 실행해도 같은 결과로 수렴한다. 기존 설정 파일은 기본적으로 덮어쓰지 않고, 강제 갱신이 필요하면 `--overwrite-config`를 지정한다. diff --git a/go.mod b/go.mod index 394e047..a4fd42e 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module iop go 1.24 require ( - git.toki-labs.com/toki/common-proto-socket/go v0.0.0-00010101000000-000000000000 + git.toki-labs.com/toki/proto-socket/go v0.0.0-00010101000000-000000000000 github.com/creack/pty v1.1.24 github.com/google/uuid v1.6.0 github.com/prometheus/client_golang v1.20.5 @@ -57,4 +57,4 @@ require ( nhooyr.io/websocket v1.8.17 // indirect ) -replace git.toki-labs.com/toki/common-proto-socket/go => ../proto-socket/go +replace git.toki-labs.com/toki/proto-socket/go => ../proto-socket/go diff --git a/go.work b/go.work new file mode 100644 index 0000000..f435cd1 --- /dev/null +++ b/go.work @@ -0,0 +1,6 @@ +go 1.24 + +use ( + . + ../proto-socket/go +) diff --git a/go.work.sum b/go.work.sum new file mode 100644 index 0000000..1299682 --- /dev/null +++ b/go.work.sum @@ -0,0 +1,73 @@ +cloud.google.com/go v0.112.1/go.mod h1:+Vbu+Y1UU+I1rjmzeMOb/8RfkKJK2Gyxi1X6jJCZLo4= +cloud.google.com/go/compute v1.24.0/go.mod h1:kw1/T+h/+tK2LJK0wiPPx1intgdAM3j/g3hFDlscY40= +cloud.google.com/go/compute/metadata v0.2.3/go.mod h1:VAV5nSsACxMJvgaAuX6Pk2AawlZn8kiOGuCv6gTkwuA= +cloud.google.com/go/firestore v1.15.0/go.mod h1:GWOxFXcv8GZUtYpWHw/w6IuYNux/BtmeVTMmjrm4yhk= +cloud.google.com/go/iam v1.1.5/go.mod h1:rB6P/Ic3mykPbFio+vo7403drjlgvoWfYpJhMXEbzv8= +cloud.google.com/go/longrunning v0.5.5/go.mod h1:WV2LAxD8/rg5Z1cNW6FJ/ZpX4E4VnDnoTk0yawPBB7s= +cloud.google.com/go/storage v1.35.1/go.mod h1:M6M/3V/D3KpzMTJyPOR/HU6n2Si5QdaXYEsng2xgOs8= +github.com/alecthomas/kingpin/v2 v2.4.0/go.mod h1:0gyi0zQnjuFk8xrkNKamJoyUo382HRL7ATRpFZCw6tE= +github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE= +github.com/armon/go-metrics v0.4.1/go.mod h1:E6amYzXo6aW1tqzoZGT755KkbgrJsSdpwZ+3JqfkOG4= +github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk= +github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc= +github.com/fatih/color v1.14.1/go.mod h1:2oHN61fhTpgcxD3TSWCgKDiH1+x4OiDVVGH8WlgGZGg= +github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/go-kit/log v0.2.1/go.mod h1:NwTd00d/i8cPZ3xOwwiv2PO5MOcx78fFErGNcVmBjv0= +github.com/go-logfmt/logfmt v0.5.1/go.mod h1:WYhtIu8zTZfxdn5+rREduYbwxfcBr/Vr6KEVveWlfTs= +github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= +github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/google/s2a-go v0.1.7/go.mod h1:50CgR4k1jNlWBu4UfS4AcfhVe1r6pdZPygJ3R8F0Qdw= +github.com/googleapis/enterprise-certificate-proxy v0.3.2/go.mod h1:VLSiSSBs/ksPL8kq3OBOQ6WRI2QnaFynd1DCjZ62+V0= +github.com/googleapis/gax-go/v2 v2.12.3/go.mod h1:AKloxT6GtNbaLm8QTNSidHUVsHYcBHwWRvkNFJUQcS4= +github.com/googleapis/google-cloud-go-testing v0.0.0-20210719221736-1c9a4c676720/go.mod h1:dvDLG8qkwmyD9a/MJJN3XJcT3xFxOKAvTZGvuZmac9g= +github.com/hashicorp/consul/api v1.28.2/go.mod h1:KyzqzgMEya+IZPcD65YFoOVAgPpbfERu4I/tzG6/ueE= +github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= +github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48= +github.com/hashicorp/go-hclog v1.5.0/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= +github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60= +github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM= +github.com/hashicorp/go-rootcerts v1.0.2/go.mod h1:pqUvnprVnM5bf7AOirdbb01K4ccR319Vf4pU3K5EGc8= +github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= +github.com/hashicorp/serf v0.10.1/go.mod h1:yL2t6BqATOLGc5HF7qbFkTfXoPIY0WZdWHfEvMqbG+4= +github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM= +github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg= +github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= +github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= +github.com/nats-io/nats.go v1.34.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8= +github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pkg/sftp v1.13.6/go.mod h1:tz1ryNURKu77RL+GuCzmoJYxQczL3wLNNpPWagdg4Qk= +github.com/sagikazarmark/crypt v0.19.0/go.mod h1:c6vimRziqqERhtSe0MhIvzE1w54FrCHtrXb5NH/ja78= +github.com/xhit/go-str2duration/v2 v2.1.0/go.mod h1:ohY8p+0f07DiV6Em5LKB0s2YpLtXVyJfNt1+BlmyAsU= +go.etcd.io/etcd/api/v3 v3.5.12/go.mod h1:Ot+o0SWSyT6uHhA56al1oCED0JImsRiU9Dc26+C2a+4= +go.etcd.io/etcd/client/pkg/v3 v3.5.12/go.mod h1:seTzl2d9APP8R5Y2hFL3NVlD6qC/dOT+3kvrqPyTas4= +go.etcd.io/etcd/client/v2 v2.305.12/go.mod h1:aQ/yhsxMu+Oht1FOupSr60oBvcS9cKXHrzBpDsPTf9E= +go.etcd.io/etcd/client/v3 v3.5.12/go.mod h1:tSbBCakoWmmddL+BKVAJHa9km+O/E+bumDe9mSbPiqw= +go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0/go.mod h1:Mjt1i1INqiaoZOMGR1RIUJN+i3ChKoFRqzrRQhlkbs0= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0/go.mod h1:p8pYQP+m5XfbZm9fxtSKAbM6oIllS7s2AfxrChvc7iw= +go.opentelemetry.io/otel v1.24.0/go.mod h1:W7b9Ozg4nkF5tWI5zsXkaKKDjdVjpD4oAt9Qi/MArHo= +go.opentelemetry.io/otel/metric v1.24.0/go.mod h1:VYhLe1rFfxuTXLgj4CBiyz+9WYBA8pNGJgDcSFRKBco= +go.opentelemetry.io/otel/trace v1.24.0/go.mod h1:HPc3Xr/cOApsBI154IU0OI0HJexz+aw5uPdbs3UCjNU= +go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs= +golang.org/x/net v0.26.0/go.mod h1:5YKkiSynbBIh3p6iOc/vibscux0x38BZDkn8sCUPxHE= +golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI= +golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= +golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8= +google.golang.org/api v0.171.0/go.mod h1:Hnq5AHm4OTMt2BUVjael2CWZFD6vksJdWCWiUAmjC9o= +google.golang.org/appengine v1.6.8/go.mod h1:1jJ3jBArFh5pcgW8gCtRJnepW8FzD1V44FJffLiz/Ds= +google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9/go.mod h1:mqHbVIp48Muh7Ywss/AD6I5kNVKZMmAa/QEW58Gxp2s= +google.golang.org/genproto/googleapis/api v0.0.0-20240311132316-a219d84964c2/go.mod h1:O1cOfN1Cy6QEYr7VxtjOyP5AdAuR0aJ/MYZaaof623Y= +google.golang.org/genproto/googleapis/rpc v0.0.0-20240314234333-6e1732d8331c/go.mod h1:WtryC6hu0hhx87FDGxWCDptyssuo68sk10vYjF+T9fY= +google.golang.org/grpc v1.62.1/go.mod h1:IWTG0VlJLCh1SkC58F7np9ka9mx/WNkjl4PGJaiq+QE= +gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= diff --git a/packages/config/config.go b/packages/config/config.go index 2ee87a2..da65997 100644 --- a/packages/config/config.go +++ b/packages/config/config.go @@ -14,6 +14,7 @@ type EdgeConfig struct { Edge EdgeInfo `mapstructure:"edge" yaml:"edge"` Server EdgeServerConf `mapstructure:"server" yaml:"server"` OpenAI EdgeOpenAIConf `mapstructure:"openai" yaml:"openai"` + A2A EdgeA2AConf `mapstructure:"a2a" yaml:"a2a"` TLS TLSConf `mapstructure:"tls" yaml:"tls"` Logging LoggingConf `mapstructure:"logging" yaml:"logging"` Metrics MetricsConf `mapstructure:"metrics" yaml:"metrics"` @@ -57,6 +58,18 @@ type EdgeOpenAIConf struct { TimeoutSec int `mapstructure:"timeout_sec" yaml:"timeout_sec"` } +type EdgeA2AConf struct { + Enabled bool `mapstructure:"enabled" yaml:"enabled"` + Listen string `mapstructure:"listen" yaml:"listen"` + Path string `mapstructure:"path" yaml:"path"` + NodeRef string `mapstructure:"node" yaml:"node"` + Adapter string `mapstructure:"adapter" yaml:"adapter"` + Target string `mapstructure:"target" yaml:"target"` + SessionID string `mapstructure:"session_id" yaml:"session_id"` + TimeoutSec int `mapstructure:"timeout_sec" yaml:"timeout_sec"` + BearerToken string `mapstructure:"bearer_token" yaml:"bearer_token"` +} + type EdgeConsoleConf struct { Adapter string `mapstructure:"adapter" yaml:"adapter"` Target string `mapstructure:"target" yaml:"target"` @@ -205,6 +218,12 @@ func setEdgeDefaults(v *viper.Viper) { v.SetDefault("openai.adapter", "ollama") v.SetDefault("openai.session_id", "openai") v.SetDefault("openai.timeout_sec", 120) + v.SetDefault("a2a.enabled", false) + v.SetDefault("a2a.listen", "0.0.0.0:8081") + v.SetDefault("a2a.path", "/a2a") + v.SetDefault("a2a.adapter", "cli") + v.SetDefault("a2a.session_id", "a2a") + v.SetDefault("a2a.timeout_sec", 120) v.SetDefault("logging.level", "info") v.SetDefault("metrics.port", 9092) v.SetDefault("tls.enabled", false) diff --git a/packages/config/config_test.go b/packages/config/config_test.go index 9aa6dcd..6a17e11 100644 --- a/packages/config/config_test.go +++ b/packages/config/config_test.go @@ -587,6 +587,89 @@ nodes: } } +func TestLoadEdge_A2ADefaults(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "edge.yaml") + if err := os.WriteFile(f, []byte("server:\n listen: \"0.0.0.0:9090\"\n"), 0o600); err != nil { + t.Fatalf("write yaml: %v", err) + } + cfg, err := config.LoadEdge(f) + if err != nil { + t.Fatalf("load: %v", err) + } + if cfg.A2A.Enabled { + t.Fatal("expected a2a.enabled=false by default") + } + if cfg.A2A.Listen != "0.0.0.0:8081" { + t.Fatalf("expected a2a.listen default, got %q", cfg.A2A.Listen) + } + if cfg.A2A.Path != "/a2a" { + t.Fatalf("expected a2a.path=%q, got %q", "/a2a", cfg.A2A.Path) + } + if cfg.A2A.Adapter != "cli" { + t.Fatalf("expected a2a.adapter=%q, got %q", "cli", cfg.A2A.Adapter) + } + if cfg.A2A.SessionID != "a2a" { + t.Fatalf("expected a2a.session_id=%q, got %q", "a2a", cfg.A2A.SessionID) + } + if cfg.A2A.TimeoutSec != 120 { + t.Fatalf("expected a2a.timeout_sec=120, got %d", cfg.A2A.TimeoutSec) + } + if cfg.A2A.BearerToken != "" { + t.Fatalf("expected a2a.bearer_token empty by default, got %q", cfg.A2A.BearerToken) + } +} + +func TestLoadEdge_A2AOverride(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "edge.yaml") + yaml := ` +server: + listen: "0.0.0.0:9090" +a2a: + enabled: true + listen: "127.0.0.1:8082" + path: "/agent" + node: "node0" + adapter: "cli" + target: "claude" + session_id: "nomad" + timeout_sec: 60 + bearer_token: "secret-token" +` + if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { + t.Fatalf("write yaml: %v", err) + } + cfg, err := config.LoadEdge(f) + if err != nil { + t.Fatalf("load: %v", err) + } + if !cfg.A2A.Enabled { + t.Fatal("expected a2a.enabled=true") + } + if cfg.A2A.Listen != "127.0.0.1:8082" { + t.Fatalf("expected a2a.listen=%q, got %q", "127.0.0.1:8082", cfg.A2A.Listen) + } + if cfg.A2A.Path != "/agent" { + t.Fatalf("expected a2a.path=%q, got %q", "/agent", cfg.A2A.Path) + } + if cfg.A2A.NodeRef != "node0" { + t.Fatalf("expected a2a.node=%q, got %q", "node0", cfg.A2A.NodeRef) + } + if cfg.A2A.Target != "claude" { + t.Fatalf("expected a2a.target=%q, got %q", "claude", cfg.A2A.Target) + } + if cfg.A2A.SessionID != "nomad" { + t.Fatalf("expected a2a.session_id=%q, got %q", "nomad", cfg.A2A.SessionID) + } + if cfg.A2A.TimeoutSec != 60 { + t.Fatalf("expected a2a.timeout_sec=60, got %d", cfg.A2A.TimeoutSec) + } + if cfg.A2A.BearerToken != "secret-token" { + t.Fatalf("expected a2a.bearer_token=%q, got %q", "secret-token", cfg.A2A.BearerToken) + } +} + func TestCompletionMarkerConf_RegexAndLineUsage(t *testing.T) { lineOnly := config.CompletionMarkerConf{Line: "<>"} if !strings.Contains(lineOnly.Line, "<>") { diff --git a/packages/hostsetup/setup.go b/packages/hostsetup/setup.go index 29a6a6f..a9bd72a 100644 --- a/packages/hostsetup/setup.go +++ b/packages/hostsetup/setup.go @@ -56,6 +56,9 @@ func Run(ctx context.Context, spec AppSpec, opts SetupOptions, out io.Writer) er if err := applyDefaults(spec, &opts); err != nil { return err } + if err := validateDataDir(opts.DataDir); err != nil { + return err + } plan := buildPlan(spec, opts) fmt.Fprintln(out, plan) @@ -85,6 +88,9 @@ func Run(ctx context.Context, spec AppSpec, opts SetupOptions, out io.Writer) er if err := writeUnit(opts, unit); err != nil { return err } + if err := chownServicePaths(ctx, opts); err != nil { + return err + } if _, err := opts.Runner.Run(ctx, "systemctl", "daemon-reload"); err != nil { return fmt.Errorf("systemctl daemon-reload: %w", err) @@ -181,6 +187,50 @@ func ensureUserGroup(ctx context.Context, opts SetupOptions, out io.Writer) erro return nil } +var unsafeDataDirs = map[string]bool{ + "/": true, + "/etc": true, + "/home": true, + "/opt": true, + "/root": true, + "/srv": true, + "/tmp": true, + "/usr": true, + "/usr/lib": true, + "/usr/local": true, + "/var": true, + "/var/lib": true, + "/var/log": true, + "/var/run": true, +} + +func ensureTraversable(path string) error { + info, err := os.Stat(path) + if err != nil { + return fmt.Errorf("stat %s: %w", path, err) + } + cur := info.Mode().Perm() + wanted := cur | 0o011 // g+x, o+x — minimum to traverse for service user + if wanted == cur { + return nil + } + if err := os.Chmod(path, wanted); err != nil { + return fmt.Errorf("chmod %s: %w", path, err) + } + return nil +} + +func validateDataDir(p string) error { + if p == "" { + return nil + } + cleaned := filepath.Clean(p) + if unsafeDataDirs[cleaned] { + return fmt.Errorf("hostsetup: data dir %q is not an app-specific path; refuse to recursively chown", cleaned) + } + return nil +} + func ensureDirs(opts SetupOptions) error { if dir := filepath.Dir(opts.ConfigPath); dir != "" && dir != "." { if err := os.MkdirAll(dir, 0o755); err != nil { @@ -188,6 +238,14 @@ func ensureDirs(opts SetupOptions) error { } } if opts.DataDir != "" { + if parent := filepath.Dir(opts.DataDir); parent != "" && parent != "." && parent != "/" { + if err := os.MkdirAll(parent, 0o755); err != nil { + return fmt.Errorf("mkdir %s: %w", parent, err) + } + if err := ensureTraversable(parent); err != nil { + return err + } + } if err := os.MkdirAll(opts.DataDir, 0o750); err != nil { return fmt.Errorf("mkdir %s: %w", opts.DataDir, err) } @@ -219,3 +277,18 @@ func ensureConfig(spec AppSpec, opts SetupOptions, out io.Writer) error { func writeUnit(opts SetupOptions, unit string) error { return os.WriteFile(opts.UnitPath, []byte(unit), 0o644) } + +func chownServicePaths(ctx context.Context, opts SetupOptions) error { + owner := opts.User + ":" + opts.Group + if opts.DataDir != "" { + if _, err := opts.Runner.Run(ctx, "chown", "-R", owner, opts.DataDir); err != nil { + return fmt.Errorf("chown %s %s: %w", owner, opts.DataDir, err) + } + } + if opts.ConfigPath != "" { + if _, err := opts.Runner.Run(ctx, "chown", owner, opts.ConfigPath); err != nil { + return fmt.Errorf("chown %s %s: %w", owner, opts.ConfigPath, err) + } + } + return nil +} diff --git a/packages/hostsetup/setup_test.go b/packages/hostsetup/setup_test.go index b1281cd..cba944c 100644 --- a/packages/hostsetup/setup_test.go +++ b/packages/hostsetup/setup_test.go @@ -167,6 +167,12 @@ func TestRunWritesMissingConfigAndUnit(t *testing.T) { if !runner.called("useradd", "--system", "--gid", "iop", "--home-dir", spec.DefaultDataDir, "--shell", "/usr/sbin/nologin", "iop") { t.Fatalf("useradd not called: %v", runner.calls) } + if !runner.called("chown", "-R", "iop:iop", spec.DefaultDataDir) { + t.Fatalf("chown data dir not called: %v", runner.calls) + } + if !runner.called("chown", "iop:iop", spec.DefaultConfig) { + t.Fatalf("chown config not called: %v", runner.calls) + } if !runner.called("systemctl", "daemon-reload") { t.Fatalf("daemon-reload not called") } @@ -219,6 +225,103 @@ func TestRunDoesNotOverwriteExistingConfig(t *testing.T) { } } +func TestRunCreatesTraversableParentForNestedDataDir(t *testing.T) { + dir := t.TempDir() + spec := testSpec(dir) + // Mimic real Edge/Node layout: leaf nested under a shared parent. + spec.DefaultDataDir = filepath.Join(dir, "var", "lib", "iop", "edge") + runner := newFakeRunner() + runner.missing["getent group iop"] = true + runner.missing["id -u iop"] = true + opts := baseOpts(runner) + + var out bytes.Buffer + if err := Run(context.Background(), spec, opts, &out); err != nil { + t.Fatalf("Run: %v", err) + } + + parent := filepath.Join(dir, "var", "lib", "iop") + info, err := os.Stat(parent) + if err != nil { + t.Fatalf("stat parent: %v", err) + } + // World-traversable (at least the +x bit for other) so the iop service user + // can traverse the shared parent into its leaf data dir. + if info.Mode().Perm()&0o001 == 0 { + t.Fatalf("parent %s not traversable by others: mode=%o", parent, info.Mode().Perm()) + } + + leaf, err := os.Stat(spec.DefaultDataDir) + if err != nil { + t.Fatalf("stat leaf: %v", err) + } + if leaf.Mode().Perm() != 0o750 { + t.Fatalf("leaf data dir mode: want 0750 got %o", leaf.Mode().Perm()) + } + + if !runner.called("chown", "-R", "iop:iop", spec.DefaultDataDir) { + t.Fatalf("chown -R should target leaf data dir, calls=%v", runner.calls) + } +} + +func TestRunFixesExistingNonTraversableParent(t *testing.T) { + dir := t.TempDir() + spec := testSpec(dir) + spec.DefaultDataDir = filepath.Join(dir, "var", "lib", "iop", "edge") + + parent := filepath.Join(dir, "var", "lib", "iop") + if err := os.MkdirAll(parent, 0o700); err != nil { + t.Fatal(err) + } + if err := os.Chmod(parent, 0o750); err != nil { + t.Fatal(err) + } + + runner := newFakeRunner() + runner.missing["getent group iop"] = true + runner.missing["id -u iop"] = true + opts := baseOpts(runner) + + if err := Run(context.Background(), spec, opts, &bytes.Buffer{}); err != nil { + t.Fatalf("Run: %v", err) + } + + info, err := os.Stat(parent) + if err != nil { + t.Fatalf("stat parent: %v", err) + } + if info.Mode().Perm()&0o001 == 0 { + t.Fatalf("parent %s still not traversable: mode=%o", parent, info.Mode().Perm()) + } +} + +func TestRunRejectsUnsafeDataDir(t *testing.T) { + for _, bad := range []string{"/", "/var", "/var/lib", "/etc", "/usr", "/usr/lib", "/var/log"} { + t.Run(bad, func(t *testing.T) { + spec := AppSpec{ + Name: "iop-test", + DefaultConfig: "/tmp/does-not-matter.yaml", + DefaultDataDir: bad, + DefaultUnit: "/tmp/does-not-matter.service", + ConfigTemplate: "", + } + runner := newFakeRunner() + opts := baseOpts(runner) + opts.DryRun = true // ensure we don't try to write to those paths anyway + err := Run(context.Background(), spec, opts, &bytes.Buffer{}) + if err == nil { + t.Fatalf("expected error for unsafe data dir %q", bad) + } + if !strings.Contains(err.Error(), "not an app-specific path") { + t.Fatalf("unexpected error: %v", err) + } + if len(runner.calls) != 0 { + t.Fatalf("runner should not be invoked for unsafe data dir, got %v", runner.calls) + } + }) + } +} + func TestRenderUnit(t *testing.T) { spec := AppSpec{Name: "iop-edge", Description: "IOP Edge"} opts := SetupOptions{