diff --git a/go-runtime/compile/build.go b/go-runtime/compile/build.go
index 52a69731ce..0a49b1a2ba 100644
--- a/go-runtime/compile/build.go
+++ b/go-runtime/compile/build.go
@@ -211,9 +211,6 @@ var scaffoldFuncs = scaffolder.FuncMap{
 		}
 		panic(fmt.Sprintf("unsupported value %T", v))
 	},
-	"isSink": func(v schema.Verb) string {
-		return "wes"
-	},
 }
 
 func genType(module *schema.Module, t schema.Type) string {
diff --git a/go.mod b/go.mod
index 1f8f4bdb79..4cf08aa9af 100644
--- a/go.mod
+++ b/go.mod
@@ -33,7 +33,6 @@ require (
 	github.com/radovskyb/watcher v1.0.7
 	github.com/rs/cors v1.10.1
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
-	github.com/stretchr/testify v1.8.4
 	github.com/swaggest/jsonschema-go v0.3.69
 	github.com/titanous/json5 v1.0.0
 	github.com/tmc/langchaingo v0.1.5
@@ -58,13 +57,10 @@ require (
 
 require (
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/ncruces/go-strftime v0.1.9 // indirect
 	github.com/pkoukk/tiktoken-go v0.1.2 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.1.0 // indirect
-	gopkg.in/yaml.v3 v3.0.1 // indirect
 	modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect
 )
 
diff --git a/integration/integration_test.go b/integration/integration_test.go
index 5598f0d6cd..70d93f46a7 100644
--- a/integration/integration_test.go
+++ b/integration/integration_test.go
@@ -213,6 +213,22 @@ func TestExternalCalls(t *testing.T) {
 		return tests
 	})
 }
+func TestPubSub(t *testing.T) {
+	runForRuntimes(t, func(modulesDir string, runtime string, rd runtimeData) []test {
+		return []test{
+			{name: fmt.Sprintf("PubSub%s", rd.testSuffix), assertions: assertions{
+				run("ftl", rd.initOpts...),
+				scaffoldTestData(runtime, "pubsub", rd.modulePath),
+				run("ftl", "deploy", rd.moduleRoot),
+				call(rd.moduleName, "echo", obj{}, func(t testing.TB, resp obj) {
+					name, ok := resp["name"].(string)
+					assert.True(t, ok, "name is not a string")
+					assert.Equal(t, "source", name)
+				}),
+			}},
+		}
+	})
+}
 
 func TestSchemaGenerate(t *testing.T) {
 	tests := []test{
diff --git a/integration/testdata/go/pubsub/echo.go b/integration/testdata/go/pubsub/echo.go
new file mode 100644
index 0000000000..d950af8835
--- /dev/null
+++ b/integration/testdata/go/pubsub/echo.go
@@ -0,0 +1,49 @@
+//ftl:module echo
+package echo
+
+import (
+	"context"
+
+	"github.com/TBD54566975/ftl/go-runtime/ftl"
+	// Import the FTL SDK.
+)
+
+type EchoRequest struct{}
+type EchoResponse struct {
+	Name string
+}
+
+//ftl:verb
+func Echo(ctx context.Context, req EchoRequest) (EchoResponse, error) {
+	err := ftl.CallSink(ctx, Sink, SinkRequest{})
+	if err != nil {
+		return EchoResponse{}, err
+	}
+	resp, err := ftl.CallSource(ctx, Source)
+	if err != nil {
+		return EchoResponse{}, err
+	}
+
+	name := resp.Name
+	return EchoResponse{
+		Name: name,
+	}, nil
+}
+
+type SourceResponse struct {
+	Name string
+}
+
+//ftl:verb
+func Source(ctx context.Context) (SourceResponse, error) {
+	return SourceResponse{
+		Name: "source",
+	}, nil
+}
+
+type SinkRequest struct{}
+
+//ftl:verb
+func Sink(ctx context.Context, req SinkRequest) error {
+	return nil
+}
diff --git a/integration/testdata/kotlin/pubsub/Echo.kt b/integration/testdata/kotlin/pubsub/Echo.kt
new file mode 100644
index 0000000000..be3b16f5ae
--- /dev/null
+++ b/integration/testdata/kotlin/pubsub/Echo.kt
@@ -0,0 +1,25 @@
+package ftl.echo
+
+import ftl.builtin.Empty
+import xyz.block.ftl.Context
+import xyz.block.ftl.Verb
+
+data class EchoResponse(val name: String)
+
+@Verb
+fun echo(context: Context, req: Empty): EchoResponse {
+  context.callSink(::sink, Empty())
+  val resp = context.callSource(::source)
+  return EchoResponse(name = resp.name)
+}
+
+data class SourceResponse(val name: String)
+
+@Verb
+fun source(context: Context): EchoResponse {
+  return EchoResponse(name = "source")
+}
+
+@Verb
+fun sink(context: Context, req: Empty) {
+}