toki/iop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: console emitter interface, persistent status, and profile proto message updates

Browse source 
- Add console emitter interface for event-driven console output
- Implement persistent cancel reason and explicit completion tracking
- Add profile proto message definitions
- Update edge and node transport layers with adapter execution terminology
- Add new packages/events module
- Update architecture documentation and README files
This commit is contained in:
toki 2026-05-16 21:16:48 +09:00
parent 5131dc9e2c
commit cf3760f394
24 changed files with 1346 additions and 333 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
README.md
View file

@ -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는 단계적으로 확장한다.

15
apps/edge/README.md
View file

@ -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 여러 개

216
apps/edge/cmd/edge/console.go
View file

@ -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 <id|alias>")
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
}

107
apps/edge/cmd/edge/console_events.go
View file

@ -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

105
apps/edge/cmd/edge/console_test.go
View file

@ -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)
}
}

14
apps/edge/internal/bootstrap/module.go
View file

@ -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
}

152
apps/edge/internal/events/bus.go Normal file
View file

@ -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:
}
}

62
apps/edge/internal/events/bus_test.go Normal file
View file

@ -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")
}
}

282
apps/edge/internal/service/service.go Normal file
View file

@ -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
}

45
apps/edge/internal/service/service_test.go Normal file
View file

@ -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)
}
}

31
apps/edge/internal/transport/integration_test.go
View file

@ -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")
}
}

73
apps/edge/internal/transport/server.go
View file

@ -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] + "..."

34
apps/edge/internal/transport/server_test.go
View file

@ -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" {

13
apps/node/README.md
View file

@ -56,6 +56,18 @@ edge에서 실행 요청을 받으면 node는 해당 입력을 선택된 adapter
[node-message] <adapter output>
```
## 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초 간격

18
apps/node/internal/bootstrap/module.go
View file

@ -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())
}
}

2
apps/node/internal/transport/client.go
View file

@ -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()),

20
apps/node/internal/transport/integration_test.go
View file

@ -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")
}
}

4
apps/node/internal/transport/parser.go
View file

@ -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)

34
apps/node/internal/transport/parser_test.go
View file

@ -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{

67
apps/node/internal/transport/session.go
View file

@ -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
}

6
docs/architecture.md
View file

@ -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`으로 표현한다.

36
packages/events/events.go Normal file
View file

@ -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(),
}
}

323
proto/gen/iop/runtime.pb.go
View file

@ -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,
},

13
proto/iop/runtime.proto
View file

@ -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<string, string> metadata = 7;
int64 timestamp = 8; // unix nano
}
message Usage {
int32 input_tokens = 1;
int32 output_tokens = 2;