From ccaa1d795dbc4b1409285dba617a7a055d0da9cd Mon Sep 17 00:00:00 2001 From: mintaka Date: Wed, 26 Aug 2026 00:33:17 -0400 Subject: [PATCH] feat(linearagent): responder routing + async dispatcher (RIG-2717 T4/T6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T4 (routing.go): ResolveResponder maps a verified SessionEvent to the stable Manager + home channel via the recorded DL-055 ownership index (AuthoredArtifactByCoordinate), walking the recorded authoring agent to its owning Manager; unstamped issue or bare @mention falls back to the supervisor + dedicated routing channel. Never parses routing from forge text (DL-050/DL-094). Two narrow seams: OwnershipIndex (*store.Store satisfies) + ManagerResolver (agent-tree walk, wired at assembly). T6 (dispatcher.go): a single-goroutine drain of a bounded event channel. created -> resolve -> ensure @linear membership -> get-or-create topic -> upsert association -> ack thought + external-URL deep link (the 10s SLA leg, before the post) -> promptContext post with client_request_id="linear-delivery:". prompted -> follow-up post into the recorded topic (miss synthesizes via the resolver). No relay. A per-event failure (incl. a recovered panic) emits an error activity and the loop never crashes; a full queue returns ErrQueueFull -> 500 so Linear retries. Injected seams only (CommsPoster, Memberships, Topics, Associations, Client, DeepLinkFor) so the package never imports go/server. Assembly note: Topics.GetOrCreateTopic has no landed concrete store backer yet (only the unexported resolveTopicForAppend) — T7 wires it. Ref: RIG-2717 Co-authored-by: Matt Wilkinson --- go/internal/linearagent/dispatcher.go | 304 ++++++++++++++ go/internal/linearagent/dispatcher_test.go | 457 +++++++++++++++++++++ go/internal/linearagent/routing.go | 143 +++++++ go/internal/linearagent/routing_test.go | 246 +++++++++++ 4 files changed, 1150 insertions(+) create mode 100644 go/internal/linearagent/dispatcher.go create mode 100644 go/internal/linearagent/dispatcher_test.go create mode 100644 go/internal/linearagent/routing.go create mode 100644 go/internal/linearagent/routing_test.go diff --git a/go/internal/linearagent/dispatcher.go b/go/internal/linearagent/dispatcher.go new file mode 100644 index 00000000..6764e3a1 --- /dev/null +++ b/go/internal/linearagent/dispatcher.go @@ -0,0 +1,304 @@ +package linearagent + +import ( + "context" + "errors" + "log/slog" + + "github.com/google/uuid" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/store" +) + +// ackThoughtBody is the receipt the dispatcher emits on a `created` event — the +// 10-second liveness SLA leg (linear.app/developers/agent-interaction §Session +// webhooks): Linear marks a session unresponsive unless the agent emits within +// 10s. It tells the human what happened and pairs with the session external URL +// (the "Open in Compass" deep link) as the whole Option B return path (§Part 3). +const ackThoughtBody = "Compass received the session; opening in Compass\u2026" + +// externalURLLabel is the label on the session external URL entry — the +// "Open in Compass" deep link to the resolved Manager's home channel (§Part 3). +const externalURLLabel = "Open in Compass" + +// clientRequestIDPrefix namespaces the comms-rail idempotency key the dispatcher +// stamps on every PostAsAccount so a redelivered webhook never double-posts +// (§Part 1 message-level dedup). The full key is "linear-delivery:". +const clientRequestIDPrefix = "linear-delivery:" + +// ErrQueueFull is returned by Enqueue when the bounded channel is full. The HTTP +// handler maps it to a 500 so Linear retries the delivery rather than the event +// being silently dropped (§T6: full -> 500 so Linear retries). +var ErrQueueFull = errors.New("linearagent: dispatch queue full") + +// ResolveFunc is the routing seam (T4's ResolveResponder): it maps a verified +// session event to the Manager account that owns the delegated work and that +// Manager's home channel. Injected as a func so T6 never imports T4's concrete +// routing type — the driver wires the real ResolveResponder at assembly. The +// signature matches the shared contract byte-for-byte. +type ResolveFunc func(ctx context.Context, ev *SessionEvent) (managerAccountID store.AccountID, homeChannelID string, err error) + +// CommsPoster posts a message as an account into the resolved topic. *comms.Comms +// satisfies it via PostAsAccount. +type CommsPoster interface { + PostAsAccount(ctx context.Context, account store.AccountID, req *compassv1.PostMessageRequest) (*compassv1.PostMessageResponse, error) +} + +// Memberships ensures the @linear bridge account is a member of a channel — the +// postSetupThread precondition. *store.Store satisfies it via EnsureChannelMember +// (which takes a store.ChannelID; the dispatcher converts the resolver's string +// home-channel id at the call site). +type Memberships interface { + EnsureChannelMember(ctx context.Context, channelID store.ChannelID, account store.AccountID) error +} + +// Topics get-or-creates the comms topic the Linear conversation lands in, +// returning its id. Named for the issue identifier (else the session id). +type Topics interface { + GetOrCreateTopic(ctx context.Context, channelID, name string, author store.AccountID) (topicID string, err error) +} + +// Associations is the T3 store seam: the durable link between a Linear session +// and the Compass conversation it routed to. *store.Store satisfies it. +type Associations interface { + UpsertLinearAgentSession(ctx context.Context, row store.LinearAgentSessionRow) (created bool, err error) + LinearAgentSession(ctx context.Context, linearSessionID string) (store.LinearAgentSessionRow, error) +} + +// DispatcherParams carries every dependency the Dispatcher needs, all narrow +// seams (never concrete server types) so the drain loop depends on behavior, not +// packages. The driver wires the concrete implementations at assembly. +type DispatcherParams struct { + // Buffer is the bounded channel capacity. A full channel makes Enqueue + // return ErrQueueFull (-> HTTP 500 -> Linear retries). + Buffer int + // Resolve is T4's ResolveResponder. + Resolve ResolveFunc + // Poster is the comms post seam (*comms.Comms). + Poster CommsPoster + // Members is the channel-membership seam (*store.Store). + Members Memberships + // Topics is the topic get-or-create seam. + Topics Topics + // Associations is the T3 store association seam (*store.Store). + Associations Associations + // Client is the T2 Linear API client (CreateActivity + UpdateSession). + Client Client + // DeepLinkFor is T5's deep-link builder, taken as a func seam because T5's + // builder lives in go/server (no import from this package). + DeepLinkFor func(channelID string) string + // Bridge is the seeded @linear bridge system account id (T3a). + Bridge store.AccountID + // NewRequestID mints the uuid half of a client_request_id. Injectable for + // deterministic tests; defaults to uuid.NewString. + NewRequestID func() string +} + +// Dispatcher drains a bounded channel of verified session events on a single +// goroutine, routing each to a Manager, emitting the Linear-side return path, +// and posting into Compass. It never crashes on a per-event failure — a bad +// event is logged (and an `error` activity emitted to Linear) and the loop +// moves on. There is NO relay: the dispatcher does not observe or mirror agent +// output; the return path is the two `created` emits only (§Part 3). +type Dispatcher struct { + ch chan *SessionEvent + resolve ResolveFunc + poster CommsPoster + members Memberships + topics Topics + assoc Associations + client Client + deepLinkFor func(channelID string) string + bridge store.AccountID + newRequestID func() string +} + +// NewDispatcher builds a Dispatcher from params. Buffer defaults to 1 when +// non-positive; NewRequestID defaults to uuid.NewString. +func NewDispatcher(p DispatcherParams) *Dispatcher { + buf := p.Buffer + if buf <= 0 { + buf = 1 + } + newRequestID := p.NewRequestID + if newRequestID == nil { + newRequestID = uuid.NewString + } + return &Dispatcher{ + ch: make(chan *SessionEvent, buf), + resolve: p.Resolve, + poster: p.Poster, + members: p.Members, + topics: p.Topics, + assoc: p.Associations, + client: p.Client, + deepLinkFor: p.DeepLinkFor, + bridge: p.Bridge, + newRequestID: newRequestID, + } +} + +// Enqueue offers a verified event to the bounded channel without blocking. A +// full channel returns ErrQueueFull so the HTTP handler returns 500 and Linear +// retries — an event is never silently dropped. +func (d *Dispatcher) Enqueue(ev *SessionEvent) error { + select { + case d.ch <- ev: + return nil + default: + return ErrQueueFull + } +} + +// Run drains the channel until ctx is cancelled, handling one event at a time. +// It returns ctx.Err() on cancel. A per-event failure never stops the loop. +func (d *Dispatcher) Run(ctx context.Context) error { + for { + select { + case <-ctx.Done(): + return ctx.Err() + case ev := <-d.ch: + d.handle(ctx, ev) + } + } +} + +// handle processes one event, converting any failure (including a panic in a +// seam) into a logged `error` activity to Linear so a single bad event can never +// crash the drain loop. +func (d *Dispatcher) handle(ctx context.Context, ev *SessionEvent) { + defer func() { + if r := recover(); r != nil { + slog.ErrorContext(ctx, "linearagent dispatcher: recovered from panic", + "linear_session_id", ev.AgentSession.ID, "action", ev.Action, "panic", r) + d.emitError(ctx, ev.AgentSession.ID, errors.New("internal error handling event")) + } + }() + if err := d.process(ctx, ev); err != nil { + slog.ErrorContext(ctx, "linearagent dispatcher: event failed", + "linear_session_id", ev.AgentSession.ID, "action", ev.Action, "error", err) + d.emitError(ctx, ev.AgentSession.ID, err) + } +} + +// process routes one event by action. Unknown actions are ignored (no error) — +// only created/prompted drive the responder. +func (d *Dispatcher) process(ctx context.Context, ev *SessionEvent) error { + switch ev.Action { + case "created": + return d.handleCreated(ctx, ev) + case "prompted": + return d.handlePrompted(ctx, ev) + default: + return nil + } +} + +// handleCreated runs the `created` chain: resolve -> ensure @linear membership +// -> get-or-create the topic -> upsert the association -> emit the ack thought +// AND the session external URL (the 10s SLA leg, BEFORE the post) -> post the +// prompt context into the topic with the dedup client_request_id. +func (d *Dispatcher) handleCreated(ctx context.Context, ev *SessionEvent) error { + manager, homeChannel, err := d.resolve(ctx, ev) + if err != nil { + return err + } + if err := d.members.EnsureChannelMember(ctx, store.ChannelID(homeChannel), d.bridge); err != nil { + return err + } + topicID, err := d.topics.GetOrCreateTopic(ctx, homeChannel, topicName(ev), d.bridge) + if err != nil { + return err + } + if _, err := d.assoc.UpsertLinearAgentSession(ctx, store.LinearAgentSessionRow{ + LinearSessionID: ev.AgentSession.ID, + ManagerAccountID: manager, + ChannelID: store.ChannelID(homeChannel), + TopicID: topicID, + LinearIssueID: ev.AgentSession.Issue.ID, + }); err != nil { + return err + } + // The 10s SLA leg: ack thought + session external URL, BEFORE the post. + if err := d.client.CreateActivity(ctx, ev.AgentSession.ID, ActivityContent{Type: "thought", Body: ackThoughtBody}); err != nil { + return err + } + if err := d.client.UpdateSession(ctx, ev.AgentSession.ID, []ExternalURL{{ + Label: externalURLLabel, + URL: d.deepLinkFor(homeChannel), + }}); err != nil { + return err + } + return d.post(ctx, homeChannel, topicID, ev.PromptContext) +} + +// handlePrompted routes a follow-up to the recorded conversation: look up the +// association and post into its channel/topic. On a miss (a prompted event with +// no `created` on record) it synthesizes the association via the resolver from +// the payload's agentSession, then posts. +func (d *Dispatcher) handlePrompted(ctx context.Context, ev *SessionEvent) error { + row, err := d.assoc.LinearAgentSession(ctx, ev.AgentSession.ID) + switch { + case err == nil: + return d.post(ctx, string(row.ChannelID), row.TopicID, ev.AgentActivity.Body) + case errors.Is(err, store.ErrNotFound): + manager, homeChannel, resErr := d.resolve(ctx, ev) + if resErr != nil { + return resErr + } + if memErr := d.members.EnsureChannelMember(ctx, store.ChannelID(homeChannel), d.bridge); memErr != nil { + return memErr + } + topicID, topErr := d.topics.GetOrCreateTopic(ctx, homeChannel, topicName(ev), d.bridge) + if topErr != nil { + return topErr + } + if _, upErr := d.assoc.UpsertLinearAgentSession(ctx, store.LinearAgentSessionRow{ + LinearSessionID: ev.AgentSession.ID, + ManagerAccountID: manager, + ChannelID: store.ChannelID(homeChannel), + TopicID: topicID, + LinearIssueID: ev.AgentSession.Issue.ID, + }); upErr != nil { + return upErr + } + return d.post(ctx, homeChannel, topicID, ev.AgentActivity.Body) + default: + return err + } +} + +// post writes one message as the @linear bridge account into channel/topic with +// a fresh dedup client_request_id. +func (d *Dispatcher) post(ctx context.Context, channelID, topicID, body string) error { + _, err := d.poster.PostAsAccount(ctx, d.bridge, &compassv1.PostMessageRequest{ + Container: &compassv1.PostMessageRequest_ChannelId{ChannelId: channelID}, + Topic: &compassv1.PostMessageRequest_TopicId{TopicId: topicID}, + Blocks: []*compassv1.MessageBlock{{Block: &compassv1.MessageBlock_Text{Text: body}}}, + ClientRequestId: d.clientRequestID(), + }) + return err +} + +// emitError best-effort posts an `error` activity to Linear for a failed event. +// It never propagates: the drain loop keeps going regardless. +func (d *Dispatcher) emitError(ctx context.Context, sessionID string, cause error) { + if err := d.client.CreateActivity(ctx, sessionID, ActivityContent{Type: "error", Body: cause.Error()}); err != nil { + slog.ErrorContext(ctx, "linearagent dispatcher: emitting error activity failed", + "linear_session_id", sessionID, "error", err) + } +} + +// clientRequestID mints the comms-rail idempotency key: "linear-delivery:". +func (d *Dispatcher) clientRequestID() string { + return clientRequestIDPrefix + d.newRequestID() +} + +// topicName is the issue identifier when present, else the session id. +func topicName(ev *SessionEvent) string { + if ev.AgentSession.Issue.Identifier != "" { + return ev.AgentSession.Issue.Identifier + } + return ev.AgentSession.ID +} diff --git a/go/internal/linearagent/dispatcher_test.go b/go/internal/linearagent/dispatcher_test.go new file mode 100644 index 00000000..5e4ae7a8 --- /dev/null +++ b/go/internal/linearagent/dispatcher_test.go @@ -0,0 +1,457 @@ +package linearagent + +import ( + "context" + "errors" + "fmt" + "slices" + "strings" + "sync" + "testing" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/store" +) + +// The fakes below are the narrow seams the Dispatcher depends on. Each records +// its calls and (where a test gates on completion) signals on a channel so the +// test blocks on the real event, never a clock. + +// fakeResolver records calls and returns a fixed manager/home channel or a +// preset error. +type fakeResolver struct { + manager store.AccountID + homeChannel string + err error + calls int +} + +func (f *fakeResolver) resolve(_ context.Context, _ *SessionEvent) (store.AccountID, string, error) { + f.calls++ + if f.err != nil { + return "", "", f.err + } + return f.manager, f.homeChannel, nil +} + +// recordingComms records each PostAsAccount and signals on posted. reqIDs +// captures the client_request_id of every post. +type recordingComms struct { + mu sync.Mutex + reqIDs []string + topics []string + bodies []string + err error + posted chan struct{} +} + +func (c *recordingComms) PostAsAccount(_ context.Context, _ store.AccountID, req *compassv1.PostMessageRequest) (*compassv1.PostMessageResponse, error) { + c.mu.Lock() + c.reqIDs = append(c.reqIDs, req.GetClientRequestId()) + c.topics = append(c.topics, req.GetTopicId()) + if len(req.GetBlocks()) > 0 { + c.bodies = append(c.bodies, req.GetBlocks()[0].GetText()) + } + c.mu.Unlock() + if c.posted != nil { + c.posted <- struct{}{} + } + return &compassv1.PostMessageResponse{}, c.err +} + +// fakeMembers records EnsureChannelMember calls. +type fakeMembers struct { + mu sync.Mutex + channels []string + accounts []store.AccountID + err error +} + +func (m *fakeMembers) EnsureChannelMember(_ context.Context, channelID store.ChannelID, account store.AccountID) error { + m.mu.Lock() + m.channels = append(m.channels, string(channelID)) + m.accounts = append(m.accounts, account) + m.mu.Unlock() + return m.err +} + +// fakeTopics returns a fixed topic id. +type fakeTopics struct { + topicID string + names []string + err error +} + +func (tk *fakeTopics) GetOrCreateTopic(_ context.Context, _, name string, _ store.AccountID) (string, error) { + tk.names = append(tk.names, name) + if tk.err != nil { + return "", tk.err + } + return tk.topicID, nil +} + +// fakeAssoc is the T3 association seam. rows holds every upsert; lookup returns +// lookupRow/lookupErr on LinearAgentSession. +type fakeAssoc struct { + mu sync.Mutex + rows []store.LinearAgentSessionRow + lookupRow store.LinearAgentSessionRow + lookupErr error +} + +func (a *fakeAssoc) UpsertLinearAgentSession(_ context.Context, row store.LinearAgentSessionRow) (bool, error) { + a.mu.Lock() + a.rows = append(a.rows, row) + a.mu.Unlock() + return true, nil +} + +func (a *fakeAssoc) LinearAgentSession(_ context.Context, _ string) (store.LinearAgentSessionRow, error) { + if a.lookupErr != nil { + return store.LinearAgentSessionRow{}, a.lookupErr + } + return a.lookupRow, nil +} + +// recordingClient records the ordered sequence of Linear-side emits ("thought", +// "external-url", "error") and signals thoughts/errors on channels a test gates +// on. +type recordingClient struct { + mu sync.Mutex + events []string + bodies []string + errCh chan struct{} +} + +func (c *recordingClient) CreateActivity(_ context.Context, _ string, content ActivityContent) error { + c.mu.Lock() + c.events = append(c.events, content.Type) + c.bodies = append(c.bodies, content.Body) + c.mu.Unlock() + if content.Type == "error" && c.errCh != nil { + c.errCh <- struct{}{} + } + return nil +} + +func (c *recordingClient) UpdateSession(_ context.Context, _ string, _ []ExternalURL) error { + c.mu.Lock() + c.events = append(c.events, "external-url") + c.mu.Unlock() + return nil +} + +func (c *recordingClient) seq() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.events...) +} + +const testBridge store.AccountID = "acct-linear" + +// runDispatcher starts d.Run on a fresh goroutine and returns a stop func that +// cancels it and waits for exit — the deterministic lifecycle every test uses. +func runDispatcher(t *testing.T, d *Dispatcher) func() { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + _ = d.Run(ctx) // returns ctx.Err() on cancel; not an assertion target + close(done) + }() + return func() { + cancel() + <-done + } +} + +// TestDispatcherCreatedHappyPath pins the created chain's ORDER: membership +// first, then the two Linear-side emits (thought, external-url) BEFORE the post, +// and the dedup client_request_id format. +func TestDispatcherCreatedHappyPath(t *testing.T) { + res := &fakeResolver{manager: "mgr-1", homeChannel: "chan-1"} + comms := &recordingComms{posted: make(chan struct{}, 1)} + members := &fakeMembers{} + topics := &fakeTopics{topicID: "topic-1"} + assoc := &fakeAssoc{} + client := &recordingClient{} + + d := NewDispatcher(DispatcherParams{ + Buffer: 4, + Resolve: res.resolve, + Poster: comms, + Members: members, + Topics: topics, + Associations: assoc, + Client: client, + DeepLinkFor: func(ch string) string { return "https://compass.rigel.build/c/" + ch }, + Bridge: testBridge, + NewRequestID: func() string { return "fixed-uuid" }, + }) + stop := runDispatcher(t, d) + defer stop() + + if err := d.Enqueue(&SessionEvent{ + Action: "created", + PromptContext: "please do the thing", + AgentSession: AgentSession{ID: "sess-1", Issue: Issue{ID: "iss-1", Identifier: "RIG-9"}}, + }); err != nil { + t.Fatalf("Enqueue: %v", err) + } + + <-comms.posted // gate on the post completing + + // Membership ensured for @linear into the manager's home channel. + if len(members.channels) != 1 || members.channels[0] != "chan-1" || members.accounts[0] != testBridge { + t.Fatalf("EnsureChannelMember = %v/%v, want chan-1/%s", members.channels, members.accounts, testBridge) + } + // Emit order: thought, then external-url, then the post (post is gated, so + // both emits must already be recorded). + if got := client.seq(); len(got) != 2 || got[0] != "thought" || got[1] != "external-url" { + t.Fatalf("emit sequence = %v, want [thought external-url] before the post", got) + } + // Association upserted with the resolved manager/channel/topic. + if len(assoc.rows) != 1 { + t.Fatalf("association rows = %d, want 1", len(assoc.rows)) + } + row := assoc.rows[0] + if row.ManagerAccountID != "mgr-1" || row.ChannelID != "chan-1" || row.TopicID != "topic-1" || row.LinearIssueID != "iss-1" { + t.Fatalf("association row = %+v, want mgr-1/chan-1/topic-1/iss-1", row) + } + // Topic named for the issue identifier. + if len(topics.names) != 1 || topics.names[0] != "RIG-9" { + t.Fatalf("topic names = %v, want [RIG-9]", topics.names) + } + // Post carried the prompt context into the topic with the dedup key. + if comms.topics[0] != "topic-1" || comms.bodies[0] != "please do the thing" { + t.Fatalf("post topic/body = %q/%q, want topic-1/please do the thing", comms.topics[0], comms.bodies[0]) + } + assertReqID(t, comms.reqIDs[0], "fixed-uuid") +} + +// TestDispatcherCreatedTopicNameFallsBackToSession pins the topic-name fallback: +// no issue identifier → the session id. +func TestDispatcherCreatedTopicNameFallsBackToSession(t *testing.T) { + comms := &recordingComms{posted: make(chan struct{}, 1)} + topics := &fakeTopics{topicID: "topic-x"} + d := newTestDispatcher(t, dispatcherDeps{ + res: &fakeResolver{manager: "mgr", homeChannel: "chan"}, + comms: comms, + topics: topics, + assoc: &fakeAssoc{}, + client: &recordingClient{}, + }) + stop := runDispatcher(t, d) + defer stop() + + if err := d.Enqueue(&SessionEvent{Action: "created", AgentSession: AgentSession{ID: "sess-noissue"}}); err != nil { + t.Fatalf("Enqueue: %v", err) + } + <-comms.posted + if len(topics.names) != 1 || topics.names[0] != "sess-noissue" { + t.Fatalf("topic names = %v, want [sess-noissue] (session-id fallback)", topics.names) + } +} + +// TestDispatcherPromptedFollowUp pins the prompted path: a hit posts the +// agentActivity body into the RECORDED channel/topic, with a fresh dedup key, +// and never re-runs the created-side emits. +func TestDispatcherPromptedFollowUp(t *testing.T) { + comms := &recordingComms{posted: make(chan struct{}, 1)} + client := &recordingClient{} + assoc := &fakeAssoc{lookupRow: store.LinearAgentSessionRow{ + LinearSessionID: "sess-1", ManagerAccountID: "mgr-1", ChannelID: "chan-1", TopicID: "topic-1", + }} + d := newTestDispatcher(t, dispatcherDeps{ + res: &fakeResolver{}, + comms: comms, + topics: &fakeTopics{topicID: "should-not-be-used"}, + assoc: assoc, + client: client, + reqID: func() string { return "uuid-2" }, + }) + stop := runDispatcher(t, d) + defer stop() + + if err := d.Enqueue(&SessionEvent{ + Action: "prompted", + AgentSession: AgentSession{ID: "sess-1"}, + AgentActivity: AgentActivity{Body: "the follow-up prompt"}, + }); err != nil { + t.Fatalf("Enqueue: %v", err) + } + <-comms.posted + + if comms.topics[0] != "topic-1" || comms.bodies[0] != "the follow-up prompt" { + t.Fatalf("post topic/body = %q/%q, want topic-1/the follow-up prompt", comms.topics[0], comms.bodies[0]) + } + assertReqID(t, comms.reqIDs[0], "uuid-2") + // No created-side emits on a follow-up. + if got := client.seq(); len(got) != 0 { + t.Fatalf("prompted emitted Linear activities %v, want none", got) + } +} + +// TestDispatcherPromptedMissSynthesizes pins the prompted-miss path: a lookup +// miss (ErrNotFound) synthesizes the association via the resolver, then posts. +func TestDispatcherPromptedMissSynthesizes(t *testing.T) { + comms := &recordingComms{posted: make(chan struct{}, 1)} + res := &fakeResolver{manager: "mgr-9", homeChannel: "chan-9"} + assoc := &fakeAssoc{lookupErr: fmt.Errorf("%w: no such session", store.ErrNotFound)} + d := newTestDispatcher(t, dispatcherDeps{ + res: res, + comms: comms, + topics: &fakeTopics{topicID: "topic-9"}, + assoc: assoc, + client: &recordingClient{}, + }) + stop := runDispatcher(t, d) + defer stop() + + if err := d.Enqueue(&SessionEvent{ + Action: "prompted", + AgentSession: AgentSession{ID: "sess-orphan"}, + AgentActivity: AgentActivity{Body: "orphaned follow-up"}, + }); err != nil { + t.Fatalf("Enqueue: %v", err) + } + <-comms.posted + + if res.calls != 1 { + t.Fatalf("resolver calls = %d, want 1 (synthesis on miss)", res.calls) + } + if len(assoc.rows) != 1 || assoc.rows[0].ChannelID != "chan-9" || assoc.rows[0].TopicID != "topic-9" { + t.Fatalf("synthesized association = %+v, want chan-9/topic-9", assoc.rows) + } + if comms.topics[0] != "topic-9" || comms.bodies[0] != "orphaned follow-up" { + t.Fatalf("post topic/body = %q/%q, want topic-9/orphaned follow-up", comms.topics[0], comms.bodies[0]) + } +} + +// TestDispatcherEnqueueWhenFull pins the backpressure contract: a full bounded +// channel makes Enqueue return ErrQueueFull (→ HTTP 500 → Linear retries). +func TestDispatcherEnqueueWhenFull(t *testing.T) { + // Buffer 1, no Run goroutine draining: the first Enqueue fills the channel, + // the second must fail rather than block. + d := NewDispatcher(DispatcherParams{ + Buffer: 1, + Resolve: (&fakeResolver{}).resolve, + Poster: &recordingComms{}, + Members: &fakeMembers{}, + Topics: &fakeTopics{}, + Associations: &fakeAssoc{}, + Client: &recordingClient{}, + DeepLinkFor: func(string) string { return "" }, + Bridge: testBridge, + }) + if err := d.Enqueue(&SessionEvent{Action: "created"}); err != nil { + t.Fatalf("first Enqueue: %v", err) + } + if err := d.Enqueue(&SessionEvent{Action: "created"}); !errors.Is(err, ErrQueueFull) { + t.Fatalf("second Enqueue error = %v, want ErrQueueFull", err) + } +} + +// TestDispatcherPerEventFailureKeepsDraining pins the never-crash contract: a +// failing event emits an `error` activity AND the loop keeps draining the next, +// good event. A panic/return on the bad event would hang the good event's post +// and fail the test. +func TestDispatcherPerEventFailureKeepsDraining(t *testing.T) { + comms := &recordingComms{posted: make(chan struct{}, 1)} + client := &recordingClient{errCh: make(chan struct{}, 1)} + // First event resolves with an error; the resolver flips to success after. + res := &flakyResolver{errFor: "bad", manager: "mgr", homeChannel: "chan"} + d := newTestDispatcher(t, dispatcherDeps{ + res: res, + comms: comms, + topics: &fakeTopics{topicID: "topic"}, + assoc: &fakeAssoc{}, + client: client, + }) + stop := runDispatcher(t, d) + defer stop() + + if err := d.Enqueue(&SessionEvent{Action: "created", AgentSession: AgentSession{ID: "bad"}}); err != nil { + t.Fatalf("Enqueue(bad): %v", err) + } + <-client.errCh // the failing event emitted an `error` activity + + if err := d.Enqueue(&SessionEvent{Action: "created", AgentSession: AgentSession{ID: "good", Issue: Issue{Identifier: "RIG-1"}}}); err != nil { + t.Fatalf("Enqueue(good): %v", err) + } + <-comms.posted // the loop kept draining and handled the good event + + if got := lastErrorBody(client); got == "" { + t.Fatal("failing event did not emit an `error` activity") + } +} + +// flakyResolver errors for the event whose session id equals errFor, else +// returns manager/homeChannel. +type flakyResolver struct { + errFor string + manager store.AccountID + homeChannel string +} + +func (f *flakyResolver) resolve(_ context.Context, ev *SessionEvent) (store.AccountID, string, error) { + if ev.AgentSession.ID == f.errFor { + return "", "", errors.New("routing failed") + } + return f.manager, f.homeChannel, nil +} + +func lastErrorBody(c *recordingClient) string { + c.mu.Lock() + defer c.mu.Unlock() + for i := range slices.Backward(c.events) { + if c.events[i] == "error" { + return c.bodies[i] + } + } + return "" +} + +// assertReqID checks the dedup client_request_id scheme: "linear-delivery:". +func assertReqID(t *testing.T, got, wantUUID string) { + t.Helper() + want := clientRequestIDPrefix + wantUUID + if got != want { + t.Fatalf("client_request_id = %q, want %q", got, want) + } + if !strings.HasPrefix(got, "linear-delivery:") { + t.Fatalf("client_request_id = %q, want the linear-delivery: dedup prefix", got) + } +} + +// dispatcherDeps + newTestDispatcher cut the boilerplate for the common wiring. +type dispatcherDeps struct { + res interface { + resolve(ctx context.Context, ev *SessionEvent) (store.AccountID, string, error) + } + comms CommsPoster + topics Topics + assoc Associations + client Client + reqID func() string +} + +func newTestDispatcher(t *testing.T, deps dispatcherDeps) *Dispatcher { + t.Helper() + reqID := deps.reqID + if reqID == nil { + reqID = func() string { return "fixed-uuid" } + } + return NewDispatcher(DispatcherParams{ + Buffer: 4, + Resolve: deps.res.resolve, + Poster: deps.comms, + Members: &fakeMembers{}, + Topics: deps.topics, + Associations: deps.assoc, + Client: deps.client, + DeepLinkFor: func(ch string) string { return "https://compass.rigel.build/c/" + ch }, + Bridge: testBridge, + NewRequestID: reqID, + }) +} diff --git a/go/internal/linearagent/routing.go b/go/internal/linearagent/routing.go new file mode 100644 index 00000000..390c7321 --- /dev/null +++ b/go/internal/linearagent/routing.go @@ -0,0 +1,143 @@ +package linearagent + +// Responder routing resolution (RIG-2717 T4, design +// docs/designs/product/compass-linear-agent-responder/design.md §Part 2 / §T4). +// +// A Linear delegation or @mention names the Compass app, never a specific +// Manager, so the bridge must resolve which stable Manager runs the session. +// The trusted routing source is Compass's own recorded ownership truth — the +// DL-055 forge_authored_artifacts index — never a header parsed from forge text +// (DL-050 / DL-094): an owner claim parsed from a body is untrusted display +// metadata that must never reach a routing decision. +// +// - A delegated issue with a recorded ownership row resolves to the stable +// Manager for that recorded work. The row records the AUTHORING agent, which +// may be a transient peer/sub-agent (not itself a Manager), so the resolver +// walks from the recorded agent to its owning Manager and returns that +// Manager's home channel. +// - No recorded row (a human-filed issue delegated cold, any coordinate +// Compass has never authored) — or a bare @mention carrying no issue +// coordinate — routes to the supervisor / top-level Manager via the +// dedicated routing channel, where the lane is decided and stamped. + +import ( + "context" + "errors" + "strconv" + "strings" + + "github.com/RigelBuild/compass/go/internal/store" +) + +// OwnershipIndex is the narrow read T4 defines over the DL-055 ownership index +// (store.forge_authored_artifacts). It is backed by *store.Store's landed +// AuthoredArtifactByCoordinate — individual coordinate params, full-artifact +// return carrying the recorded .AgentAccountID — so the concrete store +// satisfies it directly and the resolver never embeds the store's query shape. +// +// The returned AuthoredArtifact's AgentAccountID is the recorded AUTHORING +// agent (may be a peer, not a Manager); store.ErrNotFound means Compass has +// never authored the coordinate. Recorded truth only — never a header parsed +// from forge text (DL-050 / DL-094). +type OwnershipIndex interface { + AuthoredArtifactByCoordinate(ctx context.Context, provider store.ForgeProvider, host, repo string, kind store.ForgeArtifactKind, number uint64) (store.AuthoredArtifact, error) +} + +// ManagerResolver walks a recorded authoring agent (possibly a peer) to its +// owning Manager and returns that Manager's account id and home channel id. +// It is its own narrow seam because no single store method spans the tree walk +// (up parent_agent_id to a role="manager" agent) AND the home-channel read; the +// driver backs it with the store's agent-tree + account reads at assembly. +// store.ErrNotFound when the agent (or a walk ancestor) does not resolve. +type ManagerResolver interface { + OwningManager(ctx context.Context, agent store.AccountID) (managerAccountID store.AccountID, homeChannelID string, err error) +} + +// Resolver resolves a delegated Linear session to a stable (Manager, home +// channel). It holds the ownership-index and manager-walk seams plus the +// config-resolved fallback target (the supervisor / top-level Manager account +// and the dedicated routing channel id). +type Resolver struct { + ownership OwnershipIndex + managers ManagerResolver + + // forgeHost is the forge coordinate host recorded for Linear-authored + // artifacts, config-resolved at construction. It completes the coordinate + // key the ownership index is queried on (provider is always Linear here). + forgeHost string + + // supervisorAccountID and routingChannelID are the fallback target: the + // supervisor / top-level Manager and the dedicated routing channel a + // no-recorded-row (or coordinate-less) event routes to. + supervisorAccountID store.AccountID + routingChannelID string +} + +// NewResolver constructs a Resolver over its two seams and the config-resolved +// fallback target. forgeHost is the coordinate host Linear-authored rows carry; +// supervisorAccountID and routingChannelID are the dedicated routing fallback. +func NewResolver(ownership OwnershipIndex, managers ManagerResolver, forgeHost string, supervisorAccountID store.AccountID, routingChannelID string) *Resolver { + return &Resolver{ + ownership: ownership, + managers: managers, + forgeHost: forgeHost, + supervisorAccountID: supervisorAccountID, + routingChannelID: routingChannelID, + } +} + +// ResolveResponder resolves the stable Manager and home channel that should run +// ev's session (design §Part 2). A recorded ownership row for the delegated +// issue's forge coordinate walks the recorded authoring agent to its owning +// Manager; a missing row (store.ErrNotFound) or an event with no issue +// coordinate falls back to the supervisor + dedicated routing channel. +func (r *Resolver) ResolveResponder(ctx context.Context, ev *SessionEvent) (managerAccountID store.AccountID, homeChannelID string, err error) { + provider, host, repo, number, ok := r.coordinate(ev) + if !ok { + // A bare @mention with no issue coordinate: route to the supervisor. + return r.supervisorAccountID, r.routingChannelID, nil + } + + art, err := r.ownership.AuthoredArtifactByCoordinate(ctx, provider, host, repo, store.ForgeArtifactKindIssue, number) + if errors.Is(err, store.ErrNotFound) { + // No recorded row: a cold delegation Compass has never authored. + return r.supervisorAccountID, r.routingChannelID, nil + } + if err != nil { + return "", "", err + } + + // Recorded row: walk the AUTHORING agent (possibly a peer) to its owning + // Manager and that Manager's home channel. + return r.managers.OwningManager(ctx, art.AgentAccountID) +} + +// coordinate extracts the delegated issue's forge coordinate from ev. A Linear +// issue identifier is "TEAM-NUMBER" (e.g. "RIG-2717"): the team key is the +// forge repo, the number is the artifact number, the provider is Linear, and +// the host is config-resolved. ok=false when ev carries no parseable issue +// identifier (a bare @mention), which routes to the supervisor fallback. +func (r *Resolver) coordinate(ev *SessionEvent) (provider store.ForgeProvider, host, repo string, number uint64, ok bool) { + team, num, ok := parseIssueIdentifier(ev.AgentSession.Issue.Identifier) + if !ok { + return 0, "", "", 0, false + } + return store.ForgeProviderLinear, r.forgeHost, team, num, true +} + +// parseIssueIdentifier splits a Linear issue identifier "TEAM-NUMBER" into its +// team key and issue number. It splits on the LAST '-' so a team key may itself +// contain a dash. ok=false for an empty identifier, a missing separator, an +// empty team key, or an unparseable / zero number. +func parseIssueIdentifier(identifier string) (team string, number uint64, ok bool) { + i := strings.LastIndex(identifier, "-") + if i <= 0 || i == len(identifier)-1 { + return "", 0, false + } + team = identifier[:i] + n, err := strconv.ParseUint(identifier[i+1:], 10, 64) + if err != nil || n == 0 { + return "", 0, false + } + return team, n, true +} diff --git a/go/internal/linearagent/routing_test.go b/go/internal/linearagent/routing_test.go new file mode 100644 index 00000000..67050a79 --- /dev/null +++ b/go/internal/linearagent/routing_test.go @@ -0,0 +1,246 @@ +package linearagent + +// Unit tests for ResolveResponder (RIG-2717 T4). Table-driven over fakes for +// the two seams (OwnershipIndex + ManagerResolver), asserting the four routing +// outcomes of design §Part 2: a recorded coordinate whose authoring agent IS a +// Manager, a recorded coordinate whose authoring agent is a PEER (the walk must +// resolve peer -> owning Manager, never peer -> peer), an unknown coordinate +// (store.ErrNotFound -> supervisor + routing channel), and a bare @mention with +// no issue coordinate (-> supervisor + routing channel). +// +// context.Background() here is the test root — the sanctioned exemption to the +// thread-ctx rule. + +import ( + "context" + "errors" + "testing" + + "github.com/RigelBuild/compass/go/internal/store" +) + +const ( + testForgeHost = "linear.app" + testSupervisor = store.AccountID("acct-supervisor") + testRoutingChan = "chan-routing" +) + +// fakeOwnershipIndex scripts AuthoredArtifactByCoordinate: a single recorded +// artifact keyed by the (repo, number) coordinate, else store.ErrNotFound. It +// records the coordinate it was queried on so a test can assert the extraction. +type fakeOwnershipIndex struct { + // row is returned when the queried number matches wantNumber (and wantRepo); + // any other coordinate is a miss. + wantRepo string + wantNumber uint64 + row store.AuthoredArtifact + + gotProvider store.ForgeProvider + gotHost string + gotRepo string + gotKind store.ForgeArtifactKind + gotNumber uint64 + calls int +} + +func (f *fakeOwnershipIndex) AuthoredArtifactByCoordinate(_ context.Context, provider store.ForgeProvider, host, repo string, kind store.ForgeArtifactKind, number uint64) (store.AuthoredArtifact, error) { + f.calls++ + f.gotProvider, f.gotHost, f.gotRepo, f.gotKind, f.gotNumber = provider, host, repo, kind, number + if repo == f.wantRepo && number == f.wantNumber { + return f.row, nil + } + return store.AuthoredArtifact{}, store.ErrNotFound +} + +// fakeManagerResolver models the agent-tree walk: a map from a recorded +// authoring agent to its owning Manager + that Manager's home channel. A miss +// is store.ErrNotFound. It records the agent it was walked from so a test can +// assert the walk was invoked on the recorded authoring agent (not skipped). +type fakeManagerResolver struct { + owners map[store.AccountID]managerHome + gotFrom store.AccountID + calls int +} + +type managerHome struct { + manager store.AccountID + homeChannel string +} + +func (f *fakeManagerResolver) OwningManager(_ context.Context, agent store.AccountID) (store.AccountID, string, error) { + f.calls++ + f.gotFrom = agent + mh, ok := f.owners[agent] + if !ok { + return "", "", store.ErrNotFound + } + return mh.manager, mh.homeChannel, nil +} + +func sessionEvent(identifier string) *SessionEvent { + ev := &SessionEvent{Type: "AgentSessionEvent", Action: "created"} + ev.AgentSession.Issue.Identifier = identifier + return ev +} + +func TestResolveResponder(t *testing.T) { + ctx := context.Background() + + const ( + managerAgent = store.AccountID("acct-manager") + peerAgent = store.AccountID("acct-peer") + mgrHome = "chan-manager-home" + ) + + // A recorded row whose authoring agent is a Manager: the walk resolves the + // Manager to itself (an agent whose owning Manager is itself). + rowByManager := store.AuthoredArtifact{ + Provider: store.ForgeProviderLinear, Host: testForgeHost, Repo: "RIG", + Kind: store.ForgeArtifactKindIssue, Number: 2717, AgentAccountID: managerAgent, + } + // A recorded row whose authoring agent is a transient peer: the walk must + // climb to the peer's OWNING Manager, not return the peer. + rowByPeer := store.AuthoredArtifact{ + Provider: store.ForgeProviderLinear, Host: testForgeHost, Repo: "RIG", + Kind: store.ForgeArtifactKindIssue, Number: 2717, AgentAccountID: peerAgent, + } + + tests := []struct { + name string + identifier string + ownership *fakeOwnershipIndex + managers *fakeManagerResolver + wantManager store.AccountID + wantChannel string + wantWalk store.AccountID // "" => the walk must NOT be invoked + }{ + { + name: "recorded coordinate, authoring agent is a Manager", + identifier: "RIG-2717", + ownership: &fakeOwnershipIndex{wantRepo: "RIG", wantNumber: 2717, row: rowByManager}, + managers: &fakeManagerResolver{owners: map[store.AccountID]managerHome{ + managerAgent: {manager: managerAgent, homeChannel: mgrHome}, + }}, + wantManager: managerAgent, + wantChannel: mgrHome, + wantWalk: managerAgent, + }, + { + name: "recorded coordinate, authoring agent is a PEER (the walk)", + identifier: "RIG-2717", + ownership: &fakeOwnershipIndex{wantRepo: "RIG", wantNumber: 2717, row: rowByPeer}, + managers: &fakeManagerResolver{owners: map[store.AccountID]managerHome{ + peerAgent: {manager: managerAgent, homeChannel: mgrHome}, + }}, + wantManager: managerAgent, // peer -> owning Manager, NOT the peer + wantChannel: mgrHome, + wantWalk: peerAgent, + }, + { + name: "unknown coordinate -> supervisor + routing channel", + identifier: "RIG-9999", + ownership: &fakeOwnershipIndex{wantRepo: "RIG", wantNumber: 2717, row: rowByManager}, + managers: &fakeManagerResolver{owners: map[store.AccountID]managerHome{}}, + wantManager: testSupervisor, + wantChannel: testRoutingChan, + wantWalk: "", + }, + { + name: "bare @mention, no issue -> supervisor + routing channel", + identifier: "", + ownership: &fakeOwnershipIndex{wantRepo: "RIG", wantNumber: 2717, row: rowByManager}, + managers: &fakeManagerResolver{owners: map[store.AccountID]managerHome{}}, + wantManager: testSupervisor, + wantChannel: testRoutingChan, + wantWalk: "", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + r := NewResolver(tc.ownership, tc.managers, testForgeHost, testSupervisor, testRoutingChan) + + gotManager, gotChannel, err := r.ResolveResponder(ctx, sessionEvent(tc.identifier)) + if err != nil { + t.Fatalf("ResolveResponder: unexpected error: %v", err) + } + if gotManager != tc.wantManager { + t.Errorf("manager = %q, want %q", gotManager, tc.wantManager) + } + if gotChannel != tc.wantChannel { + t.Errorf("home channel = %q, want %q", gotChannel, tc.wantChannel) + } + + // The walk must be invoked exactly when a row is recorded, and on the + // RECORDED authoring agent — the peer case proves peer -> Manager, not + // peer -> peer (a walk skipped or walked from the wrong agent reddens). + if tc.wantWalk == "" { + if tc.managers.calls != 0 { + t.Errorf("manager walk invoked %d times, want 0 (fallback path)", tc.managers.calls) + } + } else { + if tc.managers.calls != 1 { + t.Fatalf("manager walk invoked %d times, want 1", tc.managers.calls) + } + if tc.managers.gotFrom != tc.wantWalk { + t.Errorf("walk started from %q, want %q (must walk the RECORDED authoring agent)", tc.managers.gotFrom, tc.wantWalk) + } + } + }) + } +} + +// TestResolveResponderExtractsLinearCoordinate pins the coordinate extraction: +// a "TEAM-NUMBER" identifier maps to (Linear, config host, team=repo, number) +// and queries the issue kind. A regression that swaps team/number or drops the +// host would reroute or miss the ownership row. +func TestResolveResponderExtractsLinearCoordinate(t *testing.T) { + ctx := context.Background() + own := &fakeOwnershipIndex{wantRepo: "RIG", wantNumber: 2717, row: store.AuthoredArtifact{AgentAccountID: "acct-x"}} + mgr := &fakeManagerResolver{owners: map[store.AccountID]managerHome{"acct-x": {manager: "m", homeChannel: "c"}}} + r := NewResolver(own, mgr, testForgeHost, testSupervisor, testRoutingChan) + + if _, _, err := r.ResolveResponder(ctx, sessionEvent("RIG-2717")); err != nil { + t.Fatalf("ResolveResponder: %v", err) + } + if own.gotProvider != store.ForgeProviderLinear { + t.Errorf("queried provider = %d, want Linear (%d)", own.gotProvider, store.ForgeProviderLinear) + } + if own.gotHost != testForgeHost { + t.Errorf("queried host = %q, want %q", own.gotHost, testForgeHost) + } + if own.gotRepo != "RIG" { + t.Errorf("queried repo = %q, want team key %q", own.gotRepo, "RIG") + } + if own.gotNumber != 2717 { + t.Errorf("queried number = %d, want 2717", own.gotNumber) + } + if own.gotKind != store.ForgeArtifactKindIssue { + t.Errorf("queried kind = %d, want issue (%d)", own.gotKind, store.ForgeArtifactKindIssue) + } +} + +// TestResolveResponderPropagatesOwnershipError proves a non-NotFound store +// failure is surfaced, not swallowed into the supervisor fallback (which would +// silently misroute on a transient DB fault). +func TestResolveResponderPropagatesOwnershipError(t *testing.T) { + ctx := context.Background() + boom := errors.New("store: connection reset") + own := &erroringOwnershipIndex{err: boom} + mgr := &fakeManagerResolver{owners: map[store.AccountID]managerHome{}} + r := NewResolver(own, mgr, testForgeHost, testSupervisor, testRoutingChan) + + _, _, err := r.ResolveResponder(ctx, sessionEvent("RIG-2717")) + if !errors.Is(err, boom) { + t.Fatalf("error = %v, want the store failure propagated", err) + } + if mgr.calls != 0 { + t.Errorf("manager walk invoked %d times on ownership error, want 0", mgr.calls) + } +} + +type erroringOwnershipIndex struct{ err error } + +func (e *erroringOwnershipIndex) AuthoredArtifactByCoordinate(_ context.Context, _ store.ForgeProvider, _, _ string, _ store.ForgeArtifactKind, _ uint64) (store.AuthoredArtifact, error) { + return store.AuthoredArtifact{}, e.err +}