Add a Transport

What "add a transport" means

Every transport in rousseau is an adapter over an upstream protocol client. The adapter implements a small interface, exposes a Deliver method for the cron scheduler, and is registered as a Cobra subcommand in internal/cli/.

The agent core does not move. That is the invariant — one interface implementation plus one CLI wire-up equals a new transport.

The interface

internal/transport/transport.go:

type IncomingMessage struct {
    From string
    Body string
    At   time.Time
}

type Handler interface {
    Handle(ctx context.Context, msg IncomingMessage) (string, error)
}

type Transport interface {
    Name() string
    Start(ctx context.Context, handler Handler) error
    Stop() error
}

Start blocks until ctx is cancelled or Stop is called. Handle receives inbound messages and returns the reply text — the router owns per-sender session isolation and allowlist enforcement.

Implementations also typically expose a Deliver(ctx context.Context, target, body string) error method so the cron scheduler can send messages that did not originate from an inbound turn.

Skeleton

Let's add a hypothetical XMPP transport.

Step 1 — Directory and adapter

Create internal/transport/xmpp/ mirroring the shape of internal/transport/slack/:

internal/transport/xmpp/
├── client.go        # Config, New, Start, Stop, Deliver
├── client_test.go   # Tests using injected fakes
├── dispatch.go      # onMessage → Handler.Handle → send reply
└── dispatch_test.go

Step 2 — `client.go`

// Package xmpp is the XMPP transport adapter.
package xmpp

import (
    "context"
    "fmt"
    "log/slog"
    "sync"

    "github.com/sebastienrousseau/rousseau-agent/internal/transport"
)

// Config configures the XMPP transport.
type Config struct {
    Server      string
    JID         string
    Password    string
    ReplyHeader string
}

// XMPPClient is the abstract interface the adapter needs from the
// upstream client. Kept small so tests can inject a fake.
type XMPPClient interface {
    Connect(ctx context.Context) error
    OnMessage(func(from, body string))
    Send(to, body string) error
    Close() error
}

// Client is the transport adapter.
type Client struct {
    cfg     Config
    logger  *slog.Logger
    upstream XMPPClient

    mu      sync.Mutex
    handler transport.Handler
    stopped bool
}

// New constructs a Client. In production, upstream is a real xmpp.Client;
// in tests, it is a fake.
func New(cfg Config, logger *slog.Logger, upstream XMPPClient) (*Client, error) {
    if cfg.Server == "" || cfg.JID == "" {
        return nil, fmt.Errorf("xmpp: server and JID required")
    }
    return &Client{cfg: cfg, logger: logger, upstream: upstream}, nil
}

// Name satisfies transport.Transport.
func (*Client) Name() string { return "xmpp" }

// Start satisfies transport.Transport.
func (c *Client) Start(ctx context.Context, handler transport.Handler) error {
    c.mu.Lock()
    c.handler = handler
    c.mu.Unlock()

    if err := c.upstream.Connect(ctx); err != nil {
        return fmt.Errorf("xmpp: connect: %w", err)
    }
    c.upstream.OnMessage(func(from, body string) {
        c.onMessage(ctx, from, body)
    })
    <-ctx.Done()
    return c.Stop()
}

// Stop satisfies transport.Transport.
func (c *Client) Stop() error {
    c.mu.Lock()
    if c.stopped {
        c.mu.Unlock()
        return nil
    }
    c.stopped = true
    c.mu.Unlock()
    return c.upstream.Close()
}

// Deliver sends a message outside the request/response loop (e.g. cron).
func (c *Client) Deliver(_ context.Context, target, body string) error {
    return c.upstream.Send(target, c.cfg.ReplyHeader+body)
}

Step 3 — `dispatch.go`

package xmpp

import (
    "context"
    "log/slog"
    "time"

    "github.com/sebastienrousseau/rousseau-agent/internal/transport"
)

func (c *Client) onMessage(ctx context.Context, from, body string) {
    c.mu.Lock()
    handler := c.handler
    c.mu.Unlock()
    if handler == nil {
        return
    }

    reply, err := handler.Handle(ctx, transport.IncomingMessage{
        From: from,
        Body: body,
        At:   time.Now().UTC(),
    })
    if err != nil {
        c.logger.Warn("xmpp.handle_failed", slog.String("err", err.Error()))
        return
    }
    if reply == "" {
        return
    }
    if err := c.upstream.Send(from, c.cfg.ReplyHeader+reply); err != nil {
        c.logger.Warn("xmpp.send_failed", slog.String("err", err.Error()))
    }
}

Step 4 — Config surface

Add an XMPPConfig type to internal/config/config.go:

type XMPPConfig struct {
    Server      string   `mapstructure:"server"`
    JID         string   `mapstructure:"jid"`
    Password    string   `mapstructure:"password"`
    ReplyHeader string   `mapstructure:"reply_header"`
    Allowlist   []string `mapstructure:"allowlist"`
}

And a field on Config:

XMPP XMPPConfig `mapstructure:"xmpp"`

Set any defaults in setDefaults(v).

Step 5 — CLI wire-up

Add internal/cli/xmpp.go:

package cli

import (
    "github.com/spf13/cobra"

    "github.com/sebastienrousseau/rousseau-agent/internal/transport/xmpp"
)

func newXMPPCmd(opts *Options) *cobra.Command {
    var allow []string
    cmd := &cobra.Command{
        Use:   "xmpp",
        Short: "Run the XMPP bridge",
        RunE: func(cmd *cobra.Command, _ []string) error {
            setUnattendedPermissionDefault(opts, "xmpp")
            ctx := cmd.Context()

            wiring, err := assembleDaemon(ctx, opts, allow)
            if err != nil {
                return err
            }
            defer func() { _ = wiring.Sessions.Close() }() //nolint:errcheck

            upstream, err := newRealXMPPClient(opts.Config.XMPP)
            if err != nil {
                return err
            }

            client, err := xmpp.New(xmpp.Config{
                Server:      opts.Config.XMPP.Server,
                JID:         opts.Config.XMPP.JID,
                Password:    opts.Config.XMPP.Password,
                ReplyHeader: opts.Config.XMPP.ReplyHeader,
            }, opts.Logger, upstream)
            if err != nil {
                return err
            }

            shutdown, err := wiring.startCron(ctx, client.Deliver, opts.Logger)
            if err != nil {
                return err
            }
            defer shutdown()

            return client.Start(ctx, wiring.Router)
        },
    }
    cmd.Flags().StringSliceVar(&allow, "allow", nil, "restrict inbound to these XMPP JIDs")
    return cmd
}

root.AddCommand(newXMPPCmd(opts))

Step 6 — Tests

Follow the fake-injection pattern used by internal/transport/whatsapp/client_test.go and internal/transport/slack/*_test.go:

type fakeXMPP struct {
    connectErr error
    onMsg      func(from, body string)
    sent       []struct{ To, Body string }
}

func (f *fakeXMPP) Connect(context.Context) error { return f.connectErr }
func (f *fakeXMPP) OnMessage(fn func(from, body string)) { f.onMsg = fn }
func (f *fakeXMPP) Send(to, body string) error {
    f.sent = append(f.sent, struct{ To, Body string }{to, body})
    return nil
}
func (f *fakeXMPP) Close() error { return nil }

func TestClient_ReplyRoundTrip(t *testing.T) {
    up := &fakeXMPP{}
    c, err := xmpp.New(xmpp.Config{Server: "s", JID: "j"}, slog.Default(), up)
    require.NoError(t, err)

    handler := transport.HandlerFunc(func(_ context.Context, m transport.IncomingMessage) (string, error) {
        return "hello " + m.From, nil
    })

    ctx, cancel := context.WithCancel(context.Background())
    go func() { _ = c.Start(ctx, handler) }()
    // Simulate an inbound message.
    up.onMsg("alice@x", "hi")
    // Give the goroutine a moment; usually via a channel signal in real tests.
    cancel()

    require.Len(t, up.sent, 1)
    require.Equal(t, "hello alice@x", up.sent[0].Body)
}

Step 7 — Docs

Add content/transports/xmpp.md in the docs site (this repo). Follow the same layout as content/transports/slack.md — description, config surface, wiring recipe, caveats.

Step 8 — CI

The lint gate will catch missing godoc comments on exported identifiers, missing tests, unused imports, and the coverage floor. Run:

make check

If everything passes locally, the CI matrix will pass too (Ubuntu + macOS run the same command).

Common pitfalls

Forgetting Stop() idempotency. The interface allows Stop to be called multiple times. Use a mutex + stopped bool.
Not honouring ctx in Start. The daemon signal handler cancels the root context; every long-lived component must return when it does.
Baking in the upstream client type. Define a small interface for the upstream, take it as a constructor parameter, inject a real client in cli/xmpp.go and a fake in tests. This is the load-bearing testability pattern.
Blocking the callback. OnMessage callbacks typically run on the upstream's goroutine. Route to the handler via a channel or a bounded goroutine pool so a slow model call cannot back up the upstream.

Troubleshooting

Interface satisfaction fails at compile

Add var _ transport.Transport = (*Client)(nil) at the bottom of your client.go. The compiler will point at the missing method.

`Start` never returns

You forgot to <-ctx.Done() before returning. The daemon signal handler cancels the root context on SIGTERM / SIGINT; every Start must respect it.

Tests hang because the callback runs on a background goroutine

Use a channel + require.Eventually (or a t.Deadline) to wait for the reply. Do not rely on time.Sleep.

Coverage floor fails

The gate is 75% overall, higher for core packages. Add tests for at least the happy-path resolve + reply loop and one skip-reason (loop prevention or non-text messages).

Router allowlist blocks your test

Tests should use a transport.HandlerFunc directly, not go through the router. If you must exercise router behaviour, pass an empty allowlist (or one that includes the test sender).

Developer Guide: Architecture — where Transport fits.
Developer Guide: Testing — the injection pattern in depth.
Developer Guide: Add a provider — same shape, different interface.
Developer Guide: Contributing — commit style and PR checklist.
Transports — the shipped adapters as reference implementations.

What "add a transport" means

The interface

Skeleton

Step 1 — Directory and adapter

Step 2 — `client.go`

Step 3 — `dispatch.go`

Step 4 — Config surface

Step 5 — CLI wire-up

Step 6 — Tests

Step 7 — Docs

Step 8 — CI

Common pitfalls

Troubleshooting

Interface satisfaction fails at compile

`Start` never returns

Tests hang because the callback runs on a background goroutine

Coverage floor fails

Router allowlist blocks your test

Related pages

Further reading

What "add a transport" means

The interface

Skeleton

Step 1 — Directory and adapter

Step 2 — client.go

Step 3 — dispatch.go

Step 4 — Config surface

Step 5 — CLI wire-up

Step 6 — Tests

Step 7 — Docs

Step 8 — CI

Common pitfalls

Troubleshooting

Interface satisfaction fails at compile

Start never returns

Tests hang because the callback runs on a background goroutine

Coverage floor fails

Router allowlist blocks your test

Related pages

Further reading

Step 2 — `client.go`

Step 3 — `dispatch.go`

`Start` never returns