Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 14 additions & 4 deletions docs/specs/product/compass.md
Original file line number Diff line number Diff line change
Expand Up @@ -725,9 +725,12 @@ the authored fields only.
### Requirement: `ListAgentBoards` returns visible boards and derived refs

`ListAgentBoards` SHALL return each matching agent's whole board when the
caller can see that agent, with the matching task keys and derived refs. The
request's optional handle, text, ref, waiting, state, and halt filters SHALL be
combined with AND; the default limit is 20 and the maximum is 100.
caller can see that agent, with the matching task keys and derived refs. Each
board SHALL carry that agent's asks the caller can see: every pending ask and
asks answered in the last 24 hours, at most 20, pending asks first and each
group newest first, with each ask's forge mirror state and URL. The request's
optional handle, text, ref, waiting, state, and halt filters SHALL be combined
with AND; the default board limit is 20 and the maximum is 100.

#### Scenario: A peer lists a matching board

Expand All @@ -736,6 +739,12 @@ combined with AND; the default limit is 20 and the maximum is 100.
- **Then** the response contains the whole board and identifies the matching
task key.

#### Scenario: A peer cannot see an ask in an invisible channel

- **Given** a caller who can see an agent but cannot see one of its ask channels
- **When** the caller lists boards
- **Then** the response contains the agent's board without that ask.

### Requirement: The caller is the connection's account, never a request field

Every `CommsService` RPC SHALL authorize against the account authenticated on
Expand Down Expand Up @@ -1055,7 +1064,8 @@ after its message commits. An ask with no ref, or whose coordinate is OFF or
MANUAL, SHALL NOT be automatically posted. The server SHALL check mode again
immediately before every post; a coordinate turned OFF while queued SHALL not
be posted. Answering an ask SHALL update the same forge comment with its answer
and answerer rather than create a second comment.
and answerer rather than create a second comment. A pending or failed mirror
SHALL be re-driven on the agent's next board write.

### Requirement: Channel membership carries a separate subscribe opt-in

Expand Down
77 changes: 73 additions & 4 deletions go/internal/comms/board.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,10 @@ import (
"google.golang.org/protobuf/proto"
)

const boardStaleAfter = 30 * time.Minute
const (
boardStaleAfter = 30 * time.Minute
boardAskWindow = 24 * time.Hour
)

// UpdateBoardAsAccount validates and applies an agent patch before publishing the committed board.
func (c *Comms) UpdateBoardAsAccount(
Expand Down Expand Up @@ -53,6 +56,9 @@ func (c *Comms) UpdateBoardAsAccount(
if err != nil {
return nil, edgeError(err)
}
if c.driver != nil {
go c.driver.DriveAgent(context.WithoutCancel(ctx), caller)
}
authored := boardAuthoredToWire(account, row, nil)
if authored.decodeErr != nil {
slog.ErrorContext(ctx, "comms: board decode failed after committed update", "agent_account_id", string(caller), "error", authored.decodeErr)
Expand All @@ -73,12 +79,13 @@ func (c *Comms) UpdateBoardAsAccount(
}
}
wire := authored.wire
completed, completeErr := c.completeBoardWire(ctx, account, row, authored.wire, authored.refs)
completed, completeErr := c.completeBoardWire(ctx, caller, account, row, authored.wire, authored.refs)
if completeErr != nil {
slog.WarnContext(ctx, "comms: board refs unavailable after update", "agent_account_id", string(caller), "error", completeErr)
} else {
wire = completed
}
c.fillAskMirrors(ctx, []*compassv1.AgentBoard{wire})
return &compassv1.UpdateAgentBoardResponse{Board: wire}, nil
}

Expand Down Expand Up @@ -510,14 +517,15 @@ func (c *Comms) ListBoardsAsAccount(
slog.WarnContext(ctx, "comms: board decode failed while listing", "agent_account_id", string(id), "error", authored.decodeErr)
}
wire := authored.wire
completed, err := c.completeBoardWire(ctx, account, rows[id], wire, authored.refs)
completed, err := c.completeBoardWire(ctx, caller, account, rows[id], wire, authored.refs)
if err != nil {
slog.WarnContext(ctx, "comms: board refs unavailable while listing", "agent_account_id", string(id), "error", err)
} else {
wire = completed
}
boards = append(boards, wire)
}
c.fillAskMirrors(ctx, boards)
return &compassv1.ListAgentBoardsResponse{Boards: boards}, nil
}

Expand Down Expand Up @@ -604,14 +612,20 @@ func boardAuthoredToWire(account store.Account, row store.BoardRow, matched []st
return authoredBoard{wire: out, refs: allRefs, decodeErr: decodeErr}
}

func (c *Comms) completeBoardWire(ctx context.Context, account store.Account, row store.BoardRow, out *compassv1.AgentBoard, allRefs []string) (*compassv1.AgentBoard, error) {
func (c *Comms) completeBoardWire(ctx context.Context, viewer store.AccountID, account store.Account, row store.BoardRow, out *compassv1.AgentBoard, allRefs []string) (*compassv1.AgentBoard, error) {
presence := compassv1.AgentPresence_AGENT_PRESENCE_OFFLINE
if c.presence != nil {
if live := c.presence.PresenceFor([]store.AccountID{account.ID}); live[account.ID] != compassv1.AgentPresence_AGENT_PRESENCE_UNSPECIFIED {
presence = live[account.ID]
}
}
out.Stale = presence == compassv1.AgentPresence_AGENT_PRESENCE_WORKING && row.WrittenAtUnixMs < time.Now().Add(-boardStaleAfter).UnixMilli()
asks, err := c.boardAsks(ctx, viewer, account.ID)
if err != nil {
slog.WarnContext(ctx, "comms: board asks unavailable", "agent_account_id", string(account.ID), "error", err)
} else {
out.Asks = asks
}
derived, untracked, err := board.DeriveRefs(ctx, c.store, c.boardForges, account.ID, allRefs)
if err != nil {
return out, err
Expand All @@ -620,6 +634,61 @@ func (c *Comms) completeBoardWire(ctx context.Context, account store.Account, ro
return out, nil
}

func (c *Comms) boardAsks(ctx context.Context, viewer, agent store.AccountID) ([]*compassv1.BoardAsk, error) {
rows, err := c.store.AsksByAuthor(ctx, viewer, agent, time.Now().Add(-boardAskWindow), 0)
if err != nil {
return nil, edgeError(err)
}
out := make([]*compassv1.BoardAsk, len(rows))
for i, r := range rows {
out[i] = &compassv1.BoardAsk{
AskId: r.AskID, MessageId: r.MessageID, ChannelId: r.ChannelID,
Ref: r.Ref, Header: r.Header, Answered: r.Answered, PostedAtUnixMs: r.PostedAtUnixMs,
}
}
return out, nil
}

// fillAskMirrors sets each ask's forge mirror state with one lookup for all
// boards; a failed lookup leaves the asks without mirror state.
func (c *Comms) fillAskMirrors(ctx context.Context, boards []*compassv1.AgentBoard) {
var askIDs []string
for _, b := range boards {
for _, ask := range b.GetAsks() {
askIDs = append(askIDs, ask.GetAskId())
}
}
if len(askIDs) == 0 {
return
}
mirrors, err := c.store.AskForgeMirrors(ctx, askIDs)
if err != nil {
slog.WarnContext(ctx, "comms: board ask mirrors unavailable", "error", err)
return
}
byAsk := make(map[string]store.ForgeMirrorRow, len(askIDs))
for _, mirror := range mirrors {
current, found := byAsk[mirror.AskID]
// The agent's automatic row wins over a user's click row, then the newest row.
auto, curAuto := mirror.Agent != "", current.Agent != ""
if !found || (auto && !curAuto) || (auto == curAuto && mirror.UpdatedAt.After(current.UpdatedAt)) {
byAsk[mirror.AskID] = mirror
}
}
for _, b := range boards {
for _, ask := range b.GetAsks() {
mirror, ok := byAsk[ask.GetAskId()]
if !ok {
continue
}
ask.MirrorState = compassv1.BoardMirrorState(mirror.State)
if ask.GetMirrorState() == compassv1.BoardMirrorState_BOARD_MIRROR_STATE_DONE {
ask.MirrorUrl = mirror.CommentURL
}
}
}
}

func haltToWire(halt store.BoardHalt) *compassv1.BoardHalt {
if halt.Kind == 0 && halt.Ref == "" && halt.Reason == "" {
return nil
Expand Down
220 changes: 220 additions & 0 deletions go/internal/comms/board_asks_pgtest_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,220 @@
//go:build pgtest

package comms

import (
"context"
"testing"
"time"

"github.com/jackc/pgx/v5/pgxpool"

compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1"
"github.com/RigelBuild/compass/go/internal/store"
)

func TestListAgentBoardsIncludesVisibleAsks(t *testing.T) {
svc, st := newHandler(t)
ctx := t.Context()
owner := mustUser(t, st, "board-asks-owner")
viewer := mustUser(t, st, "board-asks-viewer")
agent := mustAgent(t, st, owner.ID, "board-asks-agent")
private, err := st.CreateChannel(ctx, owner.ID, store.NewChannel{
Name: "private-asks", Kind: store.ChannelKindChannel,
MemberAccountIDs: []store.AccountID{owner.ID, agent.ID},
})
if err != nil {
t.Fatalf("CreateChannel(private): %v", err)
}
shared, err := st.CreateChannel(ctx, owner.ID, store.NewChannel{
Name: "roster-room", Kind: store.ChannelKindChannel,
MemberAccountIDs: []store.AccountID{owner.ID, viewer.ID, agent.ID},
})
if err != nil {
t.Fatalf("CreateChannel(shared): %v", err)
}
if _, err := svc.UpdateBoardAsAccount(ctx, agent.ID, &compassv1.UpdateAgentBoardRequest{}); err != nil {
t.Fatalf("UpdateBoardAsAccount: %v", err)
}
postBoardAsk(t, svc, ctx, agent.ID, private.ID, "owner/repo#13")
postBoardAsk(t, svc, ctx, agent.ID, shared.ID, "owner/repo#12")

viewerBoards, err := svc.ListBoardsAsAccount(ctx, viewer.ID, &compassv1.ListAgentBoardsRequest{})
if err != nil {
t.Fatalf("ListBoardsAsAccount(viewer): %v", err)
}
if len(viewerBoards.GetBoards()) != 1 || len(viewerBoards.GetBoards()[0].GetAsks()) != 1 {
t.Fatalf("viewer boards = %v, want visible agent with only its shared-channel ask", viewerBoards.GetBoards())
}
if got := viewerBoards.GetBoards()[0].GetAsks()[0]; got.GetRef() != "owner/repo#12" || got.GetChannelId() != string(shared.ID) {
t.Fatalf("viewer ask = %v, want only shared-channel ask", got)
}

ownerBoards, err := svc.ListBoardsAsAccount(ctx, owner.ID, &compassv1.ListAgentBoardsRequest{})
if err != nil {
t.Fatalf("ListBoardsAsAccount(owner): %v", err)
}
if len(ownerBoards.GetBoards()) != 1 {
t.Fatalf("owner boards = %v, want one agent", ownerBoards.GetBoards())
}
asks := ownerBoards.GetBoards()[0].GetAsks()
if len(asks) != 2 || asks[0].GetRef() != "owner/repo#12" || asks[1].GetRef() != "owner/repo#13" {
t.Fatalf("owner asks = %v, want both asks newest first", asks)
}
}

func TestListAgentBoardsIncludesForgeMirrorStates(t *testing.T) {
svc, st := newHandler(t)
ctx := t.Context()
owner := mustUser(t, st, "board-ask-mirror-owner")
agent := mustAgent(t, st, owner.ID, "board-ask-mirror-agent")
channel, err := st.CreateChannel(ctx, owner.ID, store.NewChannel{
Name: "ask-mirror-states", Kind: store.ChannelKindChannel,
MemberAccountIDs: []store.AccountID{owner.ID, agent.ID},
})
if err != nil {
t.Fatalf("CreateChannel: %v", err)
}
coord := store.ForgeCoord{Provider: store.ForgeProviderGitHub, Host: "github.com", Repo: "owner/mirrors", Number: 12}
if err := st.SetForgeMirrorSetting(ctx, owner.ID, store.ForgeMirrorSetting{
Provider: coord.Provider, Host: coord.Host, Repo: coord.Repo, Mode: store.ForgeMirrorAuto,
}); err != nil {
t.Fatalf("SetForgeMirrorSetting: %v", err)
}
firstID := postBoardAsk(t, svc, ctx, agent.ID, channel.ID, "owner/mirrors#12")
secondID := postBoardAsk(t, svc, ctx, agent.ID, channel.ID, "owner/mirrors#12")
rows, err := st.AskForgeMirrors(ctx, []string{firstID, secondID})
if err != nil || len(rows) != 2 {
t.Fatalf("AskForgeMirrors before update = (%+v, %v), want two auto rows", rows, err)
}
for i, row := range rows {
state, commentKey, commentURL := int16(3), "", ""
if i == 0 {
state, commentKey, commentURL = 2, "done-comment", "https://github.com/owner/mirrors/issues/12#issuecomment-1"
}
if _, ok, err := st.LeaseForgeMirror(ctx, row.ID, "board-ask-mirror-"+row.AskID, time.Minute); err != nil || !ok {
t.Fatalf("LeaseForgeMirror(%q): ok=%t err=%v", row.ID, ok, err)
}
if err := st.MarkForgeMirror(ctx, row.ID, "board-ask-mirror-"+row.AskID, store.ForgeMirrorMark{
PostedRev: 1, State: state, CommentKey: commentKey, URL: commentURL, Err: "delivery failed",
}); err != nil {
t.Fatalf("MarkForgeMirror(%q): %v", row.ID, err)
}
}
if _, err := svc.UpdateBoardAsAccount(ctx, agent.ID, &compassv1.UpdateAgentBoardRequest{}); err != nil {
t.Fatalf("UpdateBoardAsAccount: %v", err)
}
listed, err := svc.ListBoardsAsAccount(ctx, owner.ID, &compassv1.ListAgentBoardsRequest{})
if err != nil {
t.Fatalf("ListBoardsAsAccount: %v", err)
}
if len(listed.GetBoards()) != 1 {
t.Fatalf("boards = %v, want the agent's board", listed.GetBoards())
}
asks := listed.GetBoards()[0].GetAsks()
if len(asks) != 2 {
t.Fatalf("board asks = %v, want two asks", asks)
}
var done, failed *compassv1.BoardAsk
for _, ask := range asks {
switch ask.GetAskId() {
case firstID:
done = ask
case secondID:
failed = ask
}
}
if done == nil || done.GetMirrorState() != compassv1.BoardMirrorState_BOARD_MIRROR_STATE_DONE || done.GetMirrorUrl() != "https://github.com/owner/mirrors/issues/12#issuecomment-1" {
t.Fatalf("done ask mirror = %v, want done with its URL", done)
}
if failed == nil || failed.GetMirrorState() != compassv1.BoardMirrorState_BOARD_MIRROR_STATE_FAILED || failed.GetMirrorUrl() != "" {
t.Fatalf("failed ask mirror = %v, want failed without URL", failed)
}
}

func TestListAgentBoardsOmitsOldAnsweredAsk(t *testing.T) {
st, dsn := newTestStoreDSN(t)
ctx := t.Context()
bus := newBus(t)
admin, err := st.BootstrapAdmin(ctx, store.NewUser{Handle: "board-ask-age-root", DisplayName: "Root"})
if err != nil {
t.Fatalf("BootstrapAdmin: %v", err)
}
svc := NewComms(t.Context(), st, bus, nil, admin.ID)
owner := mustUser(t, st, "board-ask-age-owner")
agent := mustAgent(t, st, owner.ID, "board-ask-age-agent")
channel, err := st.CreateChannel(ctx, owner.ID, store.NewChannel{
Name: "answer-window", Kind: store.ChannelKindChannel,
MemberAccountIDs: []store.AccountID{owner.ID, agent.ID},
})
if err != nil {
t.Fatalf("CreateChannel: %v", err)
}
if _, err := svc.UpdateBoardAsAccount(ctx, agent.ID, &compassv1.UpdateAgentBoardRequest{}); err != nil {
t.Fatalf("UpdateBoardAsAccount: %v", err)
}
oldAsk := postBoardAsk(t, svc, ctx, agent.ID, channel.ID, "owner/repo#13")
recentAsk := postBoardAsk(t, svc, ctx, agent.ID, channel.ID, "owner/repo#12")
answer := []store.AskAnswer{{QuestionID: "q1", ChosenOptionIDs: []string{"yes"}}}
_, oldAnswer, err := st.AnswerAsk(ctx, owner.ID, oldAsk, answer)
if err != nil {
t.Fatalf("AnswerAsk(old): %v", err)
}
_, recentAnswer, err := st.AnswerAsk(ctx, owner.ID, recentAsk, answer)
if err != nil {
t.Fatalf("AnswerAsk(recent): %v", err)
}
oldSource, err := st.AskForForgeMirror(ctx, owner.ID, oldAsk)
if err != nil {
t.Fatalf("AskForForgeMirror(old): %v", err)
}
recentSource, err := st.AskForForgeMirror(ctx, owner.ID, recentAsk)
if err != nil {
t.Fatalf("AskForForgeMirror(recent): %v", err)
}
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
t.Fatalf("connect to test schema: %v", err)
}
defer pool.Close()
now := time.Now()
for _, aged := range []struct {
id string
at time.Time
}{
{id: string(oldSource.Message.ID), at: now.Add(-48 * time.Hour)},
{id: string(oldAnswer.ID), at: now.Add(-36 * time.Hour)},
{id: string(recentSource.Message.ID), at: now.Add(-3 * time.Hour)},
{id: string(recentAnswer.ID), at: now.Add(-2 * time.Hour)},
} {
if _, err := pool.Exec(ctx, "UPDATE messages SET at_unix_ms = $1 WHERE id = $2", aged.at.UnixMilli(), aged.id); err != nil {
t.Fatalf("age message %q: %v", aged.id, err)
}
}

listed, err := svc.ListBoardsAsAccount(ctx, owner.ID, &compassv1.ListAgentBoardsRequest{})
if err != nil {
t.Fatalf("ListBoardsAsAccount: %v", err)
}
if len(listed.GetBoards()) != 1 || len(listed.GetBoards()[0].GetAsks()) != 1 {
t.Fatalf("board asks = %v, want only the recent answered ask", listed.GetBoards())
}
ask := listed.GetBoards()[0].GetAsks()[0]
if !ask.GetAnswered() || ask.GetRef() != "owner/repo#12" {
t.Fatalf("recent answered ask = %v, want answered ask owner/repo#12", ask)
}
}

func postBoardAsk(t *testing.T, svc *Comms, ctx context.Context, agent store.AccountID, channel store.ChannelID, ref string) string {
t.Helper()
posted, err := svc.PostAsAccount(ctx, agent, &compassv1.PostMessageRequest{
Container: &compassv1.PostMessageRequest_ChannelId{ChannelId: string(channel)},
Topic: &compassv1.PostMessageRequest_TopicName{TopicName: testTopic},
CreateTopic: true,
Blocks: []*compassv1.MessageBlock{askWithRef(ref)},
})
if err != nil {
t.Fatalf("PostAsAccount(ask %s): %v", ref, err)
}
return posted.GetMessage().GetBlocks()[0].GetAsk().GetAskId()
}
Loading
Loading