package service import ( "context" "fmt" "sync" "time" "google.golang.org/protobuf/types/known/structpb" eventpkg "iop/packages/go/events" iop "iop/proto/gen/iop" ) type SubmitRunRequest struct { NodeRef string RunID string Adapter string Target string SessionID string Workspace string Prompt string Input map[string]any Background bool TimeoutSec int Metadata map[string]string } // RunDispatch describes a dispatched run in surface-neutral terms. It is the // metadata any caller (console, HTTP, future RPC) needs after submission. type RunDispatch struct { RunID string NodeID string NodeLabel string Adapter string Target string SessionID string Background bool TimeoutSec int } // RunStream carries asynchronous events for a dispatched foreground run. // Background runs leave both channels nil. type RunStream struct { Events <-chan *iop.RunEvent NodeEvents <-chan *iop.EdgeNodeEvent } // RunResult describes the interface for a submitted execution run, allowing // surface-neutral consumption of its dispatch info and event stream. type RunResult interface { Dispatch() RunDispatch Stream() RunStream Close() WaitTimeout() time.Duration } // RunHandle is the legacy combined surface kept for existing callers; it // embeds the surface-neutral DTOs so HTTP/API code can consume RunDispatch // or RunStream directly without depending on this struct. type RunHandle struct { RunDispatch RunStream closeOnce sync.Once close func() } func (h *RunHandle) Close() { if h != nil && h.close != nil { h.closeOnce.Do(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 (h *RunHandle) Dispatch() RunDispatch { if h == nil { return RunDispatch{} } return h.RunDispatch } func (h *RunHandle) Stream() RunStream { if h == nil { return RunStream{} } return h.RunStream } func (s *Service) SubmitRun(_ context.Context, req SubmitRunRequest) (RunResult, 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{ RunDispatch: RunDispatch{ RunID: runID, NodeID: entry.NodeID, NodeLabel: nodeLabel(entry), Adapter: runReq.GetAdapter(), Target: runReq.GetTarget(), SessionID: runReq.GetSessionId(), Background: runReq.GetBackground(), TimeoutSec: int(runReq.GetTimeoutSec()), }, RunStream: RunStream{ Events: runEvents, NodeEvents: nodeEvents, }, close: func() { if unregisterRun != nil { unregisterRun() } if unregisterNode != nil { unregisterNode() } }, }, nil } type CancelRunRequest struct { NodeRef string RunID string Adapter string Target string SessionID string } func BuildCancelRunRequest(req CancelRunRequest) *iop.CancelRequest { return &iop.CancelRequest{ RunId: req.RunID, Adapter: req.Adapter, Target: req.Target, SessionId: NormalizeSessionID(req.SessionID), Action: iop.CancelAction_CANCEL_ACTION_CANCEL_RUN, } } func (s *Service) CancelRun(_ context.Context, req CancelRunRequest) (CommandResult, error) { entry, err := s.ResolveNode(req.NodeRef) if err != nil { return CommandResult{}, err } cancelReq := BuildCancelRunRequest(req) if err := entry.Client.Send(cancelReq); err != nil { return CommandResult{}, err } return CommandResult{ NodeID: entry.NodeID, NodeLabel: nodeLabel(entry), SessionID: cancelReq.GetSessionId(), }, nil } type TerminateSessionRequest struct { NodeRef string Adapter string Target string SessionID string } // CommandResult is the surface-neutral acknowledgement for one-shot node // commands (terminate-session, future control RPCs). type CommandResult struct { NodeID string NodeLabel string SessionID string } // TerminateSessionResult is retained as an alias for backward compatibility. type TerminateSessionResult = CommandResult 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 } 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) { inputMap := make(map[string]any, len(req.Input)+1) for k, v := range req.Input { inputMap[k] = v } if req.Prompt != "" { if _, ok := inputMap["prompt"]; !ok { inputMap["prompt"] = req.Prompt } } input, err := structpb.NewStruct(inputMap) if err != nil { return nil, "", err } runID := req.RunID if runID == "" { runID = NewRunID() } metadata := make(map[string]string, len(req.Metadata)) for k, v := range req.Metadata { metadata[k] = v } return &iop.RunRequest{ RunId: runID, Adapter: req.Adapter, Target: req.Target, Workspace: req.Workspace, 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 }