Support '!' prefix in broker messages as inline interrupt (#375)

* Support "!" prefix in broker messages as inline interrupt signal

Messages arriving via the broker (Telegram, webhooks, direct messages)
that start with "!" are now treated as interrupt messages: the "!" is
stripped and the message is delivered with urgent/interrupt semantics,
equivalent to --interrupt on the CLI.

* Fix pointer mutation of event-bus message in interrupt prefix handling

deliverToAgent mutates msg.Msg and msg.Urgent in place, but at the
direct-subscriber call site the pointer comes from the event bus and is
shared across all matching subscribers. Shallow-copy before mutating so
the original bus message stays intact.

* Handle whitespace and empty-content edge cases in interrupt prefix

TrimSpace the message before checking for the "!" prefix so leading
whitespace (e.g. "  !restart") is handled correctly. TrimSpace the
content after stripping the prefix so "!  restart" works too. Default
to "interrupt" when the stripped content is empty (bare "!" or "!   ")
to satisfy the Msg-required invariant in StructuredMessage.Validate.

Adds table-driven tests covering all edge cases.

* Fix CI failures: errcheck lint and no_sqlite build tag

- Use t.Cleanup with explicit error discard for eventbus Close() calls
  in new test functions to satisfy the errcheck linter.
- Add //go:build !no_sqlite constraint to signing_key_shared_test.go
  so newTestStore is available when `go vet -tags no_sqlite` runs.

* Fix gofmt struct field alignment in edge case test

---------

Co-authored-by: Scion Agent (fix-interrupt-prefix) <agent@scion.dev>

Author: ptone

pkg/hub/messagebroker.go

@@ -495,6 +495,21 @@ func (p *MessageBrokerProxy) deliverToAgent(ctx context.Context, projectID, agen
 		return
 	}
 
+	// A leading "!" in the message body acts as an inline interrupt signal:
+	// strip the prefix and promote to urgent so the harness is interrupted
+	// before delivery — equivalent to --interrupt on the CLI.
+	// Shallow-copy to avoid mutating the event-bus pointer shared across subscribers.
+	if trimmed := strings.TrimSpace(msg.Msg); strings.HasPrefix(trimmed, "!") {
+		stripped := *msg
+		content := strings.TrimSpace(trimmed[1:])
+		if content == "" {
+			content = "interrupt"
+		}
+		stripped.Msg = content
+		stripped.Urgent = true
+		msg = &stripped
+	}
+
 	dispatcher := p.getDispatcher()
 	if dispatcher == nil {
 		p.log.Warn("No dispatcher available, cannot deliver broker message",

pkg/hub/messagebroker_test.go

@@ -195,6 +195,190 @@ func TestMessageBrokerProxy_DirectMessage(t *testing.T) {
 	}
 }
 
+func TestMessageBrokerProxy_InterruptPrefix(t *testing.T) {
+	s := newBrokerTestStore(t)
+	projectID := setupBrokerTestProject(t, s)
+	setupBrokerTestAgent(t, s, projectID, "test-agent", "running")
+
+	events := NewChannelEventPublisher()
+	defer events.Close()
+
+	b := eventbus.NewInProcessEventBus(slog.Default())
+	t.Cleanup(func() { _ = b.Close() })
+
+	dispatcher := &brokerMockDispatcher{}
+
+	proxy := NewMessageBrokerProxy(b, s, events, func() AgentDispatcher { return dispatcher }, slog.Default())
+	proxy.Start()
+	defer proxy.Stop()
+
+	proxy.subscribeAgent(projectID, "test-agent")
+
+	msg := messages.NewInstruction("user:alice", "agent:test-agent", "!restart now")
+	if err := proxy.PublishMessage(context.Background(), projectID, msg); err != nil {
+		t.Fatal(err)
+	}
+
+	time.Sleep(100 * time.Millisecond)
+
+	dispatched := dispatcher.getMessages()
+	if len(dispatched) != 1 {
+		t.Fatalf("expected 1 dispatched message, got %d", len(dispatched))
+	}
+	if dispatched[0].msg != "restart now" {
+		t.Errorf("expected message 'restart now' (! stripped), got %q", dispatched[0].msg)
+	}
+	if !dispatched[0].interrupt {
+		t.Error("expected interrupt=true for !-prefixed message")
+	}
+	if !dispatched[0].structured.Urgent {
+		t.Error("expected structured message Urgent=true for !-prefixed message")
+	}
+}
+
+func TestMessageBrokerProxy_InterruptPrefixNotStrippedWithoutBang(t *testing.T) {
+	s := newBrokerTestStore(t)
+	projectID := setupBrokerTestProject(t, s)
+	setupBrokerTestAgent(t, s, projectID, "test-agent", "running")
+
+	events := NewChannelEventPublisher()
+	defer events.Close()
+
+	b := eventbus.NewInProcessEventBus(slog.Default())
+	t.Cleanup(func() { _ = b.Close() })
+
+	dispatcher := &brokerMockDispatcher{}
+
+	proxy := NewMessageBrokerProxy(b, s, events, func() AgentDispatcher { return dispatcher }, slog.Default())
+	proxy.Start()
+	defer proxy.Stop()
+
+	proxy.subscribeAgent(projectID, "test-agent")
+
+	msg := messages.NewInstruction("user:alice", "agent:test-agent", "hello agent")
+	if err := proxy.PublishMessage(context.Background(), projectID, msg); err != nil {
+		t.Fatal(err)
+	}
+
+	time.Sleep(100 * time.Millisecond)
+
+	dispatched := dispatcher.getMessages()
+	if len(dispatched) != 1 {
+		t.Fatalf("expected 1 dispatched message, got %d", len(dispatched))
+	}
+	if dispatched[0].msg != "hello agent" {
+		t.Errorf("expected message 'hello agent' unchanged, got %q", dispatched[0].msg)
+	}
+	if dispatched[0].interrupt {
+		t.Error("expected interrupt=false for non-!-prefixed message")
+	}
+}
+
+func TestMessageBrokerProxy_InterruptPrefixEdgeCases(t *testing.T) {
+	tests := []struct {
+		name          string
+		input         string
+		wantMsg       string
+		wantInterrupt bool
+		wantUrgent    bool
+	}{
+		{"bare bang", "!", "interrupt", true, true},
+		{"bang with trailing spaces", "!   ", "interrupt", true, true},
+		{"leading whitespace before bang", "  !restart", "restart", true, true},
+		{"whitespace between bang and content", "!  restart", "restart", true, true},
+		{"leading and inner whitespace", "  !  restart now  ", "restart now", true, true},
+		{"normal message no prefix", "hello", "hello", false, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := newBrokerTestStore(t)
+			projectID := setupBrokerTestProject(t, s)
+			setupBrokerTestAgent(t, s, projectID, "test-agent", "running")
+
+			events := NewChannelEventPublisher()
+			defer events.Close()
+
+			bus := eventbus.NewInProcessEventBus(slog.Default())
+			t.Cleanup(func() { _ = bus.Close() })
+
+			dispatcher := &brokerMockDispatcher{}
+
+			proxy := NewMessageBrokerProxy(bus, s, events, func() AgentDispatcher { return dispatcher }, slog.Default())
+			proxy.Start()
+			defer proxy.Stop()
+
+			proxy.subscribeAgent(projectID, "test-agent")
+
+			msg := messages.NewInstruction("user:alice", "agent:test-agent", tt.input)
+			if err := proxy.PublishMessage(context.Background(), projectID, msg); err != nil {
+				t.Fatal(err)
+			}
+
+			time.Sleep(100 * time.Millisecond)
+
+			dispatched := dispatcher.getMessages()
+			if len(dispatched) != 1 {
+				t.Fatalf("expected 1 dispatched message, got %d", len(dispatched))
+			}
+			if dispatched[0].msg != tt.wantMsg {
+				t.Errorf("msg = %q, want %q", dispatched[0].msg, tt.wantMsg)
+			}
+			if dispatched[0].interrupt != tt.wantInterrupt {
+				t.Errorf("interrupt = %v, want %v", dispatched[0].interrupt, tt.wantInterrupt)
+			}
+			if dispatched[0].structured.Urgent != tt.wantUrgent {
+				t.Errorf("Urgent = %v, want %v", dispatched[0].structured.Urgent, tt.wantUrgent)
+			}
+		})
+	}
+}
+
+func TestMessageBrokerProxy_InterruptPrefixPersistence(t *testing.T) {
+	s := newBrokerTestStore(t)
+	projectID := setupBrokerTestProject(t, s)
+	agent := setupBrokerTestAgent(t, s, projectID, "persist-agent", "running")
+
+	events := NewChannelEventPublisher()
+	defer events.Close()
+
+	b := eventbus.NewInProcessEventBus(slog.Default())
+	t.Cleanup(func() { _ = b.Close() })
+
+	dispatcher := &brokerMockDispatcher{}
+
+	proxy := NewMessageBrokerProxy(b, s, events, func() AgentDispatcher { return dispatcher }, slog.Default())
+	proxy.Start()
+	defer proxy.Stop()
+
+	proxy.subscribeAgent(projectID, "persist-agent")
+
+	msg := messages.NewInstruction("user:alice", "agent:persist-agent", "!urgent task")
+	msg.SenderID = "user-alice-id"
+	msg.RecipientID = agent.ID
+	if err := proxy.PublishMessage(context.Background(), projectID, msg); err != nil {
+		t.Fatal(err)
+	}
+
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify the persisted message has the stripped content and urgent flag
+	ctx := context.Background()
+	result, err := s.ListMessages(ctx, store.MessageFilter{AgentID: agent.ID}, store.ListOptions{})
+	if err != nil {
+		t.Fatalf("failed to list messages: %v", err)
+	}
+	if len(result.Items) != 1 {
+		t.Fatalf("expected 1 persisted message, got %d", len(result.Items))
+	}
+	if result.Items[0].Msg != "urgent task" {
+		t.Errorf("expected persisted msg 'urgent task', got %q", result.Items[0].Msg)
+	}
+	if !result.Items[0].Urgent {
+		t.Error("expected persisted message Urgent=true")
+	}
+}
+
 func TestMessageBrokerProxy_ProjectBroadcast(t *testing.T) {
 	s := newBrokerTestStore(t)
 	projectID := setupBrokerTestProject(t, s)