Add reusable live chat Go SDK

Provide a standalone agent client for l-ai-chat with service-token auth, safe at-least-once subscriptions, ordered cumulative ACK tracking, and self-contained gRPC fake tests.

Co-authored-by: Cursor <cursoragent@cursor.com>

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "proto/l-ai-chat-proto"]
+	path = proto/l-ai-chat-proto
+	url = https://gitea.fe.kaien.ru/kaien/l-ai-chat-proto.git

README.md

@@ -0,0 +1,108 @@
+# l-ai-chat-client-go
+
+Reusable Go SDK for l-ai-chat agent/service clients. It uses the authoritative
+`l-ai-chat-proto` generated Go API through the `proto/l-ai-chat-proto`
+submodule and does not copy protobuf files manually.
+
+## Install / Build
+
+Clone with submodules:
+
+```bash
+git clone --recurse-submodules <repo-url> l-ai-chat-client-go
+cd l-ai-chat-client-go
+go test ./...
+```
+
+On this workstation the server currently consumes proto commit
+`f1f44e4ed276316ef17f8deccac54da457654312`. The local sibling
+`../l-ai-chat-proto` checkout is older, so the SDK submodule is pinned to the
+same proto worktree commit as `../l-ai-chat`.
+
+## Transport and Auth
+
+`New` uses TLS by default and automatically adds service auth metadata to
+unary RPCs, `Subscribe`, reconnects, and `Acknowledge`:
+
+```text
+authorization: Bearer <token>
+```
+
+Use `WithInsecureTransport()` only for trusted/internal networks. The service
+token travels over this gRPC transport; plaintext mode is suitable only for
+trusted/internal network until service-to-service TLS/mTLS is deployed.
+
+Tokens come from a `TokenProvider`; `StaticToken` is the simple provider. Empty
+tokens and provider errors fail closed. Tokens are not included in SDK errors.
+
+## Delivery Guarantee
+
+Agent delivery is durable at-least-once:
+
+- receiving a message does not mean processing succeeded;
+- the server durable cursor changes only after successful `Acknowledge`;
+- the SDK acknowledges only after `MarkProcessed`;
+- unacked messages replay after reconnect;
+- handlers with external side effects must be idempotent.
+
+The server derives the durable consumer identity from the authenticated
+agent/service token. The SDK does not accept an arbitrary consumer id.
+
+## Cumulative ACK Warning
+
+`Acknowledge(last_ack_sequence=N)` is cumulative. If deliveries are:
+
+```text
+100, 104, 111
+```
+
+and `104` finishes before `100`, do not acknowledge `104` yet: that would mark
+`100` processed too. Use `Subscription.MarkProcessed`; the SDK tracks actual
+delivery order, not `seq+1`, so global sequence gaps inside a topic are safe.
+
+Expected ACK behavior:
+
+```text
+104 done -> no ACK
+100 done -> ACK 104
+111 done -> ACK 111
+```
+
+## Subscription Lifecycle
+
+`Subscribe` opens one server stream for one topic and waits for the first
+`ready=true` marker before returning. READY is not exposed as a chat message.
+Each successful stream registration is a new SDK generation. Deliveries carry
+internal generation/subscription fencing, so an old `Delivery` cannot ACK a new
+stream after reconnect.
+
+Transient stream failures (`Unavailable`, `ResourceExhausted`,
+`DeadlineExceeded`, EOF/transport reset) reconnect with bounded exponential
+backoff and jitter. Reconnect does not send a local receive cursor; the server
+starts from the durable ACK cursor and replays unacked messages.
+
+`Aborted` means the server intentionally replaced this active subscription
+because another stream for the same `(consumer, topic)` connected. The SDK
+treats that as terminal by default to avoid reconnect ping-pong between two
+replicas of one logical consumer. HA/leader election belongs above this SDK.
+
+First-time `(consumer, topic)` subscriptions start at the current topic head.
+Use `History` for historical context. Existing consumers resume from the
+durable ACK cursor. `Delivery.Replay` means redelivery/backlog context, not an
+error.
+
+## gRPC Surface
+
+The SDK wraps implemented read/agent RPCs:
+
+- `ListTopics`
+- `GetTopic`
+- `GetMessage`
+- `History`
+- `PostMessage` using server-side `AgentPostMessage`
+- `Subscribe`
+- `Subscription.MarkProcessed`
+
+Generic `PostMessage`, `CreateTopic`, and `UpdateTopic` are intentionally not
+exposed as ergonomic helpers because the current server returns
+`Unimplemented` for those fail-closed gRPC paths.

ack_tracker.go

@@ -0,0 +1,67 @@
+package chatclient
+
+type deliveryToken struct {
+	_ byte
+}
+
+type ackEntry struct {
+	seq   int64
+	token *deliveryToken
+	done  bool
+}
+
+type ackTracker struct {
+	entries []ackEntry
+	index   map[*deliveryToken]int
+}
+
+func newACKTracker() ackTracker {
+	return ackTracker{index: make(map[*deliveryToken]int)}
+}
+
+func (t *ackTracker) add(seq int64, token *deliveryToken) {
+	t.index[token] = len(t.entries)
+	t.entries = append(t.entries, ackEntry{seq: seq, token: token})
+}
+
+func (t *ackTracker) mark(token *deliveryToken) (int64, bool, bool) {
+	i, ok := t.index[token]
+	if !ok {
+		return 0, false, false
+	}
+	if t.entries[i].done {
+		return t.safeBoundary(), true, true
+	}
+	t.entries[i].done = true
+	return t.safeBoundary(), true, true
+}
+
+func (t *ackTracker) safeBoundary() int64 {
+	var boundary int64
+	for i := range t.entries {
+		if !t.entries[i].done {
+			break
+		}
+		boundary = t.entries[i].seq
+	}
+	return boundary
+}
+
+func (t *ackTracker) removeConfirmed(lastAck int64) {
+	n := 0
+	for n < len(t.entries) {
+		if !t.entries[n].done || t.entries[n].seq > lastAck {
+			break
+		}
+		delete(t.index, t.entries[n].token)
+		n++
+	}
+	if n == 0 {
+		return
+	}
+	copy(t.entries, t.entries[n:])
+	t.entries = t.entries[:len(t.entries)-n]
+	for i := range t.entries {
+		t.index[t.entries[i].token] = i
+	}
+}

auth.go

@@ -0,0 +1,42 @@
+package chatclient
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"google.golang.org/grpc/credentials"
+)
+
+type TokenProvider func(ctx context.Context) (string, error)
+
+func StaticToken(token string) TokenProvider {
+	return func(context.Context) (string, error) {
+		if strings.TrimSpace(token) == "" {
+			return "", fmt.Errorf("%w: empty service token", ErrInvalidConfig)
+		}
+		return token, nil
+	}
+}
+
+type tokenCredentials struct {
+	provider                 TokenProvider
+	requireTransportSecurity bool
+}
+
+func (c tokenCredentials) GetRequestMetadata(ctx context.Context, _ ...string) (map[string]string, error) {
+	token, err := c.provider(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if strings.TrimSpace(token) == "" {
+		return nil, fmt.Errorf("%w: empty service token", ErrInvalidConfig)
+	}
+	return map[string]string{"authorization": "Bearer " + token}, nil
+}
+
+func (c tokenCredentials) RequireTransportSecurity() bool {
+	return c.requireTransportSecurity
+}
+
+var _ credentials.PerRPCCredentials = tokenCredentials{}

client.go

@@ -0,0 +1,205 @@
+package chatclient
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/google/uuid"
+	"google.golang.org/grpc"
+
+	pb "git.be.kaien.ru/l-ai/l-ai-chat-client-go/proto/l-ai-chat-proto"
+)
+
+type Client struct {
+	conn  *grpc.ClientConn
+	own   bool
+	api   pb.ChatServiceClient
+	opts  options
+	mu    sync.Mutex
+	subs  map[*Subscription]struct{}
+	close bool
+}
+
+func New(ctx context.Context, target string, opts ...Option) (*Client, error) {
+	o := defaultOptions()
+	for _, opt := range opts {
+		opt(&o)
+	}
+	if err := validateOptions(o); err != nil {
+		return nil, err
+	}
+
+	var conn *grpc.ClientConn
+	own := false
+	if o.conn != nil {
+		conn = o.conn
+	} else {
+		dialOpts, err := buildDialOptions(o)
+		if err != nil {
+			return nil, err
+		}
+		conn, err = grpc.NewClient(target, dialOpts...)
+		if err != nil {
+			return nil, err
+		}
+		own = true
+	}
+
+	c := &Client{
+		conn: conn,
+		own:  own,
+		api:  pb.NewChatServiceClient(conn),
+		opts: o,
+		subs: make(map[*Subscription]struct{}),
+	}
+	if ctx != nil {
+		select {
+		case <-ctx.Done():
+			_ = c.Close()
+			return nil, ctx.Err()
+		default:
+		}
+	}
+	return c, nil
+}
+
+func (c *Client) Close() error {
+	c.mu.Lock()
+	if c.close {
+		c.mu.Unlock()
+		return nil
+	}
+	c.close = true
+	subs := make([]*Subscription, 0, len(c.subs))
+	for sub := range c.subs {
+		subs = append(subs, sub)
+	}
+	c.mu.Unlock()
+
+	for _, sub := range subs {
+		_ = sub.Close()
+	}
+	if c.own && c.conn != nil {
+		return c.conn.Close()
+	}
+	return nil
+}
+
+func (c *Client) ListTopics(ctx context.Context) ([]*pb.Topic, error) {
+	if err := c.checkOpen(); err != nil {
+		return nil, err
+	}
+	resp, err := c.api.ListTopics(ctx, &pb.ListTopicsRequest{})
+	if err != nil {
+		return nil, err
+	}
+	return resp.GetTopics(), nil
+}
+
+func (c *Client) GetTopic(ctx context.Context, req GetTopicRequest) (*pb.Topic, error) {
+	if err := c.checkOpen(); err != nil {
+		return nil, err
+	}
+	if req.ID != "" && req.Slug != "" {
+		return nil, fmt.Errorf("%w: only one of topic id or slug may be set", ErrInvalidConfig)
+	}
+	in := &pb.GetTopicRequest{}
+	switch {
+	case req.ID != "":
+		in.Selector = &pb.GetTopicRequest_Id{Id: req.ID}
+	case req.Slug != "":
+		in.Selector = &pb.GetTopicRequest_Slug{Slug: req.Slug}
+	default:
+		return nil, fmt.Errorf("%w: topic id or slug is required", ErrInvalidConfig)
+	}
+	return c.api.GetTopic(ctx, in)
+}
+
+type GetTopicRequest struct {
+	ID   string
+	Slug string
+}
+
+func (c *Client) GetMessage(ctx context.Context, messageID string) (*pb.Message, error) {
+	if err := c.checkOpen(); err != nil {
+		return nil, err
+	}
+	return c.api.GetMessage(ctx, &pb.GetMessageRequest{MessageId: messageID})
+}
+
+type PostMessageRequest struct {
+	TopicID          string
+	Content          string
+	ReplyToMessageID string
+	IdempotencyKey   string
+}
+
+func (c *Client) PostMessage(ctx context.Context, req PostMessageRequest) (*pb.PostMessageResponse, error) {
+	if err := c.checkOpen(); err != nil {
+		return nil, err
+	}
+	in := &pb.AgentPostMessageRequest{
+		TopicId:        req.TopicID,
+		Content:        req.Content,
+		IdempotencyKey: req.IdempotencyKey,
+	}
+	if req.ReplyToMessageID != "" {
+		replyTo := req.ReplyToMessageID
+		in.ReplyToMessageId = &replyTo
+	}
+	return c.api.AgentPostMessage(ctx, in)
+}
+
+func NewIdempotencyKey() string {
+	return uuid.NewString()
+}
+
+func (c *Client) Subscribe(ctx context.Context, req SubscribeRequest) (*Subscription, error) {
+	if err := c.checkOpen(); err != nil {
+		return nil, err
+	}
+	if req.TopicID == "" {
+		return nil, fmt.Errorf("%w: topic id is required", ErrInvalidConfig)
+	}
+	sub, err := newSubscription(ctx, c, req)
+	if err != nil {
+		return nil, err
+	}
+	c.mu.Lock()
+	if c.close {
+		c.mu.Unlock()
+		_ = sub.Close()
+		return nil, ErrClosed
+	}
+	c.subs[sub] = struct{}{}
+	c.mu.Unlock()
+	return sub, nil
+}
+
+type SubscribeRequest struct {
+	TopicID string
+}
+
+func (c *Client) unregister(sub *Subscription) {
+	c.mu.Lock()
+	delete(c.subs, sub)
+	c.mu.Unlock()
+}
+
+func (c *Client) checkOpen() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.close {
+		return ErrClosed
+	}
+	return nil
+}
+
+func ignoreClosed(err error) error {
+	if errors.Is(err, context.Canceled) || errors.Is(err, ErrClosed) {
+		return nil
+	}
+	return err
+}

doc.go

@@ -0,0 +1,6 @@
+// Package chatclient is a reusable Go SDK for l-ai-chat agent/service clients.
+//
+// It wraps the authoritative ChatService protobuf client with service-token
+// authentication, AgentPostMessage helpers, paged History reads, and durable
+// at-least-once subscriptions with ordered cumulative ACK tracking.
+package chatclient

errors.go

@@ -0,0 +1,12 @@
+package chatclient
+
+import "errors"
+
+var (
+	ErrClosed               = errors.New("chat client is closed")
+	ErrInvalidConfig        = errors.New("invalid chat client configuration")
+	ErrStaleDelivery        = errors.New("stale delivery")
+	ErrUnknownDelivery      = errors.New("unknown delivery")
+	ErrSubscriptionReplaced = errors.New("subscription replaced")
+	ErrProtocolViolation    = errors.New("chat service protocol violation")
+)

example_test.go

@@ -0,0 +1,99 @@
+package chatclient_test
+
+import (
+	"context"
+	"log"
+
+	chatclient "git.be.kaien.ru/l-ai/l-ai-chat-client-go"
+)
+
+func ExampleNew() {
+	ctx := context.Background()
+	client, err := chatclient.New(ctx, "chat.internal:443",
+		chatclient.WithStaticToken("service-token-from-secret-store"),
+	)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer client.Close()
+}
+
+func ExampleClient_PostMessage() {
+	ctx := context.Background()
+	client, _ := chatclient.New(ctx, "chat.internal:443",
+		chatclient.WithStaticToken("service-token-from-secret-store"),
+	)
+	defer client.Close()
+
+	key := "security-agent:logical-action-123"
+	_, err := client.PostMessage(ctx, chatclient.PostMessageRequest{
+		TopicID:        "11111111-1111-1111-1111-111111111111",
+		Content:        "scan completed",
+		IdempotencyKey: key,
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func ExampleSubscription_sequential() {
+	ctx := context.Background()
+	client, _ := chatclient.New(ctx, "chat.internal:443",
+		chatclient.WithStaticToken("service-token-from-secret-store"),
+	)
+	defer client.Close()
+
+	sub, err := client.Subscribe(ctx, chatclient.SubscribeRequest{
+		TopicID: "11111111-1111-1111-1111-111111111111",
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer sub.Close()
+
+	for delivery := range sub.Deliveries() {
+		if err := process(ctx, delivery); err != nil {
+			log.Fatal(err)
+		}
+		if err := sub.MarkProcessed(ctx, delivery); err != nil {
+			log.Fatal(err)
+		}
+	}
+	if err := sub.Err(); err != nil {
+		log.Fatal(err)
+	}
+}
+
+func ExampleSubscription_concurrentProcessing() {
+	ctx := context.Background()
+	client, _ := chatclient.New(ctx, "chat.internal:443",
+		chatclient.WithStaticToken("service-token-from-secret-store"),
+	)
+	defer client.Close()
+
+	sub, err := client.Subscribe(ctx, chatclient.SubscribeRequest{
+		TopicID: "11111111-1111-1111-1111-111111111111",
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer sub.Close()
+
+	sem := make(chan struct{}, 4)
+	for delivery := range sub.Deliveries() {
+		delivery := delivery
+		sem <- struct{}{}
+		go func() {
+			defer func() { <-sem }()
+			if err := process(ctx, delivery); err != nil {
+				_ = sub.Close()
+				return
+			}
+			// Safe even when goroutines finish out of order: MarkProcessed
+			// advances only the processed prefix in actual delivery order.
+			_ = sub.MarkProcessed(ctx, delivery)
+		}()
+	}
+}
+
+func process(context.Context, chatclient.Delivery) error { return nil }

go.mod

@@ -0,0 +1,16 @@
+module git.be.kaien.ru/l-ai/l-ai-chat-client-go
+
+go 1.25.0
+
+require (
+	github.com/google/uuid v1.6.0
+	google.golang.org/grpc v1.70.0
+	google.golang.org/protobuf v1.36.8
+)
+
+require (
+	golang.org/x/net v0.32.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20241202173237-19429a94021a // indirect
+)

go.sum

@@ -0,0 +1,32 @@
+github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
+github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+go.opentelemetry.io/otel v1.32.0 h1:WnBN+Xjcteh0zdk01SVqV55d/m62NJLJdIyb4y/WO5U=
+go.opentelemetry.io/otel v1.32.0/go.mod h1:00DCVSB0RQcnzlwyTfqtxSm+DRr9hpYrHjNGiBHVQIg=
+go.opentelemetry.io/otel/metric v1.32.0 h1:xV2umtmNcThh2/a/aCP+h64Xx5wsj8qqnkYZktzNa0M=
+go.opentelemetry.io/otel/metric v1.32.0/go.mod h1:jH7CIbbK6SH2V2wE16W05BHCtIDzauciCRLoc/SyMv8=
+go.opentelemetry.io/otel/sdk v1.32.0 h1:RNxepc9vK59A8XsgZQouW8ue8Gkb4jpWtJm9ge5lEG4=
+go.opentelemetry.io/otel/sdk v1.32.0/go.mod h1:LqgegDBjKMmb2GC6/PrTnteJG39I8/vJCAP9LlJXEjU=
+go.opentelemetry.io/otel/sdk/metric v1.32.0 h1:rZvFnvmvawYb0alrYkjraqJq0Z4ZUJAiyYCU9snn1CU=
+go.opentelemetry.io/otel/sdk/metric v1.32.0/go.mod h1:PWeZlq0zt9YkYAp3gjKZ0eicRYvOh1Gd+X99x6GHpCQ=
+go.opentelemetry.io/otel/trace v1.32.0 h1:WIC9mYrXf8TmY/EXuULKc8hR17vE+Hjv2cssQDe03fM=
+go.opentelemetry.io/otel/trace v1.32.0/go.mod h1:+i4rkvCraA+tG6AzwloGaCtkx53Fa+L+V8e9a7YvhT8=
+golang.org/x/net v0.32.0 h1:ZqPmj8Kzc+Y6e0+skZsuACbx+wzMgo5MQsJh9Qd6aYI=
+golang.org/x/net v0.32.0/go.mod h1:CwU0IoeOlnQQWJ6ioyFrfRuomB8GKF6KbYXZVyeXNfs=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20241202173237-19429a94021a h1:hgh8P4EuoxpsuKMXX/To36nOFD7vixReXgn8lPGnt+o=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20241202173237-19429a94021a/go.mod h1:5uTbfoYQed2U9p3KIj2/Zzm02PYhndfdmML0qC3q3FU=
+google.golang.org/grpc v1.70.0 h1:pWFv03aZoHzlRKHWicjsZytKAiYCtNS0dHbXnIdq7jQ=
+google.golang.org/grpc v1.70.0/go.mod h1:ofIJqVKDXx/JiXrwr2IG4/zwdH9txy3IlF40RmcJSQw=
+google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
+google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=

history.go

@@ -0,0 +1,70 @@
+package chatclient
+
+import (
+	"context"
+	"fmt"
+
+	pb "git.be.kaien.ru/l-ai/l-ai-chat-client-go/proto/l-ai-chat-proto"
+)
+
+type HistoryRequest struct {
+	TopicID        string
+	BeforeSequence int64
+	AfterSequence  int64
+	Limit          int32
+}
+
+func (c *Client) History(ctx context.Context, req HistoryRequest) (*pb.HistoryResponse, error) {
+	if err := c.checkOpen(); err != nil {
+		return nil, err
+	}
+	if req.BeforeSequence > 0 && req.AfterSequence > 0 {
+		return nil, fmt.Errorf("%w: before and after sequence are mutually exclusive", ErrInvalidConfig)
+	}
+	return c.api.History(ctx, &pb.HistoryRequest{
+		TopicId:        req.TopicID,
+		BeforeSequence: req.BeforeSequence,
+		AfterSequence:  req.AfterSequence,
+		Limit:          req.Limit,
+	})
+}
+
+type HistoryPager struct {
+	client  *Client
+	req     HistoryRequest
+	forward bool
+	done    bool
+}
+
+func (c *Client) NewHistoryPager(req HistoryRequest) (*HistoryPager, error) {
+	if req.BeforeSequence > 0 && req.AfterSequence > 0 {
+		return nil, fmt.Errorf("%w: before and after sequence are mutually exclusive", ErrInvalidConfig)
+	}
+	return &HistoryPager{
+		client:  c,
+		req:     req,
+		forward: req.AfterSequence > 0,
+	}, nil
+}
+
+func (p *HistoryPager) Next(ctx context.Context) (*pb.HistoryResponse, error) {
+	if p.done {
+		return &pb.HistoryResponse{}, nil
+	}
+	resp, err := p.client.History(ctx, p.req)
+	if err != nil {
+		return nil, err
+	}
+	if len(resp.GetMessages()) == 0 || !resp.GetHasMore() {
+		p.done = true
+	}
+	if len(resp.GetMessages()) > 0 {
+		last := resp.GetMessages()[len(resp.GetMessages())-1].GetSequence()
+		if p.forward {
+			p.req.AfterSequence = last
+		} else {
+			p.req.BeforeSequence = last
+		}
+	}
+	return resp, nil
+}

options.go

@@ -0,0 +1,153 @@
+package chatclient
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+const (
+	defaultSubscriptionBufferSize = 100
+	defaultReconnectInitial       = 200 * time.Millisecond
+	defaultReconnectMax           = 5 * time.Second
+	defaultCloseTimeout           = 5 * time.Second
+)
+
+type Sleeper func(ctx context.Context, d time.Duration) error
+type Jitter func(d time.Duration) time.Duration
+
+type Option func(*options)
+
+type options struct {
+	tokenProvider          TokenProvider
+	insecureTransport      bool
+	dialOptions            []grpc.DialOption
+	conn                   *grpc.ClientConn
+	subscriptionBufferSize int
+	reconnectInitial       time.Duration
+	reconnectMax           time.Duration
+	closeTimeout           time.Duration
+	sleeper                Sleeper
+	jitter                 Jitter
+}
+
+func defaultOptions() options {
+	return options{
+		subscriptionBufferSize: defaultSubscriptionBufferSize,
+		reconnectInitial:       defaultReconnectInitial,
+		reconnectMax:           defaultReconnectMax,
+		closeTimeout:           defaultCloseTimeout,
+		sleeper:                sleepContext,
+		jitter:                 defaultJitter,
+	}
+}
+
+func WithTokenProvider(provider TokenProvider) Option {
+	return func(o *options) { o.tokenProvider = provider }
+}
+
+func WithStaticToken(token string) Option {
+	return WithTokenProvider(StaticToken(token))
+}
+
+func WithInsecureTransport() Option {
+	return func(o *options) { o.insecureTransport = true }
+}
+
+func WithGRPCDialOptions(opts ...grpc.DialOption) Option {
+	return func(o *options) { o.dialOptions = append(o.dialOptions, opts...) }
+}
+
+func WithClientConn(conn *grpc.ClientConn) Option {
+	return func(o *options) { o.conn = conn }
+}
+
+func WithSubscriptionBufferSize(size int) Option {
+	return func(o *options) { o.subscriptionBufferSize = size }
+}
+
+func WithReconnectBackoff(initial, max time.Duration) Option {
+	return func(o *options) {
+		o.reconnectInitial = initial
+		o.reconnectMax = max
+	}
+}
+
+func withSleeperForTest(s Sleeper) Option {
+	return func(o *options) { o.sleeper = s }
+}
+
+func withJitterForTest(j Jitter) Option {
+	return func(o *options) { o.jitter = j }
+}
+
+func buildDialOptions(o options) ([]grpc.DialOption, error) {
+	if o.tokenProvider == nil {
+		return nil, fmt.Errorf("%w: token provider is required", ErrInvalidConfig)
+	}
+
+	var transport credentials.TransportCredentials
+	if o.insecureTransport {
+		transport = insecure.NewCredentials()
+	} else {
+		transport = credentials.NewTLS(&tls.Config{MinVersion: tls.VersionTLS12})
+	}
+
+	creds := tokenCredentials{
+		provider:                 o.tokenProvider,
+		requireTransportSecurity: !o.insecureTransport,
+	}
+	out := []grpc.DialOption{
+		grpc.WithTransportCredentials(transport),
+		grpc.WithPerRPCCredentials(creds),
+	}
+	out = append(out, o.dialOptions...)
+	return out, nil
+}
+
+func validateOptions(o options) error {
+	if o.tokenProvider == nil {
+		return fmt.Errorf("%w: token provider is required", ErrInvalidConfig)
+	}
+	if o.subscriptionBufferSize <= 0 {
+		return fmt.Errorf("%w: subscription buffer size must be positive", ErrInvalidConfig)
+	}
+	if o.reconnectInitial <= 0 {
+		return fmt.Errorf("%w: reconnect initial backoff must be positive", ErrInvalidConfig)
+	}
+	if o.reconnectMax < o.reconnectInitial {
+		return fmt.Errorf("%w: reconnect max backoff must be >= initial backoff", ErrInvalidConfig)
+	}
+	if o.closeTimeout <= 0 {
+		return fmt.Errorf("%w: close timeout must be positive", ErrInvalidConfig)
+	}
+	if o.sleeper == nil || o.jitter == nil {
+		return fmt.Errorf("%w: backoff hooks must be configured", ErrInvalidConfig)
+	}
+	return nil
+}
+
+func sleepContext(ctx context.Context, d time.Duration) error {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-timer.C:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func defaultJitter(d time.Duration) time.Duration {
+	// Deterministic half-jitter keeps reconnects bounded without pulling in a
+	// retry framework. Tests replace this hook.
+	if d <= 1 {
+		return d
+	}
+	return d - d/4
+}

proto/l-ai-chat-proto

@@ -0,0 +1 @@
+Subproject commit f1f44e4ed276316ef17f8deccac54da457654312

subscription.go

@@ -0,0 +1,406 @@
+package chatclient
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	pb "git.be.kaien.ru/l-ai/l-ai-chat-client-go/proto/l-ai-chat-proto"
+)
+
+type Delivery struct {
+	Message *pb.Message
+	Replay  bool
+
+	subscriptionID string
+	generation     uint64
+	sub            *Subscription
+	token          *deliveryToken
+}
+
+type Handler func(context.Context, Delivery) error
+
+type Subscription struct {
+	client *Client
+	topic  string
+
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	deliveries chan Delivery
+	done       chan struct{}
+	ackNotify  chan struct{}
+
+	mu             sync.Mutex
+	err            error
+	closed         bool
+	generation     uint64
+	subscriptionID string
+	tracker        ackTracker
+	ackTarget      int64
+	ackConfirmed   int64
+
+	readyStream pb.ChatService_SubscribeClient
+}
+
+func newSubscription(ctx context.Context, c *Client, req SubscribeRequest) (*Subscription, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	subCtx, cancel := context.WithCancel(ctx)
+	s := &Subscription{
+		client:     c,
+		topic:      req.TopicID,
+		ctx:        subCtx,
+		cancel:     cancel,
+		deliveries: make(chan Delivery, c.opts.subscriptionBufferSize),
+		done:       make(chan struct{}),
+		ackNotify:  make(chan struct{}, 1),
+		tracker:    newACKTracker(),
+	}
+
+	stream, err := s.openReady(subCtx)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	s.readyStream = stream
+	s.generation = 1
+
+	go s.run()
+	go s.runACK()
+	return s, nil
+}
+
+func (s *Subscription) Deliveries() <-chan Delivery { return s.deliveries }
+func (s *Subscription) Done() <-chan struct{}       { return s.done }
+
+func (s *Subscription) Err() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.err
+}
+
+func (s *Subscription) Close() error {
+	s.mu.Lock()
+	if s.closed {
+		s.mu.Unlock()
+		return nil
+	}
+	s.closed = true
+	s.mu.Unlock()
+	s.cancel()
+
+	timer := time.NewTimer(s.client.opts.closeTimeout)
+	defer timer.Stop()
+	select {
+	case <-s.done:
+		return nil
+	case <-timer.C:
+		return context.DeadlineExceeded
+	}
+}
+
+func (s *Subscription) MarkProcessed(ctx context.Context, d Delivery) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	if d.sub != s || d.token == nil {
+		return ErrUnknownDelivery
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.closed {
+		return ErrClosed
+	}
+	if d.generation != s.generation {
+		return ErrStaleDelivery
+	}
+	boundary, ok, _ := s.tracker.mark(d.token)
+	if !ok {
+		return ErrUnknownDelivery
+	}
+	if boundary > s.ackTarget {
+		s.ackTarget = boundary
+		s.signalACKLocked()
+	}
+	return nil
+}
+
+func (s *Subscription) Run(ctx context.Context, handler Handler) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case d, ok := <-s.deliveries:
+			if !ok {
+				return ignoreClosed(s.Err())
+			}
+			if err := handler(ctx, d); err != nil {
+				return err
+			}
+			if err := s.MarkProcessed(ctx, d); err != nil {
+				return err
+			}
+		}
+	}
+}
+
+func (s *Subscription) run() {
+	defer func() {
+		s.cancel()
+		close(s.deliveries)
+		close(s.done)
+		s.client.unregister(s)
+	}()
+
+	stream := s.readyStream
+	s.readyStream = nil
+	backoff := s.client.opts.reconnectInitial
+
+	for {
+		err := s.recvUntilError(stream)
+		if s.ctx.Err() != nil {
+			s.setErr(nil)
+			return
+		}
+		if isReplacement(err) {
+			s.setErr(fmt.Errorf("%w: %v", ErrSubscriptionReplaced, err))
+			return
+		}
+		if !isRetryableStreamError(err) {
+			s.setErr(err)
+			return
+		}
+		if err := s.client.opts.sleeper(s.ctx, s.client.opts.jitter(backoff)); err != nil {
+			s.setErr(nil)
+			return
+		}
+		backoff = nextBackoff(backoff, s.client.opts.reconnectMax)
+
+		next, err := s.openReady(s.ctx)
+		if err != nil {
+			if s.ctx.Err() != nil {
+				s.setErr(nil)
+				return
+			}
+			if isReplacement(err) {
+				s.setErr(fmt.Errorf("%w: %v", ErrSubscriptionReplaced, err))
+				return
+			}
+			if !isRetryableStreamError(err) {
+				s.setErr(err)
+				return
+			}
+			continue
+		}
+		s.resetGeneration()
+		stream = next
+		backoff = s.client.opts.reconnectInitial
+	}
+}
+
+func (s *Subscription) recvUntilError(stream pb.ChatService_SubscribeClient) error {
+	for {
+		dm, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		if dm.GetReady() {
+			return fmt.Errorf("%w: unexpected ready marker after subscription registration", ErrProtocolViolation)
+		}
+		if dm.GetMessage() == nil {
+			return fmt.Errorf("%w: delivery message is missing message", ErrProtocolViolation)
+		}
+		token := &deliveryToken{}
+		delivery := Delivery{
+			Message:        dm.GetMessage(),
+			Replay:         dm.GetReplay(),
+			subscriptionID: dm.GetSubscriptionId(),
+			token:          token,
+			sub:            s,
+		}
+		s.mu.Lock()
+		if s.closed {
+			s.mu.Unlock()
+			return ErrClosed
+		}
+		if s.subscriptionID == "" {
+			s.subscriptionID = dm.GetSubscriptionId()
+		}
+		if dm.GetSubscriptionId() == "" || dm.GetSubscriptionId() != s.subscriptionID {
+			s.mu.Unlock()
+			return fmt.Errorf("%w: inconsistent subscription_id", ErrProtocolViolation)
+		}
+		delivery.generation = s.generation
+		s.tracker.add(dm.GetMessage().GetSequence(), token)
+		s.mu.Unlock()
+
+		select {
+		case s.deliveries <- delivery:
+		case <-s.ctx.Done():
+			return ErrClosed
+		}
+	}
+}
+
+func (s *Subscription) openReady(ctx context.Context) (pb.ChatService_SubscribeClient, error) {
+	stream, err := s.client.api.Subscribe(ctx, &pb.SubscribeRequest{TopicId: s.topic})
+	if err != nil {
+		return nil, err
+	}
+	dm, err := stream.Recv()
+	if err != nil {
+		return nil, err
+	}
+	if !dm.GetReady() || dm.GetMessage() != nil {
+		return nil, fmt.Errorf("%w: subscribe stream did not start with ready marker", ErrProtocolViolation)
+	}
+	return stream, nil
+}
+
+func (s *Subscription) resetGeneration() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.generation++
+	s.subscriptionID = ""
+	s.tracker = newACKTracker()
+	s.ackTarget = 0
+	s.ackConfirmed = 0
+}
+
+func (s *Subscription) runACK() {
+	for {
+		select {
+		case <-s.ctx.Done():
+			return
+		case <-s.ackNotify:
+		}
+
+		for {
+			gen, subID, target, ok := s.nextACK()
+			if !ok {
+				break
+			}
+			resp, err := s.client.api.Acknowledge(s.ctx, &pb.AcknowledgeRequest{
+				TopicId:         s.topic,
+				LastAckSequence: target,
+				SubscriptionId:  subID,
+			})
+			if err != nil {
+				if s.ctx.Err() != nil || !isRetryableACKError(err) || !s.sameGeneration(gen) {
+					if s.ctx.Err() == nil && !isRetryableACKError(err) {
+						s.setErr(err)
+						s.cancel()
+					}
+					break
+				}
+				_ = s.client.opts.sleeper(s.ctx, s.client.opts.jitter(s.client.opts.reconnectInitial))
+				continue
+			}
+			s.confirmACK(gen, resp.GetLastAckSequence())
+		}
+	}
+}
+
+func (s *Subscription) nextACK() (uint64, string, int64, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.closed || s.subscriptionID == "" || s.ackTarget <= s.ackConfirmed {
+		return 0, "", 0, false
+	}
+	return s.generation, s.subscriptionID, s.ackTarget, true
+}
+
+func (s *Subscription) sameGeneration(gen uint64) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.generation == gen && !s.closed
+}
+
+func (s *Subscription) confirmACK(gen uint64, confirmed int64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.generation != gen || s.closed {
+		return
+	}
+	if confirmed > s.ackConfirmed {
+		s.ackConfirmed = confirmed
+		s.tracker.removeConfirmed(confirmed)
+	}
+	if s.ackTarget > s.ackConfirmed {
+		s.signalACKLocked()
+	}
+}
+
+func (s *Subscription) signalACKLocked() {
+	select {
+	case s.ackNotify <- struct{}{}:
+	default:
+	}
+}
+
+func (s *Subscription) setErr(err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.err == nil {
+		s.err = err
+	}
+	s.closed = true
+}
+
+func nextBackoff(cur, max time.Duration) time.Duration {
+	if cur <= 0 {
+		return max
+	}
+	next := cur * 2
+	if next < cur || next > max {
+		return max
+	}
+	return next
+}
+
+func isReplacement(err error) bool {
+	return status.Code(err) == codes.Aborted
+}
+
+func isRetryableStreamError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, io.EOF) {
+		return true
+	}
+	switch status.Code(err) {
+	case codes.Unavailable, codes.ResourceExhausted, codes.DeadlineExceeded:
+		return true
+	case codes.Canceled:
+		return false
+	default:
+		return false
+	}
+}
+
+func isRetryableACKError(err error) bool {
+	switch status.Code(err) {
+	case codes.Unavailable, codes.ResourceExhausted, codes.DeadlineExceeded:
+		return true
+	default:
+		return false
+	}
+}

subscription_test.go

@@ -0,0 +1,605 @@
+package chatclient
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+
+	pb "git.be.kaien.ru/l-ai/l-ai-chat-client-go/proto/l-ai-chat-proto"
+)
+
+type fakeChatServer struct {
+	pb.UnimplementedChatServiceServer
+
+	mu             sync.Mutex
+	authByMethod   map[string][]string
+	postReqs       []*pb.AgentPostMessageRequest
+	genericPosts   int
+	ackCalls       []ackCall
+	ackErrors      map[int64][]error
+	subscribeCount int
+	subscribeFunc  func(*pb.SubscribeRequest, pb.ChatService_SubscribeServer) error
+	historyPages   []*pb.HistoryResponse
+	historyReqs    []*pb.HistoryRequest
+}
+
+type ackCall struct {
+	topic string
+	subID string
+	seq   int64
+}
+
+func newFake() *fakeChatServer {
+	return &fakeChatServer{
+		authByMethod: make(map[string][]string),
+		ackErrors:    make(map[int64][]error),
+	}
+}
+
+func (f *fakeChatServer) record(ctx context.Context, method string) {
+	md, _ := metadata.FromIncomingContext(ctx)
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.authByMethod[method] = append([]string(nil), md.Get("authorization")...)
+}
+
+func (f *fakeChatServer) ListTopics(ctx context.Context, _ *pb.ListTopicsRequest) (*pb.ListTopicsResponse, error) {
+	f.record(ctx, "ListTopics")
+	return &pb.ListTopicsResponse{Topics: []*pb.Topic{{Id: "topic-1", Slug: "general"}}}, nil
+}
+
+func (f *fakeChatServer) GetTopic(ctx context.Context, _ *pb.GetTopicRequest) (*pb.Topic, error) {
+	f.record(ctx, "GetTopic")
+	return &pb.Topic{Id: "topic-1", Slug: "general"}, nil
+}
+
+func (f *fakeChatServer) GetMessage(ctx context.Context, req *pb.GetMessageRequest) (*pb.Message, error) {
+	f.record(ctx, "GetMessage")
+	return &pb.Message{Id: req.GetMessageId(), Sequence: 1}, nil
+}
+
+func (f *fakeChatServer) PostMessage(context.Context, *pb.PostMessageRequest) (*pb.PostMessageResponse, error) {
+	f.mu.Lock()
+	f.genericPosts++
+	f.mu.Unlock()
+	return nil, status.Error(codes.Unimplemented, "generic post is fail-closed")
+}
+
+func (f *fakeChatServer) AgentPostMessage(ctx context.Context, req *pb.AgentPostMessageRequest) (*pb.PostMessageResponse, error) {
+	f.record(ctx, "AgentPostMessage")
+	f.mu.Lock()
+	f.postReqs = append(f.postReqs, req)
+	f.mu.Unlock()
+	return &pb.PostMessageResponse{
+		Message: &pb.Message{Id: "posted", TopicId: req.GetTopicId(), Content: req.GetContent(), Sequence: 10},
+	}, nil
+}
+
+func (f *fakeChatServer) History(ctx context.Context, req *pb.HistoryRequest) (*pb.HistoryResponse, error) {
+	f.record(ctx, "History")
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.historyReqs = append(f.historyReqs, req)
+	if len(f.historyPages) == 0 {
+		return &pb.HistoryResponse{}, nil
+	}
+	page := f.historyPages[0]
+	f.historyPages = f.historyPages[1:]
+	return page, nil
+}
+
+func (f *fakeChatServer) Subscribe(req *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+	f.record(stream.Context(), "Subscribe")
+	f.mu.Lock()
+	f.subscribeCount++
+	fn := f.subscribeFunc
+	f.mu.Unlock()
+	if fn == nil {
+		return status.Error(codes.Unavailable, "no subscribe script")
+	}
+	return fn(req, stream)
+}
+
+func (f *fakeChatServer) Acknowledge(ctx context.Context, req *pb.AcknowledgeRequest) (*pb.AcknowledgeResponse, error) {
+	f.record(ctx, "Acknowledge")
+	f.mu.Lock()
+	f.ackCalls = append(f.ackCalls, ackCall{topic: req.GetTopicId(), subID: req.GetSubscriptionId(), seq: req.GetLastAckSequence()})
+	errs := f.ackErrors[req.GetLastAckSequence()]
+	if len(errs) > 0 {
+		err := errs[0]
+		f.ackErrors[req.GetLastAckSequence()] = errs[1:]
+		f.mu.Unlock()
+		return nil, err
+	}
+	f.mu.Unlock()
+	return &pb.AcknowledgeResponse{LastAckSequence: req.GetLastAckSequence()}, nil
+}
+
+func newClientForTest(t *testing.T, fake *fakeChatServer, opts ...Option) *Client {
+	t.Helper()
+	lis := bufconn.Listen(1 << 20)
+	srv := grpc.NewServer()
+	pb.RegisterChatServiceServer(srv, fake)
+	go func() { _ = srv.Serve(lis) }()
+	t.Cleanup(func() {
+		srv.Stop()
+		_ = lis.Close()
+	})
+
+	all := []Option{
+		WithStaticToken("test-token"),
+		WithInsecureTransport(),
+		WithGRPCDialOptions(grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
+			return lis.DialContext(ctx)
+		})),
+		withJitterForTest(func(d time.Duration) time.Duration { return d }),
+		withSleeperForTest(func(ctx context.Context, d time.Duration) error {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			default:
+				return nil
+			}
+		}),
+	}
+	all = append(all, opts...)
+	c, err := New(context.Background(), "passthrough:///bufnet", all...)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = c.Close() })
+	return c
+}
+
+func msg(seq int64) *pb.DeliveryMessage {
+	return &pb.DeliveryMessage{
+		Message:        &pb.Message{Id: fmt.Sprintf("m-%d", seq), TopicId: "topic", Sequence: seq},
+		SubscriptionId: "sub-1",
+	}
+}
+
+func recvDelivery(t *testing.T, sub *Subscription) Delivery {
+	t.Helper()
+	select {
+	case d := <-sub.Deliveries():
+		return d
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for delivery")
+		return Delivery{}
+	}
+}
+
+func waitDone(t *testing.T, sub *Subscription) {
+	t.Helper()
+	select {
+	case <-sub.Done():
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for subscription done")
+	}
+}
+
+func waitACKs(t *testing.T, fake *fakeChatServer, want ...int64) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		fake.mu.Lock()
+		got := make([]int64, len(fake.ackCalls))
+		for i, c := range fake.ackCalls {
+			got[i] = c.seq
+		}
+		fake.mu.Unlock()
+		if len(got) >= len(want) {
+			ok := true
+			for i := range want {
+				if got[i] != want[i] {
+					ok = false
+				}
+			}
+			if ok {
+				return
+			}
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	t.Fatalf("ack calls = %+v, want prefix %v", fake.ackCalls, want)
+}
+
+func TestAuthMetadataUnarySubscribeAcknowledgeAndEmptyTokenFailsClosed(t *testing.T) {
+	fake := newFake()
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		if err := stream.Send(&pb.DeliveryMessage{Ready: true}); err != nil {
+			return err
+		}
+		if err := stream.Send(msg(100)); err != nil {
+			return err
+		}
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	c := newClientForTest(t, fake)
+	if _, err := c.ListTopics(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	d := recvDelivery(t, sub)
+	if err := sub.MarkProcessed(context.Background(), d); err != nil {
+		t.Fatal(err)
+	}
+	waitACKs(t, fake, 100)
+
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	for _, method := range []string{"ListTopics", "Subscribe", "Acknowledge"} {
+		got := fake.authByMethod[method]
+		if len(got) != 1 || got[0] != "Bearer test-token" {
+			t.Fatalf("%s auth metadata = %v", method, got)
+		}
+	}
+
+	empty := newClientForTest(t, newFake(), WithStaticToken(""))
+	_, err = empty.ListTopics(context.Background())
+	if status.Code(err) != codes.Unauthenticated {
+		t.Fatalf("empty token: got %v", err)
+	}
+	if err != nil && strings.Contains(err.Error(), "test-token") {
+		t.Fatalf("error leaked token: %v", err)
+	}
+}
+
+func TestPostMessageUsesAgentPostMessageAndPreservesIdempotencyKey(t *testing.T) {
+	fake := newFake()
+	c := newClientForTest(t, fake)
+	_, err := c.PostMessage(context.Background(), PostMessageRequest{
+		TopicID:        "topic",
+		Content:        "payload",
+		IdempotencyKey: "logical-key",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	if fake.genericPosts != 0 {
+		t.Fatal("generic PostMessage must not be called")
+	}
+	if len(fake.postReqs) != 1 || fake.postReqs[0].GetIdempotencyKey() != "logical-key" {
+		t.Fatalf("AgentPostMessage reqs = %+v", fake.postReqs)
+	}
+}
+
+func TestOrderedCumulativeACKWithSequenceGaps(t *testing.T) {
+	fake := newFake()
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		for _, seq := range []int64{100, 104, 111} {
+			_ = stream.Send(msg(seq))
+		}
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	c := newClientForTest(t, fake)
+	sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	d100, d104, d111 := recvDelivery(t, sub), recvDelivery(t, sub), recvDelivery(t, sub)
+	if err := sub.MarkProcessed(context.Background(), d104); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(30 * time.Millisecond)
+	fake.mu.Lock()
+	if len(fake.ackCalls) != 0 {
+		t.Fatalf("ACK advanced over unprocessed 100: %+v", fake.ackCalls)
+	}
+	fake.mu.Unlock()
+	if err := sub.MarkProcessed(context.Background(), d100); err != nil {
+		t.Fatalf("mark 100: %v", err)
+	}
+	waitACKs(t, fake, 104)
+	if err := sub.MarkProcessed(context.Background(), d111); err != nil {
+		t.Fatalf("mark 111: %v", err)
+	}
+	waitACKs(t, fake, 104, 111)
+}
+
+func TestLargerOutOfOrderCompletionDoesNotSkipEarliestPending(t *testing.T) {
+	fake := newFake()
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		for _, seq := range []int64{100, 104, 111, 118, 130} {
+			_ = stream.Send(msg(seq))
+		}
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	c := newClientForTest(t, fake)
+	sub, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	deliveries := map[int64]Delivery{}
+	for i := 0; i < 5; i++ {
+		d := recvDelivery(t, sub)
+		deliveries[d.Message.GetSequence()] = d
+	}
+	for _, seq := range []int64{118, 104} {
+		_ = sub.MarkProcessed(context.Background(), deliveries[seq])
+	}
+	time.Sleep(30 * time.Millisecond)
+	fake.mu.Lock()
+	if len(fake.ackCalls) != 0 {
+		t.Fatalf("ACK skipped earliest pending: %+v", fake.ackCalls)
+	}
+	fake.mu.Unlock()
+	if err := sub.MarkProcessed(context.Background(), deliveries[100]); err != nil {
+		t.Fatalf("mark 100: %v", err)
+	}
+	waitACKs(t, fake, 104)
+	if err := sub.MarkProcessed(context.Background(), deliveries[130]); err != nil {
+		t.Fatalf("mark 130: %v", err)
+	}
+	time.Sleep(30 * time.Millisecond)
+	fake.mu.Lock()
+	if len(fake.ackCalls) != 1 {
+		t.Fatalf("ACK skipped 111: %+v", fake.ackCalls)
+	}
+	fake.mu.Unlock()
+	if err := sub.MarkProcessed(context.Background(), deliveries[111]); err != nil {
+		t.Fatalf("mark 111: %v", err)
+	}
+	waitACKs(t, fake, 104, 130)
+}
+
+func TestACKTransientFailureIsRetriedWithoutLosingBoundary(t *testing.T) {
+	fake := newFake()
+	fake.ackErrors[104] = []error{status.Error(codes.Unavailable, "try again")}
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		_ = stream.Send(msg(100))
+		_ = stream.Send(msg(104))
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	c := newClientForTest(t, fake)
+	sub, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	d100, d104 := recvDelivery(t, sub), recvDelivery(t, sub)
+	if err := sub.MarkProcessed(context.Background(), d100); err != nil {
+		t.Fatalf("mark 100: %v", err)
+	}
+	if err := sub.MarkProcessed(context.Background(), d104); err != nil {
+		t.Fatalf("mark 104: %v", err)
+	}
+	waitACKs(t, fake, 104, 104)
+}
+
+func TestReconnectReplayAndStaleGenerationFencing(t *testing.T) {
+	fake := newFake()
+	var calls int
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		calls++
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		if calls == 1 {
+			_ = stream.Send(msg(100))
+			_ = stream.Send(msg(104))
+			return status.Error(codes.Unavailable, "drop")
+		}
+		dm := msg(104)
+		dm.Replay = true
+		dm.SubscriptionId = "sub-2"
+		_ = stream.Send(dm)
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	c := newClientForTest(t, fake)
+	sub, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	d100, old104 := recvDelivery(t, sub), recvDelivery(t, sub)
+	if err := sub.MarkProcessed(context.Background(), d100); err != nil {
+		t.Fatalf("mark 100: %v", err)
+	}
+	waitACKs(t, fake, 100)
+	new104 := recvDelivery(t, sub)
+	if !new104.Replay {
+		t.Fatal("reconnected delivery must expose replay=true")
+	}
+	if err := sub.MarkProcessed(context.Background(), old104); !errors.Is(err, ErrStaleDelivery) {
+		t.Fatalf("old generation MarkProcessed = %v", err)
+	}
+	if err := sub.MarkProcessed(context.Background(), new104); err != nil {
+		t.Fatal(err)
+	}
+	waitACKs(t, fake, 100, 104)
+}
+
+func TestStreamReplacementAbortedIsTerminalNoReconnect(t *testing.T) {
+	fake := newFake()
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		return status.Error(codes.Aborted, "subscription replaced")
+	}
+	c := newClientForTest(t, fake)
+	sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitDone(t, sub)
+	if !errors.Is(sub.Err(), ErrSubscriptionReplaced) {
+		t.Fatalf("Err = %v", sub.Err())
+	}
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	if fake.subscribeCount != 1 {
+		t.Fatalf("replacement triggered reconnect storm: %d subscribes", fake.subscribeCount)
+	}
+}
+
+func TestAuthTerminalErrorsDoNotEndlesslyReconnect(t *testing.T) {
+	for _, code := range []codes.Code{codes.Unauthenticated, codes.PermissionDenied} {
+		t.Run(code.String(), func(t *testing.T) {
+			fake := newFake()
+			fake.subscribeFunc = func(_ *pb.SubscribeRequest, _ pb.ChatService_SubscribeServer) error {
+				return status.Error(code, "terminal")
+			}
+			c := newClientForTest(t, fake)
+			if _, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"}); status.Code(err) != code {
+				t.Fatalf("Subscribe error = %v", err)
+			}
+			fake.mu.Lock()
+			defer fake.mu.Unlock()
+			if fake.subscribeCount != 1 {
+				t.Fatalf("terminal auth reconnects: %d", fake.subscribeCount)
+			}
+		})
+	}
+}
+
+func TestTransientReconnect(t *testing.T) {
+	fake := newFake()
+	var calls int
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		calls++
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		if calls == 1 {
+			return status.Error(codes.Unavailable, "transient")
+		}
+		_ = stream.Send(msg(200))
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	c := newClientForTest(t, fake)
+	sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	d := recvDelivery(t, sub)
+	if d.Message.GetSequence() != 200 {
+		t.Fatalf("delivery seq = %d", d.Message.GetSequence())
+	}
+}
+
+func TestBoundedBufferShutdownDoesNotDropOrPanic(t *testing.T) {
+	fake := newFake()
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		for _, seq := range []int64{1, 2, 3} {
+			_ = stream.Send(msg(seq))
+		}
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	c := newClientForTest(t, fake, WithSubscriptionBufferSize(1))
+	sub, err := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := sub.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestMultipleTopicSubscriptionsAreIndependent(t *testing.T) {
+	fake := newFake()
+	fake.subscribeFunc = func(req *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		dm := msg(1)
+		dm.Message.TopicId = req.GetTopicId()
+		dm.SubscriptionId = "sub-" + req.GetTopicId()
+		_ = stream.Send(dm)
+		if req.GetTopicId() == "general" {
+			return status.Error(codes.Unavailable, "one topic failed")
+		}
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	c := newClientForTest(t, fake)
+	general, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "general"})
+	security, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "security"})
+	if recvDelivery(t, general).Message.GetTopicId() != "general" {
+		t.Fatal("general delivery mismatch")
+	}
+	ds := recvDelivery(t, security)
+	if ds.Message.GetTopicId() != "security" {
+		t.Fatal("security delivery mismatch")
+	}
+	if err := security.MarkProcessed(context.Background(), ds); err != nil {
+		t.Fatal(err)
+	}
+	waitACKs(t, fake, 1)
+}
+
+func TestCloseRacesWithReconnectRecvAckAndMarkProcessed(t *testing.T) {
+	fake := newFake()
+	ackEntered := make(chan struct{})
+	releaseACK := make(chan struct{})
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		_ = stream.Send(msg(1))
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	fake.ackErrors[1] = []error{status.Error(codes.Unavailable, "retry")}
+	c := newClientForTest(t, fake, withSleeperForTest(func(ctx context.Context, d time.Duration) error {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			return nil
+		}
+	}))
+	origAck := fake.Acknowledge
+	_ = origAck
+	sub, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	d := recvDelivery(t, sub)
+	go func() { _ = sub.MarkProcessed(context.Background(), d) }()
+	close(ackEntered)
+	<-ackEntered
+	close(releaseACK)
+	go func() { _ = sub.MarkProcessed(context.Background(), d) }()
+	if err := sub.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestHistoryPagerUsesReturnedSequenceAcrossGaps(t *testing.T) {
+	fake := newFake()
+	fake.historyPages = []*pb.HistoryResponse{
+		{Messages: []*pb.Message{{Sequence: 100}, {Sequence: 104}, {Sequence: 111}}, HasMore: true},
+		{Messages: []*pb.Message{{Sequence: 118}}, HasMore: false},
+	}
+	c := newClientForTest(t, fake)
+	pager, err := c.NewHistoryPager(HistoryRequest{TopicID: "topic", AfterSequence: 99, Limit: 3})
+	if err != nil {
+		t.Fatal(err)
+	}
+	first, err := pager.Next(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first.Messages[len(first.Messages)-1].GetSequence() != 111 {
+		t.Fatal("first page setup broken")
+	}
+	second, err := pager.Next(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(second.Messages) != 1 || second.Messages[0].GetSequence() != 118 {
+		t.Fatalf("second page = %+v", second.Messages)
+	}
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	if len(fake.historyReqs) != 2 || fake.historyReqs[1].GetAfterSequence() != 111 {
+		t.Fatalf("history reqs = %+v", fake.historyReqs)
+	}
+}