Feat v0.5.0 realtime dialog (#30)

Co-authored-by: Jeffrey Smith <jasafpro@gmail.com>
Co-committed-by: Jeffrey Smith <jasafpro@gmail.com>

server/events/bus_test.go

@@ -117,6 +117,12 @@ func TestRouteFor(t *testing.T) {
 		// Extension lifecycle
 		{"extension.loaded", DirLocal},
 		{"extension.error", DirLocal},
+		// Realtime (v0.5.0)
+		{"realtime.chat.message", DirToClient},
+		{"realtime.custom.event", DirToClient},
+		// Room management (v0.5.0)
+		{"room.subscribe", DirFromClient},
+		{"room.unsubscribe", DirFromClient},
 	}
 
 	for _, tt := range tests {
@@ -190,6 +196,27 @@ func TestWaitFor_CleansUpSubscription(t *testing.T) {
 	bus.Publish(Event{Label: "tool.result.cleanup", Payload: json.RawMessage(`{}`)})
 }
 
+func TestRealtimeRouting(t *testing.T) {
+	// realtime.* events should be sent to clients but not accepted from them
+	if !ShouldSendToClient("realtime.chat.message") {
+		t.Error("realtime.* should be sent to client")
+	}
+	if ShouldAcceptFromClient("realtime.chat.message") {
+		t.Error("realtime.* should NOT be accepted from client")
+	}
+
+	// room.subscribe/unsubscribe should be accepted from clients
+	if !ShouldAcceptFromClient("room.subscribe") {
+		t.Error("room.subscribe should be accepted from client")
+	}
+	if !ShouldAcceptFromClient("room.unsubscribe") {
+		t.Error("room.unsubscribe should be accepted from client")
+	}
+	if ShouldSendToClient("room.subscribe") {
+		t.Error("room.subscribe should NOT be sent to client")
+	}
+}
+
 func TestToolCallRouteToClient(t *testing.T) {
 	if !ShouldSendToClient("tool.call.abc123") {
 		t.Error("tool.call.* should be sent to client")

server/events/types.go

@@ -87,6 +87,13 @@ var routeTable = map[string]Direction{
 	"trigger.fired":     DirLocal,   // Trigger invocation event (observability)
 	"trigger.error":     DirLocal,   // Trigger execution error
 
+	// Realtime (v0.5.0): extension-published events, room-scoped
+	"realtime.":         DirToClient,
+
+	// Room management (v0.5.0): client room join/leave
+	"room.subscribe":    DirFromClient,
+	"room.unsubscribe":  DirFromClient,
+
 	// Heartbeat
 	"ping":  DirFromClient,
 	"pong":  DirToClient,

server/events/ws.go

@@ -316,6 +316,26 @@ func (c *Conn) readPump() {
 			continue
 		}
 
+		// v0.5.0: Room management — intercept before bus publish (like ping)
+		if event.Label == "room.subscribe" {
+			var req struct {
+				Room string `json:"room"`
+			}
+			if json.Unmarshal(event.Payload, &req) == nil && req.Room != "" && len(c.rooms) < 100 {
+				c.JoinRoom(req.Room)
+			}
+			continue
+		}
+		if event.Label == "room.unsubscribe" {
+			var req struct {
+				Room string `json:"room"`
+			}
+			if json.Unmarshal(event.Payload, &req) == nil && req.Room != "" {
+				c.LeaveRoom(req.Room)
+			}
+			continue
+		}
+
 		// Validate: only accept events allowed from clients
 		if !ShouldAcceptFromClient(event.Label) {
 			continue

server/main.go

@@ -168,6 +168,8 @@ func main() {
 	if cfg.StoragePath != "" {
 		starlarkRunner.SetPackagesDir(cfg.StoragePath + "/packages")
 	}
+	// v0.5.0: realtime module — extension publish to WebSocket channels
+	starlarkRunner.SetBus(bus)
 	// v0.38.5: allow extensions to reach private IPs (self-hosted Gitea, etc.)
 	if os.Getenv("EXT_ALLOW_PRIVATE_IPS") == "true" {
 		starlarkRunner.SetAllowPrivateIPs(true)

server/models/models_extension_perm.go

@@ -35,6 +35,7 @@ const (
 	ExtPermWorkflowAccess       = "workflow.access"   // v0.30.2: workflow definition + stage data access
 	ExtPermConnectionsRead      = "connections.read"  // v0.38.1: extension connection resolution
 	ExtPermTriggersRegister     = "triggers.register" // v0.2.2: event/webhook trigger registration
+	ExtPermRealtimePublish      = "realtime.publish"  // v0.5.0: publish events to realtime channels
 )
 
 // ValidExtensionPermissions is the set of recognized permission keys.
@@ -48,6 +49,7 @@ var ValidExtensionPermissions = map[string]bool{
 	ExtPermWorkflowAccess:       true,
 	ExtPermConnectionsRead:      true,
 	ExtPermTriggersRegister:     true,
+	ExtPermRealtimePublish:      true,
 }
 
 // ── Extension Permission Model ───────────────

server/sandbox/realtime_module.go

@@ -0,0 +1,154 @@
+// Package sandbox — realtime_module.go
+//
+// v0.5.0: Starlark realtime.publish() module.
+//
+// Allows extensions to publish events to WebSocket channels.
+// Events are scoped to rooms — only clients subscribed to the
+// channel receive the event.
+//
+// Starlark API:
+//   realtime.publish(channel, event, data={})
+//
+// Requires permission: realtime.publish
+package sandbox
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"go.starlark.net/starlark"
+	"go.starlark.net/starlarkstruct"
+
+	"switchboard-core/events"
+)
+
+// maxRealtimePayload is the payload size limit (pg_notify safe).
+const maxRealtimePayload = 7168 // 7KB
+
+// reservedPrefixes are event label prefixes that extensions cannot publish to.
+var reservedPrefixes = []string{
+	"user.", "system.", "model.", "role.", "notification.",
+	"workflow.", "plugin.", "internal.", "db.", "tool.",
+	"trigger.", "extension.", "ping", "pong", "room.",
+}
+
+// BuildRealtimeModule creates the "realtime" Starlark module for a package.
+// Requires the realtime.publish permission.
+func BuildRealtimeModule(ctx context.Context, bus *events.Bus, packageID string) *starlarkstruct.Module {
+	return MakeModule("realtime", starlark.StringDict{
+		"publish": starlark.NewBuiltin("realtime.publish", realtimePublish(ctx, bus, packageID)),
+	})
+}
+
+func realtimePublish(ctx context.Context, bus *events.Bus, packageID string) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
+	return func(_ *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
+		var channel, event string
+		var data *starlark.Dict
+
+		if err := starlark.UnpackArgs(b.Name(), args, kwargs,
+			"channel", &channel,
+			"event", &event,
+			"data?", &data,
+		); err != nil {
+			return nil, err
+		}
+
+		if channel == "" {
+			return nil, fmt.Errorf("realtime.publish: channel is required")
+		}
+		if event == "" {
+			return nil, fmt.Errorf("realtime.publish: event is required")
+		}
+
+		// Reject reserved event prefixes
+		for _, prefix := range reservedPrefixes {
+			if strings.HasPrefix(event, prefix) {
+				return nil, fmt.Errorf("realtime.publish: event %q uses reserved prefix %q", event, prefix)
+			}
+		}
+
+		// Build payload map
+		payload := map[string]any{
+			"_pkg": packageID,
+		}
+
+		if data != nil {
+			for _, item := range data.Items() {
+				k, ok := starlark.AsString(item[0])
+				if !ok {
+					return nil, fmt.Errorf("realtime.publish: dict key must be string, got %s", item[0].Type())
+				}
+				if k == "_pkg" {
+					continue // reserved field
+				}
+				payload[k] = starlarkToGoVal(item[1])
+			}
+		}
+
+		// Marshal and check size
+		jsonBytes, err := json.Marshal(payload)
+		if err != nil {
+			return nil, fmt.Errorf("realtime.publish: failed to marshal data: %w", err)
+		}
+		if len(jsonBytes) > maxRealtimePayload {
+			return nil, fmt.Errorf("realtime.publish: payload too large (%d bytes, max %d)", len(jsonBytes), maxRealtimePayload)
+		}
+
+		// Publish to event bus
+		bus.Publish(events.Event{
+			Label:   "realtime." + event,
+			Room:    channel,
+			Payload: jsonBytes,
+			Ts:      time.Now().UnixMilli(),
+		})
+
+		return starlark.True, nil
+	}
+}
+
+// starlarkToGoVal converts a Starlark value to a Go value for JSON marshaling.
+func starlarkToGoVal(v starlark.Value) any {
+	switch val := v.(type) {
+	case starlark.NoneType:
+		return nil
+	case starlark.Bool:
+		return bool(val)
+	case starlark.Int:
+		if i, ok := val.Int64(); ok {
+			return i
+		}
+		return val.String()
+	case starlark.Float:
+		return float64(val)
+	case starlark.String:
+		return string(val)
+	case *starlark.Dict:
+		m := make(map[string]any, val.Len())
+		for _, item := range val.Items() {
+			k, ok := starlark.AsString(item[0])
+			if !ok {
+				k = item[0].String()
+			}
+			m[k] = starlarkToGoVal(item[1])
+		}
+		return m
+	case *starlark.List:
+		n := val.Len()
+		s := make([]any, n)
+		for i := 0; i < n; i++ {
+			s[i] = starlarkToGoVal(val.Index(i))
+		}
+		return s
+	case starlark.Tuple:
+		s := make([]any, len(val))
+		for i, v := range val {
+			s[i] = starlarkToGoVal(v)
+		}
+		return s
+	default:
+		return v.String()
+	}
+}

server/sandbox/realtime_module_test.go

@@ -0,0 +1,158 @@
+package sandbox
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"sync"
+	"testing"
+
+	"go.starlark.net/starlark"
+
+	"switchboard-core/events"
+)
+
+func TestRealtimePublish_HappyPath(t *testing.T) {
+	bus := events.NewBus()
+
+	var mu sync.Mutex
+	var received events.Event
+
+	bus.Subscribe("realtime.*", func(e events.Event) {
+		mu.Lock()
+		received = e
+		mu.Unlock()
+	})
+
+	mod := BuildRealtimeModule(context.Background(), bus, "test-pkg")
+	sb := New(DefaultConfig())
+
+	_, err := sb.Exec(context.Background(), "test.star", `
+result = realtime.publish("room:123", "chat.message", {"text": "hello", "from": "alice"})
+`, map[string]starlark.Value{"realtime": mod})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	if received.Label != "realtime.chat.message" {
+		t.Errorf("expected label 'realtime.chat.message', got %q", received.Label)
+	}
+	if received.Room != "room:123" {
+		t.Errorf("expected room 'room:123', got %q", received.Room)
+	}
+
+	var payload map[string]any
+	if err := json.Unmarshal(received.Payload, &payload); err != nil {
+		t.Fatalf("payload unmarshal error: %v", err)
+	}
+	if payload["text"] != "hello" {
+		t.Errorf("expected text 'hello', got %v", payload["text"])
+	}
+	if payload["from"] != "alice" {
+		t.Errorf("expected from 'alice', got %v", payload["from"])
+	}
+	if payload["_pkg"] != "test-pkg" {
+		t.Errorf("expected _pkg 'test-pkg', got %v", payload["_pkg"])
+	}
+}
+
+func TestRealtimePublish_EmptyData(t *testing.T) {
+	bus := events.NewBus()
+	var received events.Event
+	bus.Subscribe("realtime.*", func(e events.Event) { received = e })
+
+	mod := BuildRealtimeModule(context.Background(), bus, "my-pkg")
+	sb := New(DefaultConfig())
+
+	_, err := sb.Exec(context.Background(), "test.star", `
+realtime.publish("ch:1", "heartbeat")
+`, map[string]starlark.Value{"realtime": mod})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	var payload map[string]any
+	json.Unmarshal(received.Payload, &payload)
+	if payload["_pkg"] != "my-pkg" {
+		t.Errorf("expected _pkg 'my-pkg', got %v", payload["_pkg"])
+	}
+	// Should only have _pkg
+	if len(payload) != 1 {
+		t.Errorf("expected 1 key (_pkg only), got %d", len(payload))
+	}
+}
+
+func TestRealtimePublish_MissingArgs(t *testing.T) {
+	bus := events.NewBus()
+	mod := BuildRealtimeModule(context.Background(), bus, "pkg")
+	sb := New(DefaultConfig())
+
+	// Missing event
+	_, err := sb.Exec(context.Background(), "test.star", `
+realtime.publish("ch:1")
+`, map[string]starlark.Value{"realtime": mod})
+	if err == nil {
+		t.Fatal("expected error for missing event arg")
+	}
+}
+
+func TestRealtimePublish_ReservedPrefix(t *testing.T) {
+	bus := events.NewBus()
+	mod := BuildRealtimeModule(context.Background(), bus, "pkg")
+	sb := New(DefaultConfig())
+
+	_, err := sb.Exec(context.Background(), "test.star", `
+realtime.publish("ch:1", "system.notify", {})
+`, map[string]starlark.Value{"realtime": mod})
+	if err == nil {
+		t.Fatal("expected error for reserved prefix")
+	}
+	if !strings.Contains(err.Error(), "reserved prefix") {
+		t.Errorf("error should mention reserved prefix, got: %v", err)
+	}
+}
+
+func TestRealtimePublish_PayloadTooLarge(t *testing.T) {
+	bus := events.NewBus()
+	mod := BuildRealtimeModule(context.Background(), bus, "pkg")
+	sb := New(DefaultConfig())
+
+	// Build a large string via repeated concatenation in Starlark
+	_, err := sb.Exec(context.Background(), "test.star", `
+big = "x" * 8000
+realtime.publish("ch:1", "big.event", {"data": big})
+`, map[string]starlark.Value{"realtime": mod})
+	if err == nil {
+		t.Fatal("expected error for payload too large")
+	}
+	if !strings.Contains(err.Error(), "too large") {
+		t.Errorf("error should mention 'too large', got: %v", err)
+	}
+}
+
+func TestStarlarkToGoVal(t *testing.T) {
+	tests := []struct {
+		name string
+		val  starlark.Value
+		want any
+	}{
+		{"none", starlark.None, nil},
+		{"true", starlark.True, true},
+		{"false", starlark.False, false},
+		{"int", starlark.MakeInt(42), int64(42)},
+		{"float", starlark.Float(3.14), float64(3.14)},
+		{"string", starlark.String("hello"), "hello"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := starlarkToGoVal(tt.val)
+			if got != tt.want {
+				t.Errorf("starlarkToGoVal(%s) = %v (%T), want %v (%T)", tt.val, got, got, tt.want, tt.want)
+			}
+		})
+	}
+}

server/sandbox/runner.go

@@ -33,6 +33,7 @@ import (
 
 	starlarkjson "go.starlark.net/lib/json"
 
+	"switchboard-core/events"
 	"switchboard-core/models"
 	"switchboard-core/store"
 )
@@ -62,6 +63,7 @@ type Runner struct {
 	db              *sql.DB              // nil = db module unavailable
 	dbPostgres      bool                 // true = use $N placeholders; false = use ?
 	allowPrivateIPs bool                 // v0.38.5: disable SSRF private IP check
+	bus             *events.Bus          // v0.5.0: event bus for realtime module
 }
 
 // NewRunner creates a runner with the given sandbox and dependencies.
@@ -97,6 +99,11 @@ func (r *Runner) SetPackagesDir(dir string) {
 	r.packagesDir = dir
 }
 
+// SetBus attaches the event bus for the realtime module (v0.5.0).
+func (r *Runner) SetBus(bus *events.Bus) {
+	r.bus = bus
+}
+
 // SetAllowPrivateIPs disables the SSRF check that blocks connections to
 // private/loopback IPs. For self-hosted environments where extensions
 // reach internal services. Controlled by EXT_ALLOW_PRIVATE_IPS env var.
@@ -327,6 +334,11 @@ func (r *Runner) buildModulesWithLibCtx(ctx context.Context, packageID string, m
 			if r.connResolver != nil && rc != nil && rc.UserID != "" {
 				modules["connections"] = BuildConnectionsModule(ctx, r.connResolver, rc.UserID)
 			}
+
+		case models.ExtPermRealtimePublish:
+			if r.bus != nil {
+				modules["realtime"] = BuildRealtimeModule(ctx, r.bus, packageID)
+			}
 		}
 	}
 

server/static/openapi.yaml

@@ -1035,6 +1035,19 @@ paths:
                   ticket:
                     type: string
 
+  # WebSocket Realtime Protocol (v0.5.0)
+  # ──────────────────────────────────────────────
+  # Room Management (client → server, intercepted before bus):
+  #   { "event": "room.subscribe",   "payload": { "room": "conversation:abc" } }
+  #   { "event": "room.unsubscribe", "payload": { "room": "conversation:abc" } }
+  #
+  # Realtime Events (server → client, room-scoped):
+  #   { "event": "realtime.<name>", "room": "<channel>", "payload": { "_pkg": "...", ... }, "ts": 1234567890 }
+  #
+  # Starlark extensions publish via: realtime.publish(channel, event, data)
+  # SDK clients subscribe via: sw.realtime.subscribe(channel, [event], callback)
+  # Max 100 rooms per connection. Payload limit: 7KB.
+
   # ──────────────────────────────────────────────
   # Extension Assets
   # ──────────────────────────────────────────────