pico

@@ -700,13 +700,15 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
 	if *block {
 		count := 0
 		for topic, channel := range handler.PubSub.GetChannels() {
-			if topic == name {
+			if topic == name || (psub.HasWildcard(topic) && psub.MatchTopic(topic, name)) {
 				for _, client := range channel.GetClients() {
 					if client.Direction == psub.ChannelDirectionOutput || client.Direction == psub.ChannelDirectionInputOutput {
 						count++
 					}
 				}
-				break
+				if topic == name {
+					break
+				}
 			}
 		}
 

@@ -735,13 +737,15 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
 					case <-time.After(1 * time.Millisecond):
 						count := 0
 						for topic, channel := range handler.PubSub.GetChannels() {
-							if topic == name {
+							if topic == name || (psub.HasWildcard(topic) && psub.MatchTopic(topic, name)) {
 								for _, client := range channel.GetClients() {
 									if client.Direction == psub.ChannelDirectionOutput || client.Direction == psub.ChannelDirectionInputOutput {
 										count++
 									}
 								}
-								break
+								if topic == name {
+									break
+								}
 							}
 						}
 

@@ -1,6 +1,7 @@
 package pipe
 
 import (
+	"bytes"
 	"context"
 	"crypto/ed25519"
 	"crypto/rand"

@@ -9,6 +10,7 @@ import (
 	"log/slog"
 	"os"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 

@@ -2200,3 +2202,107 @@ func TestMonitor_FixedWindowNonSliding(t *testing.T) {
 			monitor.WindowEnd.Format(time.RFC3339))
 	}
 }
+
+func TestSSH_WildcardSub(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	dbUser := &db.User{ID: "alice-id", Name: "alice"}
+	server.DBPool.AddUser(dbUser)
+	server.DBPool.AddPubkey(&db.PublicKey{
+		ID:     "alice-pk",
+		UserID: "alice-id",
+		Key:    user.PublicKey(),
+	})
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to dial ssh server: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	// 1. Subscribe to wildcard topic: "sub metric-drain*"
+	subSession, err := client.NewSession()
+	if err != nil {
+		t.Fatalf("failed to create sub session: %v", err)
+	}
+
+	subOut, err := subSession.StdoutPipe()
+	if err != nil {
+		t.Fatalf("failed stdout pipe: %v", err)
+	}
+
+	if err := subSession.Start("sub metric-drain*"); err != nil {
+		t.Fatalf("failed to start sub: %v", err)
+	}
+
+	var buf bytes.Buffer
+	var bufMu sync.Mutex
+	go func() {
+		b := make([]byte, 1024)
+		for {
+			n, err := subOut.Read(b)
+			if n > 0 {
+				bufMu.Lock()
+				buf.Write(b[:n])
+				bufMu.Unlock()
+			}
+			if err != nil {
+				break
+			}
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+
+	// 2. Publish to "metric-drain-pgs"
+	pubClient1, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to dial pub client 1: %v", err)
+	}
+	defer func() { _ = pubClient1.Close() }()
+
+	pubSession1, err := pubClient1.NewSession()
+	if err != nil {
+		t.Fatalf("failed pub session 1: %v", err)
+	}
+	pubIn1, _ := pubSession1.StdinPipe()
+	go func() {
+		defer func() { _ = pubIn1.Close() }()
+		_, _ = io.WriteString(pubIn1, "pgs-event\n")
+	}()
+	_ = pubSession1.Run("pub metric-drain-pgs -b=false")
+
+	// 3. Publish to "metric-drain-prose"
+	pubClient2, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to dial pub client 2: %v", err)
+	}
+	defer func() { _ = pubClient2.Close() }()
+
+	pubSession2, err := pubClient2.NewSession()
+	if err != nil {
+		t.Fatalf("failed pub session 2: %v", err)
+	}
+	pubIn2, _ := pubSession2.StdinPipe()
+	go func() {
+		defer func() { _ = pubIn2.Close() }()
+		_, _ = io.WriteString(pubIn2, "prose-event\n")
+	}()
+	_ = pubSession2.Run("pub metric-drain-prose -b=false")
+
+	time.Sleep(150 * time.Millisecond)
+	_ = subSession.Close()
+
+	bufMu.Lock()
+	output := buf.String()
+	bufMu.Unlock()
+
+	if !strings.Contains(output, "pgs-event") {
+		t.Errorf("expected SSH wildcard subscriber output to contain 'pgs-event', got: %q", output)
+	}
+	if !strings.Contains(output, "prose-event") {
+		t.Errorf("expected SSH wildcard subscriber output to contain 'prose-event', got: %q", output)
+	}
+}

@@ -73,3 +73,132 @@ func TestWildcardSubExistingAndNewTopics(t *testing.T) {
 		t.Errorf("wildcard subscriber should NOT receive other-data, got: %q", got)
 	}
 }
+
+// TestWildcardSubMultipleSubscribers verifies that multiple wildcard subscribers
+// listening on the same pattern both receive published messages.
+func TestWildcardSubMultipleSubscribers(t *testing.T) {
+	cast := NewMulticast(slog.Default())
+
+	subBuf1 := new(Buffer)
+	subBuf2 := new(Buffer)
+	subCtx, cancelSub := context.WithCancel(context.Background())
+	defer cancelSub()
+
+	wildcardChannel := NewChannel("logs-*")
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	go func() {
+		defer wg.Done()
+		_ = cast.Sub(subCtx, "sub-1", subBuf1, []*Channel{wildcardChannel}, false)
+	}()
+
+	go func() {
+		defer wg.Done()
+		_ = cast.Sub(subCtx, "sub-2", subBuf2, []*Channel{wildcardChannel}, false)
+	}()
+
+	time.Sleep(50 * time.Millisecond)
+
+	channel := NewChannel("logs-app1")
+	pubCtx, cancelPub := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancelPub()
+
+	_ = cast.Pub(pubCtx, "pub-1", &Buffer{b: *bytes.NewBufferString("app1-log\n")}, []*Channel{channel}, false)
+
+	time.Sleep(100 * time.Millisecond)
+	cancelSub()
+	wg.Wait()
+
+	if subBuf1.String() != "app1-log\n" {
+		t.Errorf("sub-1 expected app1-log, got %q", subBuf1.String())
+	}
+	if subBuf2.String() != "app1-log\n" {
+		t.Errorf("sub-2 expected app1-log, got %q", subBuf2.String())
+	}
+}
+
+// TestWildcardSubVariousPatterns verifies prefix, suffix, and middle asterisk wildcard matching.
+func TestWildcardSubVariousPatterns(t *testing.T) {
+	cast := NewMulticast(slog.Default())
+
+	prefixBuf := new(Buffer)
+	suffixBuf := new(Buffer)
+	middleBuf := new(Buffer)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var wg sync.WaitGroup
+	wg.Add(3)
+
+	go func() {
+		defer wg.Done()
+		_ = cast.Sub(ctx, "sub-prefix", prefixBuf, []*Channel{NewChannel("metric-*")}, false)
+	}()
+	go func() {
+		defer wg.Done()
+		_ = cast.Sub(ctx, "sub-suffix", suffixBuf, []*Channel{NewChannel("*-drain")}, false)
+	}()
+	go func() {
+		defer wg.Done()
+		_ = cast.Sub(ctx, "sub-middle", middleBuf, []*Channel{NewChannel("metric-*-drain")}, false)
+	}()
+
+	time.Sleep(50 * time.Millisecond)
+
+	pubCtx, cancelPub := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancelPub()
+
+	// Publish to metric-app-drain
+	_ = cast.Pub(pubCtx, "pub", &Buffer{b: *bytes.NewBufferString("event\n")}, []*Channel{NewChannel("metric-app-drain")}, false)
+
+	time.Sleep(100 * time.Millisecond)
+	cancel()
+	wg.Wait()
+
+	if prefixBuf.String() != "event\n" {
+		t.Errorf("prefix subscriber expected event, got %q", prefixBuf.String())
+	}
+	if suffixBuf.String() != "event\n" {
+		t.Errorf("suffix subscriber expected event, got %q", suffixBuf.String())
+	}
+	if middleBuf.String() != "event\n" {
+		t.Errorf("middle subscriber expected event, got %q", middleBuf.String())
+	}
+}
+
+// TestWildcardSubLiteralCharsWithoutStar verifies that '?' or '[' without '*' are treated as literal names.
+func TestWildcardSubLiteralCharsWithoutStar(t *testing.T) {
+	cast := NewMulticast(slog.Default())
+
+	subBuf := new(Buffer)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	// Subscribe to a topic with a literal '?' character
+	go func() {
+		defer wg.Done()
+		_ = cast.Sub(ctx, "sub-literal", subBuf, []*Channel{NewChannel("topic?one")}, false)
+	}()
+
+	time.Sleep(50 * time.Millisecond)
+
+	pubCtx, cancelPub := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancelPub()
+
+	// Publish to topicXone (should NOT match because '?' is not treated as a wildcard)
+	_ = cast.Pub(pubCtx, "pub", &Buffer{b: *bytes.NewBufferString("data\n")}, []*Channel{NewChannel("topicXone")}, false)
+
+	time.Sleep(100 * time.Millisecond)
+	cancel()
+	wg.Wait()
+
+	if subBuf.String() != "" {
+		t.Errorf("literal subscriber should not have matched topicXone, got: %q", subBuf.String())
+	}
+}