refactor(runtime): unify workload execution

api/openapi.yaml

@@ -14,8 +14,6 @@ servers:
 tags:
   - name: scroll
     description: Scroll management and command execution
-  - name: logs
-    description: Log streaming and retrieval
   - name: queue
     description: Command queue management
   - name: websocket
@@ -285,12 +283,6 @@ components:
         additionalProperties:
           $ref: '#/components/schemas/LockStatus'
 
-    ScrollLogMap:
-      type: object
-      additionalProperties:
-        type: array
-        items:
-          type: string
     RuntimeScroll:
       type: object
       required:
@@ -646,25 +638,6 @@ paths:
                 type: object
                 additionalProperties: true
 
-  /api/v1/scrolls/{id}/logs:
-    get:
-      operationId: getScrollLogs
-      summary: Get scroll-scoped logs
-      tags: [logs, runtime]
-      parameters:
-        - name: id
-          in: path
-          required: true
-          schema:
-            type: string
-      responses:
-        '200':
-          description: Logs keyed by procedure
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/ScrollLogMap'
-
   /api/v1/scrolls/{id}/ports:
     get:
       operationId: getScrollPorts

apps/druid/adapters/cli/daemon.go

@@ -138,14 +138,15 @@ func runRuntimeDaemon() error {
 		OperatorServiceAccount: k8sOperatorServiceAccount,
 	}
 	dockerConfig := runtimedocker.Config{WorkerImage: dockerWorkerImage, Storage: dockerStorage, BindRoot: dockerBindRoot, VolumePrefix: dockerVolumePrefix, UIS3Bucket: dockerUIS3Bucket, UIS3PublicBaseURL: dockerUIS3PublicBaseURL, UIS3Region: dockerUIS3Region, UIS3Endpoint: dockerUIS3Endpoint, UIS3Prefix: dockerUIS3Prefix, UIS3AccessKey: dockerUIS3AccessKey, UIS3SecretKey: dockerUIS3SecretKey, UIS3SessionToken: dockerUIS3SessionToken}
-	logManager := services.NewLogManager()
-	consoleService := services.NewConsoleManager(logManager)
-	runtime, err := runtimebackend.NewRuntime(runtimeBackendName, consoleService, runtimeStateDir, runtimebackend.WithKubernetesConfig(kubernetesConfig), runtimebackend.WithDockerConfig(dockerConfig))
+	runtime, err := runtimebackend.NewRuntime(runtimeBackendName, runtimeStateDir, runtimebackend.WithKubernetesConfig(kubernetesConfig), runtimebackend.WithDockerConfig(dockerConfig))
 	if err != nil {
 		return err
 	}
 	manager := services.NewRuntimeScrollManager(runtime.Store)
-	supervisor := appservices.NewRuntimeSupervisor(runtime.Store, manager, runtime.Backend)
+	supervisor, err := appservices.NewRuntimeSupervisor(runtime.Store, manager, runtime)
+	if err != nil {
+		return err
+	}
 	callbacks := appservices.NewWorkerCallbackManager()
 	supervisor.SetWorkerTimeout(runtimeWorkerTimeout)
 	callbackConfig := ports.RuntimeWorkerCallbackConfig{
@@ -181,9 +182,9 @@ func runRuntimeDaemon() error {
 	if err != nil {
 		return err
 	}
-	scrollHandler := runtimehandlers.NewScrollHandler(supervisor, consoleService, logManager, authorizer)
+	scrollHandler := runtimehandlers.NewScrollHandler(supervisor, authorizer)
 	scrollHandler.SetAllowUnauthenticatedPublic(runtimeAllowUnauthenticatedPublic)
-	websocketHandler := runtimehandlers.NewWebsocketHandler(consoleService)
+	websocketHandler := runtimehandlers.NewWebsocketHandler()
 	websocketHandler.SetScrollHandler(scrollHandler)
 	websocketHandler.SetAuthorizer(authorizer)
 	websocketHandler.SetAllowUnauthenticatedPublic(runtimeAllowUnauthenticatedPublic)

apps/druid/adapters/http/handlers/routes.go

@@ -68,8 +68,6 @@ func RegisterPublicRoutes(app *fiber.App, handlers RouteHandlers) {
 	app.Post("/:id/api/v1/command", handlers.Server.RunDaemonCommand)
 	app.Get("/:id/api/v1/queue", handlers.Server.GetDaemonQueue)
 	app.Get("/:id/api/v1/consoles", handlers.Server.GetDaemonConsoles)
-	app.Get("/:id/api/v1/logs", handlers.Server.GetDaemonLogs)
-	app.Get("/:id/api/v1/logs/:stream", handlers.Server.GetDaemonStreamLogs)
 	app.Get("/:id/api/v1/ports", handlers.Server.GetDaemonPorts)
 	app.Get("/:id/api/v1/ui/packages", handlers.Server.GetDaemonUIPackages)
 	app.Post("/:id/api/v1/ui/packages/:scope/publish", handlers.Server.PublishDaemonUIPackage)

apps/druid/adapters/http/handlers/scroll_handler.go

@@ -2,34 +2,28 @@ package handlers
 
 import (
 	"errors"
-	"strings"
 
 	"github.com/gofiber/fiber/v2"
 	appservices "github.com/highcard-dev/daemon/apps/druid/core/services"
 	"github.com/highcard-dev/daemon/internal/api"
 	"github.com/highcard-dev/daemon/internal/core/domain"
 	"github.com/highcard-dev/daemon/internal/core/ports"
-	"github.com/highcard-dev/daemon/internal/core/services"
 )
 
 type ScrollHandler struct {
 	supervisor                 *appservices.RuntimeSupervisor
-	consoleService             *services.ConsoleManager
-	logService                 *services.LogManager
 	authorizer                 ports.AuthorizerServiceInterface
 	allowUnauthenticatedPublic bool
 }
 
-func NewScrollHandler(supervisor *appservices.RuntimeSupervisor, consoleService *services.ConsoleManager, logService *services.LogManager, authorizer ...ports.AuthorizerServiceInterface) *ScrollHandler {
+func NewScrollHandler(supervisor *appservices.RuntimeSupervisor, authorizer ...ports.AuthorizerServiceInterface) *ScrollHandler {
 	var auth ports.AuthorizerServiceInterface
 	if len(authorizer) > 0 {
 		auth = authorizer[0]
 	}
 	return &ScrollHandler{
-		supervisor:     supervisor,
-		consoleService: consoleService,
-		logService:     logService,
-		authorizer:     auth,
+		supervisor: supervisor,
+		authorizer: auth,
 	}
 }
 
@@ -232,22 +226,11 @@ func (h *ScrollHandler) GetScrollConsoles(c *fiber.Ctx, id string) error {
 	if _, err := h.getScroll(id); err != nil {
 		return err
 	}
-	prefix := id + "/"
-	consoles := map[string]*domain.Console{}
-	for consoleID, console := range h.consoleService.GetConsoles() {
-		if strings.HasPrefix(consoleID, prefix) {
-			consoles[strings.TrimPrefix(consoleID, prefix)] = console
-		}
-	}
-	return c.JSON(consoles)
-}
-
-func (h *ScrollHandler) GetScrollLogs(c *fiber.Ctx, id string) error {
-	logs, err := h.scrollLogs(id)
+	consoles, err := h.supervisor.Consoles(id)
 	if err != nil {
 		return err
 	}
-	return c.JSON(logs)
+	return c.JSON(consoles)
 }
 
 func (h *ScrollHandler) GetDaemonScroll(c *fiber.Ctx) error {
@@ -285,27 +268,6 @@ func (h *ScrollHandler) GetDaemonConsoles(c *fiber.Ctx) error {
 	return h.GetScrollConsoles(c, c.Params("id"))
 }
 
-func (h *ScrollHandler) GetDaemonLogs(c *fiber.Ctx) error {
-	logs, err := h.scrollLogs(c.Params("id"))
-	if err != nil {
-		return err
-	}
-	streams := make([]map[string]any, 0, len(logs))
-	for stream, log := range logs {
-		streams = append(streams, map[string]any{"stream": stream, "log": log})
-	}
-	return c.JSON(streams)
-}
-
-func (h *ScrollHandler) GetDaemonStreamLogs(c *fiber.Ctx) error {
-	logs, err := h.scrollLogs(c.Params("id"))
-	if err != nil {
-		return err
-	}
-	stream := c.Params("stream")
-	return c.JSON(map[string]any{"stream": stream, "log": logs[stream]})
-}
-
 func (h *ScrollHandler) GetDaemonPorts(c *fiber.Ctx) error {
 	return h.GetScrollPorts(c, c.Params("id"))
 }
@@ -436,24 +398,3 @@ func (h *ScrollHandler) getScroll(id string) (*domain.RuntimeScroll, error) {
 	}
 	return runtimeScroll, err
 }
-
-func (h *ScrollHandler) scrollLogs(id string) (map[string][]string, error) {
-	if _, err := h.getScroll(id); err != nil {
-		return nil, err
-	}
-	prefix := id + "/"
-	logs := map[string][]string{}
-	for streamID, log := range h.logService.GetStreams() {
-		if !strings.HasPrefix(streamID, prefix) {
-			continue
-		}
-		response := make(chan []byte, 100)
-		log.Req <- response
-		lines := []string{}
-		for line := range response {
-			lines = append(lines, string(line))
-		}
-		logs[strings.TrimPrefix(streamID, prefix)] = lines
-	}
-	return logs, nil
-}

apps/druid/adapters/http/handlers/websocket_handler.go

@@ -1,24 +1,24 @@
 package handlers
 
 import (
+	"context"
+	"io"
 	"time"
 
 	"github.com/gofiber/contrib/websocket"
 	"github.com/highcard-dev/daemon/internal/core/ports"
-	"github.com/highcard-dev/daemon/internal/core/services"
 	"github.com/highcard-dev/daemon/internal/utils/logger"
 	"go.uber.org/zap"
 )
 
 type WebsocketHandler struct {
-	consoleService             *services.ConsoleManager
 	scrolls                    *ScrollHandler
 	authorizer                 ports.AuthorizerServiceInterface
 	allowUnauthenticatedPublic bool
 }
 
-func NewWebsocketHandler(consoleService *services.ConsoleManager) *WebsocketHandler {
-	return &WebsocketHandler{consoleService: consoleService}
+func NewWebsocketHandler() *WebsocketHandler {
+	return &WebsocketHandler{}
 }
 
 func (h *WebsocketHandler) SetScrollHandler(scrolls *ScrollHandler) {
@@ -34,11 +34,7 @@ func (h *WebsocketHandler) SetAllowUnauthenticatedPublic(allow bool) {
 }
 
 func (h *WebsocketHandler) AttachConsole(c *websocket.Conn) {
-	consoleID := c.Params("console")
-	if id := c.Params("id"); id != "" {
-		consoleID = id + "/" + consoleID
-	}
-	h.attach(c, consoleID)
+	h.attach(c, c.Params("id"), c.Params("console"))
 }
 
 func (h *WebsocketHandler) AttachScrollConsole(c *websocket.Conn) {
@@ -49,17 +45,16 @@ func (h *WebsocketHandler) AttachScrollConsole(c *websocket.Conn) {
 	h.AttachConsole(c)
 }
 
-func (h *WebsocketHandler) attach(c *websocket.Conn, consoleID string) {
+func (h *WebsocketHandler) attach(c *websocket.Conn, runtimeID string, consoleID string) {
 	defer c.Close()
-
-	console := h.consoleService.GetConsole(consoleID)
-	if console == nil {
-		logger.Log().Warn("Console not found", zap.String("console", consoleID))
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	session, err := h.scrolls.supervisor.OpenConsole(ctx, runtimeID, consoleID)
+	if err != nil {
+		logger.Log().Warn("Console unavailable", zap.String("runtime", runtimeID), zap.String("console", consoleID), zap.Error(err))
 		return
 	}
-
-	subscription := console.Channel.Subscribe()
-	defer console.Channel.Unsubscribe(subscription)
+	defer session.Close()
 
 	done := make(chan struct{})
 	go func() {
@@ -69,12 +64,31 @@ func (h *WebsocketHandler) attach(c *websocket.Conn, consoleID string) {
 			if err != nil {
 				return
 			}
-			if console.WriteInput != nil {
-				if err := console.WriteInput(string(data)); err != nil {
-					logger.Log().Debug("Failed to write console input", zap.Error(err))
+			if _, err := session.Write(data); err != nil {
+				logger.Log().Debug("Failed to write console input", zap.Error(err))
+				return
+			}
+		}
+	}()
+	output := make(chan []byte, 25)
+	go func() {
+		defer close(output)
+		buffer := make([]byte, 32*1024)
+		for {
+			n, err := session.Read(buffer)
+			if n > 0 {
+				select {
+				case output <- append([]byte(nil), buffer[:n]...):
+				case <-ctx.Done():
 					return
 				}
 			}
+			if err != nil {
+				if err != io.EOF && ctx.Err() == nil {
+					logger.Log().Debug("Console output ended", zap.Error(err))
+				}
+				return
+			}
 		}
 	}()
 
@@ -84,11 +98,11 @@ func (h *WebsocketHandler) attach(c *websocket.Conn, consoleID string) {
 		select {
 		case <-done:
 			return
-		case data, ok := <-subscription:
-			if !ok || data == nil {
+		case data, ok := <-output:
+			if !ok {
 				return
 			}
-			if err := c.WriteMessage(websocket.TextMessage, *data); err != nil {
+			if err := c.WriteMessage(websocket.TextMessage, data); err != nil {
 				return
 			}
 		case <-pingTicker.C:

apps/druid/core/services/runtime_commands.go

@@ -76,10 +76,8 @@ func (s *RuntimeSession) RemoveCommand(command string) error {
 	s.mu.Lock()
 	root := s.runtimeScroll.Root
 	s.mu.Unlock()
-	if stopper, ok := s.runtimeBackend.(interface{ StopCommand(string, string) error }); ok {
-		if err := stopper.StopCommand(root, command); err != nil {
-			return err
-		}
+	if err := s.runtimeBackend.StopCommand(root, command); err != nil {
+		return err
 	}
 
 	s.mu.Lock()

apps/druid/core/services/runtime_consoles.go

@@ -0,0 +1,56 @@
+package services
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/highcard-dev/daemon/internal/core/domain"
+)
+
+func (s *RuntimeSupervisor) Consoles(id string) (map[string]domain.Console, error) {
+	session, err := s.sessionFor(id)
+	if err != nil {
+		return nil, err
+	}
+	return session.consoles(), nil
+}
+
+func (s *RuntimeSupervisor) OpenConsole(ctx context.Context, id string, procedure string) (io.ReadWriteCloser, error) {
+	session, err := s.sessionFor(id)
+	if err != nil {
+		return nil, err
+	}
+	if _, ok := session.consoles()[procedure]; !ok {
+		return nil, fmt.Errorf("console %s is unavailable", procedure)
+	}
+	session.mu.Lock()
+	root := session.runtimeScroll.Root
+	session.mu.Unlock()
+	return session.runtimeBackend.OpenConsole(ctx, root, procedure)
+}
+
+func (s *RuntimeSession) consoles() map[string]domain.Console {
+	s.mu.Lock()
+	statuses := copyProcedureStatuses(s.runtimeScroll.Procedures)
+	s.mu.Unlock()
+	consoles := map[string]domain.Console{}
+	for commandName, command := range s.scrollService.GetFile().Commands {
+		if command == nil {
+			continue
+		}
+		for index, procedure := range command.Procedures {
+			name := domain.ProcedureName(commandName, index, procedure)
+			status, ok := statuses[commandName][name]
+			if !ok || procedure == nil || !procedure.IsContainer() {
+				continue
+			}
+			consoleType := domain.ConsoleTypeContainer
+			if procedure.TTY {
+				consoleType = domain.ConsoleTypeTTY
+			}
+			consoles[name] = domain.Console{Type: consoleType, InputMode: "stdin", Exit: status.ExitCode}
+		}
+	}
+	return consoles
+}