diff --git a/README.md b/README.md index fff5e8c..4858144 100644 --- a/README.md +++ b/README.md @@ -108,6 +108,7 @@ Edge의 핵심 역할은 다음과 같다. - Node configuration - Adapter/Profile configuration - Runtime routing +- Edge service API surface - Job assignment - Stream relay - Session handling @@ -117,6 +118,8 @@ Edge의 핵심 역할은 다음과 같다. Edge는 자신의 데이터를 자체적으로 가진다. Control Plane은 Edge의 데이터를 조회하고 조정하지만, 런타임 데이터의 원본은 Edge다. +현재 edge 내부에서는 console을 최종 사용자 인터페이스가 아니라 임시 디버그 클라이언트로 본다. Console의 `/` 명령은 `apps/edge/internal/service`를 호출하는 얇은 어댑터이며, 향후 HTTP/API handler도 같은 service를 호출하는 방향이다. 실행 이벤트와 node lifecycle 이벤트는 `apps/edge/internal/events` bus를 통해 fanout한다. + ### Node Node는 실제 실행자다. @@ -137,7 +140,7 @@ IOP 내부 실행은 model 중심이 아니라 adapter 중심으로 정리한다 - `node adapter`: Node 안에 등록되어 실제 실행을 담당하는 adapter - `adapter execution`: Node adapter가 수행하는 실행 단위 -현재 코드에는 `RunRequest`, `ExecutionSpec`, `RuntimeEvent`, `NodeCommandRequest`처럼 이 방향을 담기 위한 타입들이 있다. 세부 계약과 schema는 작업을 진행하면서 점진적으로 정리한다. +현재 코드에는 `RunRequest`, `ExecutionSpec`, `RuntimeEvent`, `NodeCommandRequest`, `EdgeNodeEvent`처럼 이 방향을 담기 위한 타입들이 있다. `RunEvent`는 adapter execution stream에, `EdgeNodeEvent`는 node 연결/해제 같은 edge-node lifecycle과 이후 제어/상태성 이벤트에 사용한다. 세부 계약과 schema는 작업을 진행하면서 점진적으로 정리한다. ### Runtime Domain @@ -177,7 +180,9 @@ NomadCode - Edge-Node 소켓 기반 구조를 우선 검증 중이다. - Node 등록, 설정 전달, 실행 요청, 스트리밍 이벤트 흐름이 점진적으로 정리되고 있다. +- Edge 내부에는 API 전환을 고려한 `apps/edge/internal/service`와 in-process event fanout인 `apps/edge/internal/events`가 있다. - CLI adapter 쪽 구현이 먼저 진행되고 있다. +- edge console의 `/` 명령은 임시 수동 테스트 UI이며, 장기 인터페이스는 별도 HTTP/API 표면으로 추가한다. - 현재 실행 이력은 Node local SQLite store에서 검증 중이다. Edge 단위 이력 집계와 로컬 실행 그룹 상태 소유권은 로드맵에 따라 정리한다. - `mock` adapter와 dummy/TODO 구현은 개발 단계에서 정상적인 구성이다. - Ollama/vLLM 등 모델 runtime adapter는 단계적으로 확장한다. diff --git a/apps/edge/README.md b/apps/edge/README.md index b5237bc..456d35c 100644 --- a/apps/edge/README.md +++ b/apps/edge/README.md @@ -7,6 +7,20 @@ Edge는 현재 단계에서 OpenAI-compatible HTTP API gateway가 아니라 Node **현재 상태: 초기 구현** node 등록/레지스트리/transport와 edge 콘솔 기반 수동 통신 테스트가 구현되어 있다. +## 내부 경계 + +edge console은 최종 API가 아니라 임시 디버그 클라이언트다. Console의 `/` 명령은 직접 transport를 다루지 않고 `apps/edge/internal/service`를 호출한다. 이후 HTTP/API handler가 추가되면 같은 service를 호출하고, console은 필요하면 수동 테스트 도구로만 남긴다. + +현재 edge 내부 흐름은 다음처럼 나뉜다. + +- `internal/transport`: node TCP/protobuf 연결, register handshake, protobuf message 수신 +- `internal/node`: node registry와 사전 등록 node store +- `internal/service`: `SubmitRun`, `TerminateSession`, `UsageStatus`, `ListNodes`, `ResolveNode` +- `internal/events`: `RunEvent`, `EdgeNodeEvent` in-process publish/subscribe fanout +- `cmd/edge`: serve/console CLI 진입점과 console 출력 포맷 + +실행 요청과 이벤트는 같은 TCP 연결을 공유하지만, 메시지 타입과 내부 파이프라인은 분리한다. `RunRequest`, `CancelRequest`, `NodeCommandRequest`는 edge가 node에 보내는 command/request 계열이고, `RunEvent`, `EdgeNodeEvent`는 node와 edge 내부에서 발생한 event 계열이다. + ## 원격 CLI adapter 수동 테스트 `bin/edge.sh`는 edge 서버를 열고 입력 프롬프트를 제공한다. @@ -60,6 +74,7 @@ edge는 여러 대의 node가 동시에 연결된 환경을 지원한다. - **Single-node Fallback**: 연결된 node가 정확히 1개일 때는 선택 없이도 해당 node가 자동 지정된다. - **Ambiguous Error**: node가 2개 이상일 때 node 선택 없이 메시지를 보내면 에러가 발생하며 선택을 요구한다. - **Node-aware Events**: 모든 이벤트와 메시지 출력에 `[node-{alias}-...]` 접두어가 붙어 출처를 식별할 수 있다. +- **Lifecycle Events**: node 등록/연결 해제는 실행 스트림과 분리된 `EdgeNodeEvent`로 인식하며, 콘솔에서는 `connected`/`disconnected` 이벤트와 reason을 출력한다. ## Transport 1개 · Logical Session 여러 개 diff --git a/apps/edge/cmd/edge/console.go b/apps/edge/cmd/edge/console.go index ece7e6e..765d212 100644 --- a/apps/edge/cmd/edge/console.go +++ b/apps/edge/cmd/edge/console.go @@ -9,11 +9,11 @@ import ( "strings" "time" - toki "git.toki-labs.com/toki/common-proto-socket/go" "go.uber.org/zap" - "google.golang.org/protobuf/types/known/structpb" + edgeevents "iop/apps/edge/internal/events" edgenode "iop/apps/edge/internal/node" + edgeservice "iop/apps/edge/internal/service" "iop/apps/edge/internal/transport" "iop/packages/config" "iop/packages/observability" @@ -37,14 +37,25 @@ func runConsole(ctx context.Context, cfg *config.EdgeConfig, in io.Reader, out i if err != nil { return err } + eventBus := edgeevents.NewBus() + edgeSvc := edgeservice.New(registry, eventBus) events := newConsoleEventRouter(out, registry, logger) - server.SetRunEventHandler(events.Handle) + server.SetRunEventHandler(eventBus.PublishRun) + server.SetNodeEventHandler(eventBus.PublishNode) if err := server.Start(ctx); err != nil { return err } defer func() { _ = server.Stop() }() + consoleCtx, stopConsoleEvents := context.WithCancel(ctx) + defer stopConsoleEvents() + runEvents, unsubscribeRuns := eventBus.SubscribeAllRuns(4096) + defer unsubscribeRuns() + nodeEvents, unsubscribeNodes := eventBus.SubscribeAllNodes(64) + defer unsubscribeNodes() + go events.Drain(consoleCtx, runEvents, nodeEvents) + go func() { if err := observability.ServeMetrics(cfg.Metrics.Port); err != nil { logger.Warn("metrics server exited", zap.Error(err)) @@ -83,14 +94,14 @@ func runConsole(ctx context.Context, cfg *config.EdgeConfig, in io.Reader, out i fmt.Fprintln(out, "bye") return nil case lower == "/nodes": - printNodes(out, registry, target.NodeRef) + printNodes(out, edgeSvc.ListNodes(), target.NodeRef) case strings.HasPrefix(lower, "/node "): parts := strings.Fields(message) if len(parts) < 2 || parts[1] == "" { fmt.Fprintln(out, "usage: /node ") continue } - if _, err := resolveConsoleNode(registry, parts[1]); err != nil { + if _, err := edgeSvc.ResolveNode(parts[1]); err != nil { fmt.Fprintf(out, "error: %v\n", err) continue } @@ -121,13 +132,13 @@ func runConsole(ctx context.Context, cfg *config.EdgeConfig, in io.Reader, out i fmt.Fprintln(out, "usage: /background on|off") } case lower == "/terminate-session": - handleTerminateSession(ctx, registry, out, target) + handleTerminateSession(ctx, edgeSvc, out, target) case lower == "/status": - if err := sendConsoleStatus(ctx, registry, out, target); err != nil { + if err := sendConsoleStatus(ctx, edgeSvc, out, target); err != nil { fmt.Fprintf(out, "error: %v\n", err) } default: - if err := sendConsoleRun(ctx, registry, events, out, target, message); err != nil { + if err := sendConsoleRun(ctx, edgeSvc, events, out, target, message); err != nil { fmt.Fprintf(out, "error: %v\n", err) } } @@ -143,12 +154,7 @@ type consoleTarget struct { TimeoutSec int } -func resolveConsoleNode(registry *edgenode.Registry, nodeRef string) (*edgenode.NodeEntry, error) { - return registry.Resolve(nodeRef) -} - -func printNodes(out io.Writer, registry *edgenode.Registry, selectedRef string) { - nodes := registry.All() +func printNodes(out io.Writer, nodes []*edgenode.NodeEntry, selectedRef string) { if len(nodes) == 0 { fmt.Fprintln(out, "no nodes connected") return @@ -162,68 +168,38 @@ func printNodes(out io.Writer, registry *edgenode.Registry, selectedRef string) } } -// buildRunRequest constructs the RunRequest for a console send. -// RunRequest.target carries the adapter execution target — for the CLI adapter -// this is the selected profile name. -func buildRunRequest(adapter, targetName, sessionID string, background bool, timeoutSec int, message string) (*iop.RunRequest, string, error) { - input, err := structpb.NewStruct(map[string]any{"prompt": message}) - if err != nil { - return nil, "", err - } - if timeoutSec <= 0 { - timeoutSec = 30 - } - runID := fmt.Sprintf("manual-%d", time.Now().UnixNano()) - req := &iop.RunRequest{ - RunId: runID, - Adapter: adapter, - Target: targetName, - SessionId: normalizeConsoleSessionID(sessionID), - SessionMode: iop.RunSessionMode_RUN_SESSION_MODE_CREATE_IF_MISSING, - Background: background, - Input: input, - TimeoutSec: int32(timeoutSec), - Metadata: map[string]string{ - "source": "edge-console", - }, - } - return req, runID, nil -} - -func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events *consoleEventRouter, out io.Writer, target *consoleTarget, message string) error { - entry, err := resolveConsoleNode(registry, target.NodeRef) - if err != nil { - return err - } - - req, runID, err := buildRunRequest(target.Adapter, target.Target, target.SessionID, target.Background, target.TimeoutSec, message) - if err != nil { - return err - } - - var runEvents <-chan *iop.RunEvent - var unregister func() +func sendConsoleRun(ctx context.Context, edgeSvc *edgeservice.Service, events *consoleEventRouter, out io.Writer, target *consoleTarget, message string) error { + runID := edgeservice.NewRunID() + var untrack func() if !target.Background { - runEvents, unregister = events.Register(runID) - defer unregister() + untrack = events.TrackForeground(runID) + defer untrack() } - nodeAlias := entry.Alias - if nodeAlias == "" { - nodeAlias = entry.NodeID - } - fmt.Fprintf(out, "[edge] sent run_id=%s node=%s adapter=%s target=%s session=%s background=%v\n", - runID, nodeAlias, target.Adapter, target.Target, req.GetSessionId(), target.Background) - if err := entry.Client.Send(req); err != nil { + handle, err := edgeSvc.SubmitRun(ctx, edgeservice.SubmitRunRequest{ + NodeRef: target.NodeRef, + RunID: runID, + Adapter: target.Adapter, + Target: target.Target, + SessionID: target.SessionID, + Prompt: message, + Background: target.Background, + TimeoutSec: target.TimeoutSec, + }) + if err != nil { return err } + defer handle.Close() + + fmt.Fprintf(out, "[edge] sent run_id=%s node=%s adapter=%s target=%s session=%s background=%v\n", + handle.RunID, handle.NodeLabel, handle.Adapter, handle.Target, handle.SessionID, handle.Background) if target.Background { fmt.Fprintf(out, "[edge] background run dispatched, events will arrive asynchronously\n") return nil } - timer := time.NewTimer(time.Duration(target.TimeoutSec+5) * time.Second) + timer := time.NewTimer(handle.WaitTimeout()) defer timer.Stop() var response *consoleResponseStream @@ -234,8 +210,12 @@ func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events *co return ctx.Err() case <-timer.C: return fmt.Errorf("timed out waiting for node response") - case event := <-runEvents: - if event.GetRunId() != runID { + case event := <-handle.NodeEvents: + if edgeservice.IsNodeDisconnected(event) { + return fmt.Errorf("node disconnected: %s", event.GetReason()) + } + case event := <-handle.Events: + if event.GetRunId() != handle.RunID { continue } label := events.nodeLabel(event) @@ -245,23 +225,23 @@ func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events *co switch event.GetType() { case "start": - fmt.Fprintf(out, "[node-%s-event] start run_id=%s\n", label, runID) + fmt.Fprintf(out, "[node-%s-event] start run_id=%s\n", label, handle.RunID) case "delta": response.Write(event.GetDelta()) case "complete": response.Finish() - fmt.Fprintf(out, "[node-%s-event] complete run_id=%s detail=%q\n", label, runID, event.GetMessage()) + fmt.Fprintf(out, "[node-%s-event] complete run_id=%s detail=%q\n", label, handle.RunID, event.GetMessage()) return nil case "cancelled": response.FinishIfStarted() - fmt.Fprintf(out, "[node-%s-event] cancelled run_id=%s\n", label, runID) + fmt.Fprintf(out, "[node-%s-event] cancelled run_id=%s\n", label, handle.RunID) return nil case "error": response.FinishIfStarted() - fmt.Fprintf(out, "[node-%s-event] error run_id=%s detail=%q\n", label, runID, event.GetError()) + fmt.Fprintf(out, "[node-%s-event] error run_id=%s detail=%q\n", label, handle.RunID, event.GetError()) return fmt.Errorf("node reported error") default: - fmt.Fprintf(out, "[node-%s-event] %s run_id=%s detail=%q\n", label, event.GetType(), runID, event.GetMessage()) + fmt.Fprintf(out, "[node-%s-event] %s run_id=%s detail=%q\n", label, event.GetType(), handle.RunID, event.GetMessage()) } } } @@ -269,88 +249,64 @@ func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events *co var sendTerminateSessionFunc = sendTerminateSession -func handleTerminateSession(ctx context.Context, registry *edgenode.Registry, out io.Writer, target *consoleTarget) { - if label, err := sendTerminateSessionFunc(ctx, registry, target); err != nil { +func handleTerminateSession(ctx context.Context, edgeSvc *edgeservice.Service, out io.Writer, target *consoleTarget) { + if label, err := sendTerminateSessionFunc(ctx, edgeSvc, target); err != nil { fmt.Fprintf(out, "error: %v\n", err) } else { fmt.Fprintf(out, "terminated session %s node=%s\n", target.SessionID, label) } } -func sendTerminateSession(ctx context.Context, registry *edgenode.Registry, target *consoleTarget) (string, error) { - entry, err := resolveConsoleNode(registry, target.NodeRef) +func sendTerminateSession(ctx context.Context, edgeSvc *edgeservice.Service, target *consoleTarget) (string, error) { + result, err := edgeSvc.TerminateSession(ctx, edgeservice.TerminateSessionRequest{ + NodeRef: target.NodeRef, + Adapter: target.Adapter, + Target: target.Target, + SessionID: target.SessionID, + }) if err != nil { return "", err } - req := &iop.CancelRequest{ - Adapter: target.Adapter, - Target: target.Target, - SessionId: normalizeConsoleSessionID(target.SessionID), - Action: iop.CancelAction_CANCEL_ACTION_TERMINATE_SESSION, - } - label := entry.Alias - if label == "" { - label = entry.NodeID - } - return label, entry.Client.Send(req) + return result.NodeLabel, nil } func normalizeConsoleSessionID(id string) string { - if id == "" { - return "default" - } - return id + return edgeservice.NormalizeSessionID(id) } func buildNodeCommandRequest(adapter, targetName, sessionID string, timeoutSec int) (*iop.NodeCommandRequest, string) { - if timeoutSec <= 0 { - timeoutSec = 30 - } - reqID := fmt.Sprintf("status-%d", time.Now().UnixNano()) - req := &iop.NodeCommandRequest{ - RequestId: reqID, - Type: iop.NodeCommandType_NODE_COMMAND_TYPE_USAGE_STATUS, - Adapter: adapter, - Target: targetName, - SessionId: normalizeConsoleSessionID(sessionID), - TimeoutSec: int32(timeoutSec), - } - return req, reqID + req := edgeservice.BuildUsageStatusRequest(adapter, targetName, sessionID, timeoutSec) + return req, req.GetRequestId() } func statusWaitTimeout(req *iop.NodeCommandRequest) time.Duration { - return time.Duration(req.GetTimeoutSec()+5) * time.Second + return edgeservice.StatusWaitTimeout(req) } -func sendConsoleStatus(ctx context.Context, registry *edgenode.Registry, out io.Writer, target *consoleTarget) error { - entry, err := resolveConsoleNode(registry, target.NodeRef) +func sendConsoleStatus(ctx context.Context, edgeSvc *edgeservice.Service, out io.Writer, target *consoleTarget) error { + entry, err := edgeSvc.ResolveNode(target.NodeRef) + if err != nil { + return err + } + nodeLabel := entry.Alias + if nodeLabel == "" { + nodeLabel = entry.NodeID + } + sessionID := edgeservice.NormalizeSessionID(target.SessionID) + fmt.Fprintf(out, "[edge] sent command=status node=%s adapter=%s target=%s session=%s\n", nodeLabel, target.Adapter, target.Target, sessionID) + + result, err := edgeSvc.UsageStatus(ctx, edgeservice.UsageStatusRequest{ + NodeRef: target.NodeRef, + Adapter: target.Adapter, + Target: target.Target, + SessionID: target.SessionID, + TimeoutSec: target.TimeoutSec, + }) if err != nil { return err } - req, _ := buildNodeCommandRequest(target.Adapter, target.Target, target.SessionID, target.TimeoutSec) - - nodeAlias := entry.Alias - if nodeAlias == "" { - nodeAlias = entry.NodeID - } - fmt.Fprintf(out, "[edge] sent command=status node=%s adapter=%s target=%s session=%s\n", nodeAlias, target.Adapter, target.Target, req.GetSessionId()) - - timeout := statusWaitTimeout(req) - resp, err := toki.SendRequestTyped[*iop.NodeCommandRequest, *iop.NodeCommandResponse]( - &entry.Client.Communicator, - req, - timeout, - ) - if err != nil { - return fmt.Errorf("transport error: %w", err) - } - - if resp.GetError() != "" { - return fmt.Errorf("node reported error: %s", resp.GetError()) - } - - formatUsageStatus(out, nodeAlias, target.Target, req.GetSessionId(), resp.GetUsageStatus()) + formatUsageStatus(out, result.NodeLabel, result.Target, result.SessionID, result.UsageStatus) return nil } diff --git a/apps/edge/cmd/edge/console_events.go b/apps/edge/cmd/edge/console_events.go index 362390c..ea207fa 100644 --- a/apps/edge/cmd/edge/console_events.go +++ b/apps/edge/cmd/edge/console_events.go @@ -1,6 +1,7 @@ package main import ( + "context" "fmt" "io" "strings" @@ -9,40 +10,37 @@ import ( "go.uber.org/zap" edgenode "iop/apps/edge/internal/node" + eventpkg "iop/packages/events" iop "iop/proto/gen/iop" ) type consoleEventRouter struct { - mu sync.Mutex - out io.Writer - registry *edgenode.Registry - logger *zap.Logger - waiters map[string]chan *iop.RunEvent - streams map[string]*consoleResponseStream + mu sync.Mutex + out io.Writer + registry *edgenode.Registry + logger *zap.Logger + foreground map[string]struct{} + streams map[string]*consoleResponseStream } func newConsoleEventRouter(out io.Writer, registry *edgenode.Registry, logger *zap.Logger) *consoleEventRouter { return &consoleEventRouter{ - out: out, - registry: registry, - logger: logger, - waiters: make(map[string]chan *iop.RunEvent), - streams: make(map[string]*consoleResponseStream), + out: out, + registry: registry, + logger: logger, + foreground: make(map[string]struct{}), + streams: make(map[string]*consoleResponseStream), } } -func (r *consoleEventRouter) Register(runID string) (<-chan *iop.RunEvent, func()) { - ch := make(chan *iop.RunEvent, 4096) +func (r *consoleEventRouter) TrackForeground(runID string) func() { r.mu.Lock() - r.waiters[runID] = ch + r.foreground[runID] = struct{}{} r.mu.Unlock() - return ch, func() { + return func() { r.mu.Lock() - if current, ok := r.waiters[runID]; ok && current == ch { - delete(r.waiters, runID) - close(ch) - } + delete(r.foreground, runID) r.mu.Unlock() } } @@ -51,14 +49,7 @@ func (r *consoleEventRouter) Handle(event *iop.RunEvent) { runID := event.GetRunId() r.mu.Lock() - if waiter, ok := r.waiters[runID]; ok { - select { - case waiter <- event: - default: - if r.logger != nil { - r.logger.Warn("console run event dropped", zap.String("run_id", runID), zap.String("type", event.GetType())) - } - } + if _, ok := r.foreground[runID]; ok { r.mu.Unlock() return } @@ -66,6 +57,36 @@ func (r *consoleEventRouter) Handle(event *iop.RunEvent) { r.mu.Unlock() } +func (r *consoleEventRouter) HandleNodeEvent(event *iop.EdgeNodeEvent) { + r.mu.Lock() + r.printNodeEventLocked(event) + r.mu.Unlock() +} + +func (r *consoleEventRouter) Drain(ctx context.Context, runEvents <-chan *iop.RunEvent, nodeEvents <-chan *iop.EdgeNodeEvent) { + for { + select { + case <-ctx.Done(): + return + case event, ok := <-runEvents: + if !ok { + runEvents = nil + } else { + r.Handle(event) + } + case event, ok := <-nodeEvents: + if !ok { + nodeEvents = nil + } else { + r.HandleNodeEvent(event) + } + } + if runEvents == nil && nodeEvents == nil { + return + } + } +} + func (r *consoleEventRouter) nodeLabel(event *iop.RunEvent) string { nodeID := event.GetNodeId() if nodeID == "" { @@ -79,6 +100,38 @@ func (r *consoleEventRouter) nodeLabel(event *iop.RunEvent) string { return nodeID } +func (r *consoleEventRouter) edgeNodeLabel(event *iop.EdgeNodeEvent) string { + if event.GetAlias() != "" { + return event.GetAlias() + } + nodeID := event.GetNodeId() + if nodeID == "" { + return "unknown" + } + if r.registry != nil { + if entry, ok := r.registry.Get(nodeID); ok && entry.Alias != "" { + return entry.Alias + } + } + return nodeID +} + +func (r *consoleEventRouter) printNodeEventLocked(event *iop.EdgeNodeEvent) { + if r.out == nil { + return + } + + label := r.edgeNodeLabel(event) + switch event.GetType() { + case eventpkg.TypeNodeConnected: + fmt.Fprintf(r.out, "[node-%s-event] connected reason=%q\n", label, event.GetReason()) + case eventpkg.TypeNodeDisconnected: + fmt.Fprintf(r.out, "[node-%s-event] disconnected reason=%q\n", label, event.GetReason()) + default: + fmt.Fprintf(r.out, "[node-%s-event] %s reason=%q\n", label, event.GetType(), event.GetReason()) + } +} + func (r *consoleEventRouter) printAsyncLocked(event *iop.RunEvent) { if r.out == nil { return diff --git a/apps/edge/cmd/edge/console_test.go b/apps/edge/cmd/edge/console_test.go index 249890f..73526bf 100644 --- a/apps/edge/cmd/edge/console_test.go +++ b/apps/edge/cmd/edge/console_test.go @@ -7,13 +7,21 @@ import ( "testing" "time" - toki "git.toki-labs.com/toki/common-proto-socket/go" edgenode "iop/apps/edge/internal/node" + edgeservice "iop/apps/edge/internal/service" + eventpkg "iop/packages/events" iop "iop/proto/gen/iop" ) func TestBuildRunRequest_SessionAndBackground(t *testing.T) { - req, runID, err := buildRunRequest("cli", "codex", "session-a", true, 30, "hello") + req, runID, err := edgeservice.BuildRunRequest(edgeservice.SubmitRunRequest{ + Adapter: "cli", + Target: "codex", + SessionID: "session-a", + Background: true, + TimeoutSec: 30, + Prompt: "hello", + }) if err != nil { t.Fatalf("buildRunRequest: %v", err) } @@ -38,7 +46,12 @@ func TestBuildRunRequest_SessionAndBackground(t *testing.T) { } func TestBuildRunRequest_EmptySessionNormalized(t *testing.T) { - req, _, err := buildRunRequest("cli", "codex", "", false, 30, "hello") + req, _, err := edgeservice.BuildRunRequest(edgeservice.SubmitRunRequest{ + Adapter: "cli", + Target: "codex", + TimeoutSec: 30, + Prompt: "hello", + }) if err != nil { t.Fatalf("buildRunRequest: %v", err) } @@ -48,7 +61,12 @@ func TestBuildRunRequest_EmptySessionNormalized(t *testing.T) { } func TestBuildRunRequest_DefaultTimeoutFallback(t *testing.T) { - req, _, err := buildRunRequest("cli", "codex", "s1", false, 0, "hello") + req, _, err := edgeservice.BuildRunRequest(edgeservice.SubmitRunRequest{ + Adapter: "cli", + Target: "codex", + SessionID: "s1", + Prompt: "hello", + }) if err != nil { t.Fatalf("buildRunRequest: %v", err) } @@ -66,22 +84,14 @@ func TestNormalizeConsoleSessionID(t *testing.T) { } } -func TestConsoleEventRouterRoutesRegisteredRun(t *testing.T) { +func TestConsoleEventRouterSuppressesTrackedForegroundRun(t *testing.T) { var out bytes.Buffer router := newConsoleEventRouter(&out, nil, nil) - events, unregister := router.Register("run-1") + unregister := router.TrackForeground("run-1") defer unregister() router.Handle(&iop.RunEvent{RunId: "run-1", Type: "start"}) - select { - case event := <-events: - if event.GetRunId() != "run-1" { - t.Fatalf("RunId: got %q want run-1", event.GetRunId()) - } - default: - t.Fatal("expected registered run event") - } if out.Len() != 0 { t.Fatalf("registered foreground event should not be printed asynchronously: %q", out.String()) } @@ -276,14 +286,15 @@ func TestResolveConsoleNode_RequiresExplicitSelectionForMultipleNodes(t *testing reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "node-1"}) reg.Register(&edgenode.NodeEntry{NodeID: "node-2"}) + svc := edgeservice.New(reg, nil) // Empty target should fail for multiple nodes - if _, err := resolveConsoleNode(reg, ""); err == nil { + if _, err := svc.ResolveNode(""); err == nil { t.Error("expected error for implicit resolve with multiple nodes") } // Explicit target should succeed - if e, err := resolveConsoleNode(reg, "node-1"); err != nil || e.NodeID != "node-1" { + if e, err := svc.ResolveNode("node-1"); err != nil || e.NodeID != "node-1" { t.Errorf("failed explicit resolve: %v", err) } } @@ -291,8 +302,9 @@ func TestResolveConsoleNode_RequiresExplicitSelectionForMultipleNodes(t *testing func TestResolveConsoleNode_AllowsSingleNodeFallback(t *testing.T) { reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "node-1"}) + svc := edgeservice.New(reg, nil) - if e, err := resolveConsoleNode(reg, ""); err != nil || e.NodeID != "node-1" { + if e, err := svc.ResolveNode(""); err != nil || e.NodeID != "node-1" { t.Errorf("expected single node fallback, got error: %v", err) } } @@ -303,7 +315,7 @@ func TestPrintNodes_ShowsSelectedNode(t *testing.T) { reg.Register(&edgenode.NodeEntry{NodeID: "node-2", Alias: "alias-2"}) var out bytes.Buffer - printNodes(&out, reg, "alias-1") + printNodes(&out, reg.All(), "alias-1") got := out.String() if !strings.Contains(got, "* node-1 (alias-1)") { @@ -314,53 +326,21 @@ func TestPrintNodes_ShowsSelectedNode(t *testing.T) { } } -func TestSendTerminateSession_ReturnsNodeLabel(t *testing.T) { - reg := edgenode.NewRegistry() - reg.Register(&edgenode.NodeEntry{ - NodeID: "node-1", - Alias: "alias-1", - Client: &toki.TcpClient{}, // mock client - }) - - target := &consoleTarget{ - NodeRef: "alias-1", - SessionID: "s1", - } - - label, _ := sendTerminateSession(context.Background(), reg, target) - if label != "alias-1" { - t.Errorf("expected label alias-1, got %q", label) - } - - // Test fallback to NodeID - reg2 := edgenode.NewRegistry() - reg2.Register(&edgenode.NodeEntry{ - NodeID: "node-2", - Client: &toki.TcpClient{}, - }) - target.NodeRef = "node-2" - label2, _ := sendTerminateSession(context.Background(), reg2, target) - if label2 != "node-2" { - t.Errorf("expected label node-2, got %q", label2) - } -} - func TestHandleTerminateSession_OutputFormat(t *testing.T) { // Mock sendTerminateSessionFunc to avoid network/toki dependency oldFunc := sendTerminateSessionFunc defer func() { sendTerminateSessionFunc = oldFunc }() - sendTerminateSessionFunc = func(ctx context.Context, registry *edgenode.Registry, target *consoleTarget) (string, error) { + sendTerminateSessionFunc = func(ctx context.Context, edgeSvc *edgeservice.Service, target *consoleTarget) (string, error) { return "mock-alias", nil } - reg := edgenode.NewRegistry() target := &consoleTarget{ SessionID: "s-test", } var out bytes.Buffer - handleTerminateSession(context.Background(), reg, &out, target) + handleTerminateSession(context.Background(), nil, &out, target) got := out.String() want := "terminated session s-test node=mock-alias\n" @@ -369,11 +349,11 @@ func TestHandleTerminateSession_OutputFormat(t *testing.T) { } // Test fallback to NodeID in output - sendTerminateSessionFunc = func(ctx context.Context, registry *edgenode.Registry, target *consoleTarget) (string, error) { + sendTerminateSessionFunc = func(ctx context.Context, edgeSvc *edgeservice.Service, target *consoleTarget) (string, error) { return "node-raw-id", nil } out.Reset() - handleTerminateSession(context.Background(), reg, &out, target) + handleTerminateSession(context.Background(), nil, &out, target) got2 := out.String() want2 := "terminated session s-test node=node-raw-id\n" if got2 != want2 { @@ -416,3 +396,20 @@ func TestConsoleEventRouterFallsBackToNodeID(t *testing.T) { t.Errorf("expected node-some-raw-id-event label, got:\n%s", got) } } + +func TestConsoleEventRouterPrintsNodeLifecycleEvents(t *testing.T) { + var out bytes.Buffer + router := newConsoleEventRouter(&out, nil, nil) + + router.HandleNodeEvent(&iop.EdgeNodeEvent{ + Type: eventpkg.TypeNodeDisconnected, + NodeId: "node-1", + Alias: "alias-1", + Reason: eventpkg.ReasonTransportClosed, + }) + + got := out.String() + if !strings.Contains(got, `[node-alias-1-event] disconnected reason="transport_closed"`) { + t.Errorf("expected disconnected lifecycle output, got:\n%s", got) + } +} diff --git a/apps/edge/internal/bootstrap/module.go b/apps/edge/internal/bootstrap/module.go index 906bc26..ba952ae 100644 --- a/apps/edge/internal/bootstrap/module.go +++ b/apps/edge/internal/bootstrap/module.go @@ -7,7 +7,9 @@ import ( "go.uber.org/fx" "go.uber.org/zap" + edgeevents "iop/apps/edge/internal/events" edgenode "iop/apps/edge/internal/node" + edgeservice "iop/apps/edge/internal/service" "iop/apps/edge/internal/transport" "iop/packages/config" "iop/packages/observability" @@ -30,12 +32,20 @@ func Module(cfg *config.EdgeConfig) fx.Option { return edgenode.NewNodeStore() }, + func() *edgeevents.Bus { + return edgeevents.NewBus() + }, + + func(reg *edgenode.Registry, bus *edgeevents.Bus) *edgeservice.Service { + return edgeservice.New(reg, bus) + }, + func(cfg *config.EdgeConfig, reg *edgenode.Registry, ns *edgenode.NodeStore, logger *zap.Logger) (*transport.Server, error) { return transport.NewServer(cfg.Server.Listen, reg, ns, logger) }, ), - fx.Invoke(func(lc fx.Lifecycle, srv *transport.Server, ns *edgenode.NodeStore, cfg *config.EdgeConfig, logger *zap.Logger) { + fx.Invoke(func(lc fx.Lifecycle, srv *transport.Server, bus *edgeevents.Bus, ns *edgenode.NodeStore, cfg *config.EdgeConfig, logger *zap.Logger) { lc.Append(fx.Hook{ OnStart: func(ctx context.Context) error { seeded, err := edgenode.LoadFromConfig(cfg.Nodes) @@ -45,6 +55,8 @@ func Module(cfg *config.EdgeConfig) fx.Option { for _, rec := range seeded.All() { ns.Add(rec) } + srv.SetRunEventHandler(bus.PublishRun) + srv.SetNodeEventHandler(bus.PublishNode) if err := srv.Start(ctx); err != nil { return err } diff --git a/apps/edge/internal/events/bus.go b/apps/edge/internal/events/bus.go new file mode 100644 index 0000000..6b97ec7 --- /dev/null +++ b/apps/edge/internal/events/bus.go @@ -0,0 +1,152 @@ +package events + +import ( + "sync" + + iop "iop/proto/gen/iop" +) + +// Bus is the in-process edge event fanout used by temporary consoles, future +// API handlers, and later event persistence. +type Bus struct { + mu sync.Mutex + runSubs map[string]map[chan *iop.RunEvent]struct{} + nodeSubs map[string]map[chan *iop.EdgeNodeEvent]struct{} + allRuns map[chan *iop.RunEvent]struct{} + allNodes map[chan *iop.EdgeNodeEvent]struct{} +} + +func NewBus() *Bus { + return &Bus{ + runSubs: make(map[string]map[chan *iop.RunEvent]struct{}), + nodeSubs: make(map[string]map[chan *iop.EdgeNodeEvent]struct{}), + allRuns: make(map[chan *iop.RunEvent]struct{}), + allNodes: make(map[chan *iop.EdgeNodeEvent]struct{}), + } +} + +func (b *Bus) SubscribeRun(runID string, buffer int) (<-chan *iop.RunEvent, func()) { + if buffer <= 0 { + buffer = 1 + } + ch := make(chan *iop.RunEvent, buffer) + b.mu.Lock() + if b.runSubs[runID] == nil { + b.runSubs[runID] = make(map[chan *iop.RunEvent]struct{}) + } + b.runSubs[runID][ch] = struct{}{} + b.mu.Unlock() + + return ch, func() { + b.mu.Lock() + if subs := b.runSubs[runID]; subs != nil { + delete(subs, ch) + if len(subs) == 0 { + delete(b.runSubs, runID) + } + } + close(ch) + b.mu.Unlock() + } +} + +func (b *Bus) SubscribeNode(nodeID string, buffer int) (<-chan *iop.EdgeNodeEvent, func()) { + if buffer <= 0 { + buffer = 1 + } + ch := make(chan *iop.EdgeNodeEvent, buffer) + b.mu.Lock() + if b.nodeSubs[nodeID] == nil { + b.nodeSubs[nodeID] = make(map[chan *iop.EdgeNodeEvent]struct{}) + } + b.nodeSubs[nodeID][ch] = struct{}{} + b.mu.Unlock() + + return ch, func() { + b.mu.Lock() + if subs := b.nodeSubs[nodeID]; subs != nil { + delete(subs, ch) + if len(subs) == 0 { + delete(b.nodeSubs, nodeID) + } + } + close(ch) + b.mu.Unlock() + } +} + +func (b *Bus) SubscribeAllRuns(buffer int) (<-chan *iop.RunEvent, func()) { + if buffer <= 0 { + buffer = 1 + } + ch := make(chan *iop.RunEvent, buffer) + b.mu.Lock() + b.allRuns[ch] = struct{}{} + b.mu.Unlock() + + return ch, func() { + b.mu.Lock() + delete(b.allRuns, ch) + close(ch) + b.mu.Unlock() + } +} + +func (b *Bus) SubscribeAllNodes(buffer int) (<-chan *iop.EdgeNodeEvent, func()) { + if buffer <= 0 { + buffer = 1 + } + ch := make(chan *iop.EdgeNodeEvent, buffer) + b.mu.Lock() + b.allNodes[ch] = struct{}{} + b.mu.Unlock() + + return ch, func() { + b.mu.Lock() + delete(b.allNodes, ch) + close(ch) + b.mu.Unlock() + } +} + +func (b *Bus) PublishRun(event *iop.RunEvent) { + if event == nil { + return + } + b.mu.Lock() + defer b.mu.Unlock() + for ch := range b.allRuns { + offerRun(ch, event) + } + for ch := range b.runSubs[event.GetRunId()] { + offerRun(ch, event) + } +} + +func (b *Bus) PublishNode(event *iop.EdgeNodeEvent) { + if event == nil { + return + } + b.mu.Lock() + defer b.mu.Unlock() + for ch := range b.allNodes { + offerNode(ch, event) + } + for ch := range b.nodeSubs[event.GetNodeId()] { + offerNode(ch, event) + } +} + +func offerRun(ch chan *iop.RunEvent, event *iop.RunEvent) { + select { + case ch <- event: + default: + } +} + +func offerNode(ch chan *iop.EdgeNodeEvent, event *iop.EdgeNodeEvent) { + select { + case ch <- event: + default: + } +} diff --git a/apps/edge/internal/events/bus_test.go b/apps/edge/internal/events/bus_test.go new file mode 100644 index 0000000..0ebde21 --- /dev/null +++ b/apps/edge/internal/events/bus_test.go @@ -0,0 +1,62 @@ +package events_test + +import ( + "testing" + + edgeevents "iop/apps/edge/internal/events" + iop "iop/proto/gen/iop" +) + +func TestBusPublishesRunToSpecificAndAllSubscribers(t *testing.T) { + bus := edgeevents.NewBus() + specific, unsubscribeSpecific := bus.SubscribeRun("run-1", 1) + defer unsubscribeSpecific() + all, unsubscribeAll := bus.SubscribeAllRuns(1) + defer unsubscribeAll() + + bus.PublishRun(&iop.RunEvent{RunId: "run-1", Type: "start"}) + + select { + case event := <-specific: + if event.GetRunId() != "run-1" { + t.Fatalf("specific run id: %q", event.GetRunId()) + } + default: + t.Fatal("expected specific run subscriber event") + } + select { + case event := <-all: + if event.GetRunId() != "run-1" { + t.Fatalf("all run id: %q", event.GetRunId()) + } + default: + t.Fatal("expected all run subscriber event") + } +} + +func TestBusPublishesNodeToSpecificAndAllSubscribers(t *testing.T) { + bus := edgeevents.NewBus() + specific, unsubscribeSpecific := bus.SubscribeNode("node-1", 1) + defer unsubscribeSpecific() + all, unsubscribeAll := bus.SubscribeAllNodes(1) + defer unsubscribeAll() + + bus.PublishNode(&iop.EdgeNodeEvent{NodeId: "node-1", Type: "node.disconnected"}) + + select { + case event := <-specific: + if event.GetNodeId() != "node-1" { + t.Fatalf("specific node id: %q", event.GetNodeId()) + } + default: + t.Fatal("expected specific node subscriber event") + } + select { + case event := <-all: + if event.GetNodeId() != "node-1" { + t.Fatalf("all node id: %q", event.GetNodeId()) + } + default: + t.Fatal("expected all node subscriber event") + } +} diff --git a/apps/edge/internal/service/service.go b/apps/edge/internal/service/service.go new file mode 100644 index 0000000..aace338 --- /dev/null +++ b/apps/edge/internal/service/service.go @@ -0,0 +1,282 @@ +package service + +import ( + "context" + "fmt" + "time" + + toki "git.toki-labs.com/toki/common-proto-socket/go" + "google.golang.org/protobuf/types/known/structpb" + + edgeevents "iop/apps/edge/internal/events" + edgenode "iop/apps/edge/internal/node" + eventpkg "iop/packages/events" + iop "iop/proto/gen/iop" +) + +const ( + DefaultSessionID = "default" + DefaultTimeoutSec = 30 +) + +type Service struct { + registry *edgenode.Registry + events *edgeevents.Bus +} + +func New(registry *edgenode.Registry, events *edgeevents.Bus) *Service { + return &Service{registry: registry, events: events} +} + +func (s *Service) ListNodes() []*edgenode.NodeEntry { + return s.registry.All() +} + +func (s *Service) ResolveNode(ref string) (*edgenode.NodeEntry, error) { + return s.registry.Resolve(ref) +} + +type SubmitRunRequest struct { + NodeRef string + RunID string + Adapter string + Target string + SessionID string + Prompt string + Background bool + TimeoutSec int + Metadata map[string]string +} + +type RunHandle struct { + RunID string + NodeID string + NodeLabel string + Adapter string + Target string + SessionID string + Background bool + TimeoutSec int + Events <-chan *iop.RunEvent + NodeEvents <-chan *iop.EdgeNodeEvent + close func() +} + +func (h *RunHandle) Close() { + if h != nil && h.close != nil { + h.close() + } +} + +func (h *RunHandle) WaitTimeout() time.Duration { + if h == nil { + return time.Duration(DefaultTimeoutSec+5) * time.Second + } + return time.Duration(normalizeTimeoutSec(h.TimeoutSec)+5) * time.Second +} + +func (s *Service) SubmitRun(_ context.Context, req SubmitRunRequest) (*RunHandle, error) { + entry, err := s.ResolveNode(req.NodeRef) + if err != nil { + return nil, err + } + + runReq, runID, err := BuildRunRequest(req) + if err != nil { + return nil, err + } + + var runEvents <-chan *iop.RunEvent + var unregisterRun func() + var nodeEvents <-chan *iop.EdgeNodeEvent + var unregisterNode func() + if !runReq.GetBackground() { + if s.events == nil { + return nil, fmt.Errorf("event bus is not configured") + } + runEvents, unregisterRun = s.events.SubscribeRun(runID, 4096) + nodeEvents, unregisterNode = s.events.SubscribeNode(entry.NodeID, 16) + } + + if err := entry.Client.Send(runReq); err != nil { + if unregisterRun != nil { + unregisterRun() + } + if unregisterNode != nil { + unregisterNode() + } + return nil, err + } + + return &RunHandle{ + RunID: runID, + NodeID: entry.NodeID, + NodeLabel: nodeLabel(entry), + Adapter: runReq.GetAdapter(), + Target: runReq.GetTarget(), + SessionID: runReq.GetSessionId(), + Background: runReq.GetBackground(), + TimeoutSec: int(runReq.GetTimeoutSec()), + Events: runEvents, + NodeEvents: nodeEvents, + close: func() { + if unregisterRun != nil { + unregisterRun() + } + if unregisterNode != nil { + unregisterNode() + } + }, + }, nil +} + +type TerminateSessionRequest struct { + NodeRef string + Adapter string + Target string + SessionID string +} + +type TerminateSessionResult struct { + NodeID string + NodeLabel string + SessionID string +} + +func (s *Service) TerminateSession(_ context.Context, req TerminateSessionRequest) (TerminateSessionResult, error) { + entry, err := s.ResolveNode(req.NodeRef) + if err != nil { + return TerminateSessionResult{}, err + } + + sessionID := NormalizeSessionID(req.SessionID) + cancelReq := &iop.CancelRequest{ + Adapter: req.Adapter, + Target: req.Target, + SessionId: sessionID, + Action: iop.CancelAction_CANCEL_ACTION_TERMINATE_SESSION, + } + if err := entry.Client.Send(cancelReq); err != nil { + return TerminateSessionResult{}, err + } + return TerminateSessionResult{ + NodeID: entry.NodeID, + NodeLabel: nodeLabel(entry), + SessionID: sessionID, + }, nil +} + +type UsageStatusRequest struct { + NodeRef string + Adapter string + Target string + SessionID string + TimeoutSec int +} + +type UsageStatusResult struct { + NodeID string + NodeLabel string + Target string + SessionID string + UsageStatus *iop.AgentUsageStatus +} + +func (s *Service) UsageStatus(_ context.Context, req UsageStatusRequest) (UsageStatusResult, error) { + entry, err := s.ResolveNode(req.NodeRef) + if err != nil { + return UsageStatusResult{}, err + } + + commandReq := BuildUsageStatusRequest(req.Adapter, req.Target, req.SessionID, req.TimeoutSec) + resp, err := toki.SendRequestTyped[*iop.NodeCommandRequest, *iop.NodeCommandResponse]( + &entry.Client.Communicator, + commandReq, + StatusWaitTimeout(commandReq), + ) + if err != nil { + return UsageStatusResult{}, fmt.Errorf("transport error: %w", err) + } + if resp.GetError() != "" { + return UsageStatusResult{}, fmt.Errorf("node reported error: %s", resp.GetError()) + } + + return UsageStatusResult{ + NodeID: entry.NodeID, + NodeLabel: nodeLabel(entry), + Target: commandReq.GetTarget(), + SessionID: commandReq.GetSessionId(), + UsageStatus: resp.GetUsageStatus(), + }, nil +} + +func BuildUsageStatusRequest(adapter, targetName, sessionID string, timeoutSec int) *iop.NodeCommandRequest { + reqID := fmt.Sprintf("status-%d", time.Now().UnixNano()) + return &iop.NodeCommandRequest{ + RequestId: reqID, + Type: iop.NodeCommandType_NODE_COMMAND_TYPE_USAGE_STATUS, + Adapter: adapter, + Target: targetName, + SessionId: NormalizeSessionID(sessionID), + TimeoutSec: int32(normalizeTimeoutSec(timeoutSec)), + } +} + +func StatusWaitTimeout(req *iop.NodeCommandRequest) time.Duration { + return time.Duration(normalizeTimeoutSec(int(req.GetTimeoutSec()))+5) * time.Second +} + +func NormalizeSessionID(id string) string { + if id == "" { + return DefaultSessionID + } + return id +} + +func NewRunID() string { + return fmt.Sprintf("manual-%d", time.Now().UnixNano()) +} + +func IsNodeDisconnected(event *iop.EdgeNodeEvent) bool { + return event.GetType() == eventpkg.TypeNodeDisconnected +} + +func BuildRunRequest(req SubmitRunRequest) (*iop.RunRequest, string, error) { + input, err := structpb.NewStruct(map[string]any{"prompt": req.Prompt}) + if err != nil { + return nil, "", err + } + runID := req.RunID + if runID == "" { + runID = NewRunID() + } + metadata := map[string]string{"source": "edge-console"} + for k, v := range req.Metadata { + metadata[k] = v + } + return &iop.RunRequest{ + RunId: runID, + Adapter: req.Adapter, + Target: req.Target, + SessionId: NormalizeSessionID(req.SessionID), + SessionMode: iop.RunSessionMode_RUN_SESSION_MODE_CREATE_IF_MISSING, + Background: req.Background, + Input: input, + TimeoutSec: int32(normalizeTimeoutSec(req.TimeoutSec)), + Metadata: metadata, + }, runID, nil +} + +func normalizeTimeoutSec(timeoutSec int) int { + if timeoutSec <= 0 { + return DefaultTimeoutSec + } + return timeoutSec +} + +func nodeLabel(entry *edgenode.NodeEntry) string { + if entry.Alias != "" { + return entry.Alias + } + return entry.NodeID +} diff --git a/apps/edge/internal/service/service_test.go b/apps/edge/internal/service/service_test.go new file mode 100644 index 0000000..328b62a --- /dev/null +++ b/apps/edge/internal/service/service_test.go @@ -0,0 +1,45 @@ +package service_test + +import ( + "testing" + "time" + + edgeservice "iop/apps/edge/internal/service" + iop "iop/proto/gen/iop" +) + +func TestBuildRunRequestNormalizesSessionAndTimeout(t *testing.T) { + req, runID, err := edgeservice.BuildRunRequest(edgeservice.SubmitRunRequest{ + Adapter: "cli", + Target: "codex", + Prompt: "hello", + }) + if err != nil { + t.Fatalf("BuildRunRequest: %v", err) + } + if runID == "" || req.GetRunId() != runID { + t.Fatalf("run id mismatch: runID=%q req=%q", runID, req.GetRunId()) + } + if req.GetSessionId() != edgeservice.DefaultSessionID { + t.Fatalf("SessionId: got %q want %q", req.GetSessionId(), edgeservice.DefaultSessionID) + } + if req.GetTimeoutSec() != edgeservice.DefaultTimeoutSec { + t.Fatalf("TimeoutSec: got %d want %d", req.GetTimeoutSec(), edgeservice.DefaultTimeoutSec) + } + if req.GetSessionMode() != iop.RunSessionMode_RUN_SESSION_MODE_CREATE_IF_MISSING { + t.Fatalf("SessionMode: got %v", req.GetSessionMode()) + } +} + +func TestBuildUsageStatusRequestAndWaitTimeout(t *testing.T) { + req := edgeservice.BuildUsageStatusRequest("cli", "gemini", "", 0) + if req.GetSessionId() != edgeservice.DefaultSessionID { + t.Fatalf("SessionId: got %q want %q", req.GetSessionId(), edgeservice.DefaultSessionID) + } + if req.GetTimeoutSec() != edgeservice.DefaultTimeoutSec { + t.Fatalf("TimeoutSec: got %d want %d", req.GetTimeoutSec(), edgeservice.DefaultTimeoutSec) + } + if got, want := edgeservice.StatusWaitTimeout(req), 35*time.Second; got != want { + t.Fatalf("StatusWaitTimeout: got %v want %v", got, want) + } +} diff --git a/apps/edge/internal/transport/integration_test.go b/apps/edge/internal/transport/integration_test.go index 577ad41..209fe3c 100644 --- a/apps/edge/internal/transport/integration_test.go +++ b/apps/edge/internal/transport/integration_test.go @@ -15,6 +15,7 @@ import ( edgenode "iop/apps/edge/internal/node" "iop/apps/edge/internal/transport" "iop/packages/config" + eventpkg "iop/packages/events" iop "iop/proto/gen/iop" ) @@ -72,6 +73,10 @@ func TestEdgeServerIntegration(t *testing.T) { if err != nil { t.Fatalf("failed to create server: %v", err) } + nodeEventCh := make(chan *iop.EdgeNodeEvent, 4) + server.SetNodeEventHandler(func(event *iop.EdgeNodeEvent) { + nodeEventCh <- event + }) if err := server.Start(ctx); err != nil { t.Fatalf("failed to start server: %v", err) } @@ -132,6 +137,17 @@ func TestEdgeServerIntegration(t *testing.T) { if len(resp.GetConfig().GetAdapters()) != 2 { t.Fatalf("expected mock and ollama adapters, got %d", len(resp.GetConfig().GetAdapters())) } + select { + case event := <-nodeEventCh: + if event.GetType() != eventpkg.TypeNodeConnected { + t.Fatalf("event type: got %q want %q", event.GetType(), eventpkg.TypeNodeConnected) + } + if event.GetNodeId() != wantNodeID || event.GetAlias() != "test-node" || event.GetReason() != eventpkg.ReasonRegistered { + t.Fatalf("unexpected connected event: %+v", event) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for node connected event") + } // 4. Registry 등록 여부 확인 entry, ok := waitForRegistryEntry(ctx, registry, wantNodeID) @@ -156,4 +172,19 @@ func TestEdgeServerIntegration(t *testing.T) { if e, err := registry.Resolve("test-node"); err != nil || e.NodeID != wantNodeID { t.Fatalf("failed to resolve by alias %q: %v", "test-node", err) } + + if err := client.Close(); err != nil { + t.Fatalf("close client: %v", err) + } + select { + case event := <-nodeEventCh: + if event.GetType() != eventpkg.TypeNodeDisconnected { + t.Fatalf("event type: got %q want %q", event.GetType(), eventpkg.TypeNodeDisconnected) + } + if event.GetNodeId() != wantNodeID || event.GetAlias() != "test-node" || event.GetReason() != eventpkg.ReasonTransportClosed { + t.Fatalf("unexpected disconnected event: %+v", event) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for node disconnected event") + } } diff --git a/apps/edge/internal/transport/server.go b/apps/edge/internal/transport/server.go index b0b0f3e..e9dd28a 100644 --- a/apps/edge/internal/transport/server.go +++ b/apps/edge/internal/transport/server.go @@ -5,12 +5,14 @@ import ( "net" "strconv" "sync" + "sync/atomic" toki "git.toki-labs.com/toki/common-proto-socket/go" "go.uber.org/zap" "google.golang.org/protobuf/proto" edgenode "iop/apps/edge/internal/node" + "iop/packages/events" iop "iop/proto/gen/iop" ) @@ -25,6 +27,10 @@ func edgeParserMap() toki.ParserMap { m := &iop.RunEvent{} return m, proto.Unmarshal(b, m) }, + toki.TypeNameOf(&iop.EdgeNodeEvent{}): func(b []byte) (proto.Message, error) { + m := &iop.EdgeNodeEvent{} + return m, proto.Unmarshal(b, m) + }, toki.TypeNameOf(&iop.RegisterRequest{}): func(b []byte) (proto.Message, error) { m := &iop.RegisterRequest{} return m, proto.Unmarshal(b, m) @@ -38,13 +44,15 @@ func edgeParserMap() toki.ParserMap { // Server wraps proto-socket TcpServer and manages node connections. type Server struct { - tcp *toki.TcpServer - listen string - registry *edgenode.Registry - nodeStore *edgenode.NodeStore - logger *zap.Logger - handlerMu sync.RWMutex - onRunEvent func(*iop.RunEvent) + tcp *toki.TcpServer + listen string + registry *edgenode.Registry + nodeStore *edgenode.NodeStore + logger *zap.Logger + handlerMu sync.RWMutex + onRunEvent func(*iop.RunEvent) + onNodeEvent func(*iop.EdgeNodeEvent) + stopping atomic.Bool } func NewServer(listen string, registry *edgenode.Registry, nodeStore *edgenode.NodeStore, logger *zap.Logger) (*Server, error) { @@ -66,6 +74,7 @@ func NewServer(listen string, registry *edgenode.Registry, nodeStore *edgenode.N } func (s *Server) Start(ctx context.Context) error { + s.stopping.Store(false) if err := s.tcp.Start(ctx); err != nil { return err } @@ -74,6 +83,7 @@ func (s *Server) Start(ctx context.Context) error { } func (s *Server) Stop() error { + s.stopping.Store(true) return s.tcp.Stop() } @@ -83,6 +93,12 @@ func (s *Server) SetRunEventHandler(handler func(*iop.RunEvent)) { s.handlerMu.Unlock() } +func (s *Server) SetNodeEventHandler(handler func(*iop.EdgeNodeEvent)) { + s.handlerMu.Lock() + s.onNodeEvent = handler + s.handlerMu.Unlock() +} + func (s *Server) onNodeConnected(client *toki.TcpClient) { s.logger.Info("node connection established") @@ -127,15 +143,44 @@ func (s *Server) onNodeConnected(client *toki.TcpClient) { Alias: rec.Alias, Client: client, } + toki.AddListenerTyped[*iop.EdgeNodeEvent](&client.Communicator, func(e *iop.EdgeNodeEvent) { + if e.NodeId == "" { + e.NodeId = rec.ID + } + if e.Alias == "" { + e.Alias = rec.Alias + } + s.emitNodeEvent(e) + }) client.AddDisconnectListener(func(_ *toki.TcpClient) { s.registry.Unregister(rec.ID) s.logger.Info("node unregistered", zap.String("node_id", rec.ID)) + reason := events.ReasonTransportClosed + if s.stopping.Load() { + reason = events.ReasonEdgeShutdown + } + s.emitNodeEvent(events.NewEdgeNodeEvent( + events.SourceEdge, + events.TypeNodeDisconnected, + rec.ID, + rec.Alias, + reason, + nil, + )) }) s.registry.Register(entry) s.logger.Info("node registered", zap.String("node_id", rec.ID), zap.String("alias", rec.Alias), ) + s.emitNodeEvent(events.NewEdgeNodeEvent( + events.SourceEdge, + events.TypeNodeConnected, + rec.ID, + rec.Alias, + events.ReasonRegistered, + nil, + )) return &iop.RegisterResponse{ Accepted: true, @@ -147,6 +192,20 @@ func (s *Server) onNodeConnected(client *toki.TcpClient) { ) } +func (s *Server) emitNodeEvent(event *iop.EdgeNodeEvent) { + s.logger.Debug("node event received", + zap.String("node_id", event.GetNodeId()), + zap.String("type", event.GetType()), + zap.String("reason", event.GetReason()), + ) + s.handlerMu.RLock() + handler := s.onNodeEvent + s.handlerMu.RUnlock() + if handler != nil { + handler(event) + } +} + func safePrefix(s string) string { if len(s) > 8 { return s[:8] + "..." diff --git a/apps/edge/internal/transport/server_test.go b/apps/edge/internal/transport/server_test.go index ae2b6da..4e55d22 100644 --- a/apps/edge/internal/transport/server_test.go +++ b/apps/edge/internal/transport/server_test.go @@ -47,6 +47,40 @@ func TestEdgeParserMap_NodeCommandResponse(t *testing.T) { } } +func TestEdgeParserMap_EdgeNodeEvent(t *testing.T) { + parsers := edgeParserMap() + original := &iop.EdgeNodeEvent{ + Type: "node.disconnected", + Source: "edge", + NodeId: "node-1", + Alias: "local-node", + Reason: "transport_closed", + } + payload, err := proto.Marshal(original) + if err != nil { + t.Fatalf("marshal: %v", err) + } + + key := toki.TypeNameOf(original) + parser, ok := parsers[key] + if !ok { + t.Fatalf("parser not found for key: %s", key) + } + + parsed, err := parser(payload) + if err != nil { + t.Fatalf("parse: %v", err) + } + got := parsed.(*iop.EdgeNodeEvent) + if got.GetType() != original.GetType() || + got.GetSource() != original.GetSource() || + got.GetNodeId() != original.GetNodeId() || + got.GetAlias() != original.GetAlias() || + got.GetReason() != original.GetReason() { + t.Fatalf("unexpected edge node event: %+v", got) + } +} + func findCLIAdapter(payload *iop.NodeConfigPayload) *iop.AdapterConfig { for _, a := range payload.GetAdapters() { if a.GetType() == "cli" { diff --git a/apps/node/README.md b/apps/node/README.md index 6a81162..8b051fa 100644 --- a/apps/node/README.md +++ b/apps/node/README.md @@ -56,6 +56,18 @@ edge에서 실행 요청을 받으면 node는 해당 입력을 선택된 adapter [node-message] ``` +## Edge-Node 메시지 경계 + +node는 같은 TCP/protobuf session 위에서 여러 메시지 계열을 처리한다. + +- `RunRequest`: adapter execution 시작. CLI adapter에서는 prompt 전달로 해석된다. +- `CancelRequest`: 실행 취소 또는 logical session 종료. +- `NodeCommandRequest`: 실행이 아닌 node/adapter 조회성 명령. 현재는 `USAGE_STATUS`를 지원한다. +- `RunEvent`: adapter execution stream. `start`, `delta`, `complete`, `error`, `cancelled` 같은 실행 이벤트를 edge로 보낸다. +- `EdgeNodeEvent`: edge-node lifecycle/control event envelope. edge 연결 해제 등 실행 스트림이 아닌 이벤트를 다룬다. + +즉 CLI에 메시지를 보내는 요청 파이프라인과 이벤트 파이프라인은 논리적으로 분리되어 있고, 물리 transport만 공유한다. + ## Logical Session (transport 1개 · session 여러 개) edge-node transport 연결은 **호스트당 1개**를 유지한다. 그 연결 위에서 CLI adapter는 `session_id`가 다른 여러 장수 worker process를 독립적으로 관리한다. @@ -102,5 +114,6 @@ Cline은 mutable `--config` 디렉터리를 프로필처럼 나눈다. `/config/ - TCP 4-byte length-prefix framing - protobuf message framing via `common-proto-socket` +- edge 연결 해제는 `EdgeNodeEvent`로 로컬 표시되며, 실행 스트림 `RunEvent`와 분리한다. - mTLS helper는 존재하지만 현재 transport 설정에는 아직 연결되지 않음 - 하트비트: 30초 간격 diff --git a/apps/node/internal/bootstrap/module.go b/apps/node/internal/bootstrap/module.go index 90be505..162dea6 100644 --- a/apps/node/internal/bootstrap/module.go +++ b/apps/node/internal/bootstrap/module.go @@ -4,6 +4,7 @@ package bootstrap import ( "context" "fmt" + "io" "os" "path/filepath" @@ -16,7 +17,9 @@ import ( "iop/apps/node/internal/store" "iop/apps/node/internal/transport" "iop/packages/config" + "iop/packages/events" "iop/packages/observability" + iop "iop/proto/gen/iop" ) // Module returns the fx options that wire the node application. @@ -67,6 +70,9 @@ func Module(cfg *config.NodeConfig) fx.Option { rtr := router.New(reg, logger) n := node.New(result.NodeID, rtr, st, os.Stdout, logger) + result.Session.SetEventHandler(func(event *iop.EdgeNodeEvent) { + printEdgeEvent(os.Stdout, event) + }) result.Session.SetHandler(n) sess = result.Session @@ -107,3 +113,15 @@ func storeDSN(workspaceRoot string) (string, error) { } return "file:" + filepath.Join(workspaceRoot, "iop.db") + "?cache=shared&mode=rwc", nil } + +func printEdgeEvent(out io.Writer, event *iop.EdgeNodeEvent) { + if out == nil { + return + } + switch event.GetType() { + case events.TypeEdgeDisconnected: + fmt.Fprintf(out, "[edge-event] disconnected reason=%q\n", event.GetReason()) + default: + fmt.Fprintf(out, "[edge-event] %s reason=%q\n", event.GetType(), event.GetReason()) + } +} diff --git a/apps/node/internal/transport/client.go b/apps/node/internal/transport/client.go index 747c9b7..edd9479 100644 --- a/apps/node/internal/transport/client.go +++ b/apps/node/internal/transport/client.go @@ -58,7 +58,7 @@ func DialEdge(ctx context.Context, addr, token string, logger *zap.Logger) (*Reg return nil, fmt.Errorf("transport: register rejected: %s", resp.GetReason()) } - sess := newSession(client, logger) + sess := newSession(client, logger, resp.GetNodeId(), resp.GetAlias()) logger.Info("registered with edge", zap.String("node_id", resp.GetNodeId()), zap.String("alias", resp.GetAlias()), diff --git a/apps/node/internal/transport/integration_test.go b/apps/node/internal/transport/integration_test.go index 9ebe56e..58b072b 100644 --- a/apps/node/internal/transport/integration_test.go +++ b/apps/node/internal/transport/integration_test.go @@ -13,6 +13,7 @@ import ( toki "git.toki-labs.com/toki/common-proto-socket/go" "iop/apps/node/internal/transport" + eventpkg "iop/packages/events" iop "iop/proto/gen/iop" ) @@ -115,6 +116,10 @@ func TestNodeClientIntegration(t *testing.T) { t.Fatalf("failed to dial edge: %v", err) } defer result.Session.Close() + edgeEventCh := make(chan *iop.EdgeNodeEvent, 1) + result.Session.SetEventHandler(func(event *iop.EdgeNodeEvent) { + edgeEventCh <- event + }) result.Session.SetHandler(handler) if result.NodeID != "test-node" { @@ -169,4 +174,19 @@ func TestNodeClientIntegration(t *testing.T) { case <-time.After(2 * time.Second): t.Fatal("timeout waiting for run event from node session") } + + if err := edgeClient.Close(); err != nil { + t.Fatalf("close edge client: %v", err) + } + select { + case event := <-edgeEventCh: + if event.GetType() != eventpkg.TypeEdgeDisconnected { + t.Fatalf("event type: got %q want %q", event.GetType(), eventpkg.TypeEdgeDisconnected) + } + if event.GetNodeId() != "test-node" || event.GetAlias() != "test-alias" || event.GetReason() != eventpkg.ReasonTransportClosed { + t.Fatalf("unexpected edge disconnected event: %+v", event) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for edge disconnected event") + } } diff --git a/apps/node/internal/transport/parser.go b/apps/node/internal/transport/parser.go index 06405d9..13992d1 100644 --- a/apps/node/internal/transport/parser.go +++ b/apps/node/internal/transport/parser.go @@ -13,6 +13,10 @@ func nodeParserMap() toki.ParserMap { m := &iop.RunRequest{} return m, proto.Unmarshal(b, m) }, + toki.TypeNameOf(&iop.EdgeNodeEvent{}): func(b []byte) (proto.Message, error) { + m := &iop.EdgeNodeEvent{} + return m, proto.Unmarshal(b, m) + }, toki.TypeNameOf(&iop.CancelRequest{}): func(b []byte) (proto.Message, error) { m := &iop.CancelRequest{} return m, proto.Unmarshal(b, m) diff --git a/apps/node/internal/transport/parser_test.go b/apps/node/internal/transport/parser_test.go index 9b306f7..82c3fe5 100644 --- a/apps/node/internal/transport/parser_test.go +++ b/apps/node/internal/transport/parser_test.go @@ -79,6 +79,40 @@ func TestNodeParserMap_CancelRequest(t *testing.T) { } } +func TestNodeParserMap_EdgeNodeEvent(t *testing.T) { + parsers := nodeParserMap() + original := &iop.EdgeNodeEvent{ + Type: "edge.disconnected", + Source: "node", + NodeId: "node-1", + Alias: "local-node", + Reason: "transport_closed", + } + payload, err := proto.Marshal(original) + if err != nil { + t.Fatalf("marshal: %v", err) + } + + key := toki.TypeNameOf(original) + parser, ok := parsers[key] + if !ok { + t.Fatalf("parser not found for key: %s", key) + } + + parsed, err := parser(payload) + if err != nil { + t.Fatalf("parse: %v", err) + } + got := parsed.(*iop.EdgeNodeEvent) + if got.GetType() != original.GetType() || + got.GetSource() != original.GetSource() || + got.GetNodeId() != original.GetNodeId() || + got.GetAlias() != original.GetAlias() || + got.GetReason() != original.GetReason() { + t.Fatalf("unexpected edge node event: %+v", got) + } +} + func TestNodeParserMap_NodeCommandRequest(t *testing.T) { parsers := nodeParserMap() original := &iop.NodeCommandRequest{ diff --git a/apps/node/internal/transport/session.go b/apps/node/internal/transport/session.go index 69c1d07..2a44202 100644 --- a/apps/node/internal/transport/session.go +++ b/apps/node/internal/transport/session.go @@ -8,6 +8,7 @@ import ( "go.uber.org/zap" "google.golang.org/protobuf/proto" + "iop/packages/events" iop "iop/proto/gen/iop" ) @@ -20,14 +21,18 @@ type Handler interface { // Session represents the node's persistent connection to edge. type Session struct { - client *toki.TcpClient - logger *zap.Logger - mu sync.RWMutex - handler Handler + client *toki.TcpClient + logger *zap.Logger + nodeID string + alias string + mu sync.RWMutex + handler Handler + eventHandler func(*iop.EdgeNodeEvent) + closeReason string } -func newSession(client *toki.TcpClient, logger *zap.Logger) *Session { - s := &Session{client: client, logger: logger} +func newSession(client *toki.TcpClient, logger *zap.Logger, nodeID, alias string) *Session { + s := &Session{client: client, logger: logger, nodeID: nodeID, alias: alias} toki.AddListenerTyped[*iop.RunRequest](&client.Communicator, func(req *iop.RunRequest) { go func() { @@ -75,8 +80,20 @@ func newSession(client *toki.TcpClient, logger *zap.Logger) *Session { return resp, nil }) + toki.AddListenerTyped[*iop.EdgeNodeEvent](&client.Communicator, func(event *iop.EdgeNodeEvent) { + s.emitEvent(event) + }) + client.AddDisconnectListener(func(_ *toki.TcpClient) { logger.Info("disconnected from edge") + s.emitEvent(events.NewEdgeNodeEvent( + events.SourceNode, + events.TypeEdgeDisconnected, + nodeID, + alias, + s.disconnectReason(), + nil, + )) }) return s @@ -89,6 +106,12 @@ func (s *Session) SetHandler(h Handler) { s.mu.Unlock() } +func (s *Session) SetEventHandler(handler func(*iop.EdgeNodeEvent)) { + s.mu.Lock() + s.eventHandler = handler + s.mu.Unlock() +} + // Send transmits a proto message to edge. func (s *Session) Send(m proto.Message) error { return s.client.Send(m) @@ -98,4 +121,34 @@ func (s *Session) Send(m proto.Message) error { func (s *Session) IsAlive() bool { return s.client.IsAlive() } // Close terminates the connection to edge. -func (s *Session) Close() error { return s.client.Close() } +func (s *Session) Close() error { + s.setCloseReason(events.ReasonLocalShutdown) + return s.client.Close() +} + +func (s *Session) emitEvent(event *iop.EdgeNodeEvent) { + s.mu.RLock() + handler := s.eventHandler + s.mu.RUnlock() + if handler != nil { + handler(event) + } +} + +func (s *Session) setCloseReason(reason string) { + s.mu.Lock() + if s.closeReason == "" { + s.closeReason = reason + } + s.mu.Unlock() +} + +func (s *Session) disconnectReason() string { + s.mu.RLock() + reason := s.closeReason + s.mu.RUnlock() + if reason == "" { + return events.ReasonTransportClosed + } + return reason +} diff --git a/docs/architecture.md b/docs/architecture.md index c23ba0d..f8a8b7e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -25,7 +25,7 @@ Control Plane은 Node에 직접 연결하지 않는다. Control Plane은 Edge Edge는 여러 Node를 묶는 실행 그룹 컨트롤러다. Edge는 Node registry, adapter/profile configuration, routing, stream relay, session handling, event aggregation을 담당한다. -Node는 실제 실행자다. Node는 Edge에 연결되어 adapter execution을 수행하고, RuntimeEvent를 Edge로 돌려준다. +Node는 실제 실행자다. Node는 Edge에 연결되어 adapter execution을 수행하고, RuntimeEvent를 Edge로 돌려준다. 연결 생명주기와 이후 제어/상태성 이벤트는 RunEvent와 분리된 EdgeNodeEvent envelope로 다룬다. ## 현재 Edge-Node 흐름 @@ -43,11 +43,13 @@ Node └─ builds adapter registry └─ resolves adapter + target └─ executes adapter - └─ emits RuntimeEvent + └─ emits RuntimeEvent / EdgeNodeEvent ``` 현재 실행 이력은 Node의 local SQLite store에서 검증 중이다. Edge 단위 이력 집계와 로컬 실행 그룹 상태 소유권은 로드맵에 따라 정리한다. +Edge console의 `/` 명령은 임시 디버그 클라이언트다. Console은 `edge/internal/service`를 호출하는 얇은 어댑터로 유지하고, 이후 HTTP/API 표면도 같은 service와 `edge/internal/events` bus를 재사용한다. + ## 내부 실행 모델 내부 실행은 `adapter + target`으로 표현한다. diff --git a/packages/events/events.go b/packages/events/events.go new file mode 100644 index 0000000..b806a2d --- /dev/null +++ b/packages/events/events.go @@ -0,0 +1,36 @@ +package events + +import ( + "time" + + "github.com/google/uuid" + + iop "iop/proto/gen/iop" +) + +const ( + SourceEdge = "edge" + SourceNode = "node" + + TypeNodeConnected = "node.connected" + TypeNodeDisconnected = "node.disconnected" + TypeEdgeDisconnected = "edge.disconnected" + + ReasonRegistered = "registered" + ReasonTransportClosed = "transport_closed" + ReasonLocalShutdown = "local_shutdown" + ReasonEdgeShutdown = "edge_shutdown" +) + +func NewEdgeNodeEvent(source, eventType, nodeID, alias, reason string, metadata map[string]string) *iop.EdgeNodeEvent { + return &iop.EdgeNodeEvent{ + EventId: uuid.NewString(), + Type: eventType, + Source: source, + NodeId: nodeID, + Alias: alias, + Reason: reason, + Metadata: metadata, + Timestamp: time.Now().UnixNano(), + } +} diff --git a/proto/gen/iop/runtime.pb.go b/proto/gen/iop/runtime.pb.go index dcb4ecc..9e7c831 100644 --- a/proto/gen/iop/runtime.pb.go +++ b/proto/gen/iop/runtime.pb.go @@ -416,6 +416,108 @@ func (x *RunEvent) GetNodeId() string { return "" } +// EdgeNodeEvent is a general edge-node lifecycle/control event envelope. +// It is separate from RunEvent, which is reserved for adapter execution streams. +type EdgeNodeEvent struct { + state protoimpl.MessageState `protogen:"open.v1"` + EventId string `protobuf:"bytes,1,opt,name=event_id,json=eventId,proto3" json:"event_id,omitempty"` + Type string `protobuf:"bytes,2,opt,name=type,proto3" json:"type,omitempty"` // node.connected | node.disconnected | edge.disconnected | ... + Source string `protobuf:"bytes,3,opt,name=source,proto3" json:"source,omitempty"` // edge | node + NodeId string `protobuf:"bytes,4,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"` + Alias string `protobuf:"bytes,5,opt,name=alias,proto3" json:"alias,omitempty"` + Reason string `protobuf:"bytes,6,opt,name=reason,proto3" json:"reason,omitempty"` + Metadata map[string]string `protobuf:"bytes,7,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + Timestamp int64 `protobuf:"varint,8,opt,name=timestamp,proto3" json:"timestamp,omitempty"` // unix nano + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *EdgeNodeEvent) Reset() { + *x = EdgeNodeEvent{} + mi := &file_proto_iop_runtime_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *EdgeNodeEvent) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*EdgeNodeEvent) ProtoMessage() {} + +func (x *EdgeNodeEvent) ProtoReflect() protoreflect.Message { + mi := &file_proto_iop_runtime_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use EdgeNodeEvent.ProtoReflect.Descriptor instead. +func (*EdgeNodeEvent) Descriptor() ([]byte, []int) { + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{2} +} + +func (x *EdgeNodeEvent) GetEventId() string { + if x != nil { + return x.EventId + } + return "" +} + +func (x *EdgeNodeEvent) GetType() string { + if x != nil { + return x.Type + } + return "" +} + +func (x *EdgeNodeEvent) GetSource() string { + if x != nil { + return x.Source + } + return "" +} + +func (x *EdgeNodeEvent) GetNodeId() string { + if x != nil { + return x.NodeId + } + return "" +} + +func (x *EdgeNodeEvent) GetAlias() string { + if x != nil { + return x.Alias + } + return "" +} + +func (x *EdgeNodeEvent) GetReason() string { + if x != nil { + return x.Reason + } + return "" +} + +func (x *EdgeNodeEvent) GetMetadata() map[string]string { + if x != nil { + return x.Metadata + } + return nil +} + +func (x *EdgeNodeEvent) GetTimestamp() int64 { + if x != nil { + return x.Timestamp + } + return 0 +} + type Usage struct { state protoimpl.MessageState `protogen:"open.v1"` InputTokens int32 `protobuf:"varint,1,opt,name=input_tokens,json=inputTokens,proto3" json:"input_tokens,omitempty"` @@ -426,7 +528,7 @@ type Usage struct { func (x *Usage) Reset() { *x = Usage{} - mi := &file_proto_iop_runtime_proto_msgTypes[2] + mi := &file_proto_iop_runtime_proto_msgTypes[3] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -438,7 +540,7 @@ func (x *Usage) String() string { func (*Usage) ProtoMessage() {} func (x *Usage) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[2] + mi := &file_proto_iop_runtime_proto_msgTypes[3] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -451,7 +553,7 @@ func (x *Usage) ProtoReflect() protoreflect.Message { // Deprecated: Use Usage.ProtoReflect.Descriptor instead. func (*Usage) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{2} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{3} } func (x *Usage) GetInputTokens() int32 { @@ -478,7 +580,7 @@ type Heartbeat struct { func (x *Heartbeat) Reset() { *x = Heartbeat{} - mi := &file_proto_iop_runtime_proto_msgTypes[3] + mi := &file_proto_iop_runtime_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -490,7 +592,7 @@ func (x *Heartbeat) String() string { func (*Heartbeat) ProtoMessage() {} func (x *Heartbeat) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[3] + mi := &file_proto_iop_runtime_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -503,7 +605,7 @@ func (x *Heartbeat) ProtoReflect() protoreflect.Message { // Deprecated: Use Heartbeat.ProtoReflect.Descriptor instead. func (*Heartbeat) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{3} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{4} } func (x *Heartbeat) GetTimestamp() int64 { @@ -527,7 +629,7 @@ type CancelRequest struct { func (x *CancelRequest) Reset() { *x = CancelRequest{} - mi := &file_proto_iop_runtime_proto_msgTypes[4] + mi := &file_proto_iop_runtime_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -539,7 +641,7 @@ func (x *CancelRequest) String() string { func (*CancelRequest) ProtoMessage() {} func (x *CancelRequest) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[4] + mi := &file_proto_iop_runtime_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -552,7 +654,7 @@ func (x *CancelRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use CancelRequest.ProtoReflect.Descriptor instead. func (*CancelRequest) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{4} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{5} } func (x *CancelRequest) GetRunId() string { @@ -605,7 +707,7 @@ type NodeCommandRequest struct { func (x *NodeCommandRequest) Reset() { *x = NodeCommandRequest{} - mi := &file_proto_iop_runtime_proto_msgTypes[5] + mi := &file_proto_iop_runtime_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -617,7 +719,7 @@ func (x *NodeCommandRequest) String() string { func (*NodeCommandRequest) ProtoMessage() {} func (x *NodeCommandRequest) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[5] + mi := &file_proto_iop_runtime_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -630,7 +732,7 @@ func (x *NodeCommandRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use NodeCommandRequest.ProtoReflect.Descriptor instead. func (*NodeCommandRequest) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{5} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{6} } func (x *NodeCommandRequest) GetRequestId() string { @@ -697,7 +799,7 @@ type NodeCommandResponse struct { func (x *NodeCommandResponse) Reset() { *x = NodeCommandResponse{} - mi := &file_proto_iop_runtime_proto_msgTypes[6] + mi := &file_proto_iop_runtime_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -709,7 +811,7 @@ func (x *NodeCommandResponse) String() string { func (*NodeCommandResponse) ProtoMessage() {} func (x *NodeCommandResponse) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[6] + mi := &file_proto_iop_runtime_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -722,7 +824,7 @@ func (x *NodeCommandResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use NodeCommandResponse.ProtoReflect.Descriptor instead. func (*NodeCommandResponse) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{6} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{7} } func (x *NodeCommandResponse) GetRequestId() string { @@ -788,7 +890,7 @@ type AgentUsageStatus struct { func (x *AgentUsageStatus) Reset() { *x = AgentUsageStatus{} - mi := &file_proto_iop_runtime_proto_msgTypes[7] + mi := &file_proto_iop_runtime_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -800,7 +902,7 @@ func (x *AgentUsageStatus) String() string { func (*AgentUsageStatus) ProtoMessage() {} func (x *AgentUsageStatus) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[7] + mi := &file_proto_iop_runtime_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -813,7 +915,7 @@ func (x *AgentUsageStatus) ProtoReflect() protoreflect.Message { // Deprecated: Use AgentUsageStatus.ProtoReflect.Descriptor instead. func (*AgentUsageStatus) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{7} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{8} } func (x *AgentUsageStatus) GetRawOutput() string { @@ -869,7 +971,7 @@ type Error struct { func (x *Error) Reset() { *x = Error{} - mi := &file_proto_iop_runtime_proto_msgTypes[8] + mi := &file_proto_iop_runtime_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -881,7 +983,7 @@ func (x *Error) String() string { func (*Error) ProtoMessage() {} func (x *Error) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[8] + mi := &file_proto_iop_runtime_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -894,7 +996,7 @@ func (x *Error) ProtoReflect() protoreflect.Message { // Deprecated: Use Error.ProtoReflect.Descriptor instead. func (*Error) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{8} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{9} } func (x *Error) GetCode() string { @@ -921,7 +1023,7 @@ type RegisterRequest struct { func (x *RegisterRequest) Reset() { *x = RegisterRequest{} - mi := &file_proto_iop_runtime_proto_msgTypes[9] + mi := &file_proto_iop_runtime_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -933,7 +1035,7 @@ func (x *RegisterRequest) String() string { func (*RegisterRequest) ProtoMessage() {} func (x *RegisterRequest) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[9] + mi := &file_proto_iop_runtime_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -946,7 +1048,7 @@ func (x *RegisterRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RegisterRequest.ProtoReflect.Descriptor instead. func (*RegisterRequest) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{9} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{10} } func (x *RegisterRequest) GetToken() string { @@ -970,7 +1072,7 @@ type RegisterResponse struct { func (x *RegisterResponse) Reset() { *x = RegisterResponse{} - mi := &file_proto_iop_runtime_proto_msgTypes[10] + mi := &file_proto_iop_runtime_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -982,7 +1084,7 @@ func (x *RegisterResponse) String() string { func (*RegisterResponse) ProtoMessage() {} func (x *RegisterResponse) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[10] + mi := &file_proto_iop_runtime_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -995,7 +1097,7 @@ func (x *RegisterResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RegisterResponse.ProtoReflect.Descriptor instead. func (*RegisterResponse) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{10} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{11} } func (x *RegisterResponse) GetAccepted() bool { @@ -1044,7 +1146,7 @@ type NodeConfigPayload struct { func (x *NodeConfigPayload) Reset() { *x = NodeConfigPayload{} - mi := &file_proto_iop_runtime_proto_msgTypes[11] + mi := &file_proto_iop_runtime_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1056,7 +1158,7 @@ func (x *NodeConfigPayload) String() string { func (*NodeConfigPayload) ProtoMessage() {} func (x *NodeConfigPayload) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[11] + mi := &file_proto_iop_runtime_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1069,7 +1171,7 @@ func (x *NodeConfigPayload) ProtoReflect() protoreflect.Message { // Deprecated: Use NodeConfigPayload.ProtoReflect.Descriptor instead. func (*NodeConfigPayload) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{11} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{12} } func (x *NodeConfigPayload) GetAdapters() []*AdapterConfig { @@ -1104,7 +1206,7 @@ type AdapterConfig struct { func (x *AdapterConfig) Reset() { *x = AdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[12] + mi := &file_proto_iop_runtime_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1116,7 +1218,7 @@ func (x *AdapterConfig) String() string { func (*AdapterConfig) ProtoMessage() {} func (x *AdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[12] + mi := &file_proto_iop_runtime_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1129,7 +1231,7 @@ func (x *AdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use AdapterConfig.ProtoReflect.Descriptor instead. func (*AdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{12} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{13} } func (x *AdapterConfig) GetType() string { @@ -1218,7 +1320,7 @@ type CLIAdapterConfig struct { func (x *CLIAdapterConfig) Reset() { *x = CLIAdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[13] + mi := &file_proto_iop_runtime_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1230,7 +1332,7 @@ func (x *CLIAdapterConfig) String() string { func (*CLIAdapterConfig) ProtoMessage() {} func (x *CLIAdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[13] + mi := &file_proto_iop_runtime_proto_msgTypes[14] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1243,7 +1345,7 @@ func (x *CLIAdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use CLIAdapterConfig.ProtoReflect.Descriptor instead. func (*CLIAdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{13} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{14} } func (x *CLIAdapterConfig) GetProfiles() map[string]*CLIProfileConfig { @@ -1272,7 +1374,7 @@ type CLIProfileConfig struct { func (x *CLIProfileConfig) Reset() { *x = CLIProfileConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[14] + mi := &file_proto_iop_runtime_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1284,7 +1386,7 @@ func (x *CLIProfileConfig) String() string { func (*CLIProfileConfig) ProtoMessage() {} func (x *CLIProfileConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[14] + mi := &file_proto_iop_runtime_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1297,7 +1399,7 @@ func (x *CLIProfileConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use CLIProfileConfig.ProtoReflect.Descriptor instead. func (*CLIProfileConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{14} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{15} } func (x *CLIProfileConfig) GetCommand() string { @@ -1387,7 +1489,7 @@ type CLICompletionMarker struct { func (x *CLICompletionMarker) Reset() { *x = CLICompletionMarker{} - mi := &file_proto_iop_runtime_proto_msgTypes[15] + mi := &file_proto_iop_runtime_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1399,7 +1501,7 @@ func (x *CLICompletionMarker) String() string { func (*CLICompletionMarker) ProtoMessage() {} func (x *CLICompletionMarker) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[15] + mi := &file_proto_iop_runtime_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1412,7 +1514,7 @@ func (x *CLICompletionMarker) ProtoReflect() protoreflect.Message { // Deprecated: Use CLICompletionMarker.ProtoReflect.Descriptor instead. func (*CLICompletionMarker) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{15} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{16} } func (x *CLICompletionMarker) GetLine() string { @@ -1438,7 +1540,7 @@ type OllamaAdapterConfig struct { func (x *OllamaAdapterConfig) Reset() { *x = OllamaAdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[16] + mi := &file_proto_iop_runtime_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1450,7 +1552,7 @@ func (x *OllamaAdapterConfig) String() string { func (*OllamaAdapterConfig) ProtoMessage() {} func (x *OllamaAdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[16] + mi := &file_proto_iop_runtime_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1463,7 +1565,7 @@ func (x *OllamaAdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use OllamaAdapterConfig.ProtoReflect.Descriptor instead. func (*OllamaAdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{16} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{17} } func (x *OllamaAdapterConfig) GetBaseUrl() string { @@ -1482,7 +1584,7 @@ type VllmAdapterConfig struct { func (x *VllmAdapterConfig) Reset() { *x = VllmAdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[17] + mi := &file_proto_iop_runtime_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1494,7 +1596,7 @@ func (x *VllmAdapterConfig) String() string { func (*VllmAdapterConfig) ProtoMessage() {} func (x *VllmAdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[17] + mi := &file_proto_iop_runtime_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1507,7 +1609,7 @@ func (x *VllmAdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use VllmAdapterConfig.ProtoReflect.Descriptor instead. func (*VllmAdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{17} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{18} } func (x *VllmAdapterConfig) GetEndpoint() string { @@ -1528,7 +1630,7 @@ type NodeRuntimeConfig struct { func (x *NodeRuntimeConfig) Reset() { *x = NodeRuntimeConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[18] + mi := &file_proto_iop_runtime_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1540,7 +1642,7 @@ func (x *NodeRuntimeConfig) String() string { func (*NodeRuntimeConfig) ProtoMessage() {} func (x *NodeRuntimeConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[18] + mi := &file_proto_iop_runtime_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1553,7 +1655,7 @@ func (x *NodeRuntimeConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use NodeRuntimeConfig.ProtoReflect.Descriptor instead. func (*NodeRuntimeConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{18} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{19} } func (x *NodeRuntimeConfig) GetConcurrency() int32 { @@ -1615,6 +1717,18 @@ const file_proto_iop_runtime_proto_rawDesc = "" + "\anode_id\x18\v \x01(\tR\x06nodeId\x1a;\n" + "\rMetadataEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xb6\x02\n" + + "\rEdgeNodeEvent\x12\x19\n" + + "\bevent_id\x18\x01 \x01(\tR\aeventId\x12\x12\n" + + "\x04type\x18\x02 \x01(\tR\x04type\x12\x16\n" + + "\x06source\x18\x03 \x01(\tR\x06source\x12\x17\n" + + "\anode_id\x18\x04 \x01(\tR\x06nodeId\x12\x14\n" + + "\x05alias\x18\x05 \x01(\tR\x05alias\x12\x16\n" + + "\x06reason\x18\x06 \x01(\tR\x06reason\x12<\n" + + "\bmetadata\x18\a \x03(\v2 .iop.EdgeNodeEvent.MetadataEntryR\bmetadata\x12\x1c\n" + + "\ttimestamp\x18\b \x01(\x03R\ttimestamp\x1a;\n" + + "\rMetadataEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"O\n" + "\x05Usage\x12!\n" + "\finput_tokens\x18\x01 \x01(\x05R\vinputTokens\x12#\n" + @@ -1742,65 +1856,68 @@ func file_proto_iop_runtime_proto_rawDescGZIP() []byte { } var file_proto_iop_runtime_proto_enumTypes = make([]protoimpl.EnumInfo, 3) -var file_proto_iop_runtime_proto_msgTypes = make([]protoimpl.MessageInfo, 24) +var file_proto_iop_runtime_proto_msgTypes = make([]protoimpl.MessageInfo, 26) var file_proto_iop_runtime_proto_goTypes = []any{ (RunSessionMode)(0), // 0: iop.RunSessionMode (CancelAction)(0), // 1: iop.CancelAction (NodeCommandType)(0), // 2: iop.NodeCommandType (*RunRequest)(nil), // 3: iop.RunRequest (*RunEvent)(nil), // 4: iop.RunEvent - (*Usage)(nil), // 5: iop.Usage - (*Heartbeat)(nil), // 6: iop.Heartbeat - (*CancelRequest)(nil), // 7: iop.CancelRequest - (*NodeCommandRequest)(nil), // 8: iop.NodeCommandRequest - (*NodeCommandResponse)(nil), // 9: iop.NodeCommandResponse - (*AgentUsageStatus)(nil), // 10: iop.AgentUsageStatus - (*Error)(nil), // 11: iop.Error - (*RegisterRequest)(nil), // 12: iop.RegisterRequest - (*RegisterResponse)(nil), // 13: iop.RegisterResponse - (*NodeConfigPayload)(nil), // 14: iop.NodeConfigPayload - (*AdapterConfig)(nil), // 15: iop.AdapterConfig - (*CLIAdapterConfig)(nil), // 16: iop.CLIAdapterConfig - (*CLIProfileConfig)(nil), // 17: iop.CLIProfileConfig - (*CLICompletionMarker)(nil), // 18: iop.CLICompletionMarker - (*OllamaAdapterConfig)(nil), // 19: iop.OllamaAdapterConfig - (*VllmAdapterConfig)(nil), // 20: iop.VllmAdapterConfig - (*NodeRuntimeConfig)(nil), // 21: iop.NodeRuntimeConfig - nil, // 22: iop.RunRequest.MetadataEntry - nil, // 23: iop.RunEvent.MetadataEntry - nil, // 24: iop.NodeCommandRequest.MetadataEntry - nil, // 25: iop.AgentUsageStatus.MetadataEntry - nil, // 26: iop.CLIAdapterConfig.ProfilesEntry - (*structpb.Struct)(nil), // 27: google.protobuf.Struct + (*EdgeNodeEvent)(nil), // 5: iop.EdgeNodeEvent + (*Usage)(nil), // 6: iop.Usage + (*Heartbeat)(nil), // 7: iop.Heartbeat + (*CancelRequest)(nil), // 8: iop.CancelRequest + (*NodeCommandRequest)(nil), // 9: iop.NodeCommandRequest + (*NodeCommandResponse)(nil), // 10: iop.NodeCommandResponse + (*AgentUsageStatus)(nil), // 11: iop.AgentUsageStatus + (*Error)(nil), // 12: iop.Error + (*RegisterRequest)(nil), // 13: iop.RegisterRequest + (*RegisterResponse)(nil), // 14: iop.RegisterResponse + (*NodeConfigPayload)(nil), // 15: iop.NodeConfigPayload + (*AdapterConfig)(nil), // 16: iop.AdapterConfig + (*CLIAdapterConfig)(nil), // 17: iop.CLIAdapterConfig + (*CLIProfileConfig)(nil), // 18: iop.CLIProfileConfig + (*CLICompletionMarker)(nil), // 19: iop.CLICompletionMarker + (*OllamaAdapterConfig)(nil), // 20: iop.OllamaAdapterConfig + (*VllmAdapterConfig)(nil), // 21: iop.VllmAdapterConfig + (*NodeRuntimeConfig)(nil), // 22: iop.NodeRuntimeConfig + nil, // 23: iop.RunRequest.MetadataEntry + nil, // 24: iop.RunEvent.MetadataEntry + nil, // 25: iop.EdgeNodeEvent.MetadataEntry + nil, // 26: iop.NodeCommandRequest.MetadataEntry + nil, // 27: iop.AgentUsageStatus.MetadataEntry + nil, // 28: iop.CLIAdapterConfig.ProfilesEntry + (*structpb.Struct)(nil), // 29: google.protobuf.Struct } var file_proto_iop_runtime_proto_depIdxs = []int32{ - 27, // 0: iop.RunRequest.policy:type_name -> google.protobuf.Struct - 27, // 1: iop.RunRequest.input:type_name -> google.protobuf.Struct - 22, // 2: iop.RunRequest.metadata:type_name -> iop.RunRequest.MetadataEntry + 29, // 0: iop.RunRequest.policy:type_name -> google.protobuf.Struct + 29, // 1: iop.RunRequest.input:type_name -> google.protobuf.Struct + 23, // 2: iop.RunRequest.metadata:type_name -> iop.RunRequest.MetadataEntry 0, // 3: iop.RunRequest.session_mode:type_name -> iop.RunSessionMode - 5, // 4: iop.RunEvent.usage:type_name -> iop.Usage - 23, // 5: iop.RunEvent.metadata:type_name -> iop.RunEvent.MetadataEntry - 1, // 6: iop.CancelRequest.action:type_name -> iop.CancelAction - 2, // 7: iop.NodeCommandRequest.type:type_name -> iop.NodeCommandType - 24, // 8: iop.NodeCommandRequest.metadata:type_name -> iop.NodeCommandRequest.MetadataEntry - 2, // 9: iop.NodeCommandResponse.type:type_name -> iop.NodeCommandType - 10, // 10: iop.NodeCommandResponse.usage_status:type_name -> iop.AgentUsageStatus - 25, // 11: iop.AgentUsageStatus.metadata:type_name -> iop.AgentUsageStatus.MetadataEntry - 14, // 12: iop.RegisterResponse.config:type_name -> iop.NodeConfigPayload - 15, // 13: iop.NodeConfigPayload.adapters:type_name -> iop.AdapterConfig - 21, // 14: iop.NodeConfigPayload.runtime:type_name -> iop.NodeRuntimeConfig - 27, // 15: iop.AdapterConfig.settings:type_name -> google.protobuf.Struct - 16, // 16: iop.AdapterConfig.cli:type_name -> iop.CLIAdapterConfig - 19, // 17: iop.AdapterConfig.ollama:type_name -> iop.OllamaAdapterConfig - 20, // 18: iop.AdapterConfig.vllm:type_name -> iop.VllmAdapterConfig - 26, // 19: iop.CLIAdapterConfig.profiles:type_name -> iop.CLIAdapterConfig.ProfilesEntry - 18, // 20: iop.CLIProfileConfig.completion_marker:type_name -> iop.CLICompletionMarker - 17, // 21: iop.CLIAdapterConfig.ProfilesEntry.value:type_name -> iop.CLIProfileConfig - 22, // [22:22] is the sub-list for method output_type - 22, // [22:22] is the sub-list for method input_type - 22, // [22:22] is the sub-list for extension type_name - 22, // [22:22] is the sub-list for extension extendee - 0, // [0:22] is the sub-list for field type_name + 6, // 4: iop.RunEvent.usage:type_name -> iop.Usage + 24, // 5: iop.RunEvent.metadata:type_name -> iop.RunEvent.MetadataEntry + 25, // 6: iop.EdgeNodeEvent.metadata:type_name -> iop.EdgeNodeEvent.MetadataEntry + 1, // 7: iop.CancelRequest.action:type_name -> iop.CancelAction + 2, // 8: iop.NodeCommandRequest.type:type_name -> iop.NodeCommandType + 26, // 9: iop.NodeCommandRequest.metadata:type_name -> iop.NodeCommandRequest.MetadataEntry + 2, // 10: iop.NodeCommandResponse.type:type_name -> iop.NodeCommandType + 11, // 11: iop.NodeCommandResponse.usage_status:type_name -> iop.AgentUsageStatus + 27, // 12: iop.AgentUsageStatus.metadata:type_name -> iop.AgentUsageStatus.MetadataEntry + 15, // 13: iop.RegisterResponse.config:type_name -> iop.NodeConfigPayload + 16, // 14: iop.NodeConfigPayload.adapters:type_name -> iop.AdapterConfig + 22, // 15: iop.NodeConfigPayload.runtime:type_name -> iop.NodeRuntimeConfig + 29, // 16: iop.AdapterConfig.settings:type_name -> google.protobuf.Struct + 17, // 17: iop.AdapterConfig.cli:type_name -> iop.CLIAdapterConfig + 20, // 18: iop.AdapterConfig.ollama:type_name -> iop.OllamaAdapterConfig + 21, // 19: iop.AdapterConfig.vllm:type_name -> iop.VllmAdapterConfig + 28, // 20: iop.CLIAdapterConfig.profiles:type_name -> iop.CLIAdapterConfig.ProfilesEntry + 19, // 21: iop.CLIProfileConfig.completion_marker:type_name -> iop.CLICompletionMarker + 18, // 22: iop.CLIAdapterConfig.ProfilesEntry.value:type_name -> iop.CLIProfileConfig + 23, // [23:23] is the sub-list for method output_type + 23, // [23:23] is the sub-list for method input_type + 23, // [23:23] is the sub-list for extension type_name + 23, // [23:23] is the sub-list for extension extendee + 0, // [0:23] is the sub-list for field type_name } func init() { file_proto_iop_runtime_proto_init() } @@ -1808,7 +1925,7 @@ func file_proto_iop_runtime_proto_init() { if File_proto_iop_runtime_proto != nil { return } - file_proto_iop_runtime_proto_msgTypes[12].OneofWrappers = []any{ + file_proto_iop_runtime_proto_msgTypes[13].OneofWrappers = []any{ (*AdapterConfig_Cli)(nil), (*AdapterConfig_Ollama)(nil), (*AdapterConfig_Vllm)(nil), @@ -1819,7 +1936,7 @@ func file_proto_iop_runtime_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_iop_runtime_proto_rawDesc), len(file_proto_iop_runtime_proto_rawDesc)), NumEnums: 3, - NumMessages: 24, + NumMessages: 26, NumExtensions: 0, NumServices: 0, }, diff --git a/proto/iop/runtime.proto b/proto/iop/runtime.proto index d4fe0ef..e9a15ec 100644 --- a/proto/iop/runtime.proto +++ b/proto/iop/runtime.proto @@ -48,6 +48,19 @@ message RunEvent { string node_id = 11; } +// EdgeNodeEvent is a general edge-node lifecycle/control event envelope. +// It is separate from RunEvent, which is reserved for adapter execution streams. +message EdgeNodeEvent { + string event_id = 1; + string type = 2; // node.connected | node.disconnected | edge.disconnected | ... + string source = 3; // edge | node + string node_id = 4; + string alias = 5; + string reason = 6; + map metadata = 7; + int64 timestamp = 8; // unix nano +} + message Usage { int32 input_tokens = 1; int32 output_tokens = 2;