리팩터: node 서버 코드 제거 및 클라이언트 구조로 전환
- transport/{server,session,frame,codec,tls}.go 삭제
- packages/protocol/protocol.go 삭제 (JSON 기반 메시지 타입)
- bootstrap/module.go에서 transport.Server 와이어링 제거
- TransportConf.Listen → EdgeAddr (node는 서버가 아닌 클라이언트)
- configs/node.yaml transport.listen → transport.edge_addr
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
2d6fde0876
commit
173d2586e8
10 changed files with 6 additions and 590 deletions
|
|
@ -36,7 +36,7 @@ func rootCmd() *cobra.Command {
|
|||
func serveCmd() *cobra.Command {
|
||||
return &cobra.Command{
|
||||
Use: "serve",
|
||||
Short: "Start the IOP node server",
|
||||
Short: "Start the IOP node (connects to edge)",
|
||||
RunE: func(cmd *cobra.Command, _ []string) error {
|
||||
cfg, err := config.Load(cfgFile)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -16,7 +16,6 @@ import (
|
|||
"iop/apps/node/internal/router"
|
||||
"iop/apps/node/internal/runtime"
|
||||
"iop/apps/node/internal/store"
|
||||
"iop/apps/node/internal/transport"
|
||||
"iop/packages/config"
|
||||
"iop/packages/observability"
|
||||
)
|
||||
|
|
@ -66,25 +65,11 @@ func Module(cfg *config.NodeConfig) fx.Option {
|
|||
) *node.Node {
|
||||
return node.New(cfg, r, reg, st, logger)
|
||||
},
|
||||
|
||||
func(n *node.Node, logger *zap.Logger) *transport.Server {
|
||||
return transport.NewServer(n, logger)
|
||||
},
|
||||
),
|
||||
|
||||
fx.Invoke(func(lc fx.Lifecycle, srv *transport.Server, cfg *config.NodeConfig, logger *zap.Logger) {
|
||||
fx.Invoke(func(lc fx.Lifecycle, cfg *config.NodeConfig, logger *zap.Logger) {
|
||||
lc.Append(fx.Hook{
|
||||
OnStart: func(ctx context.Context) error {
|
||||
tlsCfg, err := transport.TLSFromConfig(cfg.TLS)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tlsCfg != nil {
|
||||
srv.SetTLS(tlsCfg)
|
||||
}
|
||||
if err := srv.Start(ctx, cfg.Transport.Listen); err != nil {
|
||||
return err
|
||||
}
|
||||
go func() {
|
||||
if err := observability.ServeMetrics(cfg.Metrics.Port); err != nil {
|
||||
logger.Warn("metrics server exited", zap.Error(err))
|
||||
|
|
@ -92,9 +77,7 @@ func Module(cfg *config.NodeConfig) fx.Option {
|
|||
}()
|
||||
return nil
|
||||
},
|
||||
OnStop: func(_ context.Context) error {
|
||||
return srv.Stop()
|
||||
},
|
||||
OnStop: func(_ context.Context) error { return nil },
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,32 +0,0 @@
|
|||
package transport
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"iop/packages/protocol"
|
||||
)
|
||||
|
||||
// encodeEnvelope marshals env to JSON bytes (the frame payload).
|
||||
func encodeEnvelope(env *protocol.Envelope) ([]byte, error) {
|
||||
return json.Marshal(env)
|
||||
}
|
||||
|
||||
// decodeEnvelope unmarshals frame payload bytes into an Envelope.
|
||||
func decodeEnvelope(data []byte) (*protocol.Envelope, error) {
|
||||
var env protocol.Envelope
|
||||
if err := json.Unmarshal(data, &env); err != nil {
|
||||
return nil, fmt.Errorf("decode envelope: %w", err)
|
||||
}
|
||||
return &env, nil
|
||||
}
|
||||
|
||||
// encodePayload marshals an inner message to JSON bytes for Envelope.Payload.
|
||||
func encodePayload(v any) ([]byte, error) {
|
||||
return json.Marshal(v)
|
||||
}
|
||||
|
||||
// decodePayload unmarshals Envelope.Payload into dst.
|
||||
func decodePayload(payload []byte, dst any) error {
|
||||
return json.Unmarshal(payload, dst)
|
||||
}
|
||||
|
|
@ -1,63 +0,0 @@
|
|||
// Package transport implements IOP's TCP framing layer.
|
||||
//
|
||||
// Wire format (same endianness as proto-socket):
|
||||
// [4 bytes big-endian uint32: payload length] [payload bytes]
|
||||
//
|
||||
// Payload is a JSON-encoded Envelope (see codec.go).
|
||||
// TODO: switch payload encoding to protobuf once proto/gen is available.
|
||||
package transport
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"io"
|
||||
)
|
||||
|
||||
const (
|
||||
frameLenBytes = 4
|
||||
maxFrameSize = 64 << 20 // 64 MiB — same limit as proto-socket
|
||||
)
|
||||
|
||||
// readFrame reads one length-prefixed frame from r.
|
||||
func readFrame(r io.Reader) ([]byte, error) {
|
||||
hdr := make([]byte, frameLenBytes)
|
||||
if _, err := io.ReadFull(r, hdr); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
length := binary.BigEndian.Uint32(hdr)
|
||||
if length == 0 {
|
||||
return []byte{}, nil
|
||||
}
|
||||
if length > maxFrameSize {
|
||||
return nil, fmt.Errorf("frame too large: %d bytes (max %d)", length, maxFrameSize)
|
||||
}
|
||||
buf := make([]byte, length)
|
||||
if _, err := io.ReadFull(r, buf); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return buf, nil
|
||||
}
|
||||
|
||||
// writeFrame writes payload as a length-prefixed frame to w.
|
||||
func writeFrame(w io.Writer, payload []byte) error {
|
||||
hdr := make([]byte, frameLenBytes)
|
||||
binary.BigEndian.PutUint32(hdr, uint32(len(payload)))
|
||||
if err := writeFull(w, hdr); err != nil {
|
||||
return err
|
||||
}
|
||||
return writeFull(w, payload)
|
||||
}
|
||||
|
||||
func writeFull(w io.Writer, b []byte) error {
|
||||
for len(b) > 0 {
|
||||
n, err := w.Write(b)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n == 0 {
|
||||
return io.ErrShortWrite
|
||||
}
|
||||
b = b[n:]
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
@ -1,135 +0,0 @@
|
|||
package transport
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// Server is the IOP TCP transport server.
|
||||
type Server struct {
|
||||
handler Handler
|
||||
logger *zap.Logger
|
||||
tlsConfig *tls.Config
|
||||
|
||||
listener net.Listener
|
||||
started atomic.Bool
|
||||
mu sync.Mutex
|
||||
sessions map[string]*Session
|
||||
nextID atomic.Uint64
|
||||
}
|
||||
|
||||
func NewServer(handler Handler, logger *zap.Logger) *Server {
|
||||
return &Server{
|
||||
handler: handler,
|
||||
logger: logger,
|
||||
sessions: make(map[string]*Session),
|
||||
}
|
||||
}
|
||||
|
||||
// SetTLS configures mTLS for the server. Must be called before Start.
|
||||
func (s *Server) SetTLS(cfg *tls.Config) {
|
||||
s.tlsConfig = cfg
|
||||
}
|
||||
|
||||
// Start begins accepting connections on addr.
|
||||
func (s *Server) Start(ctx context.Context, addr string) error {
|
||||
ln, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("transport: listen %s: %w", addr, err)
|
||||
}
|
||||
if s.tlsConfig != nil {
|
||||
ln = tls.NewListener(ln, s.tlsConfig)
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
s.listener = ln
|
||||
s.started.Store(true)
|
||||
s.mu.Unlock()
|
||||
|
||||
s.logger.Info("transport listening",
|
||||
zap.String("addr", addr),
|
||||
zap.Bool("tls", s.tlsConfig != nil),
|
||||
)
|
||||
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
_ = s.Stop()
|
||||
}()
|
||||
go s.acceptLoop(ctx)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop closes the listener and all active sessions.
|
||||
func (s *Server) Stop() error {
|
||||
if !s.started.CompareAndSwap(true, false) {
|
||||
return nil
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
ln := s.listener
|
||||
sessions := make([]*Session, 0, len(s.sessions))
|
||||
for _, sess := range s.sessions {
|
||||
sessions = append(sessions, sess)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
var err error
|
||||
if ln != nil {
|
||||
err = ln.Close()
|
||||
}
|
||||
for _, sess := range sessions {
|
||||
sess.Close()
|
||||
}
|
||||
s.logger.Info("transport stopped")
|
||||
return err
|
||||
}
|
||||
|
||||
// SessionCount returns the number of active sessions.
|
||||
func (s *Server) SessionCount() int {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return len(s.sessions)
|
||||
}
|
||||
|
||||
func (s *Server) acceptLoop(ctx context.Context) {
|
||||
for {
|
||||
conn, err := s.listener.Accept()
|
||||
if err != nil {
|
||||
if !s.started.Load() || errors.Is(err, net.ErrClosed) {
|
||||
return
|
||||
}
|
||||
s.logger.Warn("accept error", zap.Error(err))
|
||||
continue
|
||||
}
|
||||
go s.handleConn(ctx, conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleConn(ctx context.Context, conn net.Conn) {
|
||||
id := fmt.Sprintf("sess-%d", s.nextID.Add(1))
|
||||
sess := newSession(id, conn, s.handler, s.logger)
|
||||
|
||||
s.mu.Lock()
|
||||
s.sessions[id] = sess
|
||||
s.mu.Unlock()
|
||||
|
||||
s.logger.Info("client connected",
|
||||
zap.String("session_id", id),
|
||||
zap.String("remote", conn.RemoteAddr().String()),
|
||||
)
|
||||
|
||||
sess.readLoop(ctx)
|
||||
|
||||
s.mu.Lock()
|
||||
delete(s.sessions, id)
|
||||
s.mu.Unlock()
|
||||
|
||||
s.logger.Info("client disconnected", zap.String("session_id", id))
|
||||
}
|
||||
|
|
@ -1,228 +0,0 @@
|
|||
package transport
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"iop/packages/protocol"
|
||||
)
|
||||
|
||||
const (
|
||||
heartbeatInterval = 30 * time.Second
|
||||
heartbeatTimeout = 10 * time.Second
|
||||
)
|
||||
|
||||
// Handler handles decoded messages arriving on a Session.
|
||||
type Handler interface {
|
||||
OnRunRequest(ctx context.Context, sess *Session, req *protocol.RunRequest) error
|
||||
OnCapabilityRequest(ctx context.Context, sess *Session) (*protocol.CapabilityResponse, error)
|
||||
OnCancel(ctx context.Context, sess *Session, req *protocol.CancelRequest) error
|
||||
}
|
||||
|
||||
// Session represents one active TCP connection from a client.
|
||||
type Session struct {
|
||||
id string
|
||||
conn net.Conn
|
||||
handler Handler
|
||||
logger *zap.Logger
|
||||
|
||||
writeMu sync.Mutex
|
||||
closed atomic.Bool
|
||||
|
||||
cancelFns sync.Map // run_id → context.CancelFunc
|
||||
}
|
||||
|
||||
func newSession(id string, conn net.Conn, handler Handler, logger *zap.Logger) *Session {
|
||||
return &Session{
|
||||
id: id,
|
||||
conn: conn,
|
||||
handler: handler,
|
||||
logger: logger.With(zap.String("session_id", id)),
|
||||
}
|
||||
}
|
||||
|
||||
// ID returns the session identifier.
|
||||
func (s *Session) ID() string { return s.id }
|
||||
|
||||
// Send encodes env as a frame and writes it to the connection.
|
||||
func (s *Session) Send(env *protocol.Envelope) error {
|
||||
if s.closed.Load() {
|
||||
return fmt.Errorf("session %s: closed", s.id)
|
||||
}
|
||||
payload, err := encodeEnvelope(env)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.writeMu.Lock()
|
||||
defer s.writeMu.Unlock()
|
||||
return writeFrame(s.conn, payload)
|
||||
}
|
||||
|
||||
// Close terminates the session.
|
||||
func (s *Session) Close() {
|
||||
if s.closed.CompareAndSwap(false, true) {
|
||||
s.conn.Close()
|
||||
s.logger.Info("session closed")
|
||||
}
|
||||
}
|
||||
|
||||
// RegisterCancel stores a cancel function for a run, enabling CancelRequest handling.
|
||||
func (s *Session) RegisterCancel(runID string, cancel context.CancelFunc) {
|
||||
s.cancelFns.Store(runID, cancel)
|
||||
}
|
||||
|
||||
// DeregisterCancel removes a run's cancel function after completion.
|
||||
func (s *Session) DeregisterCancel(runID string) {
|
||||
s.cancelFns.Delete(runID)
|
||||
}
|
||||
|
||||
// CancelRun invokes the cancel function for a run if one is registered.
|
||||
func (s *Session) CancelRun(runID string) {
|
||||
if v, ok := s.cancelFns.Load(runID); ok {
|
||||
v.(context.CancelFunc)()
|
||||
}
|
||||
}
|
||||
|
||||
// readLoop drives the session until the connection closes or an error occurs.
|
||||
func (s *Session) readLoop(ctx context.Context) {
|
||||
defer s.Close()
|
||||
|
||||
// Perform capability handshake: server sends capability request,
|
||||
// client responds (optional — here we accept any client proactively).
|
||||
go s.heartbeatLoop(ctx)
|
||||
|
||||
for {
|
||||
if s.closed.Load() {
|
||||
return
|
||||
}
|
||||
|
||||
raw, err := readFrame(s.conn)
|
||||
if err != nil {
|
||||
if !s.closed.Load() {
|
||||
s.logger.Debug("read frame error", zap.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
if len(raw) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
env, err := decodeEnvelope(raw)
|
||||
if err != nil {
|
||||
s.logger.Warn("decode envelope failed", zap.Error(err))
|
||||
continue
|
||||
}
|
||||
|
||||
if err := s.dispatch(ctx, env); err != nil {
|
||||
s.logger.Warn("dispatch error", zap.String("type", string(env.Type)), zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Session) dispatch(ctx context.Context, env *protocol.Envelope) error {
|
||||
switch env.Type {
|
||||
case protocol.TypeHeartbeat:
|
||||
return s.handleHeartbeat(env)
|
||||
|
||||
case protocol.TypeRunRequest:
|
||||
var req protocol.RunRequest
|
||||
if err := decodePayload(env.Payload, &req); err != nil {
|
||||
return s.sendError(env.RequestID, "DECODE_ERROR", err.Error())
|
||||
}
|
||||
go func() {
|
||||
if err := s.handler.OnRunRequest(ctx, s, &req); err != nil {
|
||||
_ = s.sendError(env.RequestID, "RUN_ERROR", err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
case protocol.TypeCapabilityRequest:
|
||||
resp, err := s.handler.OnCapabilityRequest(ctx, s)
|
||||
if err != nil {
|
||||
return s.sendError(env.RequestID, "CAPABILITY_ERROR", err.Error())
|
||||
}
|
||||
return s.sendCapabilityResponse(env.RequestID, resp)
|
||||
|
||||
case protocol.TypeCancelRequest:
|
||||
var req protocol.CancelRequest
|
||||
if err := decodePayload(env.Payload, &req); err != nil {
|
||||
return s.sendError(env.RequestID, "DECODE_ERROR", err.Error())
|
||||
}
|
||||
return s.handler.OnCancel(ctx, s, &req)
|
||||
|
||||
default:
|
||||
s.logger.Warn("unknown message type", zap.String("type", string(env.Type)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Session) handleHeartbeat(env *protocol.Envelope) error {
|
||||
// Echo heartbeat back
|
||||
resp, err := encodePayload(protocol.Heartbeat{Timestamp: time.Now().UnixNano()})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return s.Send(&protocol.Envelope{
|
||||
ProtocolVersion: protocol.ProtocolVersion,
|
||||
RequestID: env.RequestID,
|
||||
Type: protocol.TypeHeartbeat,
|
||||
Payload: resp,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Session) sendError(requestID, code, message string) error {
|
||||
payload, err := encodePayload(protocol.ErrorMsg{Code: code, Message: message})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return s.Send(&protocol.Envelope{
|
||||
ProtocolVersion: protocol.ProtocolVersion,
|
||||
RequestID: requestID,
|
||||
Type: protocol.TypeError,
|
||||
Payload: payload,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Session) sendCapabilityResponse(requestID string, resp *protocol.CapabilityResponse) error {
|
||||
payload, err := encodePayload(resp)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return s.Send(&protocol.Envelope{
|
||||
ProtocolVersion: protocol.ProtocolVersion,
|
||||
RequestID: requestID,
|
||||
Type: protocol.TypeCapabilityResponse,
|
||||
Payload: payload,
|
||||
})
|
||||
}
|
||||
|
||||
// heartbeatLoop periodically sends heartbeat frames to detect dead connections.
|
||||
func (s *Session) heartbeatLoop(ctx context.Context) {
|
||||
ticker := time.NewTicker(heartbeatInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if s.closed.Load() {
|
||||
return
|
||||
}
|
||||
payload, _ := encodePayload(protocol.Heartbeat{Timestamp: time.Now().UnixNano()})
|
||||
if err := s.Send(&protocol.Envelope{
|
||||
ProtocolVersion: protocol.ProtocolVersion,
|
||||
Type: protocol.TypeHeartbeat,
|
||||
Payload: payload,
|
||||
}); err != nil {
|
||||
s.logger.Debug("heartbeat send failed", zap.Error(err))
|
||||
s.Close()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,17 +0,0 @@
|
|||
package transport
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
|
||||
"iop/packages/auth"
|
||||
"iop/packages/config"
|
||||
)
|
||||
|
||||
// TLSFromConfig builds a server *tls.Config from NodeConfig.TLS.
|
||||
// Returns nil if TLS is disabled.
|
||||
func TLSFromConfig(cfg config.TLSConf) (*tls.Config, error) {
|
||||
if !cfg.Enabled {
|
||||
return nil, nil
|
||||
}
|
||||
return auth.LoadServerTLS(cfg.Cert, cfg.Key, cfg.CA)
|
||||
}
|
||||
|
|
@ -3,7 +3,7 @@ node:
|
|||
name: "iop-node-local"
|
||||
|
||||
transport:
|
||||
listen: "0.0.0.0:9090"
|
||||
edge_addr: "localhost:9090"
|
||||
|
||||
tls:
|
||||
enabled: false
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ type NodeInfo struct {
|
|||
}
|
||||
|
||||
type TransportConf struct {
|
||||
Listen string `mapstructure:"listen" yaml:"listen"`
|
||||
EdgeAddr string `mapstructure:"edge_addr" yaml:"edge_addr"`
|
||||
}
|
||||
|
||||
type TLSConf struct {
|
||||
|
|
@ -90,7 +90,7 @@ func Load(cfgFile string) (*NodeConfig, error) {
|
|||
}
|
||||
|
||||
func setDefaults(v *viper.Viper) {
|
||||
v.SetDefault("transport.listen", "0.0.0.0:9090")
|
||||
v.SetDefault("transport.edge_addr", "localhost:9090")
|
||||
v.SetDefault("runtime.concurrency", 4)
|
||||
v.SetDefault("runtime.workspace_root", "/tmp/iop/workspace")
|
||||
v.SetDefault("sqlite.dsn", "file:iop.db?cache=shared&mode=rwc")
|
||||
|
|
|
|||
|
|
@ -1,92 +0,0 @@
|
|||
// Package protocol defines the IOP wire-level message types.
|
||||
// The transport layer uses 4-byte big-endian length-prefixed frames.
|
||||
// Each frame carries a JSON-encoded Envelope; Envelope.Payload holds
|
||||
// the JSON-encoded inner message (RunRequest, RunEvent, etc.).
|
||||
//
|
||||
// TODO: replace JSON encoding with protobuf once proto/iop/runtime.proto
|
||||
// is compiled with protoc-gen-go.
|
||||
package protocol
|
||||
|
||||
const ProtocolVersion uint32 = 1
|
||||
|
||||
type MessageType string
|
||||
|
||||
const (
|
||||
TypeRunRequest MessageType = "run.request"
|
||||
TypeRunEvent MessageType = "run.event"
|
||||
TypeHeartbeat MessageType = "heartbeat"
|
||||
TypeCancelRequest MessageType = "cancel.request"
|
||||
TypeError MessageType = "error"
|
||||
TypeCapabilityRequest MessageType = "capability.request"
|
||||
TypeCapabilityResponse MessageType = "capability.response"
|
||||
)
|
||||
|
||||
// Envelope is the outer wrapper for every IOP TCP message.
|
||||
type Envelope struct {
|
||||
ProtocolVersion uint32 `json:"protocol_version"`
|
||||
RequestID string `json:"request_id"`
|
||||
RunID string `json:"run_id,omitempty"`
|
||||
StreamID string `json:"stream_id,omitempty"`
|
||||
Type MessageType `json:"type"`
|
||||
Payload []byte `json:"payload"`
|
||||
}
|
||||
|
||||
// RunRequest initiates a model execution on the node.
|
||||
type RunRequest struct {
|
||||
RunID string `json:"run_id"`
|
||||
Adapter string `json:"adapter,omitempty"`
|
||||
Model string `json:"model,omitempty"`
|
||||
Workspace string `json:"workspace,omitempty"`
|
||||
Policy map[string]any `json:"policy,omitempty"`
|
||||
Input map[string]any `json:"input"`
|
||||
TimeoutSec int `json:"timeout_sec,omitempty"`
|
||||
Metadata map[string]string `json:"metadata,omitempty"`
|
||||
}
|
||||
|
||||
// RunEvent is a streaming execution event sent from node to client.
|
||||
type RunEvent struct {
|
||||
RunID string `json:"run_id"`
|
||||
Type string `json:"type"` // start | delta | complete | error
|
||||
Delta string `json:"delta,omitempty"`
|
||||
Message string `json:"message,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Usage *UsageInfo `json:"usage,omitempty"`
|
||||
Metadata map[string]string `json:"metadata,omitempty"`
|
||||
Timestamp int64 `json:"timestamp"` // unix nano
|
||||
}
|
||||
|
||||
type UsageInfo struct {
|
||||
InputTokens int `json:"input_tokens"`
|
||||
OutputTokens int `json:"output_tokens"`
|
||||
}
|
||||
|
||||
// Heartbeat keeps the TCP connection alive.
|
||||
type Heartbeat struct {
|
||||
Timestamp int64 `json:"timestamp"`
|
||||
}
|
||||
|
||||
// CancelRequest asks the node to cancel a running execution.
|
||||
type CancelRequest struct {
|
||||
RunID string `json:"run_id"`
|
||||
}
|
||||
|
||||
// ErrorMsg is returned when a request fails at the transport layer.
|
||||
type ErrorMsg struct {
|
||||
Code string `json:"code"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
// CapabilityRequest asks the node what adapters/models it supports.
|
||||
type CapabilityRequest struct{}
|
||||
|
||||
// CapabilityResponse lists the node's available adapters.
|
||||
type CapabilityResponse struct {
|
||||
NodeID string `json:"node_id"`
|
||||
Adapters []AdapterInfo `json:"adapters"`
|
||||
}
|
||||
|
||||
type AdapterInfo struct {
|
||||
Name string `json:"name"`
|
||||
Models []string `json:"models"`
|
||||
MaxConcurrency int `json:"max_concurrency"`
|
||||
}
|
||||
Loading…
Reference in a new issue