Recognise the agent_instance room

internal/server/authorization.go

@@ -29,6 +29,7 @@ type WorkloadReader interface {
 // reason as WorkloadReader.
 type AgentReader interface {
 	GetAgent(ctx context.Context, req *agentsv1.GetAgentRequest, opts ...grpc.CallOption) (*agentsv1.GetAgentResponse, error)
+	GetInstance(ctx context.Context, req *agentsv1.GetInstanceRequest, opts ...grpc.CallOption) (*agentsv1.GetInstanceResponse, error)
 }
 
 const (
@@ -93,6 +94,18 @@ func (s *Server) agentOrganization(ctx context.Context, agentID uuid.UUID) (uuid
 	return parseOrganization(response.GetAgent().GetOrganizationId())
 }
 
+func (s *Server) agentInstanceOrganization(ctx context.Context, agentInstanceID uuid.UUID) (uuid.UUID, error) {
+	if s.agents == nil {
+		return uuid.UUID{}, status.Error(codes.Internal, "agents client is not configured")
+	}
+	response, err := s.agents.GetInstance(ctx, &agentsv1.GetInstanceRequest{Id: agentInstanceID.String()})
+	if err != nil {
+		s.logger.Warn("subscribe agent instance lookup failed")
+		return uuid.UUID{}, status.Error(codes.PermissionDenied, "permission denied")
+	}
+	return parseOrganization(response.GetInstance().GetOrganizationId())
+}
+
 func parseOrganization(value string) (uuid.UUID, error) {
 	id, err := parseUUID(value)
 	if err != nil {

internal/server/rooms.go

@@ -15,6 +15,7 @@ import (
 const (
 	threadParticipantRoomPrefix = "thread_participant:"
 	instanceInboxRoomPrefix     = "instance_inbox:"
+	agentInstanceRoomPrefix     = "agent_instance:"
 	workloadRoomPrefix          = "workload:"
 	organizationRoomPrefix      = "organization:"
 	agentRoomPrefix             = "agent:"
@@ -32,6 +33,7 @@ const (
 	roomKindOther roomKind = iota
 	roomKindThreadParticipant
 	roomKindInstanceInbox
+	roomKindAgentInstance
 	roomKindWorkload
 	roomKindOrganization
 	roomKindAgent
@@ -127,6 +129,14 @@ func classifyRoom(room string, callerID uuid.UUID) (roomKind, uuid.UUID, string,
 		}
 		return roomKindInstanceInbox, id, "", instanceInboxRoomPrefix + id.String(), nil
 	}
+	// Checked after instance_inbox: both begin "instance"-ish but neither is a
+	// prefix of the other, so order only matters for readability here.
+	if id, matched, err := parseRoomUUID(room, agentInstanceRoomPrefix); matched {
+		if err != nil {
+			return roomKindAgentInstance, uuid.UUID{}, "", "", fmt.Errorf("agent_instance: %w", err)
+		}
+		return roomKindAgentInstance, id, "", agentInstanceRoomPrefix + id.String(), nil
+	}
 	if id, matched, err := parseRoomUUID(room, workloadRoomPrefix); matched {
 		if err != nil {
 			return roomKindWorkload, uuid.UUID{}, "", "", fmt.Errorf("workload: %w", err)

internal/server/server.go

@@ -205,6 +205,19 @@ func (s *Server) authorizeSubscribeRooms(ctx context.Context, caller callerIdent
 			if err := s.requireOrganizationRelation(ctx, caller.id, organizationID, memberRelation); err != nil {
 				return err
 			}
+		// Instance state -- created, paused, resumed, terminated. Gated like
+		// the agent room it mirrors: member on the owning organization. That
+		// also admits the instance itself, whose identity satisfies member
+		// through its own org relation, which is how the Orchestrator watches
+		// the instances it reconciles.
+		case roomKindAgentInstance:
+			organizationID, err := s.agentInstanceOrganization(ctx, room.id)
+			if err != nil {
+				return err
+			}
+			if err := s.requireOrganizationRelation(ctx, caller.id, organizationID, memberRelation); err != nil {
+				return err
+			}
 		case roomKindAgent:
 			organizationID, err := s.agentOrganization(ctx, room.id)
 			if err != nil {

internal/server/server_test.go

@@ -458,6 +458,7 @@ func TestSubscribeDeniesUnrelatedCaller(t *testing.T) {
 		"sandbox_org:" + uuid.NewString(),
 		"workload:" + uuid.NewString(),
 		"agent:" + uuid.NewString(),
+		"agent_instance:" + uuid.NewString(),
 	} {
 		streamClient, err := client.Subscribe(ctx, &notificationsv1.SubscribeRequest{Rooms: []string{room}})
 		if err != nil {
@@ -469,6 +470,26 @@ func TestSubscribeDeniesUnrelatedCaller(t *testing.T) {
 	}
 }
 
+// The Orchestrator watches every instance it reconciles through this room, and
+// agents publishes instance.updated to it. The room was never recognised here,
+// so it fell to the deny-by-default branch and every one of those subscriptions
+// was refused -- a paused or resumed instance reached nobody.
+func TestSubscribeAgentInstanceRoomAllowsOrganizationMember(t *testing.T) {
+	t.Parallel()
+
+	hub := stream.NewHub(2, zap.NewNop())
+	client, cleanup := startTestServer(t, &publisherStub{}, hub)
+	defer cleanup()
+
+	room := "agent_instance:" + uuid.NewString()
+	ctx, cancel := context.WithTimeout(authenticatedContext(uuid.New()), time.Second)
+	defer cancel()
+
+	if _, err := client.Subscribe(ctx, &notificationsv1.SubscribeRequest{Rooms: []string{room}}); err != nil {
+		t.Fatalf("Subscribe returned error: %v", err)
+	}
+}
+
 // Sandbox owner rooms are identity equality, with no authorization call at all,
 // so a permissive authorizer must not make someone else's room readable.
 func TestSubscribeSandboxOwnerRoomIsOwnerOnly(t *testing.T) {
@@ -804,6 +825,10 @@ func (allowAll) GetAgent(_ context.Context, _ *agentsv1.GetAgentRequest, _ ...gr
 	return &agentsv1.GetAgentResponse{Agent: &agentsv1.Agent{OrganizationId: uuid.NewString()}}, nil
 }
 
+func (allowAll) GetInstance(_ context.Context, _ *agentsv1.GetInstanceRequest, _ ...grpc.CallOption) (*agentsv1.GetInstanceResponse, error) {
+	return &agentsv1.GetInstanceResponse{Instance: &agentsv1.AgentInstance{OrganizationId: uuid.NewString()}}, nil
+}
+
 // denyAll refuses every check but still resolves organizations, so a test can
 // tell "the caller has no relation" apart from "the lookup failed".
 type denyAll struct{ allowAll }