Skip to content

Commit 5aac707

Browse files
apresmoiapresmoi
committed
feat: surface agent presence events
1 parent 2a36c7d commit 5aac707

32 files changed

Lines changed: 579 additions & 103 deletions

‎internal/rooms/events.go‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,104 @@ func (s *Service) publishEvent(event protocol.Event) {
1414
s.broker.Publish(event)
1515
}
1616

17+
func (s *Service) AgentConnected(ctx context.Context, agent protocol.Actor) {
18+
eventAgent := s.setAgentConnected(agent, true)
19+
s.publishAgentEvent(ctx, protocol.EventTypeAgentConnected, eventAgent)
20+
}
21+
22+
func (s *Service) AgentDisconnected(ctx context.Context, agent protocol.Actor) {
23+
eventAgent := s.setAgentConnected(agent, false)
24+
s.publishAgentEvent(ctx, protocol.EventTypeAgentDisconnected, eventAgent)
25+
}
26+
27+
func (s *Service) AgentWakeDelivered(ctx context.Context, agent protocol.Actor, event protocol.Event) {
28+
if event.Message == nil {
29+
return
30+
}
31+
s.publishAgentEvent(ctx, protocol.EventTypeAgentWakeDelivered, protocol.AgentEvent{
32+
AgentID: strings.TrimSpace(agent.ID),
33+
NetworkID: s.networkID,
34+
FQID: protocol.AgentFQID(s.networkID, strings.TrimSpace(agent.ID)),
35+
Name: strings.TrimSpace(agent.Name),
36+
MessageID: event.Message.ID,
37+
Reason: agentWakeReason(s.networkID, agent, event),
38+
Target: &event.Message.Target,
39+
})
40+
}
41+
42+
func (s *Service) AgentWakeFailed(ctx context.Context, agent protocol.Actor, event protocol.Event, err error) {
43+
if event.Message == nil {
44+
return
45+
}
46+
failure := protocol.AgentEvent{
47+
AgentID: strings.TrimSpace(agent.ID),
48+
NetworkID: s.networkID,
49+
FQID: protocol.AgentFQID(s.networkID, strings.TrimSpace(agent.ID)),
50+
Name: strings.TrimSpace(agent.Name),
51+
MessageID: event.Message.ID,
52+
Reason: agentWakeReason(s.networkID, agent, event),
53+
Target: &event.Message.Target,
54+
}
55+
if err != nil {
56+
failure.Error = strings.TrimSpace(err.Error())
57+
}
58+
s.publishAgentEvent(ctx, protocol.EventTypeAgentWakeFailed, failure)
59+
}
60+
61+
func (s *Service) setAgentConnected(agent protocol.Actor, connected bool) protocol.AgentEvent {
62+
agentID := strings.TrimSpace(agent.ID)
63+
s.agentPresenceMu.Lock()
64+
if connected {
65+
s.connectedAgents[agentID] = true
66+
} else {
67+
delete(s.connectedAgents, agentID)
68+
}
69+
s.agentPresenceMu.Unlock()
70+
71+
return protocol.AgentEvent{
72+
AgentID: agentID,
73+
NetworkID: s.networkID,
74+
FQID: protocol.AgentFQID(s.networkID, agentID),
75+
Name: strings.TrimSpace(agent.Name),
76+
}
77+
}
78+
79+
func (s *Service) publishAgentEvent(ctx context.Context, eventType string, agent protocol.AgentEvent) {
80+
if strings.TrimSpace(agent.AgentID) == "" {
81+
return
82+
}
83+
observability.Logger(ctx, "rooms.agent", "agent_id", agent.AgentID, "event_type", eventType).
84+
Info("agent lifecycle event")
85+
s.publishEvent(protocol.Event{
86+
ID: newPrefixedID("evt"),
87+
Type: eventType,
88+
NetworkID: s.networkID,
89+
Agent: &agent,
90+
CreatedAt: time.Now().UTC(),
91+
})
92+
}
93+
94+
func (s *Service) agentConnected(agentID string) bool {
95+
s.agentPresenceMu.RLock()
96+
defer s.agentPresenceMu.RUnlock()
97+
return s.connectedAgents[strings.TrimSpace(agentID)]
98+
}
99+
100+
func agentWakeReason(networkID string, agent protocol.Actor, event protocol.Event) string {
101+
if event.Message == nil {
102+
return ""
103+
}
104+
if event.Message.Target.Kind == protocol.TargetKindDM {
105+
return "dm"
106+
}
107+
for _, mention := range event.Message.Mentions {
108+
if protocol.ActorMatches(networkID, agent.ID, mention) || mention == agent.Name {
109+
return "mention"
110+
}
111+
}
112+
return "targeted"
113+
}
114+
17115
func eventIDForMessage(messageID string) string {
18116
return deterministicPrefixedID("evt", messageID)
19117
}

‎internal/rooms/events_test.go‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,89 @@ func TestEventIDForMessageStaysCursorSafeForLongMessageIDs(t *testing.T) {
4444
}
4545
}
4646

47+
func TestAgentPresencePublishesEventsAndUpdatesAgentSummary(t *testing.T) {
48+
t.Parallel()
49+
50+
service := newTestService()
51+
if _, err := service.RegisterAgentContext(context.Background(), protocol.RegisterAgentRequest{
52+
RequestedAgentID: "luna",
53+
Name: "Luna",
54+
}); err != nil {
55+
t.Fatalf("RegisterAgentContext() error = %v", err)
56+
}
57+
58+
stream := service.Subscribe(context.Background())
59+
service.AgentConnected(context.Background(), protocol.Actor{ID: "luna", Name: "Luna"})
60+
61+
connected := <-stream
62+
if connected.Type != protocol.EventTypeAgentConnected ||
63+
connected.Agent == nil ||
64+
connected.Agent.AgentID != "luna" {
65+
t.Fatalf("unexpected connected event %#v", connected)
66+
}
67+
68+
agent, err := service.GetAgent("luna")
69+
if err != nil {
70+
t.Fatalf("GetAgent() error = %v", err)
71+
}
72+
if !agent.Connected {
73+
t.Fatalf("expected connected agent summary, got %#v", agent)
74+
}
75+
76+
service.AgentDisconnected(context.Background(), protocol.Actor{ID: "luna", Name: "Luna"})
77+
disconnected := <-stream
78+
if disconnected.Type != protocol.EventTypeAgentDisconnected ||
79+
disconnected.Agent == nil ||
80+
disconnected.Agent.AgentID != "luna" {
81+
t.Fatalf("unexpected disconnected event %#v", disconnected)
82+
}
83+
84+
agent, err = service.GetAgent("luna")
85+
if err != nil {
86+
t.Fatalf("GetAgent() after disconnect error = %v", err)
87+
}
88+
if agent.Connected {
89+
t.Fatalf("expected disconnected agent summary, got %#v", agent)
90+
}
91+
}
92+
93+
func TestAgentWakeEventsPublishMessageContext(t *testing.T) {
94+
t.Parallel()
95+
96+
service := newTestService()
97+
stream := service.Subscribe(context.Background())
98+
messageEvent := protocol.Event{
99+
ID: "evt_1",
100+
Type: protocol.EventTypeMessageCreated,
101+
NetworkID: "local",
102+
Message: &protocol.Message{
103+
ID: "msg_1",
104+
NetworkID: "local",
105+
Target: protocol.Target{Kind: protocol.TargetKindRoom, RoomID: "lab"},
106+
Mentions: []string{"luna"},
107+
},
108+
}
109+
110+
service.AgentWakeDelivered(context.Background(), protocol.Actor{ID: "luna", Name: "Luna"}, messageEvent)
111+
delivered := <-stream
112+
if delivered.Type != protocol.EventTypeAgentWakeDelivered ||
113+
delivered.Agent == nil ||
114+
delivered.Agent.MessageID != "msg_1" ||
115+
delivered.Agent.Reason != "mention" ||
116+
delivered.Agent.Target == nil ||
117+
delivered.Agent.Target.RoomID != "lab" {
118+
t.Fatalf("unexpected delivered event %#v", delivered)
119+
}
120+
121+
service.AgentWakeFailed(context.Background(), protocol.Actor{ID: "luna", Name: "Luna"}, messageEvent, context.Canceled)
122+
failed := <-stream
123+
if failed.Type != protocol.EventTypeAgentWakeFailed ||
124+
failed.Agent == nil ||
125+
failed.Agent.Error != "context canceled" {
126+
t.Fatalf("unexpected failed event %#v", failed)
127+
}
128+
}
129+
47130
type blockingEventBroker struct {
48131
published chan protocol.Event
49132
blockFirst chan struct{}

‎internal/rooms/query.go‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -124,14 +124,17 @@ func (s *Service) GetAgent(agentID string) (protocol.AgentSummary, error) {
124124
if err != nil {
125125
return protocol.AgentSummary{}, err
126126
}
127-
return registeredAgentSummary(registration, rooms), nil
127+
agent := registeredAgentSummary(registration, rooms)
128+
agent.Connected = s.agentConnected(agent.ID)
129+
return agent, nil
128130
}
129131
if s.contextAgents != nil {
130132
agent, ok, err := s.contextAgents.GetAgentContext(context.Background(), id)
131133
if err != nil {
132134
return protocol.AgentSummary{}, err
133135
}
134136
if ok {
137+
agent.Connected = s.agentConnected(agent.ID)
135138
return agent, nil
136139
}
137140
return protocol.AgentSummary{}, unknownAgentError(id)
@@ -158,6 +161,7 @@ func (s *Service) GetAgent(agentID string) (protocol.AgentSummary, error) {
158161
return protocol.AgentSummary{}, unknownAgentError(id)
159162
}
160163
slices.Sort(agent.Rooms)
164+
agent.Connected = s.agentConnected(agent.ID)
161165
return agent, nil
162166
}
163167

‎internal/rooms/service.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,8 @@ type Service struct {
5555
pairingsMu sync.RWMutex
5656
pairingPublishMu sync.Mutex
5757
pairingStatuses map[string]pairingStatus
58+
agentPresenceMu sync.RWMutex
59+
connectedAgents map[string]bool
5860
lifecycleCtx context.Context
5961
lifecycleCancel context.CancelFunc
6062
now func() time.Time
@@ -101,6 +103,7 @@ func NewService(config ServiceConfig) *Service {
101103
pairingClient: config.PairingClient,
102104
relaySlots: make(chan struct{}, 8),
103105
pairingStatuses: statuses,
106+
connectedAgents: make(map[string]bool),
104107
lifecycleCtx: lifecycleCtx,
105108
lifecycleCancel: lifecycleCancel,
106109
now: now,

‎internal/rooms/service_collections.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -263,6 +263,9 @@ func (s *Service) ListAgentsContext(ctx context.Context, page protocol.PageReque
263263
if err != nil {
264264
return protocol.AgentPage{}, err
265265
}
266+
for index := range agents {
267+
agents[index].Connected = s.agentConnected(agents[index].ID)
268+
}
266269
return paginateAgents(agents, page)
267270
}
268271

@@ -293,6 +296,7 @@ func (s *Service) ListAgentsContext(ctx context.Context, page protocol.PageReque
293296
agents := make([]protocol.AgentSummary, 0, len(agentsByID))
294297
for _, agent := range agentsByID {
295298
slices.Sort(agent.Rooms)
299+
agent.Connected = s.agentConnected(agent.ID)
296300
agents = append(agents, *agent)
297301
}
298302
slices.SortFunc(agents, func(left, right protocol.AgentSummary) int {

‎internal/transport/attach.go‎

Lines changed: 38 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -117,13 +117,17 @@ func handleAttachment(
117117
}); err != nil {
118118
return
119119
}
120+
service.AgentConnected(request.Context(), agent)
121+
defer service.AgentDisconnected(context.Background(), agent)
120122

121123
ctx, cancel := context.WithCancel(request.Context())
122124
defer cancel()
123125

124126
readErrCh := make(chan error, 1)
125127
go func() {
126-
readErrCh <- consumeAttachmentFrames(ctx, connection, writer, session, attachmentReadTimeout())
128+
readErrCh <- consumeAttachmentFrames(ctx, connection, writer, session, attachmentReadTimeout(), func(event protocol.Event) {
129+
service.AgentWakeDelivered(context.Background(), agent, event)
130+
})
127131
}()
128132

129133
heartbeatTicker := time.NewTicker(attachmentHeartbeatInterval() / 2)
@@ -140,6 +144,7 @@ func handleAttachment(
140144
observability.Logger(request.Context(), "transport.attach", "agent_id", agent.ID).
141145
Warn("attachment websocket read error", "error", err)
142146
}
147+
publishPendingWakeFailures(service, agent, session, err)
143148
return
144149
case <-heartbeatTicker.C:
145150
if err := writer.write(protocol.AttachmentFrame{
@@ -156,6 +161,9 @@ func handleAttachment(
156161
continue
157162
}
158163
session.NoteSent(event.ID)
164+
if attachmentWakeEvent(event, service.Network().ID, agent) {
165+
session.NoteWakeSent(event.ID, event)
166+
}
159167

160168
if err := writer.write(protocol.AttachmentFrame{
161169
Op: protocol.AttachmentOpEvent,
@@ -164,12 +172,19 @@ func handleAttachment(
164172
Cursor: event.ID,
165173
Event: &event,
166174
}); err != nil {
175+
publishPendingWakeFailures(service, agent, session, err)
167176
return
168177
}
169178
}
170179
}
171180
}
172181

182+
func publishPendingWakeFailures(service Service, agent protocol.Actor, session *attachmentSession, err error) {
183+
for _, event := range session.PendingWakes() {
184+
service.AgentWakeFailed(context.Background(), agent, event, err)
185+
}
186+
}
187+
173188
type attachmentWriter struct {
174189
connection *websocket.Conn
175190
mu sync.Mutex
@@ -253,6 +268,22 @@ func attachedAgentMessage(message *protocol.Message, networkID string, agentID s
253268
return false
254269
}
255270

271+
func attachmentWakeEvent(event protocol.Event, networkID string, agent protocol.Actor) bool {
272+
if event.Type != protocol.EventTypeMessageCreated || event.Message == nil {
273+
return false
274+
}
275+
message := event.Message
276+
if message.Target.Kind == protocol.TargetKindDM {
277+
return participantsIncludeAttachedAgent(message.Target.ParticipantIDs, networkID, agent.ID)
278+
}
279+
for _, mention := range message.Mentions {
280+
if protocol.ActorMatches(networkID, agent.ID, mention) || mention == agent.Name {
281+
return true
282+
}
283+
}
284+
return false
285+
}
286+
256287
func participantsIncludeAttachedAgent(participants []string, networkID string, agentID string) bool {
257288
for _, participantID := range participants {
258289
if protocol.ActorMatches(networkID, agentID, participantID) {
@@ -268,6 +299,7 @@ func consumeAttachmentFrames(
268299
writer *attachmentWriter,
269300
session *attachmentSession,
270301
readTimeout time.Duration,
302+
onWakeAck func(protocol.Event),
271303
) error {
272304
for {
273305
select {
@@ -292,12 +324,16 @@ func consumeAttachmentFrames(
292324

293325
switch frame.Op {
294326
case protocol.AttachmentOpAck:
295-
if !session.Ack(frame.Cursor) {
327+
event, wake, ok := session.Ack(frame.Cursor)
328+
if !ok {
296329
if err := writeAttachmentError(writer, "unexpected ACK cursor"); err != nil {
297330
return err
298331
}
299332
return fmt.Errorf("unexpected ACK cursor %q", frame.Cursor)
300333
}
334+
if wake && onWakeAck != nil {
335+
onWakeAck(event)
336+
}
301337
case protocol.AttachmentOpPong:
302338
continue
303339
case protocol.AttachmentOpPing:

0 commit comments

Comments
 (0)